Repository: aio-libs/aiohttp-sse

Issue:
Graceful example worked fails on browser reconnect
Gathering tasks in worker may fail all worker on a single failed connection, since `gather` will raise an exception.

https://github.com/aio-libs/aiohttp-sse/blob/be53173416f6ff820a3eb01fff53efd8b70edf33/examples/graceful_shutdown.py#L29-L46

Discussion:
codecov[bot]: ## [Codecov](https://app.codecov.io/gh/aio-libs/aiohttp-sse/pull/430?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=aio-libs) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Comparison is base [(`21cb76e`)](https://app.codecov.io/gh/aio-libs/aiohttp-sse/commit/21cb76eccc6c770fc2708531b166496f33e6114d?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=aio-libs) 99.47% compared to head [(`07ed455`)](https://app.codecov.io/gh/aio-libs/aiohttp-sse/pull/430?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=aio-libs) 99.47%.


<details><summary>Additional details and impacted files</summary>


```diff
@@           Coverage Diff           @@
##           master     #430   +/-   ##
=======================================
  Coverage   99.47%   99.47%           
=======================================
  Files           4        4           
  Lines         381      381           
  Branches       45       45           
=======================================
  Hits          379      379           
  Misses          1        1           
  Partials        1        1           
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/aio-libs/aiohttp-sse/pull/430?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=aio-libs).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=aio-libs).


Patch:
diff --git a/CHANGES/370.bugfix b/CHANGES/370.bugfix
new file mode 100644
index 0000000..f5ff993
--- /dev/null
+++ b/CHANGES/370.bugfix
@@ -0,0 +1,1 @@
+Cancel ping task on ConnectionResetError.
diff --git a/CHANGES/391.feature b/CHANGES/391.feature
new file mode 100644
index 0000000..1eae15b
--- /dev/null
+++ b/CHANGES/391.feature
@@ -0,0 +1,1 @@
+Added the ability to publicly access the connection status via `.is_connected()` method.
diff --git a/CHANGES/421.misc b/CHANGES/421.misc
new file mode 100644
index 0000000..607bd72
--- /dev/null
+++ b/CHANGES/421.misc
@@ -0,0 +1,1 @@
+Add coverage for preparing already closed connection.
diff --git a/CHANGES/439.doc b/CHANGES/439.doc
new file mode 100644
index 0000000..41ac31f
--- /dev/null
+++ b/CHANGES/439.doc
@@ -0,0 +1,2 @@
+On the `CHANGES/README.rst <https://github.com/aio-libs/aiohttp-sse/tree/master/CHANGES/README.rst>`_ page,
+a link to the ``Towncrier philosophy`` has been fixed.
diff --git a/CHANGES/README.rst b/CHANGES/README.rst
index 84eab79..9484d93 100644
--- a/CHANGES/README.rst
+++ b/CHANGES/README.rst
@@ -83,4 +83,4 @@ File :file:`CHANGES/553.feature.rst`:
 
 
 .. _Towncrier philosophy:
-   https://towncrier.readthedocs.io/en/actual-freaking-docs/#philosophy
+   https://towncrier.readthedocs.io/en/stable/#philosophy
diff --git a/aiohttp_sse/__init__.py b/aiohttp_sse/__init__.py
index 7a18da2..93119ed 100644
--- a/aiohttp_sse/__init__.py
+++ b/aiohttp_sse/__init__.py
@@ -46,6 +46,10 @@ class EventSourceResponse(StreamResponse):
         self._ping_task = None
         self._sep = sep if sep is not None else self.DEFAULT_SEPARATOR
 
+    def is_connected(self) -> bool:
+        """Check connection is prepared and ping task is not done."""
+        return self.prepared and not self._ping_task.done()
+
     async def _prepare(self, request):
         await self.prepare(request)
         return self
@@ -165,7 +169,10 @@ class EventSourceResponse(StreamResponse):
         # as ping message.
         while True:
             await asyncio.sleep(self._ping_interval)
-            await self.write(": ping{0}{0}".format(self._sep).encode("utf-8"))
+            try:
+                await self.write(": ping{0}{0}".format(self._sep).encode("utf-8"))
+            except ConnectionResetError:
+                self._ping_task.cancel()
 
     async def __aenter__(self):
         return self
diff --git a/examples/chat.py b/examples/chat.py
index 4d97516..684fd78 100644
--- a/examples/chat.py
+++ b/examples/chat.py
@@ -111,7 +111,6 @@ async def subscribe(request):
     return response
 
 
