AsyncAioPikaManager connecting to RabbitMQ

[RaphaelHippe]

Hi,
I am trying to write a SocketIO server that acts as an interface between a SPA (Vue) and a microservice backend using RabbitMQ as a communication layer. My idea was that the SocketIO server listens to an exchange and updates relevant clients. Further it should be able to publish events to a different exchange.

I took the example from this repo and modified it by what I found from other discussions from here:

from aiohttp import web

import socketio

sio = socketio.AsyncServer(async_mode='aiohttp', channel='super.test', client_manager=socketio.AsyncAioPikaManager('amqp://guest:guest@localhost:5672//'))
app = web.Application()
sio.attach(app)


async def background_task():
    """Example of how to send server generated events to clients."""
    count = 0
    while True:
        await sio.sleep(10)
        count += 1
        await sio.emit('my_response', {'data': 'Server generated event'})


async def index(request):
    with open('app.html') as f:
        return web.Response(text=f.read(), content_type='text/html')


@sio.event
async def my_event(sid, message):
    await sio.emit('my_response', {'data': message['data']}, room=sid)


@sio.event
async def my_broadcast_event(sid, message):
    await sio.emit('my_response', {'data': message['data']})


@sio.event
async def join(sid, message):
    sio.enter_room(sid, message['room'])
    await sio.emit('my_response', {'data': 'Entered room: ' + message['room']},
                   room=sid)


@sio.event
async def leave(sid, message):
    sio.leave_room(sid, message['room'])
    await sio.emit('my_response', {'data': 'Left room: ' + message['room']},
                   room=sid)


@sio.event
async def close_room(sid, message):
    await sio.emit('my_response',
                   {'data': 'Room ' + message['room'] + ' is closing.'},
                   room=message['room'])
    await sio.close_room(message['room'])


@sio.event
async def my_room_event(sid, message):
    await sio.emit('my_response', {'data': message['data']},
                   room=message['room'])


@sio.event
async def disconnect_request(sid):
    await sio.disconnect(sid)


@sio.event
async def connect(sid, environ):
    await sio.emit('my_response', {'data': 'Connected', 'count': 0}, room=sid)


@sio.event
def disconnect(sid):
    print('Client disconnected')


# app.router.add_static('/static', 'static')
# app.router.add_get('/', index)


async def init_app():
    sio.start_background_task(background_task)
    return app


if __name__ == '__main__':
    web.run_app(init_app())

This application starts and connects to my local rabbitmq server. From the documentation I thought the channel argument would be the exchange name on my rabbitmq. However, this creates an exchange called socketio and not super.test.

channel – The channel name on which the server sends and receives notifications. Must be the same in all the servers. With this manager, the channel name is the exchange name in rabbitmq

Did I missunderstood the documentation?

When I publish messages to the rabibtmq, I don't see them being accepted by the server. I figure I need to register some sort of callback? But I don't find any information about that in the documentation.

Further, I did not find any information on how to publish to the rabbitmq.

Any help is highly appreciated.

Thank you very much,
best regards
Raphael

[miguelgrinberg]

Socket.IO creates its own queue and manages it fully. You should not write to the queue directly, the queue is used internally for communication of the processes running Socket.IO code. If you want to emit to your clients from a process that is not a server, use the documented approach.

[RaphaelHippe]

Okay, thank you for your replay. I still haven't fully understood how it works. Here is my updated version:

socketio server:

import logging
import logging.handlers

import socketio
from aiohttp import web

logger = logging.getLogger("socket io server")
logger.setLevel(logging.INFO)
handler = logging.StreamHandler()
formatter = logging.Formatter("%(asctime)s [%(name)s] [%(levelname)s] %(message)s")
handler.setFormatter(formatter)
logger.addHandler(handler)


logger.info("Starting socket io...")
# create socketio server

sio = socketio.AsyncServer(async_mode='aiohttp', client_manager=socketio.AsyncAioPikaManager('amqp://guest:guest@localhost:5672//'), logger=logger)
app = web.Application()
sio.attach(app)

# define event handlers
@sio.event
def connect(sid, environ):
    logging.info("Client connected: %s", sid)

    # sid join room 'test'
    sio.enter_room(sid, 'test') 

    sio.emit("message", "Hello from the server!", room=sid)

@sio.event
def disconnect(sid):
    logging.info("Client disconnected: %s", sid)

async def init_app():
    # sio.start_background_task(background_task)
    return app

web.run_app(init_app())

rabbitmq consumer

import logging
import logging.handlers

import socketio
import pika

sio = socketio.AsyncServer(async_mode='aiohttp', client_manager=socketio.AsyncAioPikaManager('amqp://guest:guest@localhost:5672//'))

logging.basicConfig(level=logging.INFO)
logging.getLogger("pika").propagate = False

logger = logging.getLogger("Rabbit MQ consumer")
logger.setLevel(logging.INFO)
handler = logging.StreamHandler()
formatter = logging.Formatter("%(asctime)s [%(name)s] [%(levelname)s] %(message)s")
handler.setFormatter(formatter)
logger.addHandler(handler)

# Connect to RabbitMQ
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# Declare the exchange
channel.exchange_declare(exchange='super.test', exchange_type='fanout')

# Declare the queue
result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue

# Bind the queue to the exchange
channel.queue_bind(exchange='super.test', queue=queue_name)

