Python consume RabbitMQ and run SocketIO server

Question

Setup

I have a python application, which should consume messages from a RabbitMQ and act as a SocketIO server to a Vue2 APP. When it receives messages from RabbitMQ it should send out a message over SocketIO to the Vue2 APP. Therefore I wrote 2 classes RabbitMQHandler and SocketIOHandler. I am starting the RabbitMQHandler in a separate thread so that both the RabbitMQ consume and the wsgi server can run in parallel.

Code

import random
import threading
import socketio
import eventlet
import sys
import os
import uuid
import pika
from dotenv import load_dotenv
import logging

class RabbitMQHandler():
    def __init__(self, RABBITMQ_USER, RABBITMQ_PW, RABBITMQ_IP):
        self.queue_name = 'myqueue'
        self.exchange_name = 'myqueue'
        credentials = pika.PlainCredentials(RABBITMQ_USER, RABBITMQ_PW)
        self.connection = pika.BlockingConnection(pika.ConnectionParameters(RABBITMQ_IP, 5672, '/', credentials))
        self.channel = self.connection.channel()

        self.channel.queue_declare(queue=self.queue_name)
        self.channel.exchange_declare(exchange=self.exchange_name, exchange_type='fanout')
        self.channel.queue_bind(exchange=self.exchange_name, queue=self.queue_name)

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.connection.close()

    def run(self, callback):
        logging.info('start consuming messages...')
        self.channel.basic_consume(queue=self.queue_name,auto_ack=True, on_message_callback=callback)
        self.channel.start_consuming()


class SocketIOHandler():
    def __init__(self):
        self.id = str(uuid.uuid4())
        # create a Socket.IO server
        self.sio = socketio.Server(async_mode='eventlet', cors_allowed_origins='*')
        # wrap with a WSGI application
        self.app = socketio.WSGIApp(self.sio)

        self.sio.on('connect_to_backend', self.handle_connect)
        self.sio.on('random_number', self.handle_random_number)

    def handle_connect(self, sid, msg):
        logging.info('new socket io message')
        self.emit('connect_success', {
            'success': True,
        })

    def handle_random_number(self, sid, msg):
        logging.info('handle_random_number')
        self.emit('response_random_number', { 'number': random.randint(0,10)})

    def emit(self, event, msg):
        logging.info('socket server: {}'.format(self.id))
        logging.info('sending event: "{}"'.format(event))
        self.sio.emit(event, msg)
        logging.info('sent event: "{}"'.format(event))

    def run(self):
        logging.info('start web socket on port 8765...')
        eventlet.wsgi.server(eventlet.listen(('', 8765)), self.app)

def start_rabbitmq_handler(socketio_handler, RABBITMQ_USER, RABBITMQ_PW, RABBITMQ_IP):
    def callback(ch, method, properties, body):
        logging.info('rabbitmq handler')
        socketio_handler.emit('response_random_number', { 'number': random.randint(0,10)})

    with RabbitMQHandler(RABBITMQ_USER, RABBITMQ_PW, RABBITMQ_IP) as rabbitmq_handler:
        rabbitmq_handler.run(callback=callback)


threads = []

def main():
    global threads
    load_dotenv()
    RABBITMQ_USER = os.getenv('RABBITMQ_USER')
    RABBITMQ_PW = os.getenv('RABBITMQ_PW')
    RABBITMQ_IP = os.getenv('RABBITMQ_IP')

    socketio_handler = SocketIOHandler()
    rabbitmq_thread = threading.Thread(target=start_rabbitmq_handler, args=(socketio_handler, RABBITMQ_USER, RABBITMQ_PW, RABBITMQ_IP))
    threads.append(rabbitmq_thread)
    rabbitmq_thread.start()
    socketio_handler.run()

if __name__ == '__main__':
    try:
        logging.basicConfig(level=logging.INFO)
        logging.getLogger("pika").propagate = False
        main()
    except KeyboardInterrupt:
        try:
            for t in threads:
                t.exit()
        sys.exit(0)
    except SystemExit:
        for t in threads:
            t.exit()
        os._exit(0)

Problem

The Problem is, that when the RabbitMQHandler receives a message the event response_random_number does not get through to the Vue2 APP. Even though it is emited in the callback function. When I send the random_number event from the Vue2 APP to the python application I do get the response_random_number event back from the python application in the Vue2 APP.

So all connections work on their own, but not together. My guess would be, that there is some sort of threading communication error. I added the id to the SocketIOHandler class to make sure it is the same instanced object and the prints are the same.

The logs 'socket server: ...', sending event: ... and sent event: ... tell me, that the function is being called correctly.

Answer 1

In the end the solution for me was to have two python applications: a SocketIOHandler and a RabbitMQHandler.

For me the important bit was that the RabbitMQHandler can send messages to the SocketIOClients on RabbitMQ messages. For this I used the client_manager from SocketIO. There are several options to emit events from an external process. In the end I went with the RedisManager and in a different system with the KombuManager for RabbitMQ directly.

In the SocketIOHandler

sio = socketio.AsyncServer(async_mode='aiohttp', cors_allowed_origins='*', async_handlers=True, client_manager=socketio.AsyncRedisManager('redis://localhost:6379/0'))

In the RabbitMQHandler

sio = socketio.RedisManager('redis://localhost:6379/0', write_only=True)

Then you can call sio.emit in the RabbitMQHandler the same way you can in the SocketIOHandler.

See also this github issue.