Websocket client daemon

[javieralmancevo]

Hi

I am new to asyncio and I am trying to create a daemon that has a websocket client that connects and gathers information from a server until it receives a SIGTERM signal. I have working code but I keep getting a mistake on closing so I suspect I am getting something wrong. I create a task with an infinite loop where the websocket client does its thing. In this corutine I catch CancelledError and close the socket but I keep getting the error: websockets.exceptions.InvalidState: Cannot close a WebSocket connection in the OPEN state.

The code is this:

URI = 'bla bla'

async def websocket_routine():
    websocket_client = await websockets.client.connect(URI)

    try:
        #Do stuff with websocket_client
    except concurrent.futures.CancelledError:
        websocket_client.close()
        print('websocket close')

def exit(signame, loop, task):
    tasks = asyncio.Task.all_tasks()
    for task in tasks:
        task.cancel()

if __name__ == '__main__':

    loop = asyncio.get_event_loop()

    task = loop.create_task(websocket_routine())

    for signame in ('SIGTERM', 'SIGINT'):
        loop.add_signal_handler(getattr(signal, signame), functools.partial(exit, signame, loop, task))

    try:
        loop.run_until_complete(task)
    except KeyboardInterrupt as e:
        tasks = asyncio.Task.all_tasks()
        for task in tasks:
            task.cancel()
        loop.run_forever()
    except asyncio.CancelledError as e:
        # From the SIGTERM handler.
        pass
    finally:
        loop.close()

[Ivoz]

have you tried await -ing the call?

[javieralmancevo]

You mean await -ing the call to closing the websocket like this
await websocket_client.close()
?

I just tried and made no difference. I am getting same error on closing.

[aaugustin]

I'm not sure why you're catching concurrent.futures.CancelledError instead of asyncio.CancelledError.

[javieralmancevo]

I was just looking at the asyncio docs, the Task part (https://docs.python.org/3.5/library/asyncio-task.html#task) they tell you that is the signal you have to catch. If you click on the CancelledError link it takes you to https://docs.python.org/3.5/library/concurrent.futures.html#concurrent.futures.CancelledError which is described as concurrent.futures.CancelledError.

As I said I am new to asyncio and just following the docs. Is it better to catch asyncio.CancelledError? The exception gets triggered because 'websocket close' gets printed.

[aaugustin]

Indeed:

>>> import asyncio
>>> import concurrent.futures
>>> asyncio.CancelledError is concurrent.futures.CancelledError
True

[aaugustin]

You must await websocket_client.wait_closed().

This isn't particularly intuitive or well-documented. However that's how asyncio does it; I tried to simplify but couldn't.

[javieralmancevo]

Thanks, I will try that. I must say I am finding the asyncio module more difficult to use than your usual python. Maybe there is no better way of doing it but it feels somehow un-pythonic.

[aaugustin]

async is hard.

It's noticeably easier with asyncio than with callbacks, though ;-)

[rr-]

I disagree, how is this (from the documentation)

async def handler(websocket, path):
    while True:
        listener_task = asyncio.ensure_future(websocket.recv())
        producer_task = asyncio.ensure_future(producer())
        done, pending = await asyncio.wait(
            [listener_task, producer_task],
            return_when=asyncio.FIRST_COMPLETED)

        if listener_task in done:
            message = listener_task.result()
            await consumer(message)
        else:
            listener_task.cancel()

        if producer_task in done:
            message = producer_task.result()
            await websocket.send(message)
        else:
            producer_task.cancel()

easier than (taken from here)

import websocket
import thread
import time

def on_message(ws, message):
    print message

def on_error(ws, error):
    print error

def on_close(ws):
    print "### closed ###"

def on_open(ws):
    def run(*args):
        for i in range(3):
            time.sleep(1)
            ws.send("Hello %d" % i)
        time.sleep(1)
        ws.close()
        print "thread terminating..."
    thread.start_new_thread(run, ())


if __name__ == "__main__":
    websocket.enableTrace(True)
    ws = websocket.WebSocketApp("ws://echo.websocket.org/",
                              on_message = on_message,
                              on_error = on_error,
                              on_close = on_close)
    ws.on_open = on_open
    ws.run_forever()

I mean I understand both approaches, but I find the latter one much more straightforward...

[aaugustin]

I'm not interested in having the coroutines vs. callbacks debate. I'd rather let each developer make their own mind about what they're more comfortable with and what works best in their projects.

This whole point of this project is to provide a coroutine-based API for websockets. If you prefer callbacks, that's fine, but you're in the wrong place here. Just use something else.