Constantly send data to client from server

Question

Take a look at this example.

As you can see, some sort of event is constantly being sent to the client. I want to imitate this using Django-Channels, inside consumers.py. Here's a simplified version of what I have:

class ChatConsumer(AsyncConsumer):
    async def ws_connect(self, event):
        self.send = get_db_object()

        ....

        await self.send({
            "type": "websocket.accept"
        })

    # I need to CONSTANTLY receive & send data
    async def ws_receive(self, event):

        obj = ...# query DB and get the newest object

        json_obj = {
            'field_1': obj.field_1,
            'field_2': obj.field_2,
        }

        await self.send({
            "type": "websocket.send",
            "text": json.dumps(json_obj)
        })


    @database_sync_to_async
    def get_db_object(self, **kwargs):
        return Some_Model.objects.get(**kwargs)[0]

Here, I want my Django backend to constantly:

1. Query DB
2. Receive obj from DB
3. Send the received obj to Front-End WebSocket as event

How can I achieve this? The important thing is that I need to CONSTANTLY send data to the client.

Most of the Django-Channels resources on the internet cover only Chat Apps, which don't necessarily constantly send data to the client. I couldn't find any working code that does this job.

Please, no more recommendation for Redis or channels documentation... or some random 3rd party libraries that lacks good documentation... It's easy to recommend but hard to implement. For example, I found someone recommending Snorky, but it really lacks documentation on how to implement it.

However, if there's a website that specifically does this job, I might take a look at it, even if it doesn't use Django-Channels.

Answer 1

consumers.py

import asyncio
from channels.consumer import AsyncConsumer

class ChatConsumer(AsyncConsumer):

    async def websocket_connect(self, event):
        self.connected = True
        print("connected", event)
        await self.send({
            "type": "websocket.accept"
        })

        while self.connected:
            await asyncio.sleep(2)

            obj = # do_something (Ex: constantly query DB...)
 
            await self.send({
                'type': 'websocket.send',
                'text': # obj,
            })

    async def websocket_receive(self, event):
        print("receive", event)

    async def websocket_disconnect(self, event):
        print("disconnected", event)
        self.connected = False

Javascript

var loc = window.location;
var wsStart = 'ws://';
if (loc.protocol == 'https:') {
    wsStart = 'wss://'
}
var endpoint = wsStart + loc.host + loc.pathname;

var socket = new WebSocket(endpoint);

socket.onmessage = function(e){
    console.log("message", e);
};
socket.onopen = function(e){
    console.log("open", e);
};
socket.onerror = function(e){
    console.log("error", e)
};
socket.onclose = function(e){
    console.log("close", e)
};

All you need to do is just modify obj and send it. You can extend this function as much as you want. So, right now I'm interested in getting the latest inserted row in my PostgreSQL and injecting that row into my WebSocket. I can query my DB every 2 seconds as it was specified by await asyncio.sleep(2), and inject it into the Front-End socket.

Answer 2

Using channels==1.* and Django==1.* you can use the threading module for example:

# Some view.py
import threading
import time

class Publisher(threading.Thread):
    def __init__(self, reply_channel, frequency=0.5):
        super(Publisher, self).__init__()
        self._running = True
        self._reply_channel = reply_channel
        self._publish_interval = 1.0 / frequency

    def run(self):
        while self._running:
           self._reply_channel.send({'text': 'some data'})
           time.sleep(self._publish_interval)

    def stop(self):
        self._running = False

publishers = {}

def ws_connect(message):
    message.reply_channel.send({'accept': True})
    publisher = Publisher(reply_channel=message.reply_channel)
    publisher.start()
    publishers[message.reply_channel] = publisher

def ws_disconnect(message):
    publisher = publishers[message.reply_channel]
    publisher.stop()
    del publishers[message.reply_channel]