concurrency.py

Parallel

Maps each element using operation on scheduler. Order is preserved.

transform(self)

Override this transform function to implement custom functionality when subclassing Pipeline

Returns:

Type	Description
a function mapping an observable to another (default = lambda x	x)

Source code in rxpipes/concurrency.py

def transform(self):
    return rx.pipe(
        super().transform(),
        ops.to_iterable(),
        ops.map(lambda xs: [y[1] for y in sorted(xs, key=lambda x: x[0])]),
    )

ScheduleEach

Maps each element using operation on scheduler. Order is not preserved.

transform(self)

Override this transform function to implement custom functionality when subclassing Pipeline

Returns:

Type	Description
a function mapping an observable to another (default = lambda x	x)

Source code in rxpipes/concurrency.py

def transform(self):
    return ops.flat_map(
        lambda x: rx.of(x).pipe(
            ops.map(self.operation), ops.subscribe_on(self.scheduler)
        )
    )