Currently working on RxPY 3.0 which is heavily inspired by RxJS 6.0. Need help with fixing things after code refactoring. New features:

• Operators are plain and pipable functions
• New pipe operator
• No more method chaining (see timeflies_tkinter.py for working example)
• Code base reduced. With pipable operators it's very easy to create your own operators
• Backpressure removed. Can be made as extension if anyone wants to care for it.
• Result mappers (result_mapper) will be removed to make implementation simpler.

What remains needs to be done is:

• Unit tests (currently only test_map and test_filter is working). Needs to be updated to new pipeline style.
• Documentation. The current README.md is bloated. Needs to be simplified.
• Add lazy loading for the rest of the operators in rx/operators/__init__.py.
• Add lazy loading for the rest of the static creation operators in rx/.__init__.py.

PRs are very much welcome to the feature/rxpy-3.0 branch.

When running pytest locally, I'm experiencing some weird behavior with PyQt5 & Pyside2 successive tests and the skip mechanism. I assume this is due to some conflict with the skip variable declared at the module level.

Anyway, this PR replaces skip variable by pytest.importorskip function instead (as in other mainloop scheduler tests), plus replaces 'qt5' by 'pyqt5' to be consistent with 'pyside2'.

I have two concerns about the rx.concurrency.currentthreadscheduler.CurrentThreadScheduler class implementation. Both relate to the __new__ method:

    def __new__(cls) -> 'CurrentThreadScheduler':
        """Ensure that each thread has at most a single instance."""

        thread = threading.current_thread()
        self: 'CurrentThreadScheduler' = CurrentThreadScheduler._global.get(thread)
        if not self:
            self = super().__new__(cls)
            CurrentThreadScheduler._global[thread] = self
        return self

1. Is a CurrentThreadScheduler always supposed to schedule actions on the same thread? In my opinion, the CurrentThreadScheduler is attached to some thread once the trampoline is run for the first time or "reactivated".

2. When I tried to extend from CurrentThreadScheduler in rxbackpressure I couldn't do it because the __new__ method would always return a CurrentThreadScheduler.

class MyCurrentThreadScheduler(SchedulerMixin, CurrentThreadScheduler):
    pass

my_current_thread_scheduler = MyCurrentThreadScheduler()
print(type(my_current_thread_scheduler))     # would print CurrentThreadScheduler

This issue is for tracking the remaining work for RxPY v3.

• [x] Fix type hints for pipe #355
• [x] Disposables. Have been removed for RxJS 6. I think we should keep disposables, but make them simpler and remove the static Disposable object with methods and only have the simpler creation functions like create, empty etc.
• [x] Move all observer files to core/observer.
• [x] Fix testing/marbles.py.
• [x] Lazy load schedulers. We have many schedulers that are not being used. Perhaps eventloop schedulers should be imported explicitly (full path) so they don't add to startup time
• [x] Remove (result) mappers from combine_latest and with_latest_from.
• [x] Add a starmap operator based on itertools for use with combine_latest et al.
• [x] Remove BlockingObservable and methods and provide a blocking run() method instead.
• [x] Remove all first arguments that accept Iterable[Observable] for all operators. Use *args instead.

Later, i.e v3.1

• Merge internal and core to internal similar to RxJS.
• Making Observables of type Generic[T] instead of Generic[Any]. The problem is what to do with vararg operators such as zip, combine_latest etc.

Sorry if this is a silly question, but I am slightly confused as to what artifacts are being exposed from core/__init__.py. In some cases these are typings (such as Disposable, Scheduler, Observer) and in other cases these are base implementations (e.g. Observable) which extend the typings.

Would it be better to consistently expose base implementations here, which are presumably more directly relevant for users, and keep the typings only in the package of the same name, for (mostly) internal use?

I can't get proper result with all marbles and marble Diagrams related code in Getting Started.ipynb.

In [8]:

from rx.testing import marbles

xs = Observable.from_marbles("a-b-c-|") xs.to_blocking().to_marbles()

