RxPY 3.0

[dbrattli]

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.

[dbrattli]

Should we also rename the observer interface (abc) from on_next/on_errror/on_completed to RxJS next/error/complete or to Python enhanced generator interface send/throw/close?

[MainRo]

The code on the current branch seems to completely remove the method based operators. Is the plan the following :

• RxPY 2 contains method based operators, plus the pipe operator, but method based operators should be considered deprecated.
• RxPY 3 contains only the pipe method and all operators are functions.

Meaning that v2 would allow people to migrate from method operators to pipe, and v3 has only the pipe method operator ?

I think I should find some time to work on documentation. A dedicated documentation of the RxPY APIs, more than the readme is needed. When people ask me where is the doc, I can only give them the ReactiveX link, but this far from ideal. There is an open issue on that part (#101). Do we keep it or do we close it and open a new one ? To get things step by step I can propose several PR instead of a big one (e.g one for initial sphinx structure, then one per "part"). It would allow to publish things as they are available instead of waiting - potentially - for months until everything is ready.

Concerning the observer interface (an all naming in general), I prefer to stick as much as possible to the ReactiveX documentation, and in this case the observable contract (http://reactivex.io/documentation/contract.html). So I would stick with the current naming since it is the official names, made PEP8 compliant.

[dbrattli]

RxPY 2 haven't got a stable release yet, and I haven't got time for maintaining it. The pipe operator will only work for RxPY 3 since all operators needs to be partially appliable. So the current plan is to skip RxPY 2 and move directly to RxPY 3. I'm not happy with having 2 different ways of doing things (method chaining and pipelining), and it takes much more effort to maintain. So after hours of refactoring I gave up having both. It got very complicated keeping method chaining with Subjects, GroupedObservale, ConnectableObservable etc. AFAIK RxJS removed method chaining for v6, so it should be the way to go anyways. Migration is actually pretty easy.

RxPY 2 also removed the scheduler parameter for all operators, and it was replaced with a scheduler parameter for subscribe that would set the scheduler for all operators in the chain, once and for all. But I got very unsure if this was the right thing to do, so for RxPY 3, I have reintroduced the scheduler for operators again. For UI frameworks etc, you can still set the scheduler once for subscribe.

As for the docs, I'm in favor of closing #101 as the PR did way too much, and got impossible for me to merge. README.md should be kept to a minimum (what not how) and just refer to the docs. Docstrings are now in Google style, and I would not be happy changing them to something else.

[MainRo]

RxPY 2 also removed the scheduler parameter for all operators, and it was replaced with a scheduler parameter for subscribe that would set the scheduler for all operators in the chain, once and for all. But I got very unsure if this was the right thing to do, so for RxPY 3, I have reintroduced the scheduler for operators again. For UI frameworks etc, you can still set the scheduler once for subscribe.

Yes, from the rapid look that I had on RxPY 2, the scheduler implementation was much more complex than on RxPY 1. Personally I am fine with the v1 API for schedulers. So I will be happy to get the same one on v3 :)

As for the docs, I'm in favor of closing #101 as the PR did way too much, and got impossible for me to merge. README.md should be kept to a minimum (what not how) and just refer to the docs. Docstrings are now in Google style, and I would not be happy changing them to something else.

Ok, that's great, documentation seems to be already there for all operators. BTW it seems to be duplicated both at rx/operators/init.py and rx/core/operators/*.py.
With few modifications, I have sphinx that generates a doc for the operators already present in rx/operators/init.py. If you agree I can submit a first minimal PR for this, and then start to fill more content and add references to it in the readme.
Can we add marble diagrams picked from the global reactivex documentation ?

[dbrattli]

Please feel free to make the docs the way you think would give the best result :)

Yes, docstrings are currently duplicated. The API is rx/operators/init.py and rx/init.py (for creation such as from_iterable, empty, never, throw, from_range ...). We might want to document the inner function instead in rx/core/operators since that is what you will inspect after making an operator instance such as op = operators.map(lambda x: x+10). Then op will get the inner function docstring which can be useful. Currently the inner function is called partial, but considering renaming it as _map for map etc and adding docstring there instead so that code inspection makes more sense.

[dbrattli]

