rx_rust/operators/error_handling/
catch.rs

1use crate::disposable::subscription::Subscription;
2use crate::safe_lock_option;
3use crate::utils::types::{Mutable, NecessarySendSync, Shared};
4use crate::{
5    observable::Observable,
6    observer::{Observer, Termination},
7    utils::types::MarkerType,
8};
9use educe::Educe;
10use std::marker::PhantomData;
11
12/// Catches errors on the observable to be handled by returning a new observable or throwing an error.
13/// See <https://reactivex.io/documentation/operators/catch.html>
14///
15/// # Examples
16/// ```rust
17/// use rx_rust::{
18///     observable::observable_ext::ObservableExt,
19///     observer::Termination,
20///     operators::{
21///         creating::{just::Just, throw::Throw},
22///         error_handling::catch::Catch,
23///     },
24/// };
25///
26/// let mut values = Vec::new();
27/// let mut terminations = Vec::new();
28///
29/// let observable = Catch::new(Throw::new("boom").map_infallible_to_value(), |error| Just::new(error));
30/// observable.subscribe_with_callback(
31///     |value| values.push(value),
32///     |termination| terminations.push(termination),
33/// );
34///
35/// assert_eq!(values, vec!["boom"]);
36/// assert_eq!(terminations, vec![Termination::Completed]);
37/// ```
38#[derive(Educe)]
39#[educe(Debug, Clone)]
40pub struct Catch<E0, OE, F> {
41    source: OE,
42    callback: F,
43    _marker: MarkerType<E0>,
44}
45
46impl<E0, OE, F> Catch<E0, OE, F> {
47    pub fn new<'or, 'sub, T, E, OE1>(source: OE, callback: F) -> Self
48    where
49        OE: Observable<'or, 'sub, T, E0>,
50        OE1: Observable<'or, 'sub, T, E>,
51        F: FnOnce(E0) -> OE1,
52    {
53        Self {
54            source,
55            callback,
56            _marker: PhantomData,
57        }
58    }
59}
60
61impl<'or, 'sub, T, E0, E, OE, OE1, F> Observable<'or, 'sub, T, E> for Catch<E0, OE, F>
62where
63    E: 'or,
64    OE: Observable<'or, 'sub, T, E0>,
65    OE1: Observable<'or, 'sub, T, E>,
66    F: FnOnce(E0) -> OE1 + NecessarySendSync + 'or,
67    'sub: 'or,
68{
69    fn subscribe(
70        self,
71        observer: impl Observer<T, E> + NecessarySendSync + 'or,
72    ) -> Subscription<'sub> {
73        let sub = Shared::new(Mutable::new(None));
74        let onserver = CatchObserver {
75            observer,
76            callback: self.callback,
77            sub: sub.clone(),
78            _marker: PhantomData,
79        };
80        self.source.subscribe(onserver) + sub
81    }
82}
83
84struct CatchObserver<'sub, E, OR, F> {
85    observer: OR,
86    callback: F,
87    sub: Shared<Mutable<Option<Subscription<'sub>>>>,
88    _marker: MarkerType<E>,
89}
90
91impl<'or, 'sub, T, E0, E, OR, OE1, F> Observer<T, E0> for CatchObserver<'sub, E, OR, F>
92where
93    OR: Observer<T, E> + NecessarySendSync + 'or,
94    OE1: Observable<'or, 'sub, T, E>,
95    F: FnOnce(E0) -> OE1,
96{
97    fn on_next(&mut self, value: T) {
98        self.observer.on_next(value);
99    }
100
101    fn on_termination(self, termination: Termination<E0>) {
102        match termination {
103            Termination::Completed => self.observer.on_termination(Termination::Completed),
104            Termination::Error(error) => {
105                let observable = (self.callback)(error);
106                let sub = observable.subscribe(self.observer);
107                safe_lock_option!(replace: self.sub, sub);
108            }
109        }
110    }
111}
rx_rust/operators/error_handling/catch.rs

rx_rust/operators/error_handling/
catch.rs
