rx_rust/operators/others/
debug.rs

1use crate::utils::types::{MarkerType, MaybeSend};
2use crate::{
3    disposable::Disposable,
4    observable::{Observable, Subscription},
5    observer::{Flow, Observer, Termination},
6};
7use educe::Educe;
8use std::{fmt::Display, marker::PhantomData};
9
10#[derive(Educe)]
11#[educe(Debug, Clone, PartialEq, Eq)]
12pub enum DebugEvent<'a, T: 'a, E: 'a> {
13    OnNext(&'a T),
14    OnTermination(&'a Termination<E>),
15    Subscribed,
16    Disposed,
17}
18
19/// Logs all items from the source Observable to the console, and re-emits them. This is useful for debugging.
20///
21/// # Examples
22/// ```rust
23/// use rx_rust::{
24///     observable::ObservableExt,
25///     observer::Termination,
26///     operators::{
27///         creating::from_iter::FromIter,
28///         others::debug::Debug,
29///     },
30/// };
31///
32/// let mut values = Vec::new();
33/// let mut terminations = Vec::new();
34///
35/// let observable = Debug::new_default_print(FromIter::new(vec![1, 2]), "trace");
36/// observable.subscribe_with_callback(
37///     |value| values.push(value),
38///     |termination| terminations.push(termination),
39/// );
40///
41/// assert_eq!(values, vec![1, 2]);
42/// assert_eq!(terminations, vec![Termination::Completed]);
43/// ```
44#[derive(Educe)]
45#[educe(Debug, Clone)]
46pub struct Debug<OE, C, F> {
47    source: OE,
48    context: C,
49    callback: F,
50}
51
52impl<OE, C, F> Debug<OE, C, F> {
53    pub fn new<T, E>(source: OE, context: C, callback: F) -> Self
54    where
55        F: Fn(C, DebugEvent<'_, T, E>),
56    {
57        Self {
58            source,
59            context,
60            callback,
61        }
62    }
63}
64
65pub type DefaultPrintType<C, T, E> = fn(C, DebugEvent<'_, T, E>);
66
67impl<T, E, OE, C> Debug<OE, C, DefaultPrintType<C, T, E>> {
68    pub fn new_default_print(source: OE, label: C) -> Self
69    where
70        C: Display,
71        T: std::fmt::Debug,
72        E: std::fmt::Debug,
73    {
74        Self {
75            source,
76            context: label,
77            callback: |label, event| match event {
78                DebugEvent::OnNext(value) => println!("[{}]: OnNext({:?})", label, value),
79                DebugEvent::OnTermination(termination) => {
80                    println!("[{}]: OnTermination({:?})", label, termination)
81                }
82                DebugEvent::Subscribed => println!("[{}]: Subscription", label),
83                DebugEvent::Disposed => println!("[{}]: Dispose", label),
84            },
85        }
86    }
87}
88
89impl<'or, T, E, OE, C, F> Observable<'or, T, E> for Debug<OE, C, F>
90where
91    OE: Observable<'or, T, E>,
92    C: Clone + MaybeSend + 'or,
93    F: Fn(C, DebugEvent<'_, T, E>) + Clone + MaybeSend + 'or,
94{
95    type D = DebugDisposal<Subscription<OE::D>, C, F, T, E>;
96
97    fn subscribe(self, observer: impl Observer<T, E> + MaybeSend + 'or) -> Subscription<Self::D> {
98        (self.callback)(self.context.clone(), DebugEvent::Subscribed);
99        let observer = DebugObserver {
100            observer,
101            context: self.context.clone(),
102            callback: self.callback.clone(),
103        };
104        let source_disposal = self.source.subscribe(observer);
105        Subscription::new(DebugDisposal {
106            source_disposal,
107            context: self.context,
108            callback: self.callback,
109            _marker: PhantomData,
110        })
111    }
112}
113
114pub struct DebugDisposal<D, C, F, T, E> {
115    source_disposal: D,
116    context: C,
117    callback: F,
118    _marker: MarkerType<(T, E)>,
119}
120
121impl<D, C, F, T, E> Disposable for DebugDisposal<D, C, F, T, E>
122where
123    D: Disposable,
124    F: Fn(C, DebugEvent<'_, T, E>),
125{
126    fn dispose(self) {
127        self.source_disposal.dispose();
128        (self.callback)(self.context, DebugEvent::Disposed);
129    }
130}
131
132struct DebugObserver<OR, C, F> {
133    observer: OR,
134    context: C,
135    callback: F,
136}
137
138impl<T, E, OR, C, F> Observer<T, E> for DebugObserver<OR, C, F>
139where
140    OR: Observer<T, E>,
141    C: Clone,
142    F: Fn(C, DebugEvent<'_, T, E>),
143{
144    fn on_next(&mut self, value: T) -> Flow {
145        (self.callback)(self.context.clone(), DebugEvent::OnNext(&value));
146        self.observer.on_next(value)
147    }
148
149    fn on_termination(self, termination: Termination<E>) {
150        (self.callback)(self.context, DebugEvent::OnTermination(&termination));
151        self.observer.on_termination(termination);
152    }
153}
rx_rust/operators/others/debug.rs

rx_rust/operators/others/
debug.rs
