ferroui_dynamic_data/kernel/
internal_ex.rs

1//! `InternalEx` (`Kernel/InternalEx.cs`): retries with a back-off, recurring actions on a
2//! scheduler, and the internal helpers of the operators.
3//!
4//! The internal `Swap` is `std::mem::swap`, and `OnNext(ISubject<Unit>)` is `on_next(())`.
5
6use ferroui_reactive::concurrency::{IScheduler, SchedulerExt};
7use ferroui_reactive::disposables::Disposable;
8use ferroui_reactive::{Observable, ObservableExt};
9use ferroui_system::{IDisposable, IObservable, IObserver, ObservableError};
10use std::rc::Rc;
11use std::time::Duration;
12
13type BackOffStrategy<E> = Rc<dyn Fn(&E, usize) -> Option<Duration>>;
14
15fn retry<T: 'static, E: std::error::Error + 'static>(source: Rc<dyn IObservable<T>>, back_off_strategy: BackOffStrategy<E>, failure_count: usize) -> Rc<dyn IObservable<T>> {
16    let retried = source.clone();
17    source.catch_of::<E>(move |error: ObservableError| {
18        let typed = error.downcast_ref::<E>().expect("catch_of hands over the errors of its type");
19        match back_off_strategy(typed, failure_count) {
20            None => Observable::throw::<T>(error),
21            Some(delay) => {
22                let next = retry(retried.clone(), back_off_strategy.clone(), failure_count + 1);
23                Observable::timer(delay).select_many(move |_| next.clone())
24            }
25        }
26    })
27}
28
29/// The sequence, subscribed again after each error of type `E` once the delay that
30/// `back_off_strategy` gives for the error and the number of failures so far has passed; a `None`
31/// delay ends the sequence with the error (`RetryWithBackOff<TSource, TException>`).
32pub fn retry_with_back_off<T: 'static, E: std::error::Error + 'static>(
33    source: Rc<dyn IObservable<T>>,
34    back_off_strategy: impl Fn(&E, usize) -> Option<Duration> + 'static,
35) -> Rc<dyn IObservable<T>> {
36    retry(source, Rc::new(back_off_strategy), 0)
37}
38
39/// Runs `action` every `interval` on `scheduler` until the result is disposed
40/// (`ScheduleRecurringAction(IScheduler, TimeSpan, Action)`).
41pub fn schedule_recurring_action(scheduler: &Rc<dyn IScheduler>, interval: Duration, action: impl Fn() + 'static) -> Rc<dyn IDisposable> {
42    scheduler.schedule_recursive_after(interval, move |schedule_next: &dyn Fn(Duration)| {
43        action();
44        schedule_next(interval);
45    })
46}
47
48/// Runs `action` after each interval that `interval` gives, on `scheduler`, until the result is
49/// disposed (`ScheduleRecurringAction(IScheduler, Func<TimeSpan>, Action)`).
50pub fn schedule_recurring_action_with(scheduler: &Rc<dyn IScheduler>, interval: impl Fn() -> Duration + 'static, action: impl Fn() + 'static) -> Rc<dyn IDisposable> {
51    let first = interval();
52    scheduler.schedule_recursive_after(first, move |schedule_next: &dyn Fn(Duration)| {
53        action();
54        schedule_next(interval());
55    })
56}
57
58/// The sequence with each value replaced by `()` (`ToUnit`).
59pub fn to_unit<T: 'static>(source: &Rc<dyn IObservable<T>>) -> Rc<dyn IObservable<()>> {
60    source.select(|_| ())
61}
62
63/// A sequence of the one value that `source` makes on each subscription, then the end
64/// (`InternalEx.Return(Func<T>)`).
65pub(crate) fn return_with<T: 'static>(source: impl Fn() -> T + 'static) -> Rc<dyn IObservable<T>> {
66    Observable::create(move |observer: Rc<dyn IObserver<T>>| {
67        observer.on_next(source());
68        observer.on_completed();
69        Disposable::empty()
70    })
71}
72
73/// The index of `value` in `list`, or the bitwise complement of where it belongs, by the
74/// algorithm of .NET's `List<T>.BinarySearch` (which equal item it finds matters to the
75/// operators that insert at the result): `Err(index)` for the latter.
76pub(crate) fn dotnet_binary_search<I, S>(list: &[I], value: &S, comparer: impl Fn(&I, &S) -> std::cmp::Ordering) -> Result<usize, usize> {
77    let (mut lower, mut upper) = (0isize, list.len() as isize - 1);
78    while lower <= upper {
79        let middle = lower + ((upper - lower) >> 1);
80        match comparer(&list[middle as usize], value) {
81            std::cmp::Ordering::Equal => return Ok(middle as usize),
82            std::cmp::Ordering::Less => lower = middle + 1,
83            std::cmp::Ordering::Greater => upper = middle - 1,
84        }
85    }
86    Err(lower as usize)
87}
ferroui_dynamic_data/kernel/internal_ex.rs

ferroui_dynamic_data/kernel/
internal_ex.rs
