Signals and django channels chat room

Question

Hi i'm struggling with getting two things working simultaneously...

The channels2 chat room example was okay to get going, but i wanted to add a feature of knowing how many people where in the room. I did this by updating a model of the rooms.

Then I wanted to have a dashboard which would show the most popular current rooms, which again i wanted to update with using channels. I used the django signals method and this method worked for updating the model whilst nobody was using the chat.

However, when seeing if the dashboard updated when somebody joined the chat there was an error.

    2018-05-11 19:19:09,634 - ERROR - server - Exception inside application: You cannot use AsyncToSync in the same thread as an async event loop - just await the async function directly.
  File "/dev/channels_sk/channels-master/channels/consumer.py", line 54, in __call__
    await await_many_dispatch([receive, self.channel_receive], self.dispatch)
  File "/dev/channels_sk/channels-master/channels/utils.py", line 50, in await_many_dispatch
    await dispatch(result)
  File "/dev/channels_sk/channels-master/channels/consumer.py", line 67, in dispatch
    await handler(message)
  File "/dev/channels_sk/channels-master/channels/generic/websocket.py", line 173, in websocket_connect
    await self.connect()
  File "/dev/channels_sk/tables/consumers.py", line 19, in connect
    room.save()
  File "/dev/channels_sk/.env/lib/python3.6/site-packages/django/db/models/base.py", line 729, in save
    force_update=force_update, update_fields=update_fields)
  File "/dev/channels_sk/.env/lib/python3.6/site-packages/django/db/models/base.py", line 769, in save_base
    update_fields=update_fields, raw=raw, using=using,
  File "/dev/channels_sk/.env/lib/python3.6/site-packages/django/dispatch/dispatcher.py", line 178, in send
    for receiver in self._live_receivers(sender)
  File "/dev/channels_sk/.env/lib/python3.6/site-packages/django/dispatch/dispatcher.py", line 178, in <listcomp>
    for receiver in self._live_receivers(sender)
  File "/dev/channels_sk/tables/signals.py", line 20, in room_save_handler
    'update': instance.population,
  File "/dev/channels_sk/.env/lib/python3.6/site-packages/asgiref/sync.py", line 34, in __call__
    "You cannot use AsyncToSync in the same thread as an async event loop - "
  You cannot use AsyncToSync in the same thread as an async event loop - just await the async function directly. 

consumers.py

from channels.generic.websocket import AsyncWebsocketConsumer
import json
import time
from .models import Room
from django.db.models import F

class ChatConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.room_name = self.scope['url_route']['kwargs']['room_name']
        self.room_group_name = 'chat_%s' % self.room_name

        # create room or increment
        room, created = Room.objects.get_or_create(title=self.room_name)
        pop = room.population + 1
        room.population = F('population') + 1
        room.save()

        # Join room group
        await self.channel_layer.group_add(
            self.room_group_name,
            self.channel_name
        )

        await self.accept()

        # send new population to group
        await self.channel_layer.group_send(
            self.room_group_name,
            {
            'type': 'pop_message',
            'population': pop,
            }
        )

    async def disconnect(self, close_code):
        room = Room.objects.get(title=self.room_name)
        pop = room.population - 1
        if room.population == 1:
            if room.permanent == False:
                room.delete()
        else:
            room.population = F('population') - 1
            room.save()

        # send new population to room group
        await self.channel_layer.group_send(
            self.room_group_name,
            {
            'type': 'pop_message',
            'population': pop,
            }
        )

        # Leave room group
        await self.channel_layer.group_discard(
            self.room_group_name,
            self.channel_name
        )

    # Receive message from WebSocket
    async def receive(self, text_data):
        text_data_json = json.loads(text_data)
        message = text_data_json['message']


        # Send message to room group
        await self.channel_layer.group_send(
            self.room_group_name,
            {
                'type': 'chat_message',
                'message': message,
        )

    # Receive message from room group
    async def chat_message(self, event):
        message = event['message']

        # Send message to WebSocket
        await self.send(text_data=json.dumps({
            # 'type': 'chat_message',
            'message': message,
        }))

    # change in room group population
    async def pop_message(self, event):
        content = event['type']
        population = event['population']

        # Send message to WebSocket
        await self.send(text_data=json.dumps({
            # 'type': 'pop_message',
            'content': content,
            'population': population,

        }))


class RoomConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.group_name = 'chat_room_dash'
        print("joined dash room")

        # Join room group
        await self.channel_layer.group_add(
            self.group_name,
            self.channel_name
        )

        await self.accept()

    async def disconnect(self, close_code):
        print("left dash room")
        pass

    async def send_message(self, text_data=None):
        print(text_data)
        labels = []
        data = []
        for room in Room.objects.all()[:3]:
            labels.append(room.title)
            data.append(room.population)

        await self.send(text_data=json.dumps(
        {
            'labels': labels,
            'data': data,

        })) 

signals.py

from django.db.models.signals import post_save
from django.dispatch import receiver
from channels.layers import get_channel_layer
from asgiref.sync import async_to_sync



from .models import Room


@receiver(post_save, sender=Room)
def room_save_handler(sender, instance, **kwargs):
    channel_layer = get_channel_layer()
    async_to_sync(channel_layer.group_send)(
        'chat_room_dash',
        {
            'type': 'send_message',
            'update': instance.population,
        }
    )

so far

In reading the error i have attempted the suggested solution by changing the the signals.py room_save_handler() to be async and await the message to the group.

This however didn't ever send the message when i updated the model manually or when a user went into the chat room.

My guest is that when the first consumer calls room.save() that the room_save_handler() is also called which means there is an async call within an async call.

Any help would be great!

Answer 1

I have the same problem and I have solved it in the following way:

import asyncio

from django.db.models.signals import post_save
from django.dispatch import receiver
from channels.layers import get_channel_layer


from .models import Room


@receiver(post_save, sender=Room)
def room_save_handler(sender, instance, **kwargs):
    channel_layer = get_channel_layer()
    loop = asyncio.get_event_loop()

    coroutine = async_to_sync(channel_layer.group_send)(
        'chat_room_dash',
        {
            'type': 'send_message',
            'update': instance.population,
        }
    )
    loop.run_until_complete(coroutine)