Python: How to call async function inside rx.subject subscribe on_next

Question

I need to call an async function inside the on_next of a python rx subscription as this:

from rx.subject import Subject
import asyncio

async def asyncPrint(value: str):
    print(f'async print: {value}')


async def main():
    s = Subject()
    s.subscribe(
        on_error=lambda e: print(e),
        on_next=lambda value: asyncPrint(value)
    )
    s.on_next('Im from the subject')

if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())

but I get the async error:

$ python test.py 
rx\core\observer\autodetachobserver.py:26:
RuntimeWarning: coroutine 'asyncPrint' was never awaited
  self._on_next(value)
RuntimeWarning: Enable tracemalloc to get the object allocation traceback

I really don't want to use asyncio.get_event_loop().run_until_complete(...) because I'm already running the main loop and I don't want to start a new one neither use nested loops.

I searched a about it and I saw that a lambda function can't be async. And I think maybe that's the problem here, because I really don't know how to get the on_next value without a lambda funtion using rx library.

I also searched about async-rx library, but this looks like the only different thing that can be done is an await subscribe(...), but that's not what I want. I want something like subscribe(on_next=await...).

Is that possible? from my javascript background its easy to start a async function inside a subscribe so it looks like a possible task for me. Hope someone could find a solution to that.

Thank's a lot!

score 1 · Answer 1 · edited Apr 03 '21 at 04:59

finally I manage to make this work by using asyncio.run_task()

from rx.subject import Subject
import asyncio

async def asyncPrint(value: str):
    print(f'async print: {value}')

async def main():
    s = Subject()
    s.subscribe(
        on_error=lambda e: print(e),
        on_next=lambda value: asyncio.create_task(asyncPrint(value))
    )
    s.on_next('Im from the subject')

if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())

output:

$ python teste.py 
async print: Im from the subject

Python: How to call async function inside rx.subject subscribe on_next

1 Answers1