I've commited an example of documenting the parially applied function for map and delay. This way we get docstrings on both the inner and outer function. Thus in delay.py the function is now called _delay and rx.operators.delay lazy imports _delay. This is better since it tells that you should not use the now non-documented rx.core.operators._delay directly. The partially applied inner function can now be called delay which makes intellisense very nice as you get the docstring for the inner function as well (you can test with timeflies_tkinter.py)

[jcafhe]

Hi,
I have a question regarding operators which may appear in both rx.init and rx.operators.init.
Let's say for amb, in the tests it's used as:

obs = rx.amb(o1, o2, o3)

as well as:

obs = o1.pipe(amb(o2, o3))

Should we assume that every operators which takes as input one or more observables will be available in rx and rx.operators (i.e. merge_all, combine_latest, ...)?

[dbrattli]

Yes, I think we should provide them both e.g as with the imperative rx.amb() and the pipable operators.amb().

[jcafhe]

Ok, that's sound good, thank you.

[dbrattli]

One really cool thing about the new syntax is that it's very easy to create new operators based on existing ones. E.g.

def _all(predicate: Predicate) -> Callable[[Observable], Observable]:

    filtering = ops.filter(lambda v: not predicate(v))
    mapping = ops.map(lambda b: not b)
    some = ops.some()

    def all(source: Observable) -> Observable:
        return source.pipe(
            filtering,
            some,
            mapping
        )
    return all

In this case we can in fact make it "point-free" by using the pipe function instead of the method, e.g:

from rx.core import pipe

def _all(predicate: Predicate) -> Callable[[Observable], Observable]:

    filtering = ops.filter(lambda v: not predicate(v))
    mapping = ops.map(lambda b: not b)
    some = ops.some()

    return pipe(
        filtering,
        some,
        mapping
    )

This is where the new syntax really shows it's true strength. This makes the new RxPY much more extensible and can offer much more code reuse than before.

[jcafhe]

That's awesome ! It's really like plug in some pieces of pipes together.

I guess it makes easier to build new custom operators too by removing the need to monkey patch at runtime Observable with the extension decorator.

[jcafhe]

If you agree, I will try to convert the examples too during the next days.
Will you merge the feature branch soon or did you PR for checking some coverage ?

By the way, good job for the massive update ! 👍

[MainRo]

If you agree, I will try to convert the examples too during the next days.

Do you mean the ones in the examples directory or in the readme ? I will work on the readme the coming days to move parts of it in the documentation. So please do not touch the readme :)

[MainRo]

forget my comment. The examples in the readme are already using pipe.

[dbrattli]

Yes, the rewrite was so much more than I expected. Very happy to get the tests working again, and super happy for all the help. Thanks for the great work!

Added the PR to check that the branch is building at Travis CI. I have now moved master into release/v2.0.x, so we are basically ready to merge v3 into master.

Todo:

• Rename mapi and filteri to map_indexed and filter_indexed to match the naming of other indexed operators
• Re-add the scheduler parameter to all time handling and scheduling operators.
• Remove result mappers

I would also like to change time handling from using millisecond (integer) to seconds (float) to match the Python stdlib time handling, but this may be too much to rewrite, and could add to the confusion. But it's a good opportunity now that the changes are so huge anyways. What do you think?

[MainRo]

Concerning time handling, it seems that each implementation uses a different unit. So it can make sense to make it pythonic on RxPY, (seconds encoded as a float).

[jcafhe]

I agree too, this is the right oportunity to switch to seconds.

Do you mean the ones in the examples directory or in the readme ? I will work on the readme the coming days to move parts of it in the documentation. So please do not touch the readme :)

Don't worry, I don't touch readme :D Buidling documentation is far outside my competence anyway. I was rather thinking of 'demos' in RxPY/examples.

[dbrattli]

Another thing to simplify time handling would be to just remove absolute scheduling and datetimes. Do anyone ever use it? With only relative time handling we could also remove timedeltas and only use seconds (float). It would make the impl. and typing (float instead of Union[int, timedelta]) much simpler.

[jcafhe]

Personally I don't use absolute scheduling but maybe we should be careful about that ? I really don't know.

I've just checked quickly the operators relying on time and it seems the switch to float would effect only the schedulers implementation ? (and the doc for these operators of course)

Will the tests continue to work since the messages are specified in 'virtual time'?

