Handle h11.Connection.next_event() RemoteProtocolError (#524)

raise (new) ConnectionClosed exception if this occurs when
socket is closed, otherwise reuse ProtocolError exception.

Add test case for ConnectionClosed behavior.

Resolves #96

httpx/dispatch/http11.py

@@ -4,6 +4,7 @@ import h11
 
 from ..concurrency.base import BaseSocketStream, ConcurrencyBackend, TimeoutFlag
 from ..config import TimeoutConfig, TimeoutTypes
+from ..exceptions import ConnectionClosed, ProtocolError
 from ..models import AsyncRequest, AsyncResponse
 from ..utils import get_logger
 
@@ -161,7 +162,17 @@ class HTTP11Connection:
         Read a single `h11` event, reading more data from the network if needed.
         """
         while True:
-            event = self.h11_state.next_event()
+            try:
+                event = self.h11_state.next_event()
+            except h11.RemoteProtocolError as e:
+                logger.debug(
+                    "h11.RemoteProtocolError exception "
+                    + f"their_state={self.h11_state.their_state} "
+                    + f"error_status_hint={e.error_status_hint}"
+                )
+                if self.stream.is_connection_dropped():
+                    raise ConnectionClosed(e)
+                raise ProtocolError(e)
 
             if isinstance(event, h11.Data):
                 logger.trace(f"receive_event event=Data(<{len(event.data)} bytes>)")

httpx/exceptions.py

@@ -150,6 +150,12 @@ class InvalidURL(HTTPError):
     """
 
 
+class ConnectionClosed(HTTPError):
+    """
+    Expected more data from peer, but connection was closed.
+    """
+
+
 class CookieConflict(HTTPError):
     """
     Attempted to lookup a cookie by name, but multiple cookies existed.

tests/conftest.py

@@ -63,6 +63,8 @@ async def app(scope, receive, send):
     assert scope["type"] == "http"
     if scope["path"].startswith("/slow_response"):
         await slow_response(scope, receive, send)
+    elif scope["path"].startswith("/premature_close"):
+        await premature_close(scope, receive, send)
     elif scope["path"].startswith("/status"):
         await status_code(scope, receive, send)
     elif scope["path"].startswith("/echo_body"):
@@ -103,6 +105,16 @@ async def slow_response(scope, receive, send):
     await send({"type": "http.response.body", "body": b"Hello, world!"})
 
 
+async def premature_close(scope, receive, send):
+    await send(
+        {
+            "type": "http.response.start",
+            "status": 200,
+            "headers": [[b"content-type", b"text/plain"]],
+        }
+    )
+
+
 async def status_code(scope, receive, send):
     status_code = int(scope["path"].replace("/status/", ""))
     await send(

tests/dispatch/test_connections.py

@@ -1,4 +1,6 @@
-from httpx import HTTPConnection
+import pytest
+
+from httpx import HTTPConnection, exceptions
 
 
 async def test_get(server, backend):
@@ -15,6 +17,15 @@ async def test_post(server, backend):
         assert response.status_code == 200
 
 
+async def test_premature_close(server, backend):
+    with pytest.raises(exceptions.ConnectionClosed):
+        async with HTTPConnection(origin=server.url, backend=backend) as conn:
+            response = await conn.request(
+                "GET", server.url.copy_with(path="/premature_close")
+            )
+            await response.read()
+
+
 async def test_https_get_with_ssl_defaults(https_server, ca_cert_pem_file, backend):
     """
     An HTTPS request, with default SSL configuration set on the client.