Hi!
So I’m writing a real-life chat app as part of a bigger app, and I’d love to help make Sanic’s WebSockets documentation better (IMHO it can be significantly improved, eg there’s this gist by Adam https://gist.github.com/ahopkins/5b6d380560d8e9d49e25281ff964ed81 and a small section in the documentation- but this info might be not quite enough for a commercial product).
I read through the documentation and discussion, and here are two issues for which I am confused:
- While I understand that Adam said that Sanic is not opinionated(Overrride websocket events like Open, error, message,close using sanic - #5 by fyndpurav) in the context of WebSockets, I’m still not sure if my current approach is viable/principled.
In particular, since Websocket is bi-directional,
For my app I want to send messages to the client and get “acks” from the client, eg acknowledge receiving a particular message.
So for every message, there’s a unique ack id attached, and the client sends the server back an ack with the message’s specific ack if received.
For simplicity (see my confusion regarding pub/sub in my second question) and although it’s ugly, for now, let’s just use a single worker and a dict of connected_websockets.
This is how i register a websocket from the client:
@app.websocket("/register_client_websocket")
async def register_websocket(request, websocket):
print("going to wait for user's info")
message = await websocket.recv()
user_data = json.loads(message)
user_id = user_data['user_id']
async with connected_websockets_lock:
connected_websockets[user_id] = websocket
print(f'Registered user {user_id}')
try:
request.app.add_task(get_acks(user_id=user_id, websocket=websocket))
await websocket.keepalive_ping()
print(f'connection with user {user_id} was closed')
if user_id in connected_websockets and connected_websockets[user_id] == websocket:
del connected_websockets[user_id]
await websocket.close()
finally:
pass
From my understanding of Websockets documentation, when the register_websocket handler finishes, the WebSocket closes automatically. This is the reason I have this line:
await websocket.keepalive_ping()
Also notice that I’m calling the task get_acks so that someone will handle the incoming acks from the client. Here’s what it looks like:
async def get_acks(user_id, websocket):
'''
Handle acks for receiving acks from client by deleting ack id from the dict acks
:param user_id:
:param websocket:
:return:
'''
try:
while True:
message = await websocket.recv()
try:
print(f'acks got the message {message}')
ack = json.loads(message)
if ack['message_type'] != 'ack':
continue
ack_id = ack['ack_id']
async with acks_lock:
acks[user_id].remove(ack_id)
except:
pass
except:
return
I’m not sure if my approach is principled. Something feels off - and there’s no real-life example or tutorial on this matter (I’d love to write it with you guys).
- Pub/Sub:
As I read through Adam’s gist example, I realized that I don’t fully understand what will happen when using pub/sub.
-To my understanding, each worker process may get any client (in the gist example by Adam, a few workers might have the same channel with different clients for example, right?)
For simplicity let’s look at a private chat app, eg assume channel_name = <user_id> and each such “channel” may have a single user in it or no users in it.
You recommend using pub/sub, in that paradigm a publisher might not know if there are any subs. So it seems like the only way for me to know if the client got the message through a Websocket (otherwise I send the message via a different push mechanism, in my case FCM) is to use some kind of acks mechanism, where I’ll have to handle Redis synchronization(since one worker might remove some ack from the same key[user_id] while another adds some ack to the same key). And yet, this feels wrong. Your advice?
