Proxy aiohttp websession / more rebust. (#5419)

homeassistant/components/camera/__init__.py

@@ -174,7 +174,7 @@ class Camera(Entity):
 
                 yield from asyncio.sleep(.5)
 
-        except asyncio.CancelledError:
+        except (asyncio.CancelledError, ConnectionResetError):
             _LOGGER.debug("Close stream by frontend.")
             response = None
 

homeassistant/components/camera/amcrest.py

@@ -8,9 +8,6 @@ import asyncio
 import logging
 
 import aiohttp
-from aiohttp import web
-from aiohttp.web_exceptions import HTTPGatewayTimeout
-import async_timeout
 import voluptuous as vol
 
 import homeassistant.loader as loader
@@ -18,7 +15,8 @@ from homeassistant.components.camera import (Camera, PLATFORM_SCHEMA)
 from homeassistant.const import (
     CONF_HOST, CONF_NAME, CONF_USERNAME, CONF_PASSWORD, CONF_PORT)
 from homeassistant.helpers import config_validation as cv
-from homeassistant.helpers.aiohttp_client import async_create_clientsession
+from homeassistant.helpers.aiohttp_client import (
+    async_get_clientsession, async_aiohttp_proxy_stream)
 
 REQUIREMENTS = ['amcrest==1.1.0']
 
@@ -108,7 +106,6 @@ class AmcrestCam(Camera):
             device_info.get(CONF_USERNAME),
             password=device_info.get(CONF_PASSWORD)
         )
-        self._websession = async_create_clientsession(hass)
 
     def camera_image(self):
         """Return a still image reponse from the camera."""
@@ -125,44 +122,16 @@ class AmcrestCam(Camera):
             return
 
         # Otherwise, stream an MJPEG image stream directly from the camera
+        websession = async_get_clientsession(self.hass)
         streaming_url = '%s/mjpg/video.cgi?channel=0&subtype=%d' % (
             self._base_url,
             self._resolution
         )
 
-        stream = None
+        stream_coro = websession.get(
-        response = None
+            streaming_url, auth=self._token, timeout=TIMEOUT)
-        try:
-            with async_timeout.timeout(TIMEOUT, loop=self.hass.loop):
-                stream = yield from self._websession.get(
-                    streaming_url,
-                    auth=self._token,
-                    timeout=TIMEOUT
-                )
-            response = web.StreamResponse()
-            response.content_type = stream.headers.get(CONTENT_TYPE_HEADER)
 
-            yield from response.prepare(request)
+        yield from async_aiohttp_proxy_stream(self.hass, request, stream_coro)
-
-            while True:
-                data = yield from stream.content.read(16384)
-                if not data:
-                    break
-                response.write(data)
-
-        except (asyncio.TimeoutError, aiohttp.errors.ClientError):
-            _LOGGER.exception("Error on %s", streaming_url)
-            raise HTTPGatewayTimeout()
-
-        except asyncio.CancelledError:
-            _LOGGER.debug("Close stream by frontend.")
-            response = None
-
-        finally:
-            if stream is not None:
-                stream.close()
-            if response is not None:
-                yield from response.write_eof()
 
     @property
     def name(self):

homeassistant/components/camera/mjpeg.py

@@ -9,9 +9,6 @@ import logging
 from contextlib import closing
 
 import aiohttp
-from aiohttp import web
-from aiohttp.web_exceptions import HTTPGatewayTimeout
-import async_timeout
 import requests
 from requests.auth import HTTPBasicAuth, HTTPDigestAuth
 import voluptuous as vol
@@ -20,7 +17,8 @@ from homeassistant.const import (
     CONF_NAME, CONF_USERNAME, CONF_PASSWORD, CONF_AUTHENTICATION,
     HTTP_BASIC_AUTHENTICATION, HTTP_DIGEST_AUTHENTICATION)
 from homeassistant.components.camera import (PLATFORM_SCHEMA, Camera)
-from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.aiohttp_client import (
+    async_get_clientsession, async_aiohttp_proxy_stream)
 from homeassistant.helpers import config_validation as cv
 
 _LOGGER = logging.getLogger(__name__)
@@ -137,36 +135,9 @@ class MjpegCamera(Camera):
 
         # connect to stream
         websession = async_get_clientsession(self.hass)
-        stream = None
+        stream_coro = websession.get(self._mjpeg_url, auth=self._auth)
-        response = None
-        try:
-            with async_timeout.timeout(10, loop=self.hass.loop):
-                stream = yield from websession.get(self._mjpeg_url,
-                                                   auth=self._auth)
 
