79457845

Date: 2025-02-21 15:26:18
Score: 0.5
Natty:
Report link

I found a solution. The key is to pass the two apps to the worker:

import asyncio
import nest_asyncio
nest_asyncio.apply()

async def start_worker(worker: faust.Worker) -> None:
    await worker.start()

def manage_loop():
    loop = asyncio.get_event_loop_policy().get_event_loop()
    try:
        worker = faust.Worker(*[app1, app2], loop=loop)
        loop.run_until_complete(start_worker(worker))
        loop.run_forever()
    finally:
        worker.stop_and_shutdown()
        worker.close()

if __name__=='__main__':
    manage_loop()

And then to start the app you need to call

python test_faust.py worker
Reasons:
  • Long answer (-0.5):
  • Has code block (-0.5):
  • Self-answer (0.5):
  • Low reputation (1):
Posted by: VMGA