[resolved] How to send data from a background worker to the frontend client?

[jtremesay-sereema]

I want to do some computations on a worker when receiving a new connection on a HTTP endpoint and send back the result.

I was able to create the HTTP endpoint, the worker, and send events from the endpoint to the worker, but not the response from the worker to the endpoint. No errors, no events, no idea of what happening.

The full sample project is available here.

My HTTP endpoint:

class EntryPointAsyncHttpConsumer(AsyncHttpConsumer):
    async def handle(self, body):
        # Send headers
        await self.send_headers(
            headers=[
                (b"Content-Type", b"text/plain"),
            ]
        )

        # Create the task
        await self.channel_layer.send(
            "foo",
            {
                "type": "foo.do_something",
                "channel_name": self.channel_name,
                "timestamp": datetime.datetime.now().timestamp(),
            },
        )

        # Headers are only sent after the first body event.
        await self.send_body(b"", more_body=True)

    async def foo_on_something_done(self, event):
        print("on_something_done", event)

        await self.send_body(event["timestamp"].encode())

        print("done")

My worker:

class WorkerSyncConsumer(SyncConsumer):
    def foo_do_something(self, event):
        print("do_something", event)

        async_to_sync(self.channel_layer.send)(
            event["channel_name"],
            {"type": "foo.on_something_done", "timestamp": event["timestamp"]},
        )

        print("done")

My asgi.py:

import os

from channels.routing import ChannelNameRouter, ProtocolTypeRouter, URLRouter
from django.core.asgi import get_asgi_application
from django.urls import path, re_path

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "proj.settings")

from app import consumers

application = ProtocolTypeRouter(
    {
        "http": URLRouter(
            [
                path("foo", consumers.EntryPointAsyncHttpConsumer.as_asgi()),
                re_path(r"", get_asgi_application()),
            ]
        ),
        "channel": ChannelNameRouter({"foo": consumers.WorkerSyncConsumer.as_asgi()}),
    }
)

Web server logs:

$ ./manage.py runserver
Watching for file changes with StatReloader
Performing system checks...

System check identified no issues (0 silenced).
February 17, 2023 - 13:40:03
Django version 4.1.7, using settings 'proj.settings'
Starting ASGI/Daphne version 4.0.0 development server at http://127.0.0.1:8000/
Quit the server with CONTROL-C.

Worker logs:

./manage.py runworker foo
Running worker for channels ['foo']
/Users/jtremesay/projects/channels_sandbox/venv/lib/python3.10/site-packages/channels/consumer.py:52: UserWarning: async_to_sync was passed a non-async-marked callable
  self.base_send = async_to_sync(send)
do_something {'type': 'foo.do_something', 'channel_name': 'specific.bcd920756e6f454d89a28040109b60f4!4c94bf48fd4a414f887bcfccfdf088dd', 'timestamp': 1676641208.616824}
done

HTTP client, desperately waiting for his answer:

$ http http://localhost:8000/foo
HTTP/1.1 200 OK
Content-Type: text/plain
Server: daphne
Transfer-Encoding: chunked

^C
$

python & dependencies versions:

$ python --version
Python 3.10.10
$ pip freeze
asgiref==3.6.0
async-timeout==4.0.2
attrs==22.2.0
autobahn==23.1.2
Automat==22.10.0
cffi==1.15.1
channels==4.0.0
channels-redis==4.0.0
constantly==15.1.0
cryptography==39.0.1
daphne==4.0.0
Django==4.1.7
hyperlink==21.0.0
idna==3.4
incremental==22.10.0
msgpack==1.0.4
pyasn1==0.4.8
pyasn1-modules==0.2.8
pycparser==2.21
pyOpenSSL==23.0.0
redis==4.5.1
service-identity==21.1.0
six==1.16.0
sqlparse==0.4.3
Twisted==22.10.0
txaio==23.1.1
typing_extensions==4.5.0
zope.interface==5.5.2

[carltongibson]

Hi @jtremesay-sereema.

Take a look at #1302. I think basically your consumer is shutting down at the end of handle().

There's a suggestion on the ticket there for adding a keepalive flag to the consumer to allow processing to continue. This should get you going. I'm looking to adding something similar along these lines for the next release.

[jtremesay-sereema]

Thanks for your response. I will try the proposed solution.

[jtremesay-sereema]

that works ! THANKS !

[carltongibson]

Super! As I say, it's on the list for the next releases, so if you want to follow along and feedback that would be handy.