connect event not called after a delay celery task

[cris-m]

Here is the create_app method of Flask app. I have a separate file where all the app extension are initialized

# extensions.py file
from flask_socketio import SocketIO

socketio = SocketIO(manage_session=False,
                    message_queue='redis://redis:6379/0',
                    async_mode="threading",
                    cors_allowed_origins='*',
                    logger=True,
                    engineio_logger=True)

# app.py file
from flask import Flask
from app.exetensions import socketio

def create_app():
     app = Flask(__name__, instance_relative_config=True)
     socketio.init_app(app,  
          message_queue='redis://redis:6379/0', 
          async_mode='threading',  
          logger=True,
          engineio_logger=True)
   
      @socketio.event
      def connect():
          app.logger.info('[*] Connect event')
          socketio.emit('after_connect', {'data': 'some data'})

      if __name__ == '__main__':
          socketio.run(app)
     
      return app

The app is execute from docker-compose using the following command:

gunicorn -c python:app.config.gunicorn -k geventwebsocket.gunicorn.workers.GeventWebSocketWorker -w 1 app.app:create_app()

Here is some log after the celery task

emitting event "file" to bxj4rqRx5wZJRI0BAAAD [/]
[2022-10-04 11:37:22,290: INFO/ForkPoolWorker-2] emitting event "file" to bxj4rqRx5wZJRI0BAAAD [/]
pubsub message: emit
emitting event "process" to bxj4rqRx5wZJRI0BAAAD [/]
[2022-10-04 11:37:22,295: INFO/ForkPoolWorker-2] emitting event "process" to bxj4rqRx5wZJRI0BAAAD [/]
pubsub message: emit
[2022-10-04 11:37:22,303: INFO/ForkPoolWorker-2] Task app.models.files.tasks.process_images[457d3f53-4ba6-410f-a72b-42080a64984d] succeeded in 14.010606162999466s: None
ysBWzhtnu9kcr8KAAAI: Sending packet PING data None
ysBWzhtnu9kcr8KAAAI: Received packet PONG data
ysBWzhtnu9kcr8KAAAI: Sending packet PING data None
ysBWzhtnu9kcr8KAAAI: Received packet PONG data
ysBWzhtnu9kcr8KAAAI: Sending packet PING data None
ysBWzhtnu9kcr8KAAAI: Received packet PONG data

When the app execute a celery task for the fist time, everything goes well but after the celery task, if the browser is refresh, create event does not get call but the app refresh the client SID and I see the following log:

The WebSocket transport is not available, you must install a WebSocket server that is compatible with your async mode to enable it. See the documentation for details.

. The client listening for after_connect event but after the celery event, it does not receive the emit message from server.

[miguelgrinberg]

You are using the threading async mode. Why are you running gunicorn with the gevent worker?

[cris-m]

I added the Flask-Socketio-Chat app in docker to follow the structure I have on my project. I have reproduced the issues I am having by adding one link in chat route to execute a background celery task. To see the issue flow the following step:

1. Run the app
2. Join the chat room. (Everything work fine). Messaging will work perfectly
3. Click on Execute celery task link (all the joined, text event will be send but the server will not execute the event function)

flask_sockeio_test repo

[miguelgrinberg]

You are using an invalid gunicorn configuration for eventlet. See the documentation for how to run gunicorn when using eventlet.

[cris-m]

You are using an invalid gunicorn configuration for eventlet. See the documentation for how to run gunicorn when using eventlet.

Here a simple configuration I am using. I have set the worker_class to geventwebsocket

import os
from distutils.util import strtobool

from gevent import monkey
monkey.patch_all()

bind = f'0.0.0.0:{os.getenv("WEB_API_PORT", "8000")}'
loglevel = 'debug'
accesslog = '-'
access_log_format = "%(h)s %(l)s %(u)s %(t)s '%(r)s' %(s)s %(b)s '%(f)s' '%(a)s' in %(D)sµs"  # noqa: E501
workers = 1
worker_class = "geventwebsocket.gunicorn.workers.GeventWebSocketWorker"
threads = 100
reload = bool(strtobool(os.getenv('WEB_RELOAD', 'false')))

[miguelgrinberg]

I'm sorry, but we are confusing things here. You gave me an example application to debug. I'm commenting on that application.

I've found several problems in this example that you provided, too many to really even try to make it work:

• Gunicorn is started in threading mode. The Flask app is running in threading mode (which is fine), but the Celery app is started in default mode, which would be eventlet, since you are installing eventlet in your application. This will never work.
• You are not connecting the Socket.IO instance to the message queue.
• You start the celery task by clicking a link. This breaks the chat connection that was established in the page.
• You are creating socketio and celery_app instances in multiple places. You have to have a single instance of each in the Flask app, and a single instance of each in the Celery app. You cannot cross import files from one to the other, because then multiple instances are created in each process.

I suggest you look at my Flack (https://github.com/miguelgrinberg/flack) example if you want to learn a clean way to create a Socket.IO project that uses Celery.

[cris-m]

I'm sorry, but we are confusing things here. You gave me an example application to debug. I'm commenting on that application.

I've found several problems in this example that you provided, too many to really even try to make it work:

• Gunicorn is started in threading mode. The Flask app is running in threading mode (which is fine), but the Celery app is started in default mode, which would be eventlet, since you are installing eventlet in your application. This will never work.
• You are not connecting the Socket.IO instance to the message queue.
• You start the celery task by clicking a link. This breaks the chat connection that was established in the page.
• You are creating socketio and celery_app instances in multiple places. You have to have a single instance of each in the Flask app, and a single instance of each in the Celery app. You cannot cross import files from one to the other, because then multiple instances are created in each process.

I suggest you look at my Flack (https://github.com/miguelgrinberg/flack) example if you want to learn a clean way to create a Socket.IO project that uses Celery.

Yes, Socket.IO instance was not connected to message queue. I have fixed it. Thanks for the Flack link. I will check it to make a clean app