Send events from management commands

[microHoffman]

Hey, I'm trying to implement SSE notifications endpoint . I have the routing setup as follows:

# asgi.py

application = ProtocolTypeRouter({
        "http": URLRouter(
            [
                re_path(r"", get_asgi_application()),
                re_path(
                    r"^notifications/(?P<user_address>0x[0-9A-Fa-f]+)/$",
                    URLRouter(django_eventstream.routing.urlpatterns),
                    {"format-channels": ["notifications-{user_address}"]},
                ),
            ]
        ),
    })

# urls.py

urlpatterns = [
    re_path(r"^notifications/(?P<user_address>0x[0-9A-Fa-f]+)/", include(django_eventstream.urls), {"format-channels": ["notifications-{user_address}"]})
]

The problem here is that I'd like to send the event (via send_event function) from separate management command, that handles the events and sending the notification and this management command runs on different process. I tried sending it directly from the management command, but unfortunately the event was not sent back to the client. If I understand this correctly, it's due that runserver and event_listener management command run on different processes.

So if I understand properly, I need to somehow pass to the EventsConsumer class the info that this event happened using django channels and then send the event from this consumer? I was also playing around with extending the EventsConsumer, but unfortunately I was not able to implement this. If you have a spare minute and you have any idea how I could achieve this (sending events from managment commands), I'd greatly appreciate any help. Thanks!

P.S: I've tried Pushpin, seems to be working alright, but ideally we would stick to the channels for more "standard" deployment.

[microHoffman]

Hey, right now I got working the interprocess communication between management command event_listener and runserver using my custom consumer. I am not able to extend the EventsConsumer (the one from django_eventstream) though, since I am not able to access self.channel_layer on EventsConsumer. Do you think it would be possible to introduce an option to add this EventsConsumer to the channel layer group. This would allow add the EventsConsumer to the channel layer group and from management command I could send the message to this channel layer group and react accordingly. Or is there any better option? I'll greatly appreciate any answer:) Thanks!

[jkarneges]

For IPC, I think you'd want to communicate among ListenerManagers rather than EventsConsumers (and let the managers dispatch to their consumers), but I have not thought about this deeply. Open to a PR.

[tommyxd]

@microHoffman Could you share some more details as how you managed to get the interprocess communication working? Thanks!

[microHoffman]

@microHoffman Could you share some more details as how you managed to get the interprocess communication working? Thanks!

Hey @tommyxd , I've ended up not using django-eventstream, but rather just django-channels. I've utilized the channel layers for the interprocess communication. For the sse consumer itself, I got inspired here: https://github.com/django/channels/issues/1302#issuecomment-508896846 .

[tommyxd]

Hey @tommyxd , I've ended up not using django-eventstream, but rather just django-channels. I've utilized the channel layers for the interprocess communication. For the sse consumer itself, I got inspired here: django/channels#1302 (comment) .

I see, thanks for the pointer! So with that how do you send something from a management command to the consumer, are there any intricacies to it?

[microHoffman]

Hey @tommyxd , I've ended up not using django-eventstream, but rather just django-channels. I've utilized the channel layers for the interprocess communication. For the sse consumer itself, I got inspired here: django/channels#1302 (comment) .

I see, thanks for the pointer! So with that how do you send something from a management command to the consumer, are there any intricacies to it?

You just (im not an expert so hopefully this makes sense:D):

• define asgi application
• setup your subscription route (i've done it in routing in asgi.py). add a corresponding SseConsumer (the one taken form the django channels GH) to this route
• in the consumer, i am adding user to the group in handle method - like this await self.channel_layer.group_add(group_name, self.channel_name) - you can access the self.channel_layer in the SseConsumer directly
• then when i want to send events (e.g. from management command), I am using this group send:

      from channels.layers import get_channel_layer
      self.channel_layer = get_channel_layer()
      async_to_sync(self.channel_layer.group_send)(recipient_address, {"type": type, "data": data})

Hopefully this makes sense:D

[tommyxd]

You just (im not an expert so hopefully this makes sense:D):

* define asgi application

* setup your subscription route (i've done it in routing in `asgi.py`). add a corresponding `SseConsumer` (the one taken form the django channels GH) to this route

* in the consumer, i am adding user to the group in `handle` method - like this `await self.channel_layer.group_add(group_name, self.channel_name)` - you can access the `self.channel_layer` in the `SseConsumer` directly

* then when i want to send events (e.g. from management command), I am using this group send:

        from channels.layers import get_channel_layer
        self.channel_layer = get_channel_layer()
        async_to_sync(self.channel_layer.group_send)(recipient_address, {"type": type, "data": data})

Hopefully this makes sense:D

Yep, it does make sense, thanks a lot!