tvoinarovskyi
commented
8 years ago It can be done fairy easy using loop.create_task. Didn't run code below thou.
#!/usr/bin/env python3
import asyncio
import asynqp
USERS = {'betty', 'bob', 'luis', 'tony'}
TASKS = set([])
def process_msg(msg):
loop = asyncio.get_event_loop()
task = loop.create_task(process_msg_coro(msg))
task.add_done_callback(lambda t: TASKS.remove(t))
TASKS.add(task)
async def process_msg_coro(msg):
print('>> {}'.format(msg.body))
# You can now do await's here
msg.ack()
async def connect():
connection = await asynqp.connect(host='dev_queue', virtual_host='asynqp_test')
channel = await connection.open_channel()
exchange = await channel.declare_exchange('inboxes', 'direct')
# we have 10 users. Set up a queue for each of them
# use different channels to avoid any interference
# during message consumption, just in case.
for username in USERS:
user_channel = await connection.open_channel()
queue = await user_channel.declare_queue('Inbox_{}'.format(username))
await queue.bind(exchange, routing_key=username)
await queue.consume(process_msg)
# deliver 10 messages to each user
for username in USERS:
for msg_idx in range(10):
msg = asynqp.Message('Msg #{} for {}'.format(msg_idx, username))
exchange.publish(msg, routing_key=username)
loop = asyncio.get_event_loop()
loop.run_until_complete(connect())
try:
loop.run_forever()
finally:
if TASKS:
loop.run_until_complete(asyncio.wait(TASKS))- You would want add a finalization to all of your components, as closing the loop without it might get you with broken connections and undefined state. In this case we wait for pending tasks to be completed before closing the application.
farefernandez
I'm trying to read and process messages from multiple queues concurrently. Below you can find the code that I'm using to set-up queues, messages and callbacks.
The code works fine, but now I'd like to issue an http call using
aiohttpin myprocess_msgcallback. The thing is:process_msgit is not a coroutine hence I cannot await aiohttp coroutines.Is there a way to schedule the execution of a coroutine instead of a simple callback when calling
queue.consume?