2

I am developing an application where realtime data that will be pushed to clients will come from an external API. A simple version of it can be thought like a foreign exchange currency tracker. The user will specifiy which currencies she wants to track (USD, Euro, GBP etc) and receive realtime updates. Currency data will come from an external API through long polling. My question is how to integrate this data producer to channels?

In all channels examples I found worker's work is triggered by an event but in my case it will start at the beginning, work continously and instead of receving events it will just push new values to channel layer so subscribers can be notified. So I am not sure consumer pattern is the right one. To summarize my questions:

  • Shall I use a consumer for this task and how to setup it? Considering API will be accessed by long polling async or sync consumer? Start polling external API at its connect method or just send a one-time event for this? From where and when to send this "start working" event?

  • I also want to use redis to store values for supplying initial value of currencies to the user. They'll start listening for updates on connect but may be an update will come many seconds later. Can I access the redis connection instance used by channel layer or do I need to open another connection to my redis for this purpose?

Another option for the data producer can be keeping it totally outside of Django channels as described here and just push data to channel layer but I am not sure during deployment that may be problematic with daphne. I mean how can I make sure it stays up and shares resources nicely with channels?

Thanks.

2 Answers 2

2

Workers are fine for your use case. They are meant to be long running and there isn't a new instance for every request. If you want to make your consumers async you must make sure absolutely nothing you do blocks. All db queries must be wrapped in database_sync_to_async even if the db call happens 5 levels down the call stack. You could use the Django cache API to connect to Redis but you are better off working outside of it to keep everything async. Use the redis library channels uses directly since it has async methods for working with redis as a cache.

5
  • Thanks for the answer. I got something working. Do you know how can I access channel's own redis async methods?
    – pembeci
    Commented Jul 4, 2018 at 7:56
  • If you want to send messages use the methods in the channels redis package. If you want to use it as a database connect to aioredis like this aioredis.create_redis(**channels.layers.get_channel_layer().hosts[0])
    – kagronick
    Commented Jul 5, 2018 at 20:11
  • I see. Thanks a lot.
    – pembeci
    Commented Jul 6, 2018 at 7:30
  • hi @pembeci, I have exactly the same requirements as yours and the same questions. I was wondering if you can share how you set that up
    – Nasir
    Commented Nov 21, 2018 at 14:42
  • 1
    @Nasir, I tried to explain my setup as an answer. Ask any questions if it is confusing.
    – pembeci
    Commented Nov 22, 2018 at 21:54
2

(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.

2
  • Hi, I am new to django channels. I have a celery task running in background which is supposed to send data to UI once the processing is over. I couldn't find a way to send the data. Could you please advice me a solution. Thanks!
    – laplace
    Commented Jul 7, 2020 at 19:03
  • 1
    How did you make data available in channels? Second question is will you send it to all connected users or to some specific users (i.e. is the celery task triggered from UI by a user?)?
    – pembeci
    Commented Jul 9, 2020 at 8:01

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.