rx_rust/operators/error_handling/
retry.rs

1use crate::delegate_disposal;
2use crate::disposable::{
3    Disposable, chain_disposal::ChainDisposal, shared_disposal::SharedDisposal,
4};
5use crate::observable::Subscription;
6use crate::utils::types::MaybeSend;
7use crate::{
8    observable::Observable,
9    observer::{Flow, Observer, Termination},
10};
11use educe::Educe;
12
13#[derive(Educe)]
14#[educe(Debug, Clone)]
15pub enum RetryAction<E, OE1> {
16    Retry(OE1),
17    Stop(E),
18}
19
20/// Retries an Observable in case of an error, based on a retry policy.
21/// See <https://reactivex.io/documentation/operators/retry.html>
22///
23/// # Examples
24/// ```rust
25/// use rx_rust::{
26///     observable::ObservableExt,
27///     observer::Termination,
28///     operators::{
29///         creating::{just::Just, throw::Throw},
30///         error_handling::retry::{Retry, RetryAction},
31///     },
32/// };
33///
34/// let mut values = Vec::new();
35/// let mut terminations = Vec::new();
36///
37/// let observable = Retry::new(Throw::new("boom").with_item_type(), |_| RetryAction::Retry(Just::new(42).with_error_type()));
38/// observable.subscribe_with_callback(
39///     |value| values.push(value),
40///     |termination| terminations.push(termination),
41/// );
42///
43/// assert_eq!(values, vec![42]);
44/// assert_eq!(terminations, vec![Termination::Completed]);
45/// ```
46#[derive(Educe)]
47#[educe(Debug, Clone)]
48pub struct Retry<OE, F> {
49    source: OE,
50    callback: F,
51}
52
53impl<OE, F> Retry<OE, F> {
54    pub fn new<'or, T, E, OE1>(source: OE, callback: F) -> Self
55    where
56        OE: Observable<'or, T, E>,
57        OE1: Observable<'or, T, E>,
58        F: FnMut(E) -> RetryAction<E, OE1>,
59    {
60        Self { source, callback }
61    }
62}
63
64delegate_disposal!(
65    Disposal<D, D1>,
66    ChainDisposal<SharedDisposal<Subscription<D1>>, D>,
67    where D: Disposable, D1: Disposable
68);
69
70impl<'or, T, E, OE, OE1, F> Observable<'or, T, E> for Retry<OE, F>
71where
72    OE: Observable<'or, T, E>,
73    OE1: Observable<'or, T, E>,
74    OE1::D: MaybeSend + 'or,
75    F: FnMut(E) -> RetryAction<E, OE1> + MaybeSend + 'or,
76{
77    type D = Disposal<OE::D, OE1::D>;
78
79    fn subscribe(self, observer: impl Observer<T, E> + MaybeSend + 'or) -> Subscription<Self::D> {
80        let shared_disposal = SharedDisposal::default();
81        let observer = RetryObserver {
82            observer,
83            callback: self.callback,
84            shared_disposal: shared_disposal.clone(),
85        };
86        self.source
87            .subscribe(observer)
88            .preceded_by(shared_disposal)
89            .map_into()
90    }
91}
92
93struct RetryObserver<OR, F, D: Disposable> {
94    observer: OR,
95    callback: F,
96    shared_disposal: SharedDisposal<Subscription<D>>,
97}
98
99impl<'or, T, E, OR, OE1, F> Observer<T, E> for RetryObserver<OR, F, OE1::D>
100where
101    OR: Observer<T, E> + MaybeSend + 'or,
102    OE1: Observable<'or, T, E>,
103    OE1::D: MaybeSend + 'or,
104    F: FnMut(E) -> RetryAction<E, OE1> + MaybeSend + 'or,
105{
106    fn on_next(&mut self, value: T) -> Flow {
107        self.observer.on_next(value)
108    }
109
110    fn on_termination(mut self, termination: Termination<E>) {
111        match termination {
112            completion @ Termination::Completed => self.observer.on_termination(completion),
113            Termination::Error(error) => {
114                let action = (self.callback)(error);
115                match action {
116                    RetryAction::Retry(observable) => {
117                        self.shared_disposal
118                            .clone()
119                            .replace(|| observable.subscribe(self));
120                    }
121                    RetryAction::Stop(error) => {
122                        self.observer.on_termination(Termination::Error(error))
123                    }
124                }
125            }
126        }
127    }
128}
rx_rust/operators/error_handling/retry.rs

rx_rust/operators/error_handling/
retry.rs
