anddr anddr - 1 month ago 21
Python Question

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?

Answer

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():
   ...