Flask-socketio: Does not work with message_queue but in normal mode

[teslam]

When I use flask socketIO in normal mode, I can connect and send massages over socket. But when I use message_queue it seems that the client can connect but then flask socket cant send anything back to the client.

This is OK:
socket_io = SocketIO(app, async_mode=async_mode)
This is NOT OK:
socket_io = SocketIO(app, async_mode=async_mode, message_queue='redis://')

I get following message in console:
(59477) accepted ('127.0.0.1', 48810)
Cannot publish to redis... retrying
Cannot publish to redis... giving up
Cannot publish to redis... retrying
Cannot publish to redis... giving up

The reason why I believe the client connects is that I made a print out under
@socket_io.on('connect', namespace='/test') which prints out just before the error message in console.

Can anyone help me with this?

Full code:
app.py:

from flask import Flask, render_template, session
from flask_socketio import SocketIO, emit
import eventlet

eventlet.monkey_patch()

async_mode = None
app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret_key'
socket_io = SocketIO(app, async_mode=async_mode, message_queue='redis://')


@app.route('/')
def index():
    return render_template('index.html')


@socket_io.on('connect', namespace='/test')
def test_connect():
    print("New client connected")
    emit('my_response', {'data': 'Connected', 'count': 0})


@socket_io.on('my_event', namespace='/test')
def test_message(message):
    session['receive_count'] = session.get('receive_count', 0) + 1
    emit('my_response',
         {'data': message['data'], 'count': session['receive_count']})


if __name__ == '__main__':
    socket_io.run(app)

index.html:

<h1>Test Flask scocketIO</h1>
<script src="https://code.jquery.com/jquery-3.2.1.min.js"></script>
<script type="text/javascript" src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/2.1.1/socket.io.slim.js"></script>
<p id="insert"></p>
    <h2>Receive:</h2>
    <div id="log"></div>
</body>

<script>
    $(document).ready(function() {
    namespace = '/test';

    var socket = io.connect(location.protocol + '//' + document.domain + ':' + location.port + namespace);
    socket.on('connect', function() {
                socket.emit('my_event', {data: 'I\'m connected!'});
            });
    socket.on('my_response', function(msg) {
                $('#log').append('<br>' + $('<div/>').text('Received #' + msg.count + ': ' + msg.data).html());
            });
    });
</script>

[miguelgrinberg]

Have you verified that you have a healthy redis service running on localhost? The log seems to indicate that you don't have redis running.

[stratosgear]

I have a similar weird message_queue Redis error...

First off, @m8man, as help debugging your case try using a full redis://localhost:6379 reference or redis://redis:6379 (or however you called your redis instance) if you have it dockerized somehow. It is clear that the connection to Redis is not working.

@miguelgrinberg my "weird" case is like this:

This works:

socketio = SocketIO(async_mode='eventlet', logger=True, engineio_logger=True)
...
...
socketio.init_app(app)
socketio.on_namespace(SocketRoutes('/wsio'))

this does not:

socketio = SocketIO(async_mode='eventlet', message_queue='redis://localhost:6379', logger=True, engineio_logger=True)
...
...
socketio.init_app(app)
socketio.on_namespace(SocketRoutes('/wsio'))

When I say it works, it means that a single websocket call from my client of:

this.$socket.emit("join", { room: "someRoom" });

and a response from the Flask server of:

    def on_join(self, message):
        join_room(message['room'])
        emit('join_response', {'data': 'In rooms: ' + ', '.join(rooms())})
        emit('my_message', {'msg': 'Hello'})

properly gets all messages properly transferred between client and server, and the server log (among other things) displays:

402b37a1625d4356a213694184796f21: Received packet MESSAGE data 2["join",{"room":"someRoom"}]
received event "join" from 402b37a1625d4356a213694184796f21 [/]
[2018-06-19 11:37:15,712] INFO in utils: Socket event received on JOIN: {'room': 'someRoom'}
402b37a1625d4356a213694184796f21 is entering room someRoom [/]
emitting event "join_response" to 402b37a1625d4356a213694184796f21 [/]
402b37a1625d4356a213694184796f21: Sending packet MESSAGE data 2["join_response",{"data":"In rooms: 402b37a1625d4356a213694184796f21, someRoom"}]                                                                                                                              
emitting event "my_message" to 402b37a1625d4356a213694184796f21 [/]
402b37a1625d4356a213694184796f21: Sending packet MESSAGE data 2["msg",{"msg":"Hello"}]

But in the cases where it does not work the server log looks like:

1d5315baa6374ee0bbc15161f35a844f: Received packet MESSAGE data 2["join",{"room":"someRoom"}]
received event "join" from 1d5315baa6374ee0bbc15161f35a844f [/]
[2018-06-19 11:42:19,341] INFO in utils: Socket event received on JOIN: {'room': 'someRoom'}
1d5315baa6374ee0bbc15161f35a844f is entering room somRoom [/]
emitting event "join_response" to 1d5315baa6374ee0bbc15161f35a844f [/]
emitting event "my_message" to 1d5315baa6374ee0bbc15161f35a844f [/]

My connection to Redis is working because if I use a wrong connection string I do get redis connection error messages like @m8man :

[2018-06-19 11:49:55,936] INFO in utils: Socket event received on JOIN: {'room': 'someRoom'}
b368ff947c0c4dbcb7b8f296274bcdee is entering room someSchedId [/]
emitting event "join_response" to b368ff947c0c4dbcb7b8f296274bcdee [/]
(27214) accepted ('127.0.0.1', 36194)
(27214) accepted ('127.0.0.1', 36198)
Cannot publish to redis... retrying
Cannot publish to redis... giving up
emitting event "my_message" to b368ff947c0c4dbcb7b8f296274bcdee [/]
Cannot publish to redis... retrying
Cannot publish to redis... giving up

At the beginning I thought I might have something wrong on my client socket.io libraries (in my Vue app) but even when I tried a completely different library the problem persists. And it makes sense after I turned on all the server logging and I saw the messages.

I really have to be able to emit messages from external processes, so this is kinda blocking for me.

Would you have any ideas what/how I could further debug this....?

Thanks and sorry for such a long comment. I tried to be as descriptive as possible..

[stratosgear]

Something I wasn't sure and also just tried.

Getting rid of the class SocketRoutes(Namespace): class and initializing the library with a simple:

socketio = SocketIO(async_mode='eventlet', logger=True, engineio_logger=True)
...
...
socketio.init_app(app)

and adding a plain method of:

@socketio.on('join')
def on_join(message):
    join_room(message['room'])
    emit('join_response', {'data': 'In rooms: ' + ', '.join(rooms())})
    emit('my_message', {'msg': ''Hello"})

has exactly the same behavior as before. Somehow the message_queue stops it from being able to respond.

[teslam]

Thanks Miguel. I tried redis a few hours and then I switched to amqp, which worked!

[stratosgear]

Hmmm, should I open a separate issue for my case, that I thought was similar with @m8man's...?

Thanks!

[miguelgrinberg]

@stratosgear are you using one of the async frameworks for your application? If yes, did you monkey patch the standard library?

[stratosgear]

Yes I do (I mentioned so many things, I forgot to mention this):

import eventlet
eventlet.monkey_patch()

At the top-top of my server.py file where I startup Flask with a socketio.run(app, debug=True)

and Socketio is initialized as socketio = SocketIO(async_mode='eventlet', logger=True, engineio_logger=True)

[miguelgrinberg]

Is server.py your main script? If not, try moving your monkey patching lines to the main script, above all other imports. If the redis package got imported before your server.py module, then it will not use the monkey-patched functions. Redis in particular required monkey patching to work under eventlet.

Another thing you can test is remove eventlet from your virtual environment, and see if the application works fine using the message queue but without async support. This is obviously just as a test, you do not want to use this configuration in production.