Out[8]:

'a-b-c-|'

but my output is: ''

In [9]:

xs = Observable.from_marbles("1-2-3-x-5") ys = Observable.from_marbles("1-2-3-4-5") xs.merge(ys).to_blocking().to_marbles()

Out[9]:

'11-22-33-4x'

but my output is: ''

I get none output about all these two marbles and marble diagrams related sample code! My Python is 3.4.4 (v3.4.4:737efcadf5a6, Dec 20 2015, 20:20:57) [MSC v.1600 64 bit (AMD64)] on win32 and my RxPY is Rx (1.5.2). Can anyone help me？

I feel it's time to get v3 released. Any last minute changes that need to go in first?

PS: It started with a tweet: https://twitter.com/mainro/status/1072965958921277441

With v3, an official documentation should be provided. This issue can be used to follow the progress on this topic. The following things must be done

• [X] create outline
• [X] migrate part of readme in doc
• [X] reference doc from readme
• [X] add a migration section
• [X] document APIs : autodoc for all operators and classes
• [X] choose a theme : Guzzle.
• [x] publish the doc somewhere: https://rxpy.readthedocs.io
• [x] add marble diagrams for operators
• [X] add additional readings section
• [X] license section
• [X] contributing section
• [X] Fix docstring errors in init.py
• [x] spellcheck
• [ ] final pass on each section

Hey Dag, I have a few final questions before I finalize my material for the video shoot on Tuesday. Thanks again for all your responsiveness and help. I think RxPy is going to be helpful for a lot of folks I am speaking to.

QUESTION 1:

