Sending events from celery task

[ysomad]

Hello! I have a problem with sending event from Celery task using:

• python=3.9.7
• python-socketio=5.4.1
• gunicorn=20.1.0
• eventlet=0.30.2
• celery=5.1.2
• redis=3.5.3

I did move from flask-socketio and now I'm using only python-socketio and I'm confused. Following the documentation in order to use Celery with Redis I need to connect RedisManager and I did:

# app.py

import socketio
from config import cfg

mgr = socketio.RedisManager('redis://:[email protected]:6479')
sio = socketio.Server(
    client_manager=mgr,
    cors_allowed_origins='*', 
    logger=True,
    engineio_logger=True,
)
app = socketio.WSGIApp(sio)

I have simple countdown task from which I need to send an event to a client, but for some reason event is sent to every connected client.

Task code:

# tasks.py

import time
import socketio
from celery import Celery
from config import cfg

celery = Celery('v1', broker='redis://:[email protected]:6479')

@celery.task
def start_countdown(sid, namespace, seconds) -> None:
    external_sio = socketio.RedisManager('redis://:[email protected]:6479', write_only=True)
    while seconds > 0:
        seconds -= 1
        time.sleep(1)
    external_sio.emit('finished', None, to=sid, namespace=namespace)

Celery task call:

# service.py

from tasks import start_coundown

def start_time_mode(self, sid):
    start_countdown.delay(sid, '/v1', 10)

Logs:

celery   | [2021-10-24 10:51:11,226: INFO/MainProcess] Task namespaces.v1.tasks.start_countdown[81f14418-c90d-4b3c-8fa4-992e6edac927] received
app      | AO4rXW1ySTXuPxl2AAAE: Sending packet MESSAGE data 2/v1,["started",{"mode":"time","duration":10,"words_amount":10,"language":"english","extra":null}]
app      | AO4rXW1ySTXuPxl2AAAE: Received packet MESSAGE data 2/v1,["typing",{"pressedKey":"t"}]
app      | received event "typing" from PRrHMQ33CQ_OFW3NAAAF [/v1]
app      | emitting event "typing_reply" to PRrHMQ33CQ_OFW3NAAAF [/v1]
app      | pubsub message: emit
app      | AO4rXW1ySTXuPxl2AAAE: Sending packet MESSAGE data 2/v1,["typing_reply",{"pressedKey":"t"}]
app      | pubsub message: emit
celery   | [2021-10-24 10:51:21,239: INFO/ForkPoolWorker-8] Task namespaces.v1.tasks.start_countdown[81f14418-c90d-4b3c-8fa4-992e6edac927] succeeded in 10.011719650996383s: None
app      | QzJR59fjTxIe-LKtAAAA: Sending packet MESSAGE data 2/v1,["finished"]
app      | Cannot send to sid DYzMS36e4zVukXHDAAAC
app      | AO4rXW1ySTXuPxl2AAAE: Sending packet MESSAGE data 2/v1,["finished"]

[miguelgrinberg]

Sorry about this, there is a discrepancy in the argument naming that I missed. Your emit should use room instead of to for the recipient:

external_sio.emit('finished', None, room=sid, namespace=namespace)

The main server accepts both to and room for this argument, I missed updating the client manager to also accept both names. I will update it.

Logged it as a bug: #810

[ysomad]

Thank you, Miguel! Now it's finally working as intended.