Dynamic Message Queue Switching Fails When Using SocketIO.init_app() for Redis and RabbitMQ

[pratikbelote104896]

I am facing an issue with Flask-SocketIO while trying to dynamically switch between Redis and RabbitMQ as message queues.

Environment:

• Python: 3.11
• Flask-SocketIO: 5.4.1
• Python-EngineIO: 4.9.1
• Python-SocketIO: 5.11.4
• Gunicorn: 23.0.0
• Gevent: 24.2.1
• Gevent-WebSocket: 0.10.1

Deployment: Gunicorn with Nginx.

Message Queues: Redis and RabbitMQ.

Scenarios:

1. Case 1: Using Redis as the initial message queue works fine.
2. Case 2: Using RabbitMQ as the initial message queue works fine.
3. Case 3: If Redis is online initially and then goes down, the system switches to RabbitMQ, but the socket communication stops working. Redis coming back online switches back to Redis and resumes working.
4. Case 4: If RabbitMQ is online initially and Redis becomes available, switching to Redis stops socket communication. If Redis goes down again, it switches back to RabbitMQ and resumes working.

Expected Behavior
When the system dynamically switches between Redis and RabbitMQ as message queues (using SocketIO.init_app()), the socket communication should work seamlessly regardless of the current message queue.

Observed Behavior

• Switching the message queue dynamically (using SocketIO.init_app()):
• Causes socket communication to fail during the switch.
• The behavior depends on which message queue is initialized first, as detailed in the scenarios above.

Code to Reproduce

init.py

app = Flask(__name__)
CORS(app, resources={r"/*": {"origins": "*"}}, supports_credentials=True)

fujs = FlaskUtilJs(app) 

redis_host = 'xx.xx.x.xxx'
redis_port = 6379
redis_client = None
socketio = None

rabbit_host = 'xx.xx.x.xxx'
rabbit_port = 5672
rabbit_queue = 'missed_redis_commands'
rabbitmq_url = f'amqp://guest:guest@{rabbit_host}:5672'

async_mode = 'gevent'

current_queue = "redis"
redis_url = f'redis://{redis_host}:{redis_port}/0'
message_queue = redis_url  # Default to Redis

# Try to initialize Redis connection
try:
    redis_client = redis.StrictRedis.from_url(redis_url)
    redis_client.ping()  # Test Redis connection
    message_queue = redis_url
    print("Using Redis as the message queue.",flush=True)
except (redis.ConnectionError, redis.TimeoutError) as e:
    print(f"Redis connection failed: {e}. Falling back to RabbitMQ.",flush=True)
    current_queue = "rabbitmq"
    message_queue = rabbitmq_url
    print("Using RabbitMQ as the message queue.",flush=True)

socketio = SocketIO(app, async_mode='gevent', message_queue=message_queue, cors_allowed_origins="*", logger=True, engineio_logger=True, path='/api/socket')

def monitor_redis():
    """
    Background thread to monitor Redis availability.
    If Redis becomes unavailable, switch to RabbitMQ dynamically.
    """
    global current_queue, message_queue, redis_client, socketio

    while True:
        if current_queue == "redis":
            try:
                # Attempt to ping Redis
                redis_client.ping()
            except (redis.ConnectionError, redis.TimeoutError):
                print("Redis became unavailable. Switching to RabbitMQ...",flush=True)
                current_queue = "rabbitmq"
                message_queue = rabbitmq_url
                socketio.init_app(app, message_queue=message_queue)
                print('Switched to rabbitMQ',flush=True)
        elif current_queue == "rabbitmq":
            try:
                # Check if Redis is back online
                redis_client = redis.StrictRedis.from_url(redis_url)
                redis_client.ping()
                print("Redis is back online. Switching to Redis...",flush=True)
                current_queue = "redis"
                message_queue = redis_url
                socketio.init_app(app, message_queue=message_queue)
                print('Switched to Redis again',flush=True)
            except (redis.ConnectionError, redis.TimeoutError):
                pass  # Keep using RabbitMQ
        
        time.sleep(4)  


# Start the Redis monitor thread
monitor_thread = threading.Thread(target=monitor_redis, daemon=True)
monitor_thread.start()

def emit_event(event, data, room=None):
    try:
        socketio.emit(event, data, room=room)
    except Exception as e:
        print(f"Error emitting event: {e}", flush=True)

main.py

from gevent import monkey, spawn
monkey.patch_all()

from __init__ import app, Server, socketio
import sys, traceback, time
from datetime import datetime
from flask import render_template, request, url_for, g, jsonify, render_template_string
from flask_jwt_extended import (
    get_jwt_identity, get_jwt
)
from register_cust import regcust

app.register_blueprint(regcust)

if __name__ == '__main__':
    try:
        socketio.run(app, host=Server.SERVER_URL, port=Server.SERVER_PORT, debug=False, log_output= True)
    except KeyboardInterrupt:
        print("Server stopped with keyboard c c.")
    except Exception as e:
        print(f"Error is: {e}")

nginx.conf

nginx.conf

upstream app_servers {
    ip_hash;
    server gunicorn:8000;
}

server {
    listen 80;
    # General HTTP configuration
    location / {
        proxy_pass http://app_servers;
        proxy_http_version 1.1;  # Ensure HTTP/1.1 for WebSocket compatibility
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;

        # WebSocket-specific headers
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection $http_connection;

        # Timeout settings
        proxy_read_timeout 86400;  # For long-lived WebSocket connections
        proxy_send_timeout 86400;
    }
}

Steps to Reproduce

1. Start the application with Redis as the message queue.
2. Simulate a Redis failure, causing a switch to RabbitMQ.
3. Observe socket communication failing after the switch.
4. Restart Redis and observe the switch back to Redis, with communication working again.
5. Repeat the above steps, starting with RabbitMQ instead of Redis.

Debug Information
Logs when Redis fails and switches to RabbitMQ:

Redis became unavailable. Switching to RabbitMQ...
Switched to RabbitMQ.

Logs when Redis comes back online:

Redis is back online. Switching to Redis...
Switched to Redis again.

Request for Help
Please confirm if this is a bug or if there’s a workaround to ensure seamless switching of message queues in real-time.

Thank you for your support!

git_issue_raise.txt

[miguelgrinberg]

Switching the message queue for a server that is active is not supported.

If you need this to work, I recommend you implement your own ClientManager instance that can perform this switching internally. Then you can configure Flask-SocketIO using client_manager=MyClientManager() instead of using Redis or Rabbit directly.