-loop = asyncio.get_event_loop()
 app = Application()
 app[channels] = set()
 
diff --git a/examples/graceful_shutdown.py b/examples/graceful_shutdown.py
index 065c883..6c21ed1 100644
--- a/examples/graceful_shutdown.py
+++ b/examples/graceful_shutdown.py
@@ -1,5 +1,6 @@
 import asyncio
 import json
+import logging
 import weakref
 from contextlib import suppress
 from datetime import datetime
@@ -25,6 +26,13 @@ class SSEResponse(EventSourceResponse):
         await self.send(json_dumps(data), id=id, event=event, retry=retry)
 
 
+async def send_event(stream, data, event_id):
+    try:
+        await stream.send_json(data, id=str(event_id))
+    except Exception:
+        logging.exception("Exception when sending event: %s", event_id)
+
+
 async def worker(app):
     while True:
         now = datetime.now()
@@ -36,7 +44,8 @@ async def worker(app):
                 "time": f"Server Time : {now}",
                 "last_event_id": stream.last_event_id,
             }
-            fs.append(stream.send_json(data, id=now.timestamp()))
+            coro = send_event(stream, data, now.timestamp())
+            fs.append(coro)
 
         # Run in parallel
         await asyncio.gather(*fs)
@@ -47,7 +56,7 @@ async def worker(app):
 
 async def on_startup(app):
     app[streams] = weakref.WeakSet()
-    app[worker] = app.loop.create_task(worker(app))
+    app[worker] = asyncio.create_task(worker(app))
 
 
 async def clean_up(app):
@@ -62,7 +71,7 @@ async def on_shutdown(app):
         stream.stop_streaming()
         waiters.append(stream.wait())
 
-    await asyncio.gather(*waiters)
+    await asyncio.gather(*waiters, return_exceptions=True)
     app[streams].clear()
 
 


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index f3edd51..d64a18d 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -9,17 +9,13 @@ def debug(request):
     return request.param
 
 
-@pytest.fixture
+@pytest.fixture(autouse=True)
 def loop(event_loop, debug):
     event_loop.set_debug(debug)
     return event_loop
 
 
 @pytest.fixture
-def session(loop):
-    async def create_session(loop):
-        return aiohttp.ClientSession()
-
-    session = loop.run_until_complete(create_session(loop))
-    yield session
-    loop.run_until_complete(session.close())
+async def session():
+    async with aiohttp.ClientSession() as session:
+        yield session
diff --git a/tests/test_sse.py b/tests/test_sse.py
index 696cba5..e3ae26a 100644
--- a/tests/test_sse.py
+++ b/tests/test_sse.py
@@ -23,7 +23,7 @@ async def make_runner(app, host, port):
     (False, True),
     ids=("without_sse_response", "with_sse_response"),
 )
-async def test_func(loop, unused_tcp_port, with_sse_response, session):
+async def test_func(unused_tcp_port, with_sse_response, session):
     async def func(request):
         if with_sse_response:
             resp = await sse_response(request, headers={"X-SSE": "aiohttp_sse"})
@@ -78,7 +78,7 @@ async def test_func(loop, unused_tcp_port, with_sse_response, session):
     await runner.cleanup()
 
 
-async def test_wait_stop_streaming(loop, unused_tcp_port, session):
+async def test_wait_stop_streaming(unused_tcp_port, session):
     async def func(request):
         app = request.app
         resp = EventSourceResponse()
@@ -113,12 +113,12 @@ async def test_wait_stop_streaming(loop, unused_tcp_port, session):
     await runner.cleanup()
 
 
-async def test_retry(loop, unused_tcp_port, session):
+async def test_retry(unused_tcp_port, session):
     async def func(request):
         resp = EventSourceResponse()
         await resp.prepare(request)
         with pytest.raises(TypeError):
-            await resp.send("foo", retry="one")
+            await resp.send("foo", retry="one")  # type: ignore
         await resp.send("foo", retry=1)
         resp.stop_streaming()
         await resp.wait()
@@ -159,7 +159,7 @@ def test_compression_not_implemented():
         response.enable_compression()
 
 
-def test_ping_property(loop):
+def test_ping_property():
     response = EventSourceResponse()
     default = response.DEFAULT_PING_INTERVAL
     assert response.ping_interval == default
@@ -174,7 +174,7 @@ def test_ping_property(loop):
         response.ping_interval = -42
 
 
