Flask-SocketIO + RabbitMQ

[frankcorneliusmartin]

Hi @miguelgrinberg,

I am trying to use Flask-SocketIO in combination with RabbitMQ. However it seems like the flask server gets stuck on a client connect event. I added debug logging to socketio-client, socketio-server and rabbitmq. However it still not clear to me why the response from the server to the client is not received. If I run the example bellow without the message broker it works fine.

The setup

pip freeze
used for both the client and server

amqp==5.0.6
astroid==2.4.2
autopep8==1.5.5
bidict==0.21.4
cached-property==1.5.2
certifi==2021.10.8
charset-normalizer==2.0.7
click==8.0.3
colorama==0.4.4
dnspython==2.1.0
eventlet==0.33.0
flake8==3.8.4
Flask==2.0.2
Flask-SocketIO==5.1.1
greenlet==1.1.2
idna==3.3
importlib-metadata==3.6.0
isort==5.7.0
itsdangerous==2.0.1
jedi==0.17.2
Jinja2==3.0.3
kombu==5.2.2
lazy-object-proxy==1.4.3
MarkupSafe==2.0.1
mccabe==0.6.1
netifaces==0.10.6
parso==0.7.1
pluggy==0.13.1
pycodestyle==2.6.0
pydocstyle==5.1.1
pyflakes==2.2.0
pylint==2.6.0
python-engineio==4.3.0
python-jsonrpc-server==0.4.0
python-language-server==0.36.2
python-socketio==5.5.0
requests==2.26.0
rfc3986==1.4.0
rope==0.18.0
six==1.16.0
snowballstemmer==2.1.0
toml==0.10.2
typed-ast==1.4.2
typing-extensions==3.7.4.3
ujson==4.0.2
urllib3==1.26.7
vine==5.0.0
Werkzeug==2.0.2
wincertstore==0.2
wrapt==1.12.1
yapf==0.30.0
zipp==3.4.0

server.py

from flask import Flask
from flask_socketio import SocketIO, emit

app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret!'

socketio = SocketIO(app, message_queue='amqp://guest:guest@localhost:5672/test', logger=True, engineio_logger=True)

@app.route('/')
def index():
    return {'msg': 'hello'}

@socketio.event
def my_event(message):
    emit('my response', {'data': 'got it!'})

if __name__ == '__main__':
    socketio.run(app, debug=True, port=5000)

client.py

import socketio

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

@sio.event
def connect():
    print("I'm connected!")

@sio.event
def connect_error(data):
    print(f'data={data}')
    print("The connection failed!")

@sio.event
def disconnect():
    print("I'm disconnected!")

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

I am using the docker image version of Rabbit MQ and run this as:
docker run -v D:\Repositories\rabbitmq-demo\rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf -d --hostname my-rabbit --name some-rabbit -p 8080:15672 -p 5672:5672 rabbitmq:3-management
_The rabbitmq.conf only contains the setting log.file.level = debug

Outputs

1. Start the server python .\server.py

Server initialized for eventlet.
 * Restarting with stat
Server initialized for eventlet.
 * Debugger is active!
 * Debugger PIN: 145-203-696
(9884) wsgi starting up on http://127.0.0.1:5000

I can also reach the endpoint /

2. Start the client python .\client.py

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=1637246035.9185674 failed with error HTTPConnectionPool(host='localhost', port=5000): Read timed out. (read timeout=5).
data=HTTPConnectionPool(host='localhost', port=5000): Read timed out. (read timeout=5)
The connection failed!
Traceback (most recent call last):
  File ".\client.py", line 18, in <module>
    sio.connect('http://localhost:5000')
  File "C:\ProgramData\Miniconda3\envs\socketio\lib\site-packages\socketio\client.py", line 338, in connect
    raise exceptions.ConnectionError(exc.args[0]) from None
socketio.exceptions.ConnectionError: HTTPConnectionPool(host='localhost', port=5000): Read timed out. (read timeout=5)

At the server the following logs are added:

(9884) accepted ('127.0.0.1', 63404)
Osr2xfTuHX6sSmiwAAAA: Sending packet OPEN data {'sid': 'Osr2xfTuHX6sSmiwAAAA', 'upgrades': ['websocket'], 'pingTimeout': 20000, 'pingInterval': 25000}

At this point the server seems to be stuck, as it will not handle my request to / either. It seems like the connect request has not been finished yet.

In the logs of RabbitMQ:

