RxPY v3

[dbrattli]

The backpressure code in RxPY v3 (master branch) have been removed (to reduce to codebase) making this project very relevant, so it would be nice if it was updated for the new RxPY v3 pipe syntax. All operators in v3 are btw plain functions making it much easier to create extensions to RxPY than before.

[MichaelSchneeberger]

Yes, I can have a look at the master branch and upgrade this project to the newest version of rxpy. The changes you have made are certainly going in the right direction, I think.
The rxbackpressure library, however, has not left the experimental status yet. Although, I copied quite some ideas from Monix, it is not yet clear where it is exactly heading.

[MichaelSchneeberger]

My proposal is to introduce a from_rx operator that converts a rx observable to a rxbackpressure observable. The rxbackpressure observable then also implements a pipe operator that takes rxbackpressure operators equivalent to how it is done in RxPY v3. The to_rx operator then converts a rxbackpressure observable back to a rx observable.

Here is a short scripts that demonstrates the from_rx and pipe function.

# the following pseudo code checks that the measured values are close to the simulated ones.
#   This is useful to assert the behavior of a physical node, machine or controller. 

import rx
import rxbackpressure as rxbp

from rx import operators as rx_op
from rxbp import operators as rxbp_op

n_sample = 1000

perform_simulation = ...        # simulates mathematical model describing the physics 
                                #   of the measured object as a function of the input clock
perform_measurement = ...       # performs the measurement on the object
is_lower_sel = ...              # maps measured values to some simulation interval (part 1)
is_higher_sel = ...             # maps measured values to some simulation interval (part 2)
assert_sim_meas_diff = ...      # interpolates the simulated values, 
                                #   then calculates and asserts the difference

# sim is an rx.observable that emits the simulated values
sim = rx.range(0, n_sample).pipe(
    perform_simulation,
)

# meas is an rx.observable that emits the measured values
meas = rx.just(None).pipe(
    perform_measurement,
)

# observable that emits False for every measured values that is too far off the simulation,
#   otherwise it emits True
out = rxbp_op.from_rx(sim).pipe(
    rxbp_op.pairwise,
    rxbp_op.zip(rxbp_op.lift(meas), is_lower_sel, is_higher_sel),
    assert_sim_meas_diff,
)