-async def test_ping(loop, unused_tcp_port, session):
+async def test_ping(unused_tcp_port, session):
     async def func(request):
         app = request.app
         resp = EventSourceResponse()
@@ -209,7 +209,47 @@ async def test_ping(loop, unused_tcp_port, session):
     await runner.cleanup()
 
 
-async def test_context_manager(loop, unused_tcp_port, session):
+async def test_ping_reset(unused_tcp_port, session):
+    async def func(request):
+        app = request.app
+        resp = EventSourceResponse()
+        resp.ping_interval = 1
+        await resp.prepare(request)
+        await resp.send("foo")
+        app[socket].append(resp)
+        await resp.wait()
+        return resp
+
+    app = web.Application()
+    app[socket] = []
+    app.router.add_route("GET", "/", func)
+
+    host = "127.0.0.1"
+    runner = await make_runner(app, host, unused_tcp_port)
+    url = f"http://127.0.0.1:{unused_tcp_port}/"
+
+    resp_task = asyncio.create_task(session.request("GET", url))
+
+    await asyncio.sleep(1.15)
+    esource = app[socket][0]
+
+    def reset_error_write(data):
+        raise ConnectionResetError("Cannot write to closing transport")
+
+    esource.write = reset_error_write
+    await esource.wait()
+    assert esource._ping_task.cancelled()
+    resp = await resp_task
+
+    assert 200 == resp.status
+    streamed_data = await resp.text()
+
+    expected = "data: foo\r\n\r\n" + ": ping\r\n\r\n"
+    assert streamed_data == expected
+    await runner.cleanup()
+
+
+async def test_context_manager(unused_tcp_port, session):
     async def func(request):
         h = {"X-SSE": "aiohttp_sse"}
         async with sse_response(request, headers=h) as sse:
@@ -263,7 +303,7 @@ async def test_custom_response_cls(with_subclass):
 
 
 @pytest.mark.parametrize("sep", ["\n", "\r", "\r\n"], ids=("LF", "CR", "CR+LF"))
-async def test_custom_sep(loop, unused_tcp_port, session, sep):
+async def test_custom_sep(unused_tcp_port, session, sep):
     async def func(request):
         h = {"X-SSE": "aiohttp_sse"}
         async with sse_response(request, headers=h, sep=sep) as sse:
@@ -352,7 +392,7 @@ async def test_custom_sep(loop, unused_tcp_port, session, sep):
         "steam-CR+LF:line-CR+LF",
     ),
 )
-async def test_multiline_data(loop, unused_tcp_port, session, stream_sep, line_sep):
+async def test_multiline_data(unused_tcp_port, session, stream_sep, line_sep):
     async def func(request):
         h = {"X-SSE": "aiohttp_sse"}
         lines = line_sep.join(["foo", "bar", "xyz"])
@@ -390,7 +430,52 @@ async def test_multiline_data(loop, unused_tcp_port, session, stream_sep, line_s
     await runner.cleanup()
 
 
-@pytest.mark.asyncio
+async def test_sse_state(unused_tcp_port, session):
+    async def func(request):
+        async with sse_response(request) as resp:
+            assert resp.is_connected()
+
+        assert not resp.is_connected()
+        return resp
+
+    app = web.Application()
+    app.router.add_route("GET", "/", func)
+
+    host = "127.0.0.1"
+    runner = await make_runner(app, host, unused_tcp_port)
+    await session.request("GET", f"http://{host}:{unused_tcp_port}/")
+    await runner.cleanup()
+
+
+async def test_connection_is_not_alive(unused_tcp_port, session):
+    async def func(request):
+        # within context manager first preparation is already done
+        async with sse_response(request) as resp:
+            resp.ping_interval = 1
+
+            # we should sleep to switch asyncio Task
+            # and let connection to be closed
+            while resp.is_connected():
+                await asyncio.sleep(0.01)
+
+            # this call should be cancelled, cause connection is closed
+            with pytest.raises(asyncio.CancelledError):
+                await resp.prepare(request)
+
+            return resp  # pragma: no cover
+
+    app = web.Application()
+    app.router.add_route("GET", "/", func)
+
+    host = "127.0.0.1"
+    runner = await make_runner(app, host, unused_tcp_port)
+
+    async with session.get(f"http://{host}:{unused_tcp_port}/") as resp:
+        assert resp.status == 200
+
+    await runner.cleanup()
+
+
 class TestLastEventId:
     async def test_success(self, unused_tcp_port, session):
         async def func(request):