2021-11-18 14:33:57.986246+00:00 [info] <0.1724.0> Supervisor {<0.1724.0>,rabbit_connection_sup}: child helper_sup started (<0.1725.0>): {rabbit_connection_helper_sup,start_link,[]}

2021-11-18 14:33:57.986398+00:00 [info] <0.1724.0> Supervisor {<0.1724.0>,rabbit_connection_sup}: child reader started (<0.1726.0>): {rabbit_reader,start_link,[<0.1725.0>,{acceptor,{0,0,0,0,0,0,0,0},5672}]}

2021-11-18 14:33:57.986552+00:00 [info] <0.1726.0> accepting AMQP connection <0.1726.0> (172.17.0.1:57526 -> 172.17.0.2:5672)

2021-11-18 14:33:57.988458+00:00 [dbug] <0.1726.0> Raw client connection hostname during authN phase: {0,0,0,0,0,65535,44049,2}

2021-11-18 14:33:57.988548+00:00 [dbug] <0.1726.0> Resolved client hostname during authN phase: ::ffff:172.17.0.2

2021-11-18 14:33:57.988703+00:00 [dbug] <0.1726.0> User 'guest' authenticated successfully by backend rabbit_auth_backend_internal

2021-11-18 14:33:57.989419+00:00 [info] <0.1725.0> Supervisor {<0.1725.0>,rabbit_connection_helper_sup}: child collector started (<0.1727.0>): {rabbit_queue_collector,start_link,[<<"172.17.0.1:57526 -> 172.17.0.2:5672">>]}

2021-11-18 14:33:57.989936+00:00 [info] <0.1725.0> Supervisor {<0.1725.0>,rabbit_connection_helper_sup}: child channel_sup_sup started (<0.1728.0>): {rabbit_channel_sup_sup,start_link,[]}

2021-11-18 14:33:57.990048+00:00 [info] <0.1726.0> connection <0.1726.0> (172.17.0.1:57526 -> 172.17.0.2:5672): user 'guest' authenticated and granted access to vhost 'test'

2021-11-18 14:33:57.991511+00:00 [info] <0.1730.0> Supervisor {<0.1730.0>,rabbit_channel_sup}: child writer started (<0.1731.0>): {rabbit_writer,start_link,[#Port<0.242>,1,131072,rabbit_framing_amqp_0_9_1,<0.1726.0>,{<<"172.17.0.1:57526 -> 172.17.0.2:5672">>,1},true]}

2021-11-18 14:33:57.991624+00:00 [info] <0.1730.0> Supervisor {<0.1730.0>,rabbit_channel_sup}: child limiter started (<0.1732.0>): {rabbit_limiter,start_link,[{<<"172.17.0.1:57526 -> 172.17.0.2:5672">>,1}]}

2021-11-18 14:33:57.992025+00:00 [info] <0.1730.0> Supervisor {<0.1730.0>,rabbit_channel_sup}: child channel started (<0.1733.0>): {rabbit_channel,start_link,[1,<0.1726.0>,<0.1731.0>,<0.1726.0>,<<"172.17.0.1:57526 -> 172.17.0.2:5672">>,rabbit_framing_amqp_0_9_1,{user,<<"guest">>,[administrator],[{rabbit_auth_backend_internal,none}]},<<"test">>,[{<<"consumer_cancel_notify">>,bool,true},{<<"connection.blocked">>,bool,true},{<<"authentication_failure_close">>,bool,true}],<0.1727.0>,<0.1732.0>]}

2021-11-18 14:33:57.993953+00:00 [info] <0.1736.0> Supervisor {<0.1736.0>,rabbit_amqqueue_sup}: child rabbit_amqqueue started (<0.1737.0>): {rabbit_prequeue,start_link,[{amqqueue,{resource,<<"test">>,queue,<<"flask-socketio.45366e90-4f00-4df8-8d2a-acdc5af220ce">>},false,false,none,[{<<"x-expires">>,signedint,300000}],none,[],[],[],undefined,undefined,[],[],live,0,[],<<"test">>,#{user => <<"guest">>},rabbit_classic_queue,#{}},declare,<0.1735.0>]

Any help / clue would be much appreciated.

Edit: formatting

[miguelgrinberg]

Well, you are using eventlet, so RabbitMQ is going to be blocking unless you monkey patch the Python standard library. See https://flask-socketio.readthedocs.io/en/latest/deployment.html#using-multiple-workers

[frankcorneliusmartin]

Gosh I feel kind-a silly for not trying that. Thanks