rx_rust/observer/
callback_observer.rs

1use super::{Flow, Observer, Termination};
2use educe::Educe;
3
4/// What a plain callback may answer in place of a [`Flow`].
5///
6/// A callback bound as `FnMut(T) -> R` with `R: IntoFlow` may return `()`, which keeps the source
7/// going — so the common `|value| { ... }` needs no trailing [`Flow::Continue`] — or a [`Flow`],
8/// which lets it end its own stream. This is what
9/// [`CallbackObserver`] relies on.
10///
11/// A callback that only diverges, such as `|_| unreachable!()`, is inferred to return `!`, which
12/// no stable impl can cover: spell its return type out, as `|_| -> () { unreachable!() }`.
13pub trait IntoFlow {
14    fn into_flow(self) -> Flow;
15}
16
17impl IntoFlow for () {
18    #[inline]
19    fn into_flow(self) -> Flow {
20        Flow::Continue
21    }
22}
23
24impl IntoFlow for Flow {
25    #[inline]
26    fn into_flow(self) -> Flow {
27        self
28    }
29}
30
31#[derive(Educe)]
32#[educe(Debug, Clone)]
33pub struct CallbackObserver<FN, FT> {
34    #[educe(Debug(ignore))]
35    on_next: FN,
36    #[educe(Debug(ignore))]
37    on_termination: FT,
38}
39
40impl<FN, FT> CallbackObserver<FN, FT> {
41    pub fn new<T, E, R>(on_next: FN, on_termination: FT) -> Self
42    where
43        FN: FnMut(T) -> R,
44        R: IntoFlow,
45        FT: FnOnce(Termination<E>),
46    {
47        Self {
48            on_next,
49            on_termination,
50        }
51    }
52}
53
54impl<T, E, R, FN, FT> Observer<T, E> for CallbackObserver<FN, FT>
55where
56    FN: FnMut(T) -> R,
57    R: IntoFlow,
58    FT: FnOnce(Termination<E>),
59{
60    fn on_next(&mut self, value: T) -> Flow {
61        // A callback that returns nothing keeps the source going; one that returns a `Flow` can
62        // end its own stream, see `IntoFlow`.
63        (self.on_next)(value).into_flow()
64    }
65
66    fn on_termination(self, termination: Termination<E>) {
67        (self.on_termination)(termination);
68    }
69}
rx_rust/observer/callback_observer.rs

rx_rust/observer/
callback_observer.rs
