Trait ObservableExt 

pub trait ObservableExt<'or, T, E>: Observable<'or, T, E> + Sized {
    // Provided methods
    fn all<F>(self, callback: F) -> All<T, Self, F>
       where F: FnMut(T) -> bool { ... }
    fn amb_with<OE1>(self, other: OE1) -> Amb<[EitherObservable<Self, OE1>; 2]>
       where OE1: Observable<'or, T, E> { ... }
    fn average(self) -> Average<T, Self> { ... }
    fn buffer<OE1>(self, boundary: OE1) -> Buffer<Self, OE1>
       where OE1: Observable<'or, (), E> { ... }
    fn buffer_with_count(self, count: NonZeroUsize) -> BufferWithCount<Self> { ... }
    fn buffer_with_time<S>(
        self,
        time_span: Duration,
        scheduler: S,
        delay: Option<Duration>,
    ) -> BufferWithTime<'or, Self, S> { ... }
    fn buffer_with_time_or_count<S>(
        self,
        count: NonZeroUsize,
        time_span: Duration,
        scheduler: S,
        delay: Option<Duration>,
    ) -> BufferWithTimeOrCount<Self, S> { ... }
    fn catch<E1, OE1, F>(self, callback: F) -> Catch<E, Self, F>
       where OE1: Observable<'or, T, E1>,
             F: FnOnce(E) -> OE1 { ... }
    fn collect<C>(self) -> Collect<C, T, Self>
       where C: Default + Extend<T> { ... }
    fn combine_latest<T1, OE2>(
        self,
        another_source: OE2,
    ) -> CombineLatest<Self, OE2>
       where OE2: Observable<'or, T1, E> { ... }
    fn concat_all<T1>(self) -> ConcatAll<Self, T>
       where T: Observable<'or, T1, E> { ... }
    fn concat_map<T1, OE1, F>(self, callback: F) -> ConcatMap<T, Self, OE1, F>
       where OE1: Observable<'or, T1, E>,
             F: FnMut(T) -> OE1 { ... }
    fn concat_with<OE2>(self, source_2: OE2) -> Concat<Self, OE2>
       where OE2: Observable<'or, T, E> { ... }
    fn contains(self, item: T) -> Contains<T, Self> { ... }
    fn count(self) -> Count<T, Self> { ... }
    fn debounce<S>(
        self,
        time_span: Duration,
        scheduler: S,
    ) -> Debounce<'or, Self, S> { ... }
    fn debug<C, F>(self, context: C, callback: F) -> Debug<Self, C, F>
       where F: Fn(C, DebugEvent<'_, T, E>) { ... }
    fn debug_default_print<L>(
        self,
        label: L,
    ) -> Debug<Self, L, DefaultPrintType<L, T, E>>
       where L: Display,
             T: Debug,
             E: Debug { ... }
    fn default_if_empty(self, default_value: T) -> DefaultIfEmpty<T, Self> { ... }
    fn delay<S>(self, delay: Duration, scheduler: S) -> Delay<'or, Self, S> { ... }
    fn dematerialize(self) -> Dematerialize<Self> { ... }
    fn distinct(self) -> Distinct<Self, fn(&T) -> T>
       where T: Clone { ... }
    fn distinct_until_changed(self) -> DistinctUntilChanged<Self, fn(&T) -> T>
       where T: Clone { ... }
    fn distinct_until_changed_with_key_selector<F, K>(
        self,
        key_selector: F,
    ) -> DistinctUntilChanged<Self, F>
       where F: FnMut(&T) -> K { ... }
    fn distinct_with_key_selector<F, K>(
        self,
        key_selector: F,
    ) -> Distinct<Self, F>
       where F: FnMut(&T) -> K { ... }
    fn do_after_disposal<F>(self, callback: F) -> DoAfterDisposal<Self, F>
       where F: FnOnce() { ... }
    fn do_after_next<F>(self, callback: F) -> DoAfterNext<Self, F>
       where F: FnMut(T) { ... }
    fn do_after_subscription<F>(
        self,
        callback: F,
    ) -> DoAfterSubscription<Self, F>
       where F: FnOnce() { ... }
    fn do_after_termination<F>(self, callback: F) -> DoAfterTermination<Self, F>
       where F: FnOnce(Termination<E>) { ... }
    fn do_before_disposal<F>(self, callback: F) -> DoBeforeDisposal<Self, F>
       where F: FnOnce() { ... }
    fn do_before_next<F>(self, callback: F) -> DoBeforeNext<Self, F>
       where F: FnMut(&T) { ... }
    fn do_before_subscription<F>(
        self,
        callback: F,
    ) -> DoBeforeSubscription<Self, F>
       where F: FnOnce() { ... }
    fn do_before_termination<F>(
        self,
        callback: F,
    ) -> DoBeforeTermination<Self, F>
       where F: FnOnce(&Termination<E>) { ... }
    fn element_at(self, index: usize) -> ElementAt<Self> { ... }
    fn filter<F>(self, callback: F) -> Filter<Self, F>
       where F: FnMut(&T) -> bool { ... }
    fn first(self) -> First<Self> { ... }
    fn flat_map<T1, OE1, F>(self, callback: F) -> FlatMap<T, Self, OE1, F>
       where OE1: Observable<'or, T1, E>,
             F: FnMut(T) -> OE1 { ... }
    fn group_by<F, K>(self, key_selector: F) -> GroupBy<Self, F, K>
       where F: FnMut(&T) -> K { ... }
    fn hook_on_next<F>(self, callback: F) -> HookOnNext<Self, F>
       where F: FnMut(&mut dyn Observer<T, E>, T) -> Flow { ... }
    fn hook_on_subscription<D, F>(
        self,
        callback: F,
    ) -> HookOnSubscription<Self, F>
       where D: Disposable,
             F: FnOnce(Self, BoxedObserver<'or, T, E>) -> Subscription<D> { ... }
    fn hook_on_termination<F>(self, callback: F) -> HookOnTermination<Self, F>
       where F: FnOnce(BoxedObserver<'or, T, E>, Termination<E>) { ... }
    fn ignore_elements(self) -> IgnoreElements<Self> { ... }
    fn into_boxed<'sub, 'oe>(self) -> BoxedObservable<'or, 'sub, 'oe, T, E>
       where T: 'or,
             E: 'or,
             Self: MaybeSend + 'oe,
             Self::D: MaybeSend + 'sub { ... }
    fn into_cloneable_boxed<'sub, 'oe>(
        self,
    ) -> CloneableBoxedObservable<'or, 'sub, 'oe, T, E>
       where T: 'or,
             E: 'or,
             Self: MaybeSend + MaybeSync + Clone + 'oe,
             Self::D: MaybeSend + 'sub { ... }
    fn into_future(self) -> ObservableFuture<'or, T, Self> 
       where Self: Observable<'or, T, Infallible> { ... }
    fn into_stream(self) -> ObservableStream<'or, T, Self>
       where Self: Observable<'or, T, Infallible> { ... }
    fn into_stream_with<B>(self, buffer: B) -> ObservableStream<'or, T, Self, B>
       where Self: Observable<'or, T, Infallible>,
             B: StreamBuffer<T> { ... }
    fn into_try_future(self) -> ObservableTryFuture<'or, T, E, Self>  { ... }
    fn into_try_stream(self) -> ObservableTryStream<'or, T, E, Self> { ... }
    fn into_try_stream_with<B>(
        self,
        buffer: B,
    ) -> ObservableTryStream<'or, T, E, Self, B>
       where B: StreamBuffer<T> { ... }
    fn last(self) -> Last<Self> { ... }
    fn map<T1, F>(self, callback: F) -> Map<T, Self, F>
       where F: FnMut(T) -> T1 { ... }
    fn map_err<E1, F>(self, callback: F) -> MapErr<E, Self, F>
       where F: FnOnce(E) -> E1 { ... }
    fn materialize(self) -> Materialize<Self> { ... }
    fn max(self) -> Max<Self> { ... }
    fn merge_all<T1>(self) -> MergeAll<Self, T>
       where T: Observable<'or, T1, E> { ... }
    fn merge_with<OE2>(self, source_2: OE2) -> Merge<Self, OE2>
       where OE2: Observable<'or, T, E> { ... }
    fn min(self) -> Min<Self> { ... }
    fn multicast<S, F>(self, subject_maker: F) -> ConnectableController<Self, S>
       where F: FnOnce() -> S { ... }
    fn observe_on<S>(self, scheduler: S) -> ObserveOn<'or, Self, S> { ... }
    fn publish(self) -> ConnectableController<Self, PublishSubject<'or, T, E>> { ... }
    fn publish_last(
        self,
    ) -> ConnectableController<Self, AsyncSubject<'or, T, E>> { ... }
    fn reduce<T0, F>(
        self,
        initial_value: T0,
        callback: F,
    ) -> Reduce<T0, T, Self, F>
       where F: FnMut(T0, T) -> T0 { ... }
    fn replay(
        self,
        buffer_size: Option<usize>,
    ) -> ConnectableController<Self, ReplaySubject<'or, T, E>> { ... }
    fn retry<OE1, F>(self, callback: F) -> Retry<Self, F>
       where OE1: Observable<'or, T, E>,
             F: FnMut(E) -> RetryAction<E, OE1> { ... }
    fn sample<OE1>(self, sampler: OE1) -> Sample<Self, OE1>
       where OE1: Observable<'or, (), E> { ... }
    fn scan<T0, F>(self, initial_value: T0, callback: F) -> Scan<T0, T, Self, F>
       where F: FnMut(T0, T) -> T0 { ... }
    fn sequence_equal<OE2>(
        self,
        another_source: OE2,
    ) -> SequenceEqual<T, Self, OE2>
       where OE2: Observable<'or, T, E> { ... }
    fn share(self) -> RefCount<'or, T, E, Self, PublishSubject<'or, T, E>> { ... }
    fn share_last(self) -> RefCount<'or, T, E, Self, AsyncSubject<'or, T, E>> { ... }
    fn share_replay(
        self,
        buffer_size: Option<usize>,
    ) -> RefCount<'or, T, E, Self, ReplaySubject<'or, T, E>> { ... }
    fn skip(self, count: usize) -> Skip<Self> { ... }
    fn skip_last(self, count: usize) -> SkipLast<Self> { ... }
    fn skip_until<OE1>(self, start: OE1) -> SkipUntil<Self, OE1>
       where OE1: Observable<'or, (), E> { ... }
    fn skip_while<F>(self, callback: F) -> SkipWhile<Self, F>
       where F: FnMut(&T) -> bool { ... }
    fn start_with<I>(self, values: I) -> StartWith<Self, I>
       where I: IntoIterator<Item = T> { ... }
    fn subscribe_on<S>(self, scheduler: S) -> SubscribeOn<'or, Self, S> { ... }
    fn subscribe_with_callback<FN, FT, R>(
        self,
        on_next: FN,
        on_termination: FT,
    ) -> Subscription<Self::D>
       where FN: FnMut(T) -> R + MaybeSend + 'or,
             R: IntoFlow,
             FT: FnOnce(Termination<E>) + MaybeSend + 'or { ... }
    fn sum(self) -> Sum<Self> { ... }
    fn switch<T1>(self) -> Switch<Self, T>
       where T: Observable<'or, T1, E> { ... }
    fn switch_map<T1, OE1, F>(self, callback: F) -> SwitchMap<T, Self, OE1, F>
       where OE1: Observable<'or, T1, E>,
             F: FnMut(T) -> OE1 { ... }
    fn take(self, count: usize) -> Take<Self> { ... }
    fn take_last(self, count: usize) -> TakeLast<Self> { ... }
    fn take_until<OE1>(self, stop: OE1) -> TakeUntil<Self, OE1>
       where OE1: Observable<'or, (), E> { ... }
    fn take_while<F>(self, callback: F) -> TakeWhile<Self, F>
       where F: FnMut(&T) -> bool { ... }
    fn throttle(self, time_span: Duration) -> Throttle<Self> { ... }
    fn time_interval(self) -> TimeInterval<Self> { ... }
    fn timeout<S>(
        self,
        duration: Duration,
        scheduler: S,
    ) -> Timeout<'or, Self, S> { ... }
    fn timestamp(self) -> Timestamp<Self> { ... }
    fn to_vec(self) -> Collect<Vec<T>, T, Self> { ... }
    fn window<OE1>(self, boundary: OE1) -> Window<Self, OE1>
       where OE1: Observable<'or, (), E> { ... }
    fn window_with_count(self, count: NonZeroUsize) -> WindowWithCount<Self> { ... }
    fn with_error_type<E1>(self) -> WithErrorType<E1, Self> { ... }
    fn with_item_type<T1>(self) -> WithItemType<T1, Self> { ... }
    fn zip<T1, OE2>(self, another_source: OE2) -> Zip<Self, OE2>
       where OE2: Observable<'or, T1, E> { ... }
}

Extension trait that exposes the full suite of RxRust operators on any type that implements Observable. Each method forwards to the corresponding operator constructor, allowing a fluent, ergonomic style when composing observable pipelines.

Provided Methods

fn all<F>(self, callback: F) -> All<T, Self, F>

where F: FnMut(T) -> bool,

Emits a single bool indicating whether every item satisfies the provided predicate.

fn amb_with<OE1>(self, other: OE1) -> Amb<[EitherObservable<Self, OE1>; 2]>

where OE1: Observable<'or, T, E>,

Competes two observables and mirrors whichever one produces an item or error first.

fn average(self) -> Average<T, Self>

Calculates the arithmetic mean of all numeric items emitted by the source.

fn buffer<OE1>(self, boundary: OE1) -> Buffer<Self, OE1>

where OE1: Observable<'or, (), E>,

Collects the items emitted by the source into buffers delimited by another observable. Terminating the boundary terminates the outer observable: completing it emits the pending buffer when non-empty and then completes, while an error from it discards the pending buffer and errors. Unlike window, a completed boundary does not leave the current buffer open.

fn buffer_with_count(self, count: NonZeroUsize) -> BufferWithCount<Self>

Collects items into fixed-size buffers and emits each buffer as soon as it fills up.

fn buffer_with_time<S>( self, time_span: Duration, scheduler: S, delay: Option<Duration>, ) -> BufferWithTime<'or, Self, S>

Collects items into time-based buffers driven by the provided scheduler.

fn buffer_with_time_or_count<S>( self, count: NonZeroUsize, time_span: Duration, scheduler: S, delay: Option<Duration>, ) -> BufferWithTimeOrCount<Self, S>

Collects items into buffers using both size and time boundaries whichever occurs first.

fn catch<E1, OE1, F>(self, callback: F) -> Catch<E, Self, F>

where OE1: Observable<'or, T, E1>, F: FnOnce(E) -> OE1,

Recovers from errors by switching to another observable yielded by the callback.

fn collect<C>(self) -> Collect<C, T, Self>

where C: Default + Extend<T>,

Gathers all the items into a collection built with Default and Extend, and emits it when the source completes. See to_vec for the Vec<T> case.

fn combine_latest<T1, OE2>( self, another_source: OE2, ) -> CombineLatest<Self, OE2>

where OE2: Observable<'or, T1, E>,

Combines the latest values from both observables whenever either produces a new item.

fn concat_all<T1>(self) -> ConcatAll<Self, T>

where T: Observable<'or, T1, E>,

Flattens an observable-of-observables by concatenating each inner observable sequentially.

fn concat_map<T1, OE1, F>(self, callback: F) -> ConcatMap<T, Self, OE1, F>

where OE1: Observable<'or, T1, E>, F: FnMut(T) -> OE1,

Maps each item to an observable and concatenates the resulting inner sequences.

fn concat_with<OE2>(self, source_2: OE2) -> Concat<Self, OE2>

where OE2: Observable<'or, T, E>,

Concatenates the source with another observable, waiting for the first to complete.

fn contains(self, item: T) -> Contains<T, Self>

Emits true if the sequence contains the provided item, false otherwise.

fn count(self) -> Count<T, Self>

Counts the number of items emitted and emits that count as a single value.

fn debounce<S>( self, time_span: Duration, scheduler: S, ) -> Debounce<'or, Self, S>

Emits an item from the source Observable only after a particular time span has passed without another source emission.

fn debug<C, F>(self, context: C, callback: F) -> Debug<Self, C, F>

where F: Fn(C, DebugEvent<'_, T, E>),

Attaches a label to the stream and logs lifecycle events for debugging purposes using the provided callback.

fn debug_default_print<L>( self, label: L, ) -> Debug<Self, L, DefaultPrintType<L, T, E>>

where L: Display, T: Debug, E: Debug,

Attaches a label to the stream and logs lifecycle events for debugging purposes using the default print.

fn default_if_empty(self, default_value: T) -> DefaultIfEmpty<T, Self>

Emits a default value if the source completes without emitting any items.

fn delay<S>(self, delay: Duration, scheduler: S) -> Delay<'or, Self, S>

Offsets the emission of items by the specified duration using the given scheduler.

fn dematerialize(self) -> Dematerialize<Self>

Converts a stream of notifications back into a normal observable sequence.

fn distinct(self) -> Distinct<Self, fn(&T) -> T>

where T: Clone,

Filters out duplicate items, keeping only the first occurrence of each value.

fn distinct_until_changed(self) -> DistinctUntilChanged<Self, fn(&T) -> T>

where T: Clone,

Suppresses consecutive duplicate items, comparing the values directly.

fn distinct_until_changed_with_key_selector<F, K>( self, key_selector: F, ) -> DistinctUntilChanged<Self, F>

where F: FnMut(&T) -> K,

Suppresses consecutive duplicate items using a custom key selector.

fn distinct_with_key_selector<F, K>(self, key_selector: F) -> Distinct<Self, F>

where F: FnMut(&T) -> K,

Filters out duplicates based on a key selector, keeping only unique keys.

fn do_after_disposal<F>(self, callback: F) -> DoAfterDisposal<Self, F>

where F: FnOnce(),

Invokes a callback after the downstream subscription is disposed.

fn do_after_next<F>(self, callback: F) -> DoAfterNext<Self, F>

where F: FnMut(T),

Invokes a callback after each item is forwarded downstream.

fn do_after_subscription<F>(self, callback: F) -> DoAfterSubscription<Self, F>

where F: FnOnce(),

Invokes a callback after the observer subscribes to the source.

fn do_after_termination<F>(self, callback: F) -> DoAfterTermination<Self, F>

where F: FnOnce(Termination<E>),

Invokes a callback after the source terminates, regardless of completion or error.

fn do_before_disposal<F>(self, callback: F) -> DoBeforeDisposal<Self, F>

where F: FnOnce(),

Invokes a callback right before the downstream subscription is disposed.

fn do_before_next<F>(self, callback: F) -> DoBeforeNext<Self, F>

where F: FnMut(&T),

Invokes a callback with a reference to each item before it is sent downstream.

fn do_before_subscription<F>(self, callback: F) -> DoBeforeSubscription<Self, F>

where F: FnOnce(),

Invokes a callback just before the observer subscribes to the source.

fn do_before_termination<F>(self, callback: F) -> DoBeforeTermination<Self, F>

where F: FnOnce(&Termination<E>),

Invokes a callback before the stream terminates, receiving the termination reason.

fn element_at(self, index: usize) -> ElementAt<Self>

Emits only the item at the given zero-based index and then completes.

fn filter<F>(self, callback: F) -> Filter<Self, F>

where F: FnMut(&T) -> bool,

Filters items using a predicate, forwarding only values that return true.

fn first(self) -> First<Self>

Emits only the first item from the source, then completes.

fn flat_map<T1, OE1, F>(self, callback: F) -> FlatMap<T, Self, OE1, F>

where OE1: Observable<'or, T1, E>, F: FnMut(T) -> OE1,

Maps each item to an observable and merges the resulting inner sequences concurrently.

fn group_by<F, K>(self, key_selector: F) -> GroupBy<Self, F, K>

where F: FnMut(&T) -> K,

Groups items by key into multiple observable sequences.

fn hook_on_next<F>(self, callback: F) -> HookOnNext<Self, F>

where F: FnMut(&mut dyn Observer<T, E>, T) -> Flow,

Hooks into the emission of items, allowing mutation of the downstream observer.

The callback returns the Flow the operator answers, which is normally the one the downstream observer it was handed answered.

fn hook_on_subscription<D, F>(self, callback: F) -> HookOnSubscription<Self, F>

where D: Disposable, F: FnOnce(Self, BoxedObserver<'or, T, E>) -> Subscription<D>,

Hooks into subscription, letting you override how the source subscribes observers.

fn hook_on_termination<F>(self, callback: F) -> HookOnTermination<Self, F>

where F: FnOnce(BoxedObserver<'or, T, E>, Termination<E>),

Hooks into termination, providing access to the observer and termination payload.

fn ignore_elements(self) -> IgnoreElements<Self>

Ignores all items from the source, only relaying termination events.

fn into_boxed<'sub, 'oe>(self) -> BoxedObservable<'or, 'sub, 'oe, T, E>

where T: 'or, E: 'or, Self: MaybeSend + 'oe, Self::D: MaybeSend + 'sub,

Boxes the observable, erasing its concrete type while preserving lifetime bounds.

fn into_cloneable_boxed<'sub, 'oe>( self, ) -> CloneableBoxedObservable<'or, 'sub, 'oe, T, E>

where T: 'or, E: 'or, Self: MaybeSend + MaybeSync + Clone + 'oe, Self::D: MaybeSend + 'sub,

Boxes the observable and makes it cloneable, erasing its concrete type while preserving lifetime bounds.

fn into_future(self) -> ObservableFuture<'or, T, Self>

where Self: Observable<'or, T, Infallible>,

Converts the observable into a future of its first item: Some(item), or None when the source completes without one. The source is stopped as soon as the item is in.

This is only for a source that cannot fail; a fallible one goes through into_try_future. An operator that picks another item, such as last, or one that always emits, such as collect, goes in front of it.

fn into_stream(self) -> ObservableStream<'or, T, Self>

where Self: Observable<'or, T, Infallible>,

Converts the observable into an async stream.

A Stream has no error channel, so this is only for a source that cannot fail; a fallible one goes through into_try_stream.

The items that arrive between two polls are all kept, so a source faster than the consumer grows the buffer without bound; into_stream_with takes a buffer that bounds it.

fn into_stream_with<B>(self, buffer: B) -> ObservableStream<'or, T, Self, B>

where Self: Observable<'or, T, Infallible>, B: StreamBuffer<T>,

Converts the observable into an async stream that keeps the items arriving between two polls in buffer, which decides what a source faster than the consumer costs.

Latest keeps only the newest item, Bounded a fixed number of them and Unbounded — what into_stream uses — everything; a StreamBuffer of your own can fold them instead. Whatever the buffer, the source is never slowed down: a Stream only pulls from the buffer, not from the source.

Examples

use futures::{FutureExt, StreamExt};
use rx_rust::{
    observable::ObservableExt, observer::Observer,
    operators::others::observable_try_stream::Latest,
    subject::publish_subject::PublishSubject,
};
use std::convert::Infallible;

let mut subject = PublishSubject::<_, Infallible>::new();
let mut stream = subject.clone().into_stream_with(Latest::new());
assert_eq!(stream.next().now_or_never(), None); // subscribes

subject.on_next(1);
subject.on_next(2);
subject.on_next(3);
assert_eq!(stream.next().now_or_never(), Some(Some(3)));
assert_eq!(stream.next().now_or_never(), None);

fn into_try_future(self) -> ObservableTryFuture<'or, T, E, Self>

Converts the observable into a future of its first item: Ok(Some(item)), Ok(None) when the source completes without one, or Err(error) when it fails first. The source is stopped as soon as the item is in.

The output is the Maybe of ReactiveX; an operator that always emits, such as collect, in front of it makes it a Single, and last picks the last item instead of the first.

fn into_try_stream(self) -> ObservableTryStream<'or, T, E, Self>

Converts the observable into an async stream of Results: each item as Ok, and an error as the last item, Err, before the stream ends.

The items that arrive between two polls are all kept, so a source faster than the consumer grows the buffer without bound; into_try_stream_with takes a buffer that bounds it.

fn into_try_stream_with<B>( self, buffer: B, ) -> ObservableTryStream<'or, T, E, Self, B>

where B: StreamBuffer<T>,

Converts the observable into an async stream of Results that keeps the items arriving between two polls in buffer. This is into_stream_with for a source that can fail; see there for the buffers.

fn last(self) -> Last<Self>

Emits only the final item produced by the source before completion.

fn map<T1, F>(self, callback: F) -> Map<T, Self, F>

where F: FnMut(T) -> T1,

Transforms each item by applying a user-supplied mapping function.

fn map_err<E1, F>(self, callback: F) -> MapErr<E, Self, F>

where F: FnOnce(E) -> E1,

Transforms an error emitted by the source while leaving its items unchanged.

fn materialize(self) -> Materialize<Self>

Wraps each item into a notification, turning the stream into explicit events.

fn max(self) -> Max<Self>

Emits the maximum item produced by the source according to the natural order.

fn merge_all<T1>(self) -> MergeAll<Self, T>

where T: Observable<'or, T1, E>,

Merges an observable-of-observables by interleaving items from inner streams.

fn merge_with<OE2>(self, source_2: OE2) -> Merge<Self, OE2>

where OE2: Observable<'or, T, E>,

Merges the source with another observable, interleaving both streams concurrently.

fn min(self) -> Min<Self>

Emits the minimum item produced by the source according to the natural order.

fn multicast<S, F>(self, subject_maker: F) -> ConnectableController<Self, S>

where F: FnOnce() -> S,

Converts the source into a connectable observable using a subject factory.

fn observe_on<S>(self, scheduler: S) -> ObserveOn<'or, Self, S>

Schedules downstream observation on the provided scheduler.

fn publish(self) -> ConnectableController<Self, PublishSubject<'or, T, E>>

Multicasts the source using a PublishSubject.

fn publish_last(self) -> ConnectableController<Self, AsyncSubject<'or, T, E>>

Multicasts the source using an AsyncSubject, emitting only the last value.

fn reduce<T0, F>(self, initial_value: T0, callback: F) -> Reduce<T0, T, Self, F>

where F: FnMut(T0, T) -> T0,

Aggregates the sequence using an initial seed and an accumulator function.

fn replay( self, buffer_size: Option<usize>, ) -> ConnectableController<Self, ReplaySubject<'or, T, E>>

Multicasts the source using a ReplaySubject configured with the given buffer size.

fn retry<OE1, F>(self, callback: F) -> Retry<Self, F>

where OE1: Observable<'or, T, E>, F: FnMut(E) -> RetryAction<E, OE1>,

Re-subscribes to the source based on the retry strategy returned by the callback.

fn sample<OE1>(self, sampler: OE1) -> Sample<Self, OE1>

where OE1: Observable<'or, (), E>,

Samples the source whenever the sampler observable emits an event.

fn scan<T0, F>(self, initial_value: T0, callback: F) -> Scan<T0, T, Self, F>

where F: FnMut(T0, T) -> T0,

Accumulates values over time, emitting each intermediate result.

fn sequence_equal<OE2>(self, another_source: OE2) -> SequenceEqual<T, Self, OE2>

where OE2: Observable<'or, T, E>,

Compares two sequences element by element for equality.

fn share(self) -> RefCount<'or, T, E, Self, PublishSubject<'or, T, E>>

Shares a single subscription to the source using PublishSubject semantics.

fn share_last(self) -> RefCount<'or, T, E, Self, AsyncSubject<'or, T, E>>

Shares a single subscription, replaying only the last item to new subscribers.

fn share_replay( self, buffer_size: Option<usize>, ) -> RefCount<'or, T, E, Self, ReplaySubject<'or, T, E>>

Shares a single subscription while replaying a bounded history to future subscribers.

fn skip(self, count: usize) -> Skip<Self>

Skips the first count items before emitting the remainder of the sequence.

fn skip_last(self, count: usize) -> SkipLast<Self>

Skips the last count items emitted by the source.

fn skip_until<OE1>(self, start: OE1) -> SkipUntil<Self, OE1>

where OE1: Observable<'or, (), E>,

Ignores items from the source until the notifier observable fires.

fn skip_while<F>(self, callback: F) -> SkipWhile<Self, F>

where F: FnMut(&T) -> bool,

Skips items while the predicate returns true, then emits the remaining items.

fn start_with<I>(self, values: I) -> StartWith<Self, I>

where I: IntoIterator<Item = T>,

Pre-pends the provided values before the source starts emitting.

fn subscribe_on<S>(self, scheduler: S) -> SubscribeOn<'or, Self, S>

Subscribes to the source on the provided scheduler.

fn subscribe_with_callback<FN, FT, R>( self, on_next: FN, on_termination: FT, ) -> Subscription<Self::D>

where FN: FnMut(T) -> R + MaybeSend + 'or, R: IntoFlow, FT: FnOnce(Termination<E>) + MaybeSend + 'or,

Convenience helper for subscribing with plain callbacks instead of a full observer.

on_next may return nothing, which keeps the source going, or a Flow, which lets it end its own stream with Flow::Stop: the source then stops pushing — a synchronous one stops iterating — and drops the callbacks without calling on_termination.

fn sum(self) -> Sum<Self>

Sums all numeric items and emits the accumulated total.

fn switch<T1>(self) -> Switch<Self, T>

where T: Observable<'or, T1, E>,

Switches to the most recent inner observable emitted by the source.

fn switch_map<T1, OE1, F>(self, callback: F) -> SwitchMap<T, Self, OE1, F>

where OE1: Observable<'or, T1, E>, F: FnMut(T) -> OE1,

Maps each item to an observable and switches to the latest inner sequence.

fn take(self, count: usize) -> Take<Self>

Emits only the first count items from the source before completing.

fn take_last(self, count: usize) -> TakeLast<Self>

Emits only the last count items produced by the source.

fn take_until<OE1>(self, stop: OE1) -> TakeUntil<Self, OE1>

where OE1: Observable<'or, (), E>,

Relays items until the notifier observable emits, then completes.

fn take_while<F>(self, callback: F) -> TakeWhile<Self, F>

where F: FnMut(&T) -> bool,

Emits items while the predicate holds true, then completes.

fn throttle(self, time_span: Duration) -> Throttle<Self>

Throttles emissions to at most one item per specified timespan.

Leading-edge and scheduler-free: the cooldown is decided by comparing item arrival times, so no timer is spawned.

fn time_interval(self) -> TimeInterval<Self>

Emits elapsed time between consecutive items as they flow through the stream.

fn timeout<S>(self, duration: Duration, scheduler: S) -> Timeout<'or, Self, S>

Errors if the next item does not arrive within the specified duration.

fn timestamp(self) -> Timestamp<Self>

Annotates each item with the current timestamp when it is emitted.

fn to_vec(self) -> Collect<Vec<T>, T, Self>

Gathers all the items into a Vec and emits it when the source completes. This is collect specialized to Vec<T>, which is the shape that window composes with: source.window(boundary).concat_map(|window| window.to_vec()).

fn window<OE1>(self, boundary: OE1) -> Window<Self, OE1>

where OE1: Observable<'or, (), E>,

Collects items into windows that are opened and closed by another observable. Completing the boundary stops future window rotation without terminating the source. An error from the boundary terminates the current window and the outer observable.

fn window_with_count(self, count: NonZeroUsize) -> WindowWithCount<Self>

Collects items into windows containing a fixed number of elements.

fn with_error_type<E1>(self) -> WithErrorType<E1, Self>

Gives an Observable whose error type is Infallible a concrete error type.

fn with_item_type<T1>(self) -> WithItemType<T1, Self>

Gives an Observable whose item type is Infallible a concrete item type.

fn zip<T1, OE2>(self, another_source: OE2) -> Zip<Self, OE2>

where OE2: Observable<'or, T1, E>,

Pairs items from both observables by index and emits tuples of corresponding values.

Dyn Compatibility

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors

impl<'or, T, E, OE> ObservableExt<'or, T, E> for OE

where OE: Observable<'or, T, E>,