rx_rust/observer/
mod.rs

1pub mod boxed_observer;
2pub mod callback_observer;
3
4use crate::{observer::boxed_observer::BoxedObserver, utils::types::MaybeSend};
5use educe::Educe;
6
7/// Represents the termination state of an operation, which can either be completed successfully or with an error.
8#[derive(Educe)]
9#[educe(Debug, Clone, PartialEq, Eq)]
10pub enum Termination<E> {
11    /// Indicates that the operation has completed successfully.
12    Completed,
13    /// Indicates that the operation has completed with an error.
14    Error(E),
15}
16
17/// Whether an observer still accepts events, as reported by [`Observer::on_next`].
18///
19/// This is the only way an observer can tell the source that is pushing into it to stop, while
20/// that push is running. It matters most for a synchronous source, which delivers its whole
21/// sequence before `subscribe` returns: until it does, nobody holds the subscription yet, so
22/// disposing it is not yet possible, and an infinite synchronous source would never end.
23///
24/// It does not replace disposal, and no source may rely on it alone: an observer can also go away
25/// between two events, which no return value can report. See the variants for what each one
26/// promises.
27#[derive(Debug, Clone, Copy, PartialEq, Eq)]
28#[must_use = "an upstream that ignores the flow keeps pushing into an observer that stopped"]
29pub enum Flow {
30    /// The observer is not known to have stopped, so the source may keep pushing.
31    ///
32    /// This is a hint, not a guarantee: it means "no stop has been observed here", and the
33    /// observer may still be disposed before the next event, or already have been disposed
34    /// somewhere this call could not see — a value queued behind a delivery running elsewhere is
35    /// reported as `Continue` even when that delivery is about to stop. A source must therefore
36    /// still honor its disposal; treating `Continue` as proof that downstream is alive is wrong.
37    Continue,
38    /// The observer will never accept another event.
39    ///
40    /// The caller must not call [`Observer::on_next`] again, must not call
41    /// [`Observer::on_termination`], and should drop the observer instead: it has either already
42    /// delivered its own termination downstream or been disposed, so a termination sent now would
43    /// be a second one. Unlike [`Flow::Continue`], this is a guarantee, never a hint.
44    Stop,
45}
46
47impl Flow {
48    /// Returns whether this is [`Flow::Continue`].
49    #[inline]
50    pub fn is_continue(self) -> bool {
51        matches!(self, Flow::Continue)
52    }
53
54    /// Returns whether this is [`Flow::Stop`].
55    #[inline]
56    pub fn is_stop(self) -> bool {
57        matches!(self, Flow::Stop)
58    }
59}
60
61/// A trait for observing the progress and termination state of an operation.
62pub trait Observer<T, E> {
63    /// Called when the next value in the operation is available.
64    ///
65    /// Returns whether the observer accepts further events. [`Flow::Stop`] means it accepts none
66    /// and must not be terminated either, so the caller stops pushing and drops it; see [`Flow`]
67    /// for the exact promise each variant makes. An operator that forwards values must return what
68    /// its own downstream returned, so that the answer reaches the source at the end of the chain.
69    fn on_next(&mut self, value: T) -> Flow;
70
71    /// Called when the operation has reached its termination state.
72    fn on_termination(self, termination: Termination<E>);
73}
74
75#[derive(Educe)]
76#[educe(Debug, Clone, PartialEq, Eq)]
77pub enum Event<T, E> {
78    Next(T),
79    Termination(Termination<E>),
80}
81
82pub trait BoxedObserverExt<T, E>: Observer<T, E> + Sized {
83    fn into_boxed<'or>(self) -> BoxedObserver<'or, T, E>
84    where
85        Self: MaybeSend + 'or,
86    {
87        BoxedObserver::new(self)
88    }
89}
90
91impl<T, E, OR> BoxedObserverExt<T, E> for OR where OR: Observer<T, E> {}
rx_rust/observer/mod.rs

rx_rust/observer/
mod.rs
