How do I send channels 2.x group message from django-celery 3 task?

Question

I need to postpone sending channels message. Here is my code:

# consumers.py
class ChatConsumer(WebsocketConsumer):
    def chat_message(self, event):
        self.send(text_data=json.dumps(event['message']))

    def connect(self):
        self.channel_layer.group_add(self.room_name, self.channel_name)
        self.accept()

    def receive(self, text_data=None, bytes_data=None):
        send_message_task.apply_async(
            args=(
                self.room_name,
                {'type': 'chat_message',
                 'message': 'the message'}
            ),
            countdown=10
        )

# tasks.py
@shared_task
def send_message_task(room_name, message):
    layer = get_channel_layer()
    layer.group_send(room_name, message)

The task is being executed and I can't see any errors but message is not being sent. It works only if I send it from consumer class method.

I also tried using AsyncWebsocketConsumer and sending with AsyncToSync(layer.group_send). It errors with "You cannot use AsyncToSync in the same thread as an async event loop - just await the async function directly."

Then I tried declaring send_message_task as async and using await. Nothing happens again (with no errors) and I'm not sure if the task is executed at all.

Here are versions:

Django==1.11.13
redis==2.10.5
django-celery==3.2.2
channels==2.1.2
channels_redis==2.2.1

Settings:

REDIS_HOST = os.getenv('REDIS_HOST', '127.0.0.1')
BROKER_URL = 'redis://{}:6379/0'.format(REDIS_HOST)
CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {
            "hosts": ['redis://{}:6379/1'.format(REDIS_HOST)],
        },
    },
}

Any ideas?

UPD: Just found out that redis channel layer is retreived but it's group_send method is not called and just skipped.

UPD 2: Sending using AsyncToSync(layer.group_send) from console works. Calling task without apply_async also works. But running it with apply_async causes an error You cannot use AsyncToSync in the same thread as an async event loop - just await the async function directly. Defining task as async and using await also breaks everything of course.

Answer 1

Maybe this is not direct answer to a starting question but this might help. If you get exception "You cannot use AsyncToSync in the same thread as an async event loop - just await the async function directly" then you probably makes some of this:

1. event loop is created somewhere
2. some ASYNC code is started
3. some SYNC code is called from ASYNC code
4. SYNC code is trying to call ASYNC code with AsyncToSync that prevents this

Seems that AsyncToSync detects outer event loop and makes decision to not interfere with it.

Solution is to directly include your async call in outer event loop. Example code is below, but best is to check your situation and that outer loop is running ...

loop = asyncio.get_event_loop()
loop.create_task(layer.group_send(room_name, {'type': 'chat_message', 'message': message}))