# Define the callback function
def callback(ch, method, properties, body):
    logger.info("Received message: %s", body)
    sio.emit("message", body.decode("utf-8"))
    logger.info("Message sent to socket io")

# Consume messages from the queue
channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True)
channel.start_consuming()

Am I correct, that the idea is, that the 2 processes communicate via RabbitMQ. So when I call in my rabbitmq consumer sio.emit("message", body.decode("utf-8")) then that is sent through the rabbitMQ exchange socketio to the socketio server process? And then that is supposed to send it to the connected clients?

When I publish to the exchange super.test the messages reach my rabbitmq consumer process and emit is called on the AsyncServer object. But on the rabbitmq web ui I don't see any messages on the socketio exchange or the bound queue and I also don't get the messages to my connected client.

What am I missing?

[miguelgrinberg]

I am simply trying to follow your instructions. You said I should switch to a minimum example. So I did.

Sorry I wasn't clear. By "simple" I meant an example that did not include other things unrelated to Socket.IO. I did not suggest you should switch from async to sync or from RabbitMQ to Redis.

But in any case, the logs that you shared show the beginning of a successful connection, so I really have nothing to tell you, everything seems to work on this last example that you are running.

But you haven't tried to use an external client right? Why don't you try that now with this working server and client?

[RaphaelHippe]

I am sorry, but the client log tells me that the connection is not working:

Attempting polling connection to http://localhost:5000/socket.io/?transport=polling&EIO=4
INFO:engineio.client:Attempting polling connection to http://localhost:5000/socket.io/?transport=polling&EIO=4
HTTP GET request to http://localhost:5000/socket.io/?transport=polling&EIO=4&t=1680762621.3267539 failed with error HTTPConnectionPool(host='localhost', port=5000): Read timed out. (read timeout=5).
INFO:engineio.client:HTTP GET request to http://localhost:5000/socket.io/?transport=polling&EIO=4&t=1680762621.3267539 failed with error HTTPConnectionPool(host='localhost', port=5000): Read timed out. (read timeout=5).

Am I creating the server correctly?

mgr = socketio.RedisManager('redis://localhost:6379')
sio = socketio.Server(client_manager=mgr, logger=logger, engineio_logger=True)

# event handler etc...

if __name__ == '__main__':
    try:
        # wrap the Socket.IO server with a WSGI application
        app = socketio.WSGIApp(sio)

        logger.info('start the server')

        # run the server
        eventlet.wsgi.server(eventlet.listen(('localhost', 5000)), app)

Here the relevant client code:

sio = socketio.Client(logger=True, engineio_logger=True)

# event handler etc ...

sio.connect('http://localhost:5000')
sio.wait()

[miguelgrinberg]

I'm looking at the server log, not the client log. This line

qi3al27gMKLckPuwAAAA: Sending packet OPEN data {'sid': 'qi3al27gMKLckPuwAAAA', 'upgrades': ['websocket'], 'pingTimeout': 20000, 'pingInterval': 25000}

Indicates that you have a client that initiated a connection and the server accepted it. Note that I said that your logs "show the beginning of a successful connection". There should have been more logs after this line that I assume you decided to omit.

I missed that your client is returning an error, but that is strange, because your log does not show these connection failures and it should.

Are you sure you don't have any other service in your computer running on port 5000? I suggest you move to another port to rule that out.

[RaphaelHippe]

Hi,
wanted to come back to this and report that with the help of a friend finally got a working solution:

server.py

import socketio
import eventlet
import logging
import sys
import os

eventlet.monkey_patch()

# create a SocketIO server instance
sio = socketio.Server(
    client_manager=socketio.RedisManager(url="redis://127.0.0.1:6379/0")
)


# define the event handler for the "message" event
@sio.on("message")
def handle_message(sid, data):
    logging.info("Message received: {}".format(data))
    # send a reply back to the client
    sio.emit("reply", "Thanks for your message!", room=sid)


# create a WSGI application that combines the SocketIO server with a Flask app
app = socketio.WSGIApp(sio)


if __name__ == '__main__':
    try:
        logging.basicConfig(level=logging.INFO)
        # run the server
        eventlet.wsgi.server(eventlet.listen(("localhost", 5000)), app)

    except KeyboardInterrupt:
        try:
            sys.exit(0)
        except SystemExit:
            os._exit(0)

client.py

import socketio
import logging
import sys
import os

# create a SocketIO client instance
sio = socketio.Client()

# define the event handler for the "reply" event
@sio.on("reply")
def handle_reply(data):
    logging.info("Reply received: {}".format(data))


if __name__ == '__main__':
    try:
        logging.basicConfig(level=logging.INFO)
        # connect to the server
        sio.connect("http://127.0.0.1:5000")

        # send a message to the server
        sio.emit("message", "Hello, server!")

        # wait for a reply
        sio.wait()

    except KeyboardInterrupt:
        try:
            sys.exit(0)
        except SystemExit:
            os._exit(0)

external_process.py

import socketio
# use RedisManager
mgr = socketio.RedisManager(write_only=True)
mgr.emit("reply", {"data": "Connected", "count": 0})

I think part of the problems were also Redis running on docker and thus not being reachable through localhost, but only via 127.0.0.1:6379.

Eventually I also want to try to replace the RedisManager with the PikaManager, but I think that should work, in essence, the same way

Thanks to you @miguelgrinberg and @YYYasin19 for helping me with this.