Fastapi: Check WebSocket connection with the client

[rudmac]

How can I do a server heartbeats checks without send any information

When I use websocket.application_state it returns CONNECTED, but when send some junk text within a 'disconnected' connection it raises an exception.

This is happening when I refresh my web page and the connection is lost, but the server doesn't know. I use only server->client message, one direction.

Is there any solution for that, I´m trying not use some send_text to do a heart beat.

Thanks,

[tiangolo]

WebSockets are a quite low-level set of APIs. In many cases, you have to implement the way you handle communication directly. So, in many cases, you might end up implementing REST stuff (HTTP) over WebSockets in your code.

And the implementations in clients are not necessarily as mature as we would wish. For example, there's no way to add headers to a WebSockets connection in browsers.

But given that, WebSockets have, as part of the spec, ping-pong definitions: https://developer.mozilla.org/en-US/docs/Web/API/WebSockets_API/Writing_WebSocket_servers#Pings_and_Pongs_The_Heartbeat_of_WebSockets

But you still have to send and receive them.

On the other side, you can simply catch the starlette.websockets.Disconnect exceptions.

[songzhi]

Hey Bro,I got the same problem,too.My current solution is do a custom ping-pong operation every several seconds.Here is my code:

from starlette.websockets import WebSocket, WebSocketState
import asyncio

HEART_BEAT_INTERVAL = 5
async def is_websocket_active(ws: WebSocket) -> bool:
    if not (ws.application_state == WebSocketState.CONNECTED and ws.client_state == WebSocketState.CONNECTED):
        return False
    try:
        await asyncio.wait_for(ws.send_json({'type': 'ping'}), HEART_BEAT_INTERVAL)
        message = await asyncio.wait_for(ws.receive_json(), HEART_BEAT_INTERVAL)
        assert message['type'] == 'pong'
    except BaseException:  # asyncio.TimeoutError and ws.close()
        return False
    return True

If you have some tasks that will be awaited for long time,you should use asyncio.wait_for,too.Then catch the asyncio.TimeoutError,and await is_websocket_active again.If it is closed,you can quit the whole coroutine.There is also a important thing to be keep in mind,that is in your WebSocket client code,you should deal with the ping-pong message,too.
It works for me now,have any other idea?

[tiangolo]

Thanks for the note @songzhi !

@rudmac were you able to solve your problem?

[rudmac]

Thanks for the note @songzhi !

@rudmac were you able to solve your problem?

Yes. I will close this issue.
Thanks All