rx_rust/observable/
mod.rs

1pub mod boxed_observable;
2pub mod cloneable_boxed_observable;
3pub mod either_observable;
4
5#[cfg(feature = "futures")]
6use crate::operators::others::{
7    observable_stream::ObservableStream,
8    observable_try_stream::{ObservableTryStream, StreamBuffer},
9};
10use crate::{
11    disposable::{Disposable, bound_drop_disposal::BoundDropDisposal},
12    observable::{
13        boxed_observable::BoxedObservable, cloneable_boxed_observable::CloneableBoxedObservable,
14        either_observable::EitherObservable,
15    },
16    observer::{
17        Flow, Observer, Termination,
18        boxed_observer::BoxedObserver,
19        callback_observer::{CallbackObserver, IntoFlow},
20    },
21    operators::{
22        combining::{
23            combine_latest::CombineLatest, concat::Concat, concat_all::ConcatAll, merge::Merge,
24            merge_all::MergeAll, start_with::StartWith, switch::Switch, zip::Zip,
25        },
26        conditional_boolean::{
27            all::All, amb::Amb, contains::Contains, default_if_empty::DefaultIfEmpty,
28            sequence_equal::SequenceEqual, skip_until::SkipUntil, skip_while::SkipWhile,
29            take_until::TakeUntil, take_while::TakeWhile,
30        },
31        connectable::{connectable_controller::ConnectableController, ref_count::RefCount},
32        error_handling::{
33            catch::Catch,
34            map_err::MapErr,
35            retry::{Retry, RetryAction},
36        },
37        filtering::{
38            debounce::Debounce, distinct::Distinct, distinct_until_changed::DistinctUntilChanged,
39            element_at::ElementAt, filter::Filter, first::First, ignore_elements::IgnoreElements,
40            last::Last, sample::Sample, skip::Skip, skip_last::SkipLast, take::Take,
41            take_last::TakeLast, throttle::Throttle,
42        },
43        mathematical_aggregate::{
44            average::Average, collect::Collect, count::Count, max::Max, min::Min, reduce::Reduce,
45            sum::Sum,
46        },
47        others::{
48            debug::{Debug, DebugEvent, DefaultPrintType},
49            hook_on_next::HookOnNext,
50            hook_on_subscription::HookOnSubscription,
51            hook_on_termination::HookOnTermination,
52            observable_future::ObservableFuture,
53            observable_try_future::ObservableTryFuture,
54            with_error_type::WithErrorType,
55            with_item_type::WithItemType,
56        },
57        transforming::{
58            buffer::Buffer, buffer_with_count::BufferWithCount, buffer_with_time::BufferWithTime,
59            buffer_with_time_or_count::BufferWithTimeOrCount, concat_map::ConcatMap,
60            flat_map::FlatMap, group_by::GroupBy, map::Map, scan::Scan, switch_map::SwitchMap,
61            window::Window, window_with_count::WindowWithCount,
62        },
63        utility::{
64            delay::Delay, dematerialize::Dematerialize, do_after_disposal::DoAfterDisposal,
65            do_after_next::DoAfterNext, do_after_subscription::DoAfterSubscription,
66            do_after_termination::DoAfterTermination, do_before_disposal::DoBeforeDisposal,
67            do_before_next::DoBeforeNext, do_before_subscription::DoBeforeSubscription,
68            do_before_termination::DoBeforeTermination, materialize::Materialize,
69            observe_on::ObserveOn, subscribe_on::SubscribeOn, time_interval::TimeInterval,
70            timeout::Timeout, timestamp::Timestamp,
71        },
72    },
73    subject::{
74        async_subject::AsyncSubject, publish_subject::PublishSubject, replay_subject::ReplaySubject,
75    },
76    utils::types::{MaybeSend, MaybeSync},
77};
78use std::{fmt::Display, num::NonZeroUsize, time::Duration};
79
80pub type Subscription<D> = BoundDropDisposal<D>;
81
82/// The `Observable` trait represents a source of events that can be observed by an `Observer`.
83/// See <https://reactivex.io/documentation/observable.html>
84pub trait Observable<'or, T, E> {
85    type D: Disposable;
86
87    /// Subscribes an observer to this observable. When an observer is subscribed, it will start receiving events from the observable.
88    /// The `subscribe` method returns a `Subscription` which can be used to unsubscribe the observer from the observable.
89    /// We use `Subscription` struct instead of trait like `impl Cancellable`, because we need to cancel the subscription when the `Subscription` is dropped. It's not possible to implement Drop for a trait object.
90    fn subscribe(self, observer: impl Observer<T, E> + MaybeSend + 'or) -> Subscription<Self::D>;
91}
92
93/// Extension trait that exposes the full suite of RxRust operators on any type that
94/// implements [`Observable`]. Each method forwards to the corresponding operator
95/// constructor, allowing a fluent, ergonomic style when composing observable pipelines.
96pub trait ObservableExt<'or, T, E>: Observable<'or, T, E> + Sized {
97    /// Emits a single `bool` indicating whether every item satisfies the provided predicate.
98    fn all<F>(self, callback: F) -> All<T, Self, F>
99    where
100        F: FnMut(T) -> bool,
101    {
102        All::new(self, callback)
103    }
104
105    /// Competes two observables and mirrors whichever one produces an item or error first.
106    fn amb_with<OE1>(self, other: OE1) -> Amb<[EitherObservable<Self, OE1>; 2]>
107    where
108        OE1: Observable<'or, T, E>,
109    {
110        Amb::new([EitherObservable::Left(self), EitherObservable::Right(other)])
111    }
112
113    /// Calculates the arithmetic mean of all numeric items emitted by the source.
114    fn average(self) -> Average<T, Self> {
115        Average::new(self)
116    }
117
118    /// Collects the items emitted by the source into buffers delimited by another observable.
119    /// Terminating the boundary terminates the outer observable: completing it emits the pending
120    /// buffer when non-empty and then completes, while an error from it discards the pending
121    /// buffer and errors. Unlike [`window`](ObservableExt::window), a completed boundary does not
122    /// leave the current buffer open.
123    fn buffer<OE1>(self, boundary: OE1) -> Buffer<Self, OE1>
124    where
125        OE1: Observable<'or, (), E>,
126    {
127        Buffer::new(self, boundary)
128    }
129
130    /// Collects items into fixed-size buffers and emits each buffer as soon as it fills up.
131    fn buffer_with_count(self, count: NonZeroUsize) -> BufferWithCount<Self> {
132        BufferWithCount::new(self, count)
133    }
134
135    /// Collects items into time-based buffers driven by the provided scheduler.
136    fn buffer_with_time<S>(
137        self,
138        time_span: Duration,
139        scheduler: S,
140        delay: Option<Duration>,
141    ) -> BufferWithTime<'or, Self, S> {
142        BufferWithTime::new(self, time_span, scheduler, delay)
143    }
144
145    /// Collects items into buffers using both size and time boundaries whichever occurs first.
146    fn buffer_with_time_or_count<S>(
147        self,
148        count: NonZeroUsize,
149        time_span: Duration,
150        scheduler: S,
151        delay: Option<Duration>,
152    ) -> BufferWithTimeOrCount<Self, S> {
153        BufferWithTimeOrCount::new(self, count, time_span, scheduler, delay)
154    }
155
156    /// Recovers from errors by switching to another observable yielded by the callback.
157    fn catch<E1, OE1, F>(self, callback: F) -> Catch<E, Self, F>
158    where
159        OE1: Observable<'or, T, E1>,
160        F: FnOnce(E) -> OE1,
161    {
162        Catch::new(self, callback)
163    }
164
165    /// Gathers all the items into a collection built with `Default` and `Extend`, and emits it
166    /// when the source completes. See [`to_vec`](ObservableExt::to_vec) for the `Vec<T>` case.
167    fn collect<C>(self) -> Collect<C, T, Self>
168    where
169        C: Default + Extend<T>,
170    {
171        Collect::new(self)
172    }
173
174    /// Combines the latest values from both observables whenever either produces a new item.
175    fn combine_latest<T1, OE2>(self, another_source: OE2) -> CombineLatest<Self, OE2>
176    where
177        OE2: Observable<'or, T1, E>,
178    {
179        CombineLatest::new(self, another_source)
180    }
181
182    /// Flattens an observable-of-observables by concatenating each inner observable sequentially.
183    fn concat_all<T1>(self) -> ConcatAll<Self, T>
184    where
185        T: Observable<'or, T1, E>,
186    {
187        ConcatAll::new(self)
188    }
189
190    /// Maps each item to an observable and concatenates the resulting inner sequences.
191    fn concat_map<T1, OE1, F>(self, callback: F) -> ConcatMap<T, Self, OE1, F>
192    where
193        OE1: Observable<'or, T1, E>,
194        F: FnMut(T) -> OE1,
195    {
196        ConcatMap::new(self, callback)
197    }
198
199    /// Concatenates the source with another observable, waiting for the first to complete.
200    fn concat_with<OE2>(self, source_2: OE2) -> Concat<Self, OE2>
201    where
202        OE2: Observable<'or, T, E>,
203    {
204        Concat::new(self, source_2)
205    }
206
207    /// Emits `true` if the sequence contains the provided item, `false` otherwise.
208    fn contains(self, item: T) -> Contains<T, Self> {
209        Contains::new(self, item)
210    }
211
212    /// Counts the number of items emitted and emits that count as a single value.
213    fn count(self) -> Count<T, Self> {
214        Count::new(self)
215    }
216
217    /// Emits an item from the source Observable only after a particular time span has passed without another source emission.
218    fn debounce<S>(self, time_span: Duration, scheduler: S) -> Debounce<'or, Self, S> {
219        Debounce::new(self, time_span, scheduler)
220    }
221
222    /// Attaches a label to the stream and logs lifecycle events for debugging purposes using the provided callback.
223    fn debug<C, F>(self, context: C, callback: F) -> Debug<Self, C, F>
224    where
225        F: Fn(C, DebugEvent<'_, T, E>),
226    {
227        Debug::new(self, context, callback)
228    }
229
230    /// Attaches a label to the stream and logs lifecycle events for debugging purposes using the default print.
231    fn debug_default_print<L>(self, label: L) -> Debug<Self, L, DefaultPrintType<L, T, E>>
232    where
233        L: Display,
234        T: std::fmt::Debug,
235        E: std::fmt::Debug,
236    {
237        Debug::new_default_print(self, label)
238    }
239
240    /// Emits a default value if the source completes without emitting any items.
241    fn default_if_empty(self, default_value: T) -> DefaultIfEmpty<T, Self> {
242        DefaultIfEmpty::new(self, default_value)
243    }
244
245    /// Offsets the emission of items by the specified duration using the given scheduler.
246    fn delay<S>(self, delay: Duration, scheduler: S) -> Delay<'or, Self, S> {
247        Delay::new(self, delay, scheduler)
248    }
249
250    /// Converts a stream of notifications back into a normal observable sequence.
251    fn dematerialize(self) -> Dematerialize<Self> {
252        Dematerialize::new(self)
253    }
254
255    /// Filters out duplicate items, keeping only the first occurrence of each value.
256    fn distinct(self) -> Distinct<Self, fn(&T) -> T>
257    where
258        T: Clone,
259    {
260        Distinct::new(self)
261    }
262
263    /// Suppresses consecutive duplicate items, comparing the values directly.
264    fn distinct_until_changed(self) -> DistinctUntilChanged<Self, fn(&T) -> T>
265    where
266        T: Clone,
267    {
268        DistinctUntilChanged::new(self)
269    }
270
271    /// Suppresses consecutive duplicate items using a custom key selector.
272    fn distinct_until_changed_with_key_selector<F, K>(
273        self,
274        key_selector: F,
275    ) -> DistinctUntilChanged<Self, F>
276    where
277        F: FnMut(&T) -> K,
278    {
279        DistinctUntilChanged::new_with_key_selector(self, key_selector)
280    }
281
282    /// Filters out duplicates based on a key selector, keeping only unique keys.
283    fn distinct_with_key_selector<F, K>(self, key_selector: F) -> Distinct<Self, F>
284    where
285        F: FnMut(&T) -> K,
286    {
287        Distinct::new_with_key_selector(self, key_selector)
288    }
289
290    /// Invokes a callback after the downstream subscription is disposed.
291    fn do_after_disposal<F>(self, callback: F) -> DoAfterDisposal<Self, F>
292    where
293        F: FnOnce(),
294    {
295        DoAfterDisposal::new(self, callback)
296    }
297
298    /// Invokes a callback after each item is forwarded downstream.
299    fn do_after_next<F>(self, callback: F) -> DoAfterNext<Self, F>
300    where
301        F: FnMut(T),
302    {
303        DoAfterNext::new(self, callback)
304    }
305
306    /// Invokes a callback after the observer subscribes to the source.
307    fn do_after_subscription<F>(self, callback: F) -> DoAfterSubscription<Self, F>
308    where
309        F: FnOnce(),
310    {
311        DoAfterSubscription::new(self, callback)
312    }
313
314    /// Invokes a callback after the source terminates, regardless of completion or error.
315    fn do_after_termination<F>(self, callback: F) -> DoAfterTermination<Self, F>
316    where
317        F: FnOnce(Termination<E>),
318    {
319        DoAfterTermination::new(self, callback)
320    }
321
322    /// Invokes a callback right before the downstream subscription is disposed.
323    fn do_before_disposal<F>(self, callback: F) -> DoBeforeDisposal<Self, F>
324    where
325        F: FnOnce(),
326    {
327        DoBeforeDisposal::new(self, callback)
328    }
329
330    /// Invokes a callback with a reference to each item before it is sent downstream.
331    fn do_before_next<F>(self, callback: F) -> DoBeforeNext<Self, F>
332    where
333        F: FnMut(&T),
334    {
335        DoBeforeNext::new(self, callback)
336    }
337
338    /// Invokes a callback just before the observer subscribes to the source.
339    fn do_before_subscription<F>(self, callback: F) -> DoBeforeSubscription<Self, F>
340    where
341        F: FnOnce(),
342    {
343        DoBeforeSubscription::new(self, callback)
344    }
345
346    /// Invokes a callback before the stream terminates, receiving the termination reason.
347    fn do_before_termination<F>(self, callback: F) -> DoBeforeTermination<Self, F>
348    where
349        F: FnOnce(&Termination<E>),
350    {
351        DoBeforeTermination::new(self, callback)
352    }
353
354    /// Emits only the item at the given zero-based index and then completes.
355    fn element_at(self, index: usize) -> ElementAt<Self> {
356        ElementAt::new(self, index)
357    }
358
359    /// Filters items using a predicate, forwarding only values that return `true`.
360    fn filter<F>(self, callback: F) -> Filter<Self, F>
361    where
362        F: FnMut(&T) -> bool,
363    {
364        Filter::new(self, callback)
365    }
366
367    /// Emits only the first item from the source, then completes.
368    fn first(self) -> First<Self> {
369        First::new(self)
370    }
371
372    /// Maps each item to an observable and merges the resulting inner sequences concurrently.
373    fn flat_map<T1, OE1, F>(self, callback: F) -> FlatMap<T, Self, OE1, F>
374    where
375        OE1: Observable<'or, T1, E>,
376        F: FnMut(T) -> OE1,
377    {
378        FlatMap::new(self, callback)
379    }
380
381    /// Groups items by key into multiple observable sequences.
382    fn group_by<F, K>(self, key_selector: F) -> GroupBy<Self, F, K>
383    where
384        F: FnMut(&T) -> K,
385    {
386        GroupBy::new(self, key_selector)
387    }
388
389    /// Hooks into the emission of items, allowing mutation of the downstream observer.
390    ///
391    /// The callback returns the [`Flow`] the operator answers, which is normally the one the
392    /// downstream observer it was handed answered.
393    fn hook_on_next<F>(self, callback: F) -> HookOnNext<Self, F>
394    where
395        F: FnMut(&mut dyn Observer<T, E>, T) -> Flow,
396    {
397        HookOnNext::new(self, callback)
398    }
399
400    /// Hooks into subscription, letting you override how the source subscribes observers.
401    fn hook_on_subscription<D, F>(self, callback: F) -> HookOnSubscription<Self, F>
402    where
403        D: Disposable,
404        F: FnOnce(Self, BoxedObserver<'or, T, E>) -> Subscription<D>,
405    {
406        HookOnSubscription::new(self, callback)
407    }
408
409    /// Hooks into termination, providing access to the observer and termination payload.
410    fn hook_on_termination<F>(self, callback: F) -> HookOnTermination<Self, F>
411    where
412        F: FnOnce(BoxedObserver<'or, T, E>, Termination<E>),
413    {
414        HookOnTermination::new(self, callback)
415    }
416
417    /// Ignores all items from the source, only relaying termination events.
418    fn ignore_elements(self) -> IgnoreElements<Self> {
419        IgnoreElements::new(self)
420    }
421
422    /// Boxes the observable, erasing its concrete type while preserving lifetime bounds.
423    fn into_boxed<'sub, 'oe>(self) -> BoxedObservable<'or, 'sub, 'oe, T, E>
424    where
425        T: 'or,
426        E: 'or,
427        Self: MaybeSend + 'oe,
428        Self::D: MaybeSend + 'sub,
429    {
430        BoxedObservable::new(self)
431    }
432
433    /// Boxes the observable and makes it cloneable, erasing its concrete type while preserving lifetime bounds.
434    fn into_cloneable_boxed<'sub, 'oe>(self) -> CloneableBoxedObservable<'or, 'sub, 'oe, T, E>
435    where
436        T: 'or,
437        E: 'or,
438        Self: MaybeSend + MaybeSync + Clone + 'oe,
439        Self::D: MaybeSend + 'sub,
440    {
441        CloneableBoxedObservable::new(self)
442    }
443
444    /// Converts the observable into a future of its first item: `Some(item)`, or `None` when the
445    /// source completes without one. The source is stopped as soon as the item is in.
446    ///
447    /// This is only for a source that cannot fail; a fallible one goes through
448    /// [`into_try_future`](Self::into_try_future). An operator that picks another item, such as
449    /// `last`, or one that always emits, such as `collect`, goes in front of it.
450    fn into_future(self) -> ObservableFuture<'or, T, Self>
451    where
452        Self: Observable<'or, T, std::convert::Infallible>,
453    {
454        ObservableFuture::new(self)
455    }
456
457    /// Converts the observable into an async stream.
458    ///
459    /// A `Stream` has no error channel, so this is only for a source that cannot fail; a
460    /// fallible one goes through [`into_try_stream`](Self::into_try_stream).
461    ///
462    /// The items that arrive between two polls are all kept, so a source faster than the
463    /// consumer grows the buffer without bound; [`into_stream_with`](Self::into_stream_with)
464    /// takes a buffer that bounds it.
465    #[cfg(feature = "futures")]
466    fn into_stream(self) -> ObservableStream<'or, T, Self>
467    where
468        Self: Observable<'or, T, std::convert::Infallible>,
469    {
470        ObservableStream::new(self)
471    }
472
473    /// Converts the observable into an async stream that keeps the items arriving between two
474    /// polls in `buffer`, which decides what a source faster than the consumer costs.
475    ///
476    /// [`Latest`](crate::operators::others::observable_try_stream::Latest) keeps only the newest item,
477    /// [`Bounded`](crate::operators::others::observable_try_stream::Bounded) a fixed number of them and
478    /// [`Unbounded`](crate::operators::others::observable_try_stream::Unbounded) — what
479    /// [`into_stream`](Self::into_stream) uses — everything; a
480    /// [`StreamBuffer`] of your own can
481    /// fold them instead. Whatever the buffer, the source is never slowed down: a `Stream`
482    /// only pulls from the buffer, not from the source.
483    ///
484    /// # Examples
485    /// ```rust
486    /// use futures::{FutureExt, StreamExt};
487    /// use rx_rust::{
488    ///     observable::ObservableExt, observer::Observer,
489    ///     operators::others::observable_try_stream::Latest,
490    ///     subject::publish_subject::PublishSubject,
491    /// };
492    /// use std::convert::Infallible;
493    ///
494    /// let mut subject = PublishSubject::<_, Infallible>::new();
495    /// let mut stream = subject.clone().into_stream_with(Latest::new());
496    /// assert_eq!(stream.next().now_or_never(), None); // subscribes
497    ///
498    /// subject.on_next(1);
499    /// subject.on_next(2);
500    /// subject.on_next(3);
501    /// assert_eq!(stream.next().now_or_never(), Some(Some(3)));
502    /// assert_eq!(stream.next().now_or_never(), None);
503    /// ```
504    #[cfg(feature = "futures")]
505    fn into_stream_with<B>(self, buffer: B) -> ObservableStream<'or, T, Self, B>
506    where
507        Self: Observable<'or, T, std::convert::Infallible>,
508        B: StreamBuffer<T>,
509    {
510        ObservableStream::with_buffer(self, buffer)
511    }
512
513    /// Converts the observable into a future of its first item: `Ok(Some(item))`, `Ok(None)` when
514    /// the source completes without one, or `Err(error)` when it fails first. The source is
515    /// stopped as soon as the item is in.
516    ///
517    /// The output is the `Maybe` of ReactiveX; an operator that always emits, such as `collect`,
518    /// in front of it makes it a `Single`, and `last` picks the last item instead of the first.
519    fn into_try_future(self) -> ObservableTryFuture<'or, T, E, Self> {
520        ObservableTryFuture::new(self)
521    }
522
523    /// Converts the observable into an async stream of `Result`s: each item as `Ok`, and an error
524    /// as the last item, `Err`, before the stream ends.
525    ///
526    /// The items that arrive between two polls are all kept, so a source faster than the
527    /// consumer grows the buffer without bound;
528    /// [`into_try_stream_with`](Self::into_try_stream_with) takes a buffer that bounds it.
529    #[cfg(feature = "futures")]
530    fn into_try_stream(self) -> ObservableTryStream<'or, T, E, Self> {
531        ObservableTryStream::new(self)
532    }
533
534    /// Converts the observable into an async stream of `Result`s that keeps the items arriving
535    /// between two polls in `buffer`. This is [`into_stream_with`](Self::into_stream_with) for
536    /// a source that can fail; see there for the buffers.
537    #[cfg(feature = "futures")]
538    fn into_try_stream_with<B>(self, buffer: B) -> ObservableTryStream<'or, T, E, Self, B>
539    where
540        B: StreamBuffer<T>,
541    {
542        ObservableTryStream::with_buffer(self, buffer)
543    }
544
545    /// Emits only the final item produced by the source before completion.
546    fn last(self) -> Last<Self> {
547        Last::new(self)
548    }
549
550    /// Transforms each item by applying a user-supplied mapping function.
551    fn map<T1, F>(self, callback: F) -> Map<T, Self, F>
552    where
553        F: FnMut(T) -> T1,
554    {
555        Map::new(self, callback)
556    }
557
558    /// Transforms an error emitted by the source while leaving its items unchanged.
559    fn map_err<E1, F>(self, callback: F) -> MapErr<E, Self, F>
560    where
561        F: FnOnce(E) -> E1,
562    {
563        MapErr::new(self, callback)
564    }
565
566    /// Wraps each item into a notification, turning the stream into explicit events.
567    fn materialize(self) -> Materialize<Self> {
568        Materialize::new(self)
569    }
570
571    /// Emits the maximum item produced by the source according to the natural order.
572    fn max(self) -> Max<Self> {
573        Max::new(self)
574    }
575
576    /// Merges an observable-of-observables by interleaving items from inner streams.
577    fn merge_all<T1>(self) -> MergeAll<Self, T>
578    where
579        T: Observable<'or, T1, E>,
580    {
581        MergeAll::new(self)
582    }
583
584    /// Merges the source with another observable, interleaving both streams concurrently.
585    fn merge_with<OE2>(self, source_2: OE2) -> Merge<Self, OE2>
586    where
587        OE2: Observable<'or, T, E>,
588    {
589        Merge::new(self, source_2)
590    }
591
592    /// Emits the minimum item produced by the source according to the natural order.
593    fn min(self) -> Min<Self> {
594        Min::new(self)
595    }
596
597    /// Converts the source into a connectable observable using a subject factory.
598    fn multicast<S, F>(self, subject_maker: F) -> ConnectableController<Self, S>
599    where
600        F: FnOnce() -> S,
601    {
602        ConnectableController::new(self, subject_maker())
603    }
604
605    /// Schedules downstream observation on the provided scheduler.
606    fn observe_on<S>(self, scheduler: S) -> ObserveOn<'or, Self, S> {
607        ObserveOn::new(self, scheduler)
608    }
609
610    /// Multicasts the source using a `PublishSubject`.
611    fn publish(self) -> ConnectableController<Self, PublishSubject<'or, T, E>> {
612        self.multicast(PublishSubject::default)
613    }
614
615    /// Multicasts the source using an `AsyncSubject`, emitting only the last value.
616    fn publish_last(self) -> ConnectableController<Self, AsyncSubject<'or, T, E>> {
617        self.multicast(AsyncSubject::default)
618    }
619
620    /// Aggregates the sequence using an initial seed and an accumulator function.
621    fn reduce<T0, F>(self, initial_value: T0, callback: F) -> Reduce<T0, T, Self, F>
622    where
623        F: FnMut(T0, T) -> T0,
624    {
625        Reduce::new(self, initial_value, callback)
626    }
627
628    /// Multicasts the source using a `ReplaySubject` configured with the given buffer size.
629    fn replay(
630        self,
631        buffer_size: Option<usize>,
632    ) -> ConnectableController<Self, ReplaySubject<'or, T, E>> {
633        self.multicast(|| ReplaySubject::new(buffer_size))
634    }
635
636    /// Re-subscribes to the source based on the retry strategy returned by the callback.
637    fn retry<OE1, F>(self, callback: F) -> Retry<Self, F>
638    where
639        OE1: Observable<'or, T, E>,
640        F: FnMut(E) -> RetryAction<E, OE1>,
641    {
642        Retry::new(self, callback)
643    }
644
645    /// Samples the source whenever the sampler observable emits an event.
646    fn sample<OE1>(self, sampler: OE1) -> Sample<Self, OE1>
647    where
648        OE1: Observable<'or, (), E>,
649    {
650        Sample::new(self, sampler)
651    }
652
653    /// Accumulates values over time, emitting each intermediate result.
654    fn scan<T0, F>(self, initial_value: T0, callback: F) -> Scan<T0, T, Self, F>
655    where
656        F: FnMut(T0, T) -> T0,
657    {
658        Scan::new(self, initial_value, callback)
659    }
660
661    /// Compares two sequences element by element for equality.
662    fn sequence_equal<OE2>(self, another_source: OE2) -> SequenceEqual<T, Self, OE2>
663    where
664        OE2: Observable<'or, T, E>,
665    {
666        SequenceEqual::new(self, another_source)
667    }
668
669    /// Shares a single subscription to the source using `PublishSubject` semantics.
670    fn share(self) -> RefCount<'or, T, E, Self, PublishSubject<'or, T, E>> {
671        self.publish().ref_count()
672    }
673
674    /// Shares a single subscription, replaying only the last item to new subscribers.
675    fn share_last(self) -> RefCount<'or, T, E, Self, AsyncSubject<'or, T, E>> {
676        self.publish_last().ref_count()
677    }
678
679    /// Shares a single subscription while replaying a bounded history to future subscribers.
680    fn share_replay(
681        self,
682        buffer_size: Option<usize>,
683    ) -> RefCount<'or, T, E, Self, ReplaySubject<'or, T, E>> {
684        self.replay(buffer_size).ref_count()
685    }
686
687    /// Skips the first `count` items before emitting the remainder of the sequence.
688    fn skip(self, count: usize) -> Skip<Self> {
689        Skip::new(self, count)
690    }
691
692    /// Skips the last `count` items emitted by the source.
693    fn skip_last(self, count: usize) -> SkipLast<Self> {
694        SkipLast::new(self, count)
695    }
696
697    /// Ignores items from the source until the notifier observable fires.
698    fn skip_until<OE1>(self, start: OE1) -> SkipUntil<Self, OE1>
699    where
700        OE1: Observable<'or, (), E>,
701    {
702        SkipUntil::new(self, start)
703    }
704
705    /// Skips items while the predicate returns `true`, then emits the remaining items.
706    fn skip_while<F>(self, callback: F) -> SkipWhile<Self, F>
707    where
708        F: FnMut(&T) -> bool,
709    {
710        SkipWhile::new(self, callback)
711    }
712
713    /// Pre-pends the provided values before the source starts emitting.
714    fn start_with<I>(self, values: I) -> StartWith<Self, I>
715    where
716        I: IntoIterator<Item = T>,
717    {
718        StartWith::new(self, values)
719    }
720
721    /// Subscribes to the source on the provided scheduler.
722    fn subscribe_on<S>(self, scheduler: S) -> SubscribeOn<'or, Self, S> {
723        SubscribeOn::new(self, scheduler)
724    }
725
726    /// Convenience helper for subscribing with plain callbacks instead of a full observer.
727    ///
728    /// `on_next` may return nothing, which keeps the source going, or a [`Flow`], which lets it
729    /// end its own stream with [`Flow::Stop`]: the source then stops pushing — a synchronous one
730    /// stops iterating — and drops the callbacks without calling `on_termination`.
731    fn subscribe_with_callback<FN, FT, R>(
732        self,
733        on_next: FN,
734        on_termination: FT,
735    ) -> Subscription<Self::D>
736    where
737        FN: FnMut(T) -> R + MaybeSend + 'or,
738        R: IntoFlow,
739        FT: FnOnce(Termination<E>) + MaybeSend + 'or,
740    {
741        self.subscribe(CallbackObserver::new(on_next, on_termination))
742    }
743
744    /// Sums all numeric items and emits the accumulated total.
745    fn sum(self) -> Sum<Self> {
746        Sum::new(self)
747    }
748
749    /// Switches to the most recent inner observable emitted by the source.
750    fn switch<T1>(self) -> Switch<Self, T>
751    where
752        T: Observable<'or, T1, E>,
753    {
754        Switch::new(self)
755    }
756
757    /// Maps each item to an observable and switches to the latest inner sequence.
758    fn switch_map<T1, OE1, F>(self, callback: F) -> SwitchMap<T, Self, OE1, F>
759    where
760        OE1: Observable<'or, T1, E>,
761        F: FnMut(T) -> OE1,
762    {
763        SwitchMap::new(self, callback)
764    }
765
766    /// Emits only the first `count` items from the source before completing.
767    fn take(self, count: usize) -> Take<Self> {
768        Take::new(self, count)
769    }
770
771    /// Emits only the last `count` items produced by the source.
772    fn take_last(self, count: usize) -> TakeLast<Self> {
773        TakeLast::new(self, count)
774    }
775
776    /// Relays items until the notifier observable emits, then completes.
777    fn take_until<OE1>(self, stop: OE1) -> TakeUntil<Self, OE1>
778    where
779        OE1: Observable<'or, (), E>,
780    {
781        TakeUntil::new(self, stop)
782    }
783
784    /// Emits items while the predicate holds `true`, then completes.
785    fn take_while<F>(self, callback: F) -> TakeWhile<Self, F>
786    where
787        F: FnMut(&T) -> bool,
788    {
789        TakeWhile::new(self, callback)
790    }
791
792    /// Throttles emissions to at most one item per specified timespan.
793    ///
794    /// Leading-edge and scheduler-free: the cooldown is decided by comparing
795    /// item arrival times, so no timer is spawned.
796    fn throttle(self, time_span: Duration) -> Throttle<Self> {
797        Throttle::new(self, time_span)
798    }
799
800    /// Emits elapsed time between consecutive items as they flow through the stream.
801    fn time_interval(self) -> TimeInterval<Self> {
802        TimeInterval::new(self)
803    }
804
805    /// Errors if the next item does not arrive within the specified duration.
806    fn timeout<S>(self, duration: Duration, scheduler: S) -> Timeout<'or, Self, S> {
807        Timeout::new(self, duration, scheduler)
808    }
809
810    /// Annotates each item with the current timestamp when it is emitted.
811    fn timestamp(self) -> Timestamp<Self> {
812        Timestamp::new(self)
813    }
814
815    /// Gathers all the items into a `Vec` and emits it when the source completes. This is
816    /// [`collect`](ObservableExt::collect) specialized to `Vec<T>`, which is the shape that
817    /// [`window`](ObservableExt::window) composes with:
818    /// `source.window(boundary).concat_map(|window| window.to_vec())`.
819    fn to_vec(self) -> Collect<Vec<T>, T, Self> {
820        Collect::new(self)
821    }
822
823    /// Collects items into windows that are opened and closed by another observable.
824    /// Completing the boundary stops future window rotation without terminating the source.
825    /// An error from the boundary terminates the current window and the outer observable.
826    fn window<OE1>(self, boundary: OE1) -> Window<Self, OE1>
827    where
828        OE1: Observable<'or, (), E>,
829    {
830        Window::new(self, boundary)
831    }
832
833    /// Collects items into windows containing a fixed number of elements.
834    fn window_with_count(self, count: NonZeroUsize) -> WindowWithCount<Self> {
835        WindowWithCount::new(self, count)
836    }
837
838    /// Gives an Observable whose error type is `Infallible` a concrete error type.
839    fn with_error_type<E1>(self) -> WithErrorType<E1, Self> {
840        WithErrorType::new(self)
841    }
842
843    /// Gives an Observable whose item type is `Infallible` a concrete item type.
844    fn with_item_type<T1>(self) -> WithItemType<T1, Self> {
845        WithItemType::new(self)
846    }
847
848    /// Pairs items from both observables by index and emits tuples of corresponding values.
849    fn zip<T1, OE2>(self, another_source: OE2) -> Zip<Self, OE2>
850    where
851        OE2: Observable<'or, T1, E>,
852    {
853        Zip::new(self, another_source)
854    }
855}
856
857impl<'or, T, E, OE> ObservableExt<'or, T, E> for OE where OE: Observable<'or, T, E> {}
rx_rust/observable/mod.rs

rx_rust/observable/
mod.rs
