How to create a parallel loop using aiohttp or asyncio in Python?

I would like to use rethinkdb .changes() feature to push some messages to users. The messages should send without any requests from the users.

I am using rethinkdb with aiohttp and websockets. How it works:

1. User sends message
2. Server puts it into rethinkdb
3. What I need: an additional loop uses rethinkdb .changes function to send updates to connected users

This is how I initiate the application:

@asyncio.coroutine
def init(loop):
    app = Application(loop=loop)
    app['sockets'] = []
    app['susers'] = []
    app.router.add_route('GET', '/', wshandler)
    handler = app.make_handler()
    srv = yield from loop.create_server(handler, '127.0.0.1', 9080)
    print("Server started at http://127.0.0.1:9080")
    return app, srv, handler

In the wshandler I have a loop, which processes incoming messages:

@asyncio.coroutine
def wshandler(request):
    resp = WebSocketResponse()
    if not resp.can_prepare(request):
        return Response(
            body=bytes(json.dumps({"error_code": 401}), 'utf-8'),
            content_type='application/json'
        )
    yield from resp.prepare(request)
    request.app['sockets'].append(resp)
    print('Someone connected')
    while True:
        msg = yield from resp.receive()
        if msg.tp == MsgType.text:
            runCommand(msg, resp, request)
        else:
            break
    request.app['sockets'].remove(resp)
    print('Someone disconnected.')
    return resp

How to create a second loop sending messages to the same pool of opened connections? How to make it thread-safe?

Generally speaking, you should try to avoid threads as much as a possible whenever running an event loop.

Unfortunately rethinkdb does not support asyncio out-of-the-box, but it does support the Tornado & Twisted frameworks. So, you could bridge Tornado & asyncio and make it work without using threads.

Edit:

As Andrew pointed out rethinkdb does support asyncio. After 2.1.0 you can presumably do:

rethinkdb.set_loop_type("asyncio")

And then in your web handlers:

res = await rethinkdb.table(tbl).changes().run(connection)
while await res.fetch_next():
   ...