operators.md

Operators

rxRust provides a comprehensive set of operators to transform, filter, and combine streams. These operators are the building blocks of reactive programming, allowing you to compose complex asynchronous logic in a declarative and readable way.

Learning Operators: Marble Diagrams

To understand how each operator works, we highly recommend consulting the ReactiveX.io documentation. It provides interactive "Marble Diagrams" which are visual representations of how an Observable (represented by a stream of marbles) is transformed by an operator over time. rxRust operators generally follow the same behavior as their ReactiveX counterparts.

For exact Rust function signatures and implementation details, please refer to the API Documentation.

Rust-Specific Considerations

While the behavior of operators is largely consistent with ReactiveX, their usage in Rust comes with unique considerations due to Rust's ownership and borrowing rules, as well as its type system. When using operators, pay close attention to:

• Ownership and Moves: Many closures passed to operators will move captured variables. If you need to use a variable across multiple closures or after the Observable chain, you might need to clone() it or use Arc/Rc for shared ownership.
• Borrowing: Ensure that data borrowed by closures lives long enough. The compiler will guide you on this.
• Clone Trait: For operators that might re-emit values (e.g., multicast or buffer operations), the item type T often needs to implement Clone.
• Context Compatibility: Remember that Local context streams are optimized for single-threaded usage (using Rc), while Shared context streams require types to be Send + Sync (using Arc) to safely work in multi-threaded environments.

Operator Reference

Creation Operators

Operators that originate new Observables. These are typically factory methods available on Local, Shared, or your custom Context.

Operator	Description
of	Emits a single value and then completes.
from_iter	Converts an IntoIterator (e.g., Vec, Range) into an Observable.
from_future	Converts a Future into an Observable that emits its result. Variants exist for _with custom schedulers and _result for Future<Output=Result<T,E>>.
from_stream	Converts an async Stream into an Observable. Variants exist for _with custom schedulers and _result for Stream<Item=Result<T,E>>.
from_fn	Creates an Observable that emits a single value generated by a function at subscription time.
interval	Emits sequential numbers at regular time intervals. Use interval_with for custom schedulers.
timer	Emits a single item after a specified delay. Variants exist for _with custom schedulers and _at a specific instant.
defer	Creates an Observable that calls a factory function to generate a new Observable for each subscriber.
create	Creates an Observable from scratch by providing a function that defines the subscription logic with an Emitter.
empty	Emits no items and immediately completes.
never	Emits no items and never completes.
throw_err	Emits no items and immediately errors.
subject	Creates a new Subject (multicasting source, hot observable). Use subject_mut_ref for mutable reference broadcasting.
behavior_subject	Creates a new BehaviorSubject (multicasting, replays last value). Use behavior_subject_mut_ref for mutable reference broadcasting.
merge_observables	Merges multiple observables concurrently, subscribing to all at once.
concat_observables	Concatenates multiple observables sequentially, subscribing one at a time.

Transformation Operators

Operators that transform the items emitted by an Observable.

Operator	Description
map	Applies a function to each item emitted.
map_to	Maps every emission to a constant value.
filter_map	Maps and filters items in one step (using Option).
scan	Applies an accumulator function to each item, emitting each intermediate result.
reduce	Applies an accumulator function and emits only the final result.
flat_map	Projects each item to an Observable, then merges them all.
concat_map	Projects each item to an Observable, then concatenates them (preserving order).
buffer	Collects items into a Vec until a notifier emits.
buffer_count	Collects items into a Vec of a specific size.
buffer_time	Collects items into a Vec for a specific duration.
pairwise	Groups consecutive emissions into pairs (prev, current).
group_by	Divides an Observable into a set of Observables that each emit a different group of items.

Filtering Operators

Operators that selectively emit items from the source Observable.

Operator	Description
filter	Emits only items that satisfy a predicate.
take	Emits only the first n items.
take_last	Emits only the last n items.
take_while	Emits items as long as a predicate is true.
take_until	Emits items until another Observable emits.
skip	Skips the first n items.
skip_last	Skips the last n items.
skip_while	Skips items as long as a predicate is true.
skip_until	Skips items until another Observable emits.
distinct	Suppresses duplicate items.
distinct_until_changed	Suppresses consecutive duplicate items.
debounce	Emits an item only after a specific timespan has passed without another emission.
throttle	Emits the first item emitted during a time window.
sample	Emits the most recent item when another Observable emits.
last	Emits only the last item.
contains	Emits true if the Observable emits a specific item.

Combination Operators

Operators that work with multiple source Observables to create a single Observable.

Operator	Description
merge	Interleaves items from multiple Observables.
zip	Combines items from multiple Observables pairwise.
combine_latest	Combines the latest item from each Observable whenever any emits.
with_latest_from	Merges the current item with the latest item from another Observable.
start_with	Emits a sequence of items before beginning to emit the items from the source.
merge_all	Flattens a Higher-Order Observable by merging inner Observables.
concat_all	Flattens a Higher-Order Observable by concatenating inner Observables sequentially.

Utility Operators

Operators for observing, timing, and error handling.

Operator	Description
tap	Performs a side effect for every emission (next, error, complete).
delay	Shifts the emissions forward in time by a specified delay.
delay_subscription	Delays the moment of subscription.
observe_on	Specifies the Scheduler on which an observer will observe this Observable.
subscribe_on	Specifies the Scheduler on which the subscription will happen.
finalize	Registers a callback to be called when the Observable terminates.
retry	Resubscribes to the source Observable if it signals an error.
map_err	Transforms the error type.
on_error	Performs a side effect if an error occurs.
on_complete	Performs a side effect if the Observable completes.