(to answer Nasir's comment and for later visitors, here is my full setup)
Channels and its workers were indeed a good choice for my project and I have something working in a nice manner. It is not yet in production but working fine and code is good structured, easy to work with etc.
First of all we need to setup a worker and get it working. Let's suppose our worker class is ExternalData, we are going to setup a specific channel for the worker:
# routing.py
application = ProtocolTypeRouter({
# ...
'channel': ChannelNameRouter({
"external-data": ExternalData,
})
})
# asgi.py
from channels.layers import get_channel_layer
from asgiref.sync import async_to_sync
# ...
# add this to the end of the file
channel_layer = get_channel_layer()
logger.info("Sending start signal to ExternalData")
async_to_sync(channel_layer.send)( "external-data", { "type": "external_data.start" })
# external_data.py worker's code
# used as a singleton object
class DataStore(object):
@classmethod
async def create(cls, owner):
self = DataStore()
self.currencies = {}
self.owner = owner
# ...
return self
class ExternalData(AsyncConsumer):
started = False
# triggered from asgi.py via daphne start
async def external_data_start(self, event):
if ExternalData.started:
if settings.DEBUG:
raise RuntimeError("ExternalData already working.")
else:
logger.warning("ExternalData already working.")
return
else:
# do your initialization work here and let the data producer start listening and saving external data
ExternalData.started = True
self.store = await DataStore.create(owner=self)
DataStore in above code is not necessary of course but if you are going to do something complex it may be better to use ExternalData for just channels related things and do the other stuff in another class. With this setup you need to first run the worker:
python manage.py runworker external-data
and then start daphne (i.e. in another terminal to see output of both of them):
daphne -b 0.0.0.0 -p 8000 YOUR_PROJECT.asgi:application
In production, when you need to write a service or similar daphne should be started a little bit later (sleep 2-3 seconds for instance) to make sure worker file is processed by python and running. You may also try the asgi.py code repeatedly (i.e. in a loop with some sleep) until some environment flag is set by the worker.
Now our data provider is up but what about the clients? We need to have a consumer which will mostly act as an intermediary between our data provider and clients. For my project, data transfer requirements were covering most of the cases:
- A: when a client connects send some initial data
- B: a client can visit a page and needs to fetch some additional data related to the page
- C: a client is in a page where you need to send real-time data and update the page
- D: some new data arrived and you need to inform the client
Ours is a single page application, that's why we needed all of these. Here is the snippet which includes how I dealt with all of these cases:
# consumer.py
class FeedsConsumer(AsyncJsonWebsocketConsumer):
groups = ["broadcast"] # for requirement D
# triggered from client
async def connect(self):
await self.accept()
self.listening = set() # for requirement C
logger.info(f"New client connected: {self.channel_name}")
# for requirement A
await self.channel_layer.send("external-data",
{ "type": "external.new_client", 'client_channel': self.channel_name })
# triggered from client
async def receive_json(self, data):
# for requirement B
if data["type"] == "get_currency":
payload["type"] = "external.send_currency"
payload["client_channel"] = self.channel_name
payload["currency"] = data["currency"]
self.listen(data["currency"]) # for requirement C
await self.channel_layer.send("external-data", payload)
# for requirement C, you possibly need a counterpart unlisten to remove channel_name from the group and update self.listening set
async def listen(self, item_id):
if item_id not in self.listening:
await self.channel_layer.group_add(item_id, self.channel_name )
self.listening.add(item_id)
# below are triggered from the worker. A and B as responses. C and D as server generated messages
# for requirement A
async def init_data(self, payload):
await self.send_json(payload)
# for requirement B
async def send_currency(self, payload):
await self.send_json(payload)
# for requirement C
async def new_value(self, payload):
await self.send_json(payload)
# for requirement D
async def new_currency(self, payload):
await self.send_json(payload)
# external_data.py worker's code
class ExternalData(AsyncConsumer):
# for requirement A. triggered from consumer.
async def external_new_client(self, payload):
data_to_send = list(self.store.currencies.keys())
# prepare your data above and then send it like below
await self.channel_layer.send(payload["client_channel"], # new client
{ 'type': 'init_data',
'data': data_to_send,
})
# for requirement B. triggered from consumer.
async def external_send_currency(self, payload):
data_to_send = self.store.currencies[payload["currency"]]
# prepare your data above and then send it like below
await self.channel_layer.send(payload["client_channel"], # only the client who requested data
{ 'type': 'send_currency',
'data': data_to_send,
})
async def new_data_arrived(self, currency, value):
if currency not in self.store.currencies:
self.store.currencies[currency] = value
# requirement D. suppose this is new data so we need to notify all connected users of its availability
await self.channel_layer.group_send("broadcast", # all clients are in this group
{ 'type': 'new_currency',
'data': currency,
})
else:
# requirement C, notify listeners.
self.store.currencies[currency] = value
await self.channel_layer.group_send(currency, # all clients listening to this currency
{ 'type': 'new_value',
'currency': currency,
'value': value,
})
Hopefully I didn't mess the code and it isn't too complicated (I was lazy to paste/edit separate code for each requirement). Please ask any questions in the comments.