Detect EOF signaling remote server closed connection

httpx/concurrency.py

@@ -108,6 +108,9 @@ async def read(
 
         return data
 
+    def is_connection_dropped(self) -> bool:
+        return self.stream_reader.at_eof()
+
 
 class Writer(BaseWriter):
     def __init__(self, stream_writer: asyncio.StreamWriter, timeout: TimeoutConfig):

httpx/dispatch/http11.py

@@ -117,6 +117,8 @@ async def _send_event(self, event: H11Event, timeout: TimeoutConfig = None) -> N
         Send a single `h11` event to the network, waiting for the data to
         drain before returning.
         """
+        if self.reader.is_connection_dropped():
+            raise NotConnected
         bytes_to_send = self.h11_state.send(event)
         await self.writer.write(bytes_to_send, timeout)
 
@@ -157,6 +159,8 @@ async def _receive_event(self, timeout: TimeoutConfig = None) -> H11Event:
         """
         Read a single `h11` event, reading more data from the network if needed.
         """
+        if self.reader.is_connection_dropped():
+            raise NotConnected
         while True:
             event = self.h11_state.next_event()
             if event is h11.NEED_DATA:

httpx/interfaces.py

@@ -130,6 +130,9 @@ async def read(
     ) -> bytes:
         raise NotImplementedError()  # pragma: no cover
 
+    def is_connection_dropped(self) -> bool:
+        raise NotImplementedError()  # pragma: no cover
+
 
 class BaseWriter:
     """

tests/dispatch/test_connection_pools.py

@@ -131,3 +131,35 @@ async def test_premature_response_close(server):
         await response.close()
         assert len(http.active_connections) == 0
         assert len(http.keepalive_connections) == 0
+
+
+@pytest.mark.asyncio
+async def test_keepalive_connection_closed_by_server_is_reestablished(server):
+    """
+    Upon keep-alive connection closed by remote a new connection should be reestablished.
+    """
+    async with httpx.ConnectionPool() as http:
+        response = await http.request("GET", "http://127.0.0.1:8000/")
+        await response.read()
+
+        await server.shutdown()  # shutdown the server to close the keep-alive connection
+        await server.startup()
+
+        response = await http.request("GET", "http://127.0.0.1:8000/")
+        await response.read()
+        assert len(http.active_connections) == 0
+        assert len(http.keepalive_connections) == 1
+
+
+@pytest.mark.asyncio
+async def test_connection_closed_by_server_before_receiving_data(server):
+    """
+    If the server closes the connection after sending the response but before reading the data
+    a disconnect error is raised.
+    """
+    async with httpx.ConnectionPool() as http:
+        response = await http.request("GET", "http://127.0.0.1:8000/")
+        await server.shutdown()  # shutdown the server to close the connection before receiving the data
+
+        with pytest.raises(httpx.exceptions.NotConnected):
+            await response.read()

tests/test_multipart.py

@@ -10,11 +10,11 @@
     CertTypes,
     Client,
     Dispatcher,
-    multipart,
     Request,
     Response,
     TimeoutTypes,
     VerifyTypes,
+    multipart,
 )