Closed boholder closed 1 year ago
Since we can't smoothly connect rxpy and asyncio together, we just need to let them work separately (in the same one thread), producing data for each other, and exiting (and cleaning context) normally after all the data is processed.
There are four subtasks to implement:
My goal is to constantly switch between rxpy and asyncio, so that they are connected together and the data stream goes through the entire chain.
In my project (an automation command line tool for Twitter), the initial data flows to the asynchronous HTTP client as an Observable or a normal list, and the client response flows to the processing chain as a new Observable, where there may be a need to request the asynchronous HTTP client again... https://github.com/boholder/puntgun/issues/17 Therefore the data stream is constantly switched between the two frameworks for processing. For example, the data flow will goes like:
After searching, I learned that:
Sadly the
aioreactive
doesn't meet my needs, there are several operators (share(), buffer_with_count()...) I already use that it doesn't implement. https://github.com/dbrattli/aioreactiveIt is impossible to transfer data to an asynchronous function via an operator, because you can't call an async function in on_next(). https://github.com/ReactiveX/RxPY/issues/649#issuecomment-1159512451
But we can store the data at somewhere else (an asyncio.Queue) , writing a custom async operator which runs an infinite loop, to perform the async work, we can even add more operators to process the async work's result. https://github.com/ReactiveX/RxPY/issues/592#issue-1105185161 https://github.com/ReactiveX/RxPY/issues/571#issue-878054692 https://blog.oakbits.com/rxpy-and-asyncio.html
I experimented a bit based on this friendly answer, but there are still something where I'm not sure how to implement:
When we use asyncio at the same time, can rxpy guarantee to execute all operators on all data till observer before exiting?
(asyncio related) The logic of putting data in the "first segment" chain into the "second segment" chain cannot wait for the Future, and some tasks in the "second" chain are canceled before they are executed. How do I wait for all tasks to finish before exit? (A search shows that asyncio.gather() only cares about the coroutine being passed in, so manual control of the event loop might solve the problem?)
(still asyncio related) asyncio needs to manually string coroutines together using await (call add()), is there a way to automatically pass the data stream from rxpy into the asyncio.queue? (still have the problem of not being able to await Future) Is this example a solution? https://github.com/ReactiveX/RxPY/blob/master/examples/asyncio/toasyncgenerator.py
Thanks for reading my question.