-            response = web.StreamResponse()
+        yield from async_aiohttp_proxy_stream(self.hass, request, stream_coro)
-            response.content_type = stream.headers.get(CONTENT_TYPE_HEADER)
-
-            yield from response.prepare(request)
-
-            while True:
-                data = yield from stream.content.read(102400)
-                if not data:
-                    break
-                response.write(data)
-
-        except asyncio.TimeoutError:
-            raise HTTPGatewayTimeout()
-
-        except asyncio.CancelledError:
-            _LOGGER.debug("Close stream by frontend.")
-            response = None
-
-        finally:
-            if stream is not None:
-                stream.close()
-            if response is not None:
-                yield from response.write_eof()
 
     @property
     def name(self):

homeassistant/components/camera/synology.py

@@ -10,8 +10,6 @@ import logging
 import voluptuous as vol
 
 import aiohttp
-from aiohttp import web
-from aiohttp.web_exceptions import HTTPGatewayTimeout
 import async_timeout
 
 from homeassistant.const import (
@@ -20,7 +18,8 @@ from homeassistant.const import (
 from homeassistant.components.camera import (
     Camera, PLATFORM_SCHEMA)
 from homeassistant.helpers.aiohttp_client import (
-    async_get_clientsession, async_create_clientsession)
+    async_get_clientsession, async_create_clientsession,
+    async_aiohttp_proxy_stream)
 import homeassistant.helpers.config_validation as cv
 from homeassistant.util.async import run_coroutine_threadsafe
 
@@ -253,38 +252,10 @@ class SynologyCamera(Camera):
             'cameraId': self._camera_id,
             'format': 'mjpeg'
         }
-        stream = None
+        stream_coro = self._websession.get(
-        response = None
+            streaming_url, params=streaming_payload)
-        try:
-            with async_timeout.timeout(TIMEOUT, loop=self.hass.loop):
-                stream = yield from self._websession.get(
-                    streaming_url,
-                    params=streaming_payload
-                )
-            response = web.StreamResponse()
-            response.content_type = stream.headers.get(CONTENT_TYPE_HEADER)
 
-            yield from response.prepare(request)
+        yield from async_aiohttp_proxy_stream(self.hass, request, stream_coro)
-
-            while True:
-                data = yield from stream.content.read(102400)
-                if not data:
-                    break
-                response.write(data)
-
-        except (asyncio.TimeoutError, aiohttp.errors.ClientError):
-            _LOGGER.exception("Error on %s", streaming_url)
-            raise HTTPGatewayTimeout()
-
-        except asyncio.CancelledError:
-            _LOGGER.debug("Close stream by frontend.")
-            response = None
-
-        finally:
-            if stream is not None:
-                stream.close()
-            if response is not None:
-                yield from response.write_eof()
 
     @property
     def name(self):

homeassistant/helpers/aiohttp_client.py

@@ -1,9 +1,13 @@
 """Helper for aiohttp webclient stuff."""
-import sys
 import asyncio
-import aiohttp
+import sys
+
+import aiohttp
+from aiohttp.hdrs import USER_AGENT, CONTENT_TYPE
+from aiohttp import web
+from aiohttp.web_exceptions import HTTPGatewayTimeout
+import async_timeout
 
-from aiohttp.hdrs import USER_AGENT
 from homeassistant.core import callback
 from homeassistant.const import EVENT_HOMEASSISTANT_STOP
 from homeassistant.const import __version__
@@ -65,6 +69,43 @@ def async_create_clientsession(hass, verify_ssl=True, auto_cleanup=True,
     return clientsession
 
 
+@asyncio.coroutine
+def async_aiohttp_proxy_stream(hass, request, stream_coro, buffer_size=102400,
+                               timeout=10):
+    """Stream websession request to aiohttp web response."""
+    response = None
+    stream = None
+
+    try:
+        with async_timeout.timeout(timeout, loop=hass.loop):
+            stream = yield from stream_coro
+
+        response = web.StreamResponse()
+        response.content_type = stream.headers.get(CONTENT_TYPE)
+
+        yield from response.prepare(request)
+
+        while True:
+            data = yield from stream.content.read(buffer_size)
+            response.write(data)
+
+    except asyncio.TimeoutError:
+        raise HTTPGatewayTimeout()
+
+    except (aiohttp.errors.ClientError,
+            aiohttp.errors.ClientDisconnectedError):
+        pass
+
+    except (asyncio.CancelledError, ConnectionResetError):
+        response = None
+
+    finally:
+        if stream is not None:
+            stream.close()
+        if response is not None:
+            yield from response.write_eof()
+
+
 @callback
 # pylint: disable=invalid-name
 def _async_register_clientsession_shutdown(hass, clientsession):