AsyncHttpConsumer improvements

channels/generic/http.py

@@ -12,6 +12,8 @@ class AsyncHttpConsumer(AsyncConsumer):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.body = []
+        self.headers_sent = False
+        self.more_body = True
 
     async def send_headers(self, *, status=200, headers=None):
         """
@@ -27,6 +29,8 @@ async def send_headers(self, *, status=200, headers=None):
         elif isinstance(headers, dict):
             headers = list(headers.items())
 
+        self.headers_sent = True
+
         await self.send(
             {"type": "http.response.start", "status": status, "headers": headers}
         )
@@ -40,9 +44,12 @@ async def send_body(self, body, *, more_body=False):
         the channel will be ignored.
         """
         assert isinstance(body, bytes), "Body is not bytes"
-        await self.send(
-            {"type": "http.response.body", "body": body, "more_body": more_body}
-        )
+
+        if self.more_body:
+            self.more_body = more_body
+            await self.send(
+                {"type": "http.response.body", "body": body, "more_body": more_body}
+            )
 
     async def send_response(self, status, body, **kwargs):
         """
@@ -70,6 +77,19 @@ async def disconnect(self):
         """
         pass
 
+    async def close(self, body=b"", status=500, headers=None):
+        """
+        Closes the HTTP response from the server end.
+        """
+        if not self.more_body:
+            # HTTP Response is already closed, nothing to do.
+            return
+
+        if not self.headers_sent:
+            await self.send_headers(status=status, headers=headers)
+
+        await self.send_body(body)
+
     async def http_request(self, message):
         """
         Async entrypoint - concatenates body fragments and hands off control
@@ -80,10 +100,15 @@ async def http_request(self, message):
         if not message.get("more_body"):
             try:
                 await self.handle(b"".join(self.body))
-            finally:
-                await self.disconnect()
-                raise StopConsumer()
+            except StopConsumer:
+                await self.close(status=200)
+            except:
+                # TODO This is just a patch, after bubbling up the exception no body is calling http_disconnect.
+                await self.close(body=b"Internal Server Error", status=500)
+                raise
 
+    # TODO This should be called by daphne whenever any exception bubbles up (As it does with websockets)
+    # IMHO this should be parallel to websocket_disconnect, removing the consumer from potential channel_layer groups.
     async def http_disconnect(self, message):
         """
         Let the user do their cleanup and close the consumer.

tests/test_generic_http.py

@@ -1,9 +1,12 @@
+import asyncio
 import json
 
 import pytest
 
 from channels.generic.http import AsyncHttpConsumer
 from channels.testing import HttpCommunicator
+from django.test import override_settings
+from channels.layers import get_channel_layer
 
 
 @pytest.mark.asyncio
@@ -32,3 +35,56 @@ async def handle(self, body):
     assert response["body"] == b'{"value": 42}'
     assert response["status"] == 200
     assert response["headers"] == [(b"Content-Type", b"application/json")]
+
+
+@pytest.mark.asyncio
+async def test_async_http_consumer_with_channel_layer():
+    """
+    Tests that AsyncHttpConsumer is implemented correctly.
+    """
+
+    class TestConsumer(AsyncHttpConsumer):
+        """
+        Abstract consumer that provides a method that handles running a command and getting a response on a
+        device.
+        """
+
+        groups = ['test_group']
+
+        async def handle(self, body):
+            print("Latest Channel ID: ")
+            await self.send_headers(
+                status=200,
+                headers=[
+                    (b"Content-Type", b"application/json"),
+                ],
+            )
+
+        async def send_to_long_poll(self, event):
+            print("RunCommandConsumer: Event received on send to long poll.")
+            command_output = str(event['data']).encode('utf8')
+            await self.send_body(command_output, more_body=False)
+
+    channel_layers_setting = {
+        "default": {"BACKEND": "channels.layers.InMemoryChannelLayer"}
+    }
+    with override_settings(CHANNEL_LAYERS=channel_layers_setting):
+        # Open a connection
+        communicator = HttpCommunicator(
+            TestConsumer,
+            method="POST",
+            path="/test/",
+            body=json.dumps({"value": 42, "anything": False}).encode("utf-8"),
+        )
+
+        def send_to_channel_layer():
+            print("send to channel layer called")
+            channel_layer = get_channel_layer()
+            # Test that the websocket channel was added to the group on connect
+            message = {"type": "send.to.long.poll", "data": "hello"}
+            asyncio.ensure_future(channel_layer.group_send("chat", message))
+
+        asyncio.get_event_loop().call_later(3, send_to_channel_layer)
+        print("Making http requests.")
+        response = await communicator.get_response(timeout=10)
+        assert True