How to query the websocket state

[dschien]

Hi,

I am new to async etc and before I file a stackoverflow ticket I thought I'd ask directly...

I need to monitor a long running client. My approach is to log an alive message and create an alarm if the message is absent.

My idea was to start with the sample client in the RTD getting started section (https://websockets.readthedocs.org/en/stable/intro.html) and add a second async message that logs a message all 5 minutes or so as long as the connection is alive.

I am sure what I am asking for is trivial. I just can't figure it out.

Many thanks

[SzieberthAdam]

The example in the documentation consist of a one message exchange session. You want to keep the connection alive. This means that both the server and the client handler coroutines should have an infinite loop.

An extreme simple server side handler for your problem should look like this:

async def server_handler(websocket, path):
    while True:
        client_message = await websocket.recv()
        #server_message = await do_something_with_client_message(client_message)
        await websocket.send(server_message)

And the respective client side handler:

async def client_handler(sleep_time=300):
    async with websockets.connect('ws://localhost:8765') as websocket:
        while True:
            await websocket.send(client_message)
            server_message = await websocket.recv()
            #await do_something_with_server_message(server_message)
            await asyncio.sleep(sleep_time)

Of course the handlers should be made more sophisticated to deal with errors and exceptions, but I believe you can start with them.

[aaugustin]

Another thing you can do is:

while True:
    try:
        client_message = asyncio.wait_for(websocket.recv(), timeout=300)
    except asyncio.TimeoutError:
        # uh oh

Really your question seems to be about orchestrating asyncio tasks (which has a learning curve). It doesn't look specific to websockets.

I hope this helps!

[dschien]

@SzieberthAdam Thanks. That looks good. But how do I create the second periodic, timer triggered event handler that logs my alive message in the client? More importantly, the alive message should only be logged if the websocket connection is still connected. Thus, I need to have a reference to the websocket... any suggestions?

PS: I don't need the server in my actual code, because I am listening for updates from an external system, but it is good to have for testing puposes.

[dschien]

@aaugustin Thank you for this as well. I haven't tried this, but it looks to me as if this is only polling for a single message. In my example, I am listening for randomly incoming messages from some IoT backend end. Hence, there can be many more than one per 300s. However, every 300s I want to log to a file a statement that tells me that my websocket is still connected.

If it matters, I forward the logs to AWS CloudWatch where I have defined an alarm event to send me an email if ever the websocket did not log an alive statement within the 300s interval.

[aaugustin]

Again, the questions you're asking have little to do with websockets.

It's a matter of running two asyncio tasks, one that deals with websockets messages and one that checks regularly when the last message was received and sends an alert if it's too old. The two tasks can communicate through a global variable, to keep things simple. Really you need to learn about asyncio tasks.

[SzieberthAdam]

@dschien You can achieve this by separating the tasks of doing the communication and watching the connection. The three main Future you want to observe is ws.opening_handshake,
ws.connection_closed and ws.closing_handshake. In the code below, client_handler() opens the connection and starts the two tasks, one for the communication with the server, and another to observe the state of the connection.

async def client_handler(uri, loop, sleep_time=300):
    async with websockets.connect(uri, loop=loop) as websocket:
        comm_handler = loop.create_task(do_comm(websocket, loop))
        log_handler = loop.create_task(do_log(websocket, sleep_time, loop))
        await asyncio.wait([comm_handler, log_handler])

async def do_comm(websocket, loop):
    while True:
        await websocket.send(client_message)
        server_message = await websocket.recv()
        if not server_message:
            break

async def do_log(websocket, sleep_time, loop):
    host, port = websocket.remote_address
    opening_handshake = await websocket.opening_handshake
    assert opening_handshake == True  # I believe this must always succeed
    logger.debug('{}:{} - opening handshake done; connected'.format(host, port))
    while True:
        sleep_task = loop.create_task(asyncio.sleep(sleep_time, loop=loop))
        done, pending = await asyncio.wait(
            [sleep_task, websocket.connection_closed],
            loop=loop, return_when=asyncio.FIRST_COMPLETED)
        if websocket.connection_closed in done:
            closing_handshake = await websocket.closing_handshake
            if closing_handshake:
                logger.debug('{}:{} - disconnected ({}) with closing handshake'.format(host, port, websocket.close_code))
            else:
                logger.error('{}:{} - disconnected ({}) without closing handshake'.format(host, port, websocket.close_code))
            break
        else:
            logger.info('{}:{} - still connected...'.format(host, port))

Please note that the code is not tested, I wrote it out of my head, so there might be errors in it.

Take Aymeric's advice, and dig into asyncio documentation. It is a complex framework which is hard to pick up well.

[dschien]

Thank you both. @aaugustin is absolutely right, as my opening line also said my question wasn't specific to this websockets but about asyncio. Thanks for both of your time.

For those who might come across thread looking for a running solution, I ended up with a stackoverflow question here: http://stackoverflow.com/questions/35529754/python-async-websocket-client-with-async-timer. That works quite nicely.