[dbrattli]

Even with absolute scheduling the timeouts will be converted to seconds by the scheduler (TimeoutScheduler/EventLoopScheduler) anyways since they use threading.Timer, so it's not really different from calculating the seconds needed in your own code and then calling Rx using only seconds.

Tests will work since they are using virtual time, but we should add a ".0" to the numbers.

[jcafhe]

Alright so this should be OK to simplify.

[dbrattli]

I think so too. I'll keep the TestScheduler working with ticks (msecs) to avoid any changes there

[jcafhe]

Rename mapi and filteri to map_indexed and filter_indexed to match the naming of other indexed operators

Maybe we should rename as well flat_mapi and partitioni to flat_map_indexed and partition_indexed? (Edit: Done)

[MainRo]

In what operators would you remove absolute times ? For me an operator like timeout must support an absolute time as input. Dealing with time is always tricky, even with python (local time vs UTC, and DST). So I would keep absolute timeouts to avoid conversions to relative times when e.g. the time information is already a timedate.
Also, I am not sure this is applicable here, but when calling timers multiple times with relative values, time drift issues occur. Using absolute times allows to avoid such a drift.

[dbrattli]

I'll leave in the datetimes. Removing them turned out to be much harder than I thought.

[jcafhe]

I'm actually removing result_mapper for generate_with_relative_time and that seems OK.
However, it exists a class with static methods as chained operators:

RxPY/rx/core/observable/staticobservable.py

Line 206 in 5c2b6b5

def generate(initial_state, condition, iterate, result_mapper) -> Observable:

I was thinking this class should be obsolete now, but not 100% sure ?

[dbrattli]

Yes, obsolete. See the top comment in the file. I will delete it.

Time rewrite to seconds is now checked in, so please do some sanity checking if you have any code. The tests are also running at seconds and works since they are running in virtual time, thus 200 becomes 200.0 seconds, so the tests don't really test fractions such as 0.5 seconds.

[jcafhe]

I've just tested with small examples (e.g. with interval, timer) and it works. Can't test right now with more complex pipelines since the rest of my code uses rx1.6.

[dbrattli]

Btw. refactored the subscribe logic and use of AutoDetachObserver/AnonymousObserver to reduce the size of the callstack on exceptions. Thus stack traces between two operators that looked like this:

File "/Users/dbrattli/Developer/GitHub/RxPY-reactivex/rx/core/operators/map.py", line 31, in on_next
    obv.on_next(result)
File "/Users/dbrattli/Developer/GitHub/RxPY-reactivex/rx/core/observerbase.py", line 19, in on_next
    self._on_next_core(value)
File "/Users/dbrattli/Developer/GitHub/RxPY-reactivex/rx/core/autodetachobserver.py", line 16, in _on_next_core
    self._observer.on_next(value)
File "/Users/dbrattli/Developer/GitHub/RxPY-reactivex/rx/core/observerbase.py", line 19, in on_next
    self._on_next_core(value)
File "/Users/dbrattli/Developer/GitHub/RxPY-reactivex/rx/core/operators/filter.py", line 34, in on_next
    observer.on_next(value)

is now reduced to:

File "/Users/dbrattli/Developer/GitHub/RxPY-reactivex/rx/core/operators/map.py", line 31, in on_next
    obv.on_next(result)
File "/Users/dbrattli/Developer/GitHub/RxPY-reactivex/rx/core/autodetachobserver.py", line 24, in on_next
    self._on_next(value)
File "/Users/dbrattli/Developer/GitHub/RxPY-reactivex/rx/core/operators/filter.py", line 34, in on_next
    observer.on_next(value)

Getting rid of the AutoDetachObserver would make the operators much more complex, so it's not worth it (imo).

[jcafhe]

Good job! I get a trace two times less than before for an example involving map with a mapper throwing an exception.

[dbrattli]

Closing this now @jcafhe and @MainRo, as the feature branch has been merged with master. You can still push to master, but please make PRs if you are unsure about the changes you are making. You don't need a PR for every doc change, but please involve any stakeholders for the code you are changing. I will add a new Issue for tracking the continued work.

[lock]

This thread has been automatically locked since there has not been any recent activity after it was closed. Please open a new issue for related bugs.