In Java Concurrency in Practice, Goetz says that for computation-intensive tasks, a roughly optimal thread pool size is (# of CPU Cores) + 1. I am guessing that when creating a ThreadPoolScheduler to support computation-intensive processes, this rule-of-thumb is still valid?

# calculate number of CPU's and add 1, then create a ThreadPoolScheduler with that number of threads
optimal_thread_count = multiprocessing.cpu_count() + 1
pool_scheduler = ThreadPoolScheduler(optimal_thread_count)

QUESTION 2:

Coming from RxJava, I noticed that subscribe_on() will take one thread from a ThreadPoolScheduler as expected.

from rx import Observable
from rx.concurrency import ThreadPoolScheduler
from threading import current_thread
import multiprocessing, time, random


def intense_calculation(value):
    # sleep for a random short duration between 0.5 to 2.0 seconds to simulate a long-running calculation
    time.sleep(random.randint(5,20) * .1)
    return value

# calculate number of CPU's and add 1, then create a ThreadPoolScheduler with that number of threads
optimal_thread_count = multiprocessing.cpu_count() + 1
pool_scheduler = ThreadPoolScheduler(optimal_thread_count)

# Create Process 1
Observable.from_(["Alpha","Beta","Gamma","Delta","Epsilon"]) \
    .subscribe_on(pool_scheduler) \
    .map(lambda s: intense_calculation(s)) \
    .subscribe(on_next=lambda i: print("PROCESS 1: {0} {1}".format(current_thread().name, i)),
               on_error=lambda e: print(e),
               on_completed=lambda: print("PROCESS 1 done!"))

input("Press any key to quit\n")

OUTPUT:

Press any key to quit
PROCESS 1: Thread-1 Alpha
PROCESS 1: Thread-1 Beta
PROCESS 1: Thread-1 Gamma
PROCESS 1: Thread-1 Delta
PROCESS 1: Thread-1 Epsilon
PROCESS 1 done!

However, observe_on() behaves quite differently in how it claims threads. I understand that observe_on() will redirect emissions to a different scheduler at that point in the Observable chain, but I noticed it may use a different thread for each emission.

from rx import Observable
from rx.concurrency import ThreadPoolScheduler
from threading import current_thread
import multiprocessing, time, random


def intense_calculation(value):
    # sleep for a random short duration between 0.5 to 2.0 seconds to simulate a long-running calculation
    time.sleep(random.randint(5,20) * .1)
    return value

# calculate number of CPU's and add 1, then create a ThreadPoolScheduler with that number of threads
optimal_thread_count = multiprocessing.cpu_count() + 1
pool_scheduler = ThreadPoolScheduler(optimal_thread_count)

# Create Process 1
Observable.from_(["Alpha","Beta","Gamma","Delta","Epsilon"]) \
    .observe_on(pool_scheduler) \
    .map(lambda s: intense_calculation(s)) \
    .subscribe(on_next=lambda i: print("PROCESS 1: {0} {1}".format(current_thread().name, i)),
               on_error=lambda e: print(e),
               on_completed=lambda: print("PROCESS 1 done!"))

input("Press any key to quit\n")

OUTPUT:

Press any key to quit
PROCESS 1: Thread-1 Alpha
PROCESS 1: Thread-2 Beta
PROCESS 1: Thread-1 Gamma
PROCESS 1: Thread-3 Delta
PROCESS 1: Thread-2 Epsilon
PROCESS 1 done!

Is this expected? In RxJava, subscribeOn() and observeOn() will both only claim one thread for each subscription. I don't see an issue with observe_on() using multiple threads for a single subscription as long as the Observable contract still being adhered to and the emissions stay serialized. Is this the behavior you intended by design?

Ref: https://github.com/ReactiveX/RxPY/issues/430 https://github.com/ReactiveX/RxPY/issues/293#issuecomment-457502444

I'm a bit conflicted about submitting this -- on the one hand, it seems like too big a patch just before release, on the other hand it would change the API surface and therefore this is probably also something that could not land just after a major release...

But personally I think something like this would improve the API, and if I understand the earlier discussion properly this might be pretty close to consensus to how things ought to be. It seems to me the main reason this wasn't attempted earlier is that it would be quite a tedious thing to do. And so it was :-)

But I got things working, as far as I'm aware, and thus I've decided to just throw this out there, curious what people think.

The idea is basically to adapt the principal user-facing API (Observable) to promote the "callback style" of subscribing to an observable. The "observer style" is still supported, and used internally in two ways, but only as implementation details (namely, the operators and the observable sub-class hierarchy).

In other words, the observer style of subscribing only remains in places that could be adapted without touching the observable API.

For what it's worth, the changes here plenty, but they are almost entirely "mechanical":

• Get rid of polymorphism in Observable.subscribe, change the signature to only accept an observer. All invokations of subscribe that were using callbacks change to subscribe_.

• Rename subscribe to subscribe_observer and rename subscribe_ to subscribe.

• The observer argument to subscribe_observer does not have to be optional, if it is not given we may as well invoke the callback-style variant with no arguments. So, considering it compulsory, we can turn things around by moving Observable.subscribe_observer(observer) into Observer.subscribe_to(observable).

The last step means that the typing / ABC for Observable now only knows about the callback style. To use the observer style, you have to first have an instance of Observer and subscribe via that instance.

I've run the whole thing through Travis to make sure each of these steps resulted in a passing build, as evidenced by the commit history below.

The release of v3 is set to June 1, and I would really like to get it released. What are the remaining issues or blockers?

I don't think everything should be perfect for the release. After the release we should continue to improve RxPY and target a release v3.1 etc as soon as we get feedback on the v3 release.

Lets get it out!

Use curry flip on buffer_with_count Propose some test cleanup (more readable) Propose first draft for documentation on curry-flip usage Propose (unnecessary?) typing overload for map_ when no mapper is provided Hint on contributing - what checks to run

WIP branch for refactoring RxPY to use curry_flip from the Expression library. This will remove a lot of boilerplate code and make the operator code more readable (no nested functions).

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:

usernames: list[str] -> **async client** -> list[User] -> Observable[User]  -> (1)
(1) -> filters -> Observable[decision] (2)
(1) -> filters that need to query API -> **async client** -> Observable[decision] (2)
(2) -> actions -> **async client** -> exit

After searching, I learned that:

1. 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/aioreactive

2. It 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

3. 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:

1. When we use asyncio at the same time, can rxpy guarantee to execute all operators on all data till observer before exiting?

2. (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?)

3. (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.

import asyncio
import time
from collections import namedtuple

import reactivex as rx
from reactivex import operators as op
from reactivex.disposable import Disposable
from reactivex.scheduler.eventloop import AsyncIOScheduler
from reactivex.subject import Subject

start = time.time()

def ts():
    return f"{time.time() - start:.3f}"

ACTION_DURATION = 1.0

first_subject = Subject()
first_async_action = Subject()
second_subject = Subject()

Data = namedtuple("Data", ["api", "param", "future"])

async def async_calling_api(data: Data):
    """Some async processing, like sending/writing data."""
    print(f"{ts()} [A]sync action started  api:{data.api} param:{data.param}")
    # process the data with async function
    await asyncio.sleep(ACTION_DURATION)
    print(f"{ts()} [A]sync action finished api:{data.api} param:{data.param}")
    # process finished, return the response
    return f"[{data.param}]"

def serialize_map_async(mapper):
    def _serialize_map_async(source):
        def on_subscribe(observer, scheduler):
            # separate different api callings into different task queues
            queues = {k: asyncio.Queue() for k in range(0, 3)}

            async def infinite_loop(q: asyncio.Queue[Data]):
                try:
                    while True:
                        data = await q.get()
                        resp = await mapper(data)
                        observer.on_next(resp)
                        data.future.set_result(resp)
                except Exception as e:
                    observer.on_error(e)

            def on_next(data: Data):
                # take data from upstream ( calls on subject.on_next() trigger it )
                # synchronous -> asynchronous by putting elements into queue
                try:
                    queues[data.api].put_nowait(data)
                except Exception as e:
                    observer.on_error(e)

            tasks = [asyncio.create_task(infinite_loop(q)) for q in queues.values()]

            d = source.subscribe(
                on_next=on_next,
                on_error=observer.on_error,
                on_completed=observer.on_completed,
            )

            def dispose():
                d.dispose()
                [task.cancel() for task in tasks]

            return Disposable(dispose)

        return rx.create(on_subscribe)

    return _serialize_map_async


async def setup():
    loop = asyncio.get_event_loop()
    first_subject.pipe(
        serialize_map_async(async_calling_api),
        # The futures created here was not waited for, so it was not added to asyncio's chain,
        # resulting in the following gather only guaranteeing all the tasks of the first level,
        # and some second level task was canceled before it was executed.
        op.do_action(lambda x: second_subject.on_next(Data(2, x, asyncio.Future()))),
    ).subscribe(
        on_next=lambda param: print(f"{ts()} [O]bserver [1] received: {param}"),
        scheduler=AsyncIOScheduler(loop)
    )

    second_subject.pipe(serialize_map_async(async_calling_api), ).subscribe(
        on_next=lambda param: print(f"{ts()} [O]bserver [2] received: {param}"),
        scheduler=AsyncIOScheduler(loop)
    )

async def add(api: int, param: str):
    future = asyncio.Future()
    first_subject.on_next(Data(api, param, future))
    return await future

async def main():
    await setup()
    # I wonder if there is a way to write "await rx.from..."
    #
    # rx.from_iterable("a", "b").pipe(
    #  op.do(await ...)
    # )

    a = await asyncio.gather(add(0, "0a"), add(0, "0b"), add(1, "1a"), add(1, "1b"), )
    print(f"---> {a}")

asyncio.run(main())

0.003 [A]sync action started  api:0 param:0a
0.003 [A]sync action started  api:1 param:1a
1.007 [A]sync action finished api:0 param:0a
1.007 [O]bserver [1] received: [0a]
1.007 [A]sync action started  api:0 param:0b
1.007 [A]sync action finished api:1 param:1a
1.007 [O]bserver [1] received: [1a]
1.007 [A]sync action started  api:1 param:1b
1.007 [A]sync action started  api:2 param:[0a]
5.021 [A]sync action finished api:0 param:0b
2.021 [O]bserver [1] received: [0b]
2.021 [A]sync action finished api:1 param:1b
2.021 [O]bserver [1] received: [1b]
2.021 [A]sync action finished api:2 param:[0a]
2.021 [O]bserver [2] received: [[0a]]
2.021 [A]sync action started  api:2 param:[1a]
---> ['[0a]', '[0b]', '[1a]', '[1b]']

Hi, I want to pass data from a background thread into the main thread. I have the following code, but it seems that CurrentThreadScheduler() is not doing what I expect.

I'm using reactivex version 4.2.

from reactivex import operators as ops
from reactivex.scheduler import CurrentThreadScheduler, ThreadPoolScheduler
from reactivex import Subject
import threading
from threading import Thread

class MyThread(Thread):
    def __init__(self, callback):
        Thread.__init__(self)
        self.callback = callback

    def run(self):
        self.callback("hello")

my_subject = Subject()

def callback(data):
    ''' This is called in a separate thread '''
    print(f"In callback: {threading.current_thread().name}")
    my_subject.on_next(data)

if __name__ == "__main__":

    thread_to_execute_on = CurrentThreadScheduler()
    # thread_to_execute_on = ThreadPoolScheduler(max_workers=1)

    print(f"Before stream: {threading.current_thread().name}")
    
    background_thread = MyThread(callback=callback)
    
    my_subject.pipe(
        ops.observe_on(thread_to_execute_on)
    ).subscribe(
        lambda data: print(f"In subscription: {threading.current_thread().name}")
    )

    background_thread.start()
    
    input()

The output is:

Before stream: MainThread
In callback: Thread-7
In subscription: Thread-7

I've also tried using ThreadPoolScheduler and the data is correctly passed to the threadpool thread. In that scenario, output is:

Before stream: MainThread
In callback: Thread-7
In subscription: ThreadPoolExecutor-0_0

Is there something I can use to schedule work back on the main thread? for example: ops.observe_on(MainThreadScheduler()). This seems to be quite simple to do in C# and java.

To be clear, the output I am after is:

Before stream: MainThread
In callback: Thread-7
In subscription: MainThread

Describe the bug One of the first example of the Get Started documentation fails with:

Traceback (most recent call last):
  File "main.py", line 12, in <module>
    reactivex.of("Alpha", "Beta", "Gamma", "Delta", "Epsilon").pipe(
  File "./venv/lib/python3.8/site-packages/reactivex/observable/observable.py", line 237, in pipe
    return pipe_(self, *operators)
  File "./venv/lib/python3.8/site-packages/reactivex/pipe.py", line 214, in pipe
    return compose(*fns)(__value)
  File "./venv/lib/python3.8/site-packages/reactivex/pipe.py", line 87, in _compose
    return reduce(lambda obs, op: op(obs), operators, source)
  File "./venv/lib/python3.8/site-packages/reactivex/pipe.py", line 87, in <lambda>
    return reduce(lambda obs, op: op(obs), operators, source)
TypeError: 'Observable' object is not callable

The next example in the list also fails in the same way.

To Reproduce Copy the code from the Custom Operator part of the documentation into a file and run it.

The line return rx.pipe( needs to be replaced with return reactivex.pipe( to work with the above imports.

Expected behavior The code should run

Code or Screenshots

import reactivex
from reactivex import operators as ops

def length_more_than_5():
    return reactivex.pipe(
        ops.map(lambda s: len(s)),
        ops.filter(lambda i: i >= 5),
    )

reactivex.of("Alpha", "Beta", "Gamma", "Delta", "Epsilon").pipe(
    length_more_than_5()
).subscribe(lambda value: print("Received {0}".format(value)))

Additional context

• OS: Ubuntu 22.04
• RxPY: 4.0.4
• Python: 3.8.10 and 3.10.4