Terminate long polling event after user disconnects

[tmb-ltsdw]

I have an event that that polls a certain Redis queue and emits the results back to the client. I would like this emission to stop once the user hangs up to prevent resources from getting used up and running infinite loops.

My current understanding is this so far :-

If the client emits to the websocket event subscribe , then the subscribe function is triggered and executed till it terminates. Short of supplying a broadcast parameter, the response is emitted to the client that initiate the websocket connection

on_disconnect will be called once the client hangs up, but since I'm running the code inside a While loop it is not possible to connect them via the disconnect event being triggered

This is probably not the best way to do this ( :( )

I would appreciate your comments and ideas as I think I'm doing something wrong or anti-patterned here. In general I wish to create a user specific streaming service. I have read about the possibility of starting a background thread but using the app context in those thread functions is something I am not very clear with. Providing a basic sample of my code to help visualize below, it is not a working example.

class SocketNameSpace(Namespace):
    def on_connect(self):
      pass
 
    def on_disconnect(self):
      pass

    def on_subscribe(self, payload):
       # Retrieve relevant details from DB using MySQL Alchemy
       # Subscribe to appropriate redis pubsub object
       redisClient = r.pubsub()

       while True:
          socketio.sleep(0)
          message = redisClient.get_message(timeout)

          if message:
            socketio.emit("subscribe_response_event", message)
            #Emit to client
          
          # The part I would like to implement somehow
          if current_user.disconnect():
            break
            

[miguelgrinberg]

Yes, this is not the best way to do what you need. Unfortunately the correct way is much harder.

What you want to do in your subscribe event handler is to start a background task to monitor redis, and then return immediately.

    def on_subscribe(self, payload):
        socketio.start_background_task(watch_redis, request.sid)

Then you can have an independent redis watcher task:

def watch_redis(sid):
    redisClient = r.pubsub()
    while True:
        socketio.sleep(0)
        message = redisClient.get_message(timeout)

        if message:
            socketio.emit("subscribe_response_event", message, to=sid)
            #Emit to client

Then there is the part about stopping this background task when the client disconnects. For that you can add an Event object. I have a blog article that explains this.

[tmb-ltsdw]

Thanks for the quick reply I will try to model it like this!

[tmb-ltsdw]

Hi @miguelgrinberg :

I was able to get it to work, but I am aware that since I am using eventlet, the background task is not exactly a new thread but another greenlet. I was wondering if using Event() for this made sense...

Also a couple of questions :-

1. Is a namespace object treated like a singleton where only one instance is maintained for the lifecycle of a server? If so then is assigning state variables per connection a reasonable way to notify background processes per user?

def on_connect(self):
  self.connection_map[request.sid] = threading.Event()

def on_disconnect(self):
  self.connection_map[request.sid].set()

And my background function

# event object passed from the calling socket event
def stream_background(sid, event):
  while not event.isSet():
    socketio.sleep(0)
    # Do work

For now this seems to work for me using eventlet.

2. I was wondering if just using a redis key value pair identified by socketid would also do since we are not technically using threads. (The above implementation works for me as of now but I have no way of knowing if it might cause other issues later, as of now I cannot foresee that level of depth but using Events from the threading library in a greenlet does feel like a code smell to me.

[miguelgrinberg]

I was wondering if using Event() for this made sense...

Yes, but you should use eventlet's Event class, not Python's.

Is a namespace object treated like a singleton where only one instance is maintained for the lifecycle of a server? If so then is assigning state variables per connection a reasonable way to notify background processes per user?

The namespace is a singleton. You can't store user variables directly on it because the same instance is used for all clients, but if you use a map indexed by sid then it is fine.

I was wondering if just using a redis key value pair identified by socketid would also do

Yes, that works too. Once benefit of using Redis is that you can write your keys with a timeout. This helps keep your dictionary clean of old connections.

[tmb-ltsdw]

Thanks for the insightful replies, I am marking this discussion as answered. I eventually decided to go with Redis and it works well for me.