ferroui_reactive/linq/
task_observable.rs

1//! `TaskObservableExtensions` (`Threading/Tasks/TaskObservableExtensions.cs`): a task as a
2//! sequence and a sequence as a task; `StartAsync` and `FromAsync` (`Linq/QueryLanguage.Async.cs`),
3//! which run a future as a task (docs/porting/reactive.md, section 5.8).
4//!
5//! Deviation (DEVIATIONS.md, Reactive): upstream's functions return a `Task<T>`; here they return
6//! a future of `Result<T, ObservableError>`, which runs as a task of the thread
7//! (`Task::spawn_local`). A `Task<T>` is such a future. The forms of a `Task` without a result are
8//! the same methods with `T = ()`. `TaskObservationOptions` reduces to the scheduler (`_on`):
9//! `IgnoreExceptionsAfterUnsubscribe` concerns the unobserved exceptions of the task scheduler of
10//! .NET, which tasks here do not have.
11
12use crate::concurrency::{IScheduler, ImmediateScheduler, SchedulerExt};
13use crate::disposables::{CancellationDisposable, SingleAssignmentDisposableValue, StableCompositeDisposable};
14use crate::linq::{Observable, SequenceError};
15use crate::ObservableBase;
16use ferroui_system::threading::tasks::{Task, TaskCompletionSource};
17use ferroui_system::threading::{CancellationToken, LocalCancellationTokenRegistration};
18use ferroui_system::{IDisposable, IObservable, IObserver, ObservableError};
19use std::cell::{Cell, RefCell};
20use std::future::Future;
21use std::rc::Rc;
22
23/// The outcome of an ended task, to its observer (`EmitTaskResult`).
24fn emit_task_result<T: Clone + 'static>(task: &Task<T>, observer: &Rc<dyn IObserver<T>>) {
25    match task.result() {
26        Ok(value) => {
27            observer.on_next(value);
28            observer.on_completed();
29        }
30        Err(error) => observer.on_error(error),
31    }
32}
33
34/// A task that has not ended, as a sequence (`SlowTaskObservable<TResult>`): each subscription
35/// continues the task, on `scheduler` when there is one.
36struct SlowTaskObservable<T> {
37    task: Task<T>,
38    scheduler: Option<Rc<dyn IScheduler>>,
39}
40
41/// Removes the continuation of a subscription (the `CancellationDisposable` of upstream, whose
42/// token cancels the continuation).
43struct RemoveContinuation<T> {
44    task: Task<T>,
45    id: Cell<Option<ferroui_system::threading::tasks::ContinuationId>>,
46}
47
48impl<T: 'static> IDisposable for RemoveContinuation<T> {
49    fn dispose(&self) {
50        if let Some(id) = self.id.take() {
51            self.task.remove_continuation(id);
52        }
53    }
54}
55
56impl<T: Clone + 'static> IObservable<T> for SlowTaskObservable<T> {
57    fn subscribe(&self, observer: Rc<dyn IObserver<T>>) -> Rc<dyn IDisposable> {
58        let id = match &self.scheduler {
59            None => self.task.continue_with(move |task| emit_task_result(task, &observer)),
60            Some(scheduler) => {
61                let scheduler = scheduler.clone();
62                self.task.continue_with(move |task| {
63                    let task = task.clone();
64                    scheduler.schedule_action(move || emit_task_result(&task, &observer));
65                })
66            }
67        };
68        Rc::new(RemoveContinuation { task: self.task.clone(), id: Cell::new(id) })
69    }
70}
71
72/// `ToObservableImpl`: an ended task is `Return` or `Throw` on the scheduler (immediate by
73/// default), another one a [`SlowTaskObservable`].
74pub(crate) fn task_to_observable<T: Clone + 'static>(task: Task<T>, scheduler: Option<Rc<dyn IScheduler>>) -> Rc<dyn IObservable<T>> {
75    if task.is_completed() {
76        let scheduler = scheduler.unwrap_or_else(ImmediateScheduler::instance);
77        return match task.result() {
78            Ok(value) => Observable::return_on(value, scheduler),
79            Err(error) => Observable::throw_on(error, scheduler),
80        };
81    }
82    Rc::new(SlowTaskObservable { task, scheduler })
83}
84
85/// A task as a sequence (`TaskObservableExtensions.ToObservable`).
86pub trait TaskObservableExt<T> {
87    /// A sequence of the result of the task, or of its error; a canceled task ends it with a
88    /// `TaskCanceledError` (`ToObservable(Task<TResult>)`, `ToObservable(Task)` for `T = ()`).
89    fn to_observable(&self) -> Rc<dyn IObservable<T>>;
90
91    /// [`to_observable`](Self::to_observable) with the notifications on `scheduler`
92    /// (`ToObservable(Task<TResult>, IScheduler)`, `ToObservable(Task<TResult>, TaskObservationOptions)`).
93    fn to_observable_on(&self, scheduler: Rc<dyn IScheduler>) -> Rc<dyn IObservable<T>>;
94}
95
96impl<T: Clone + 'static> TaskObservableExt<T> for Task<T> {
97    fn to_observable(&self) -> Rc<dyn IObservable<T>> {
98        task_to_observable(self.clone(), None)
99    }
100
101    fn to_observable_on(&self, scheduler: Rc<dyn IScheduler>) -> Rc<dyn IObservable<T>> {
102        task_to_observable(self.clone(), Some(scheduler))
103    }
104}
105
106/// The observer of `ToTask` (`ToTaskObserver<TResult>`): the last value ends the task when the
107/// sequence ends, an error faults it, the cancellation of the token cancels it.
108struct ToTaskObserver<T> {
109    source: TaskCompletionSource<T>,
110    registration: RefCell<Option<LocalCancellationTokenRegistration>>,
111    last_value: RefCell<Option<T>>,
112    resource: SingleAssignmentDisposableValue,
113    is_stopped: Cell<bool>,
114}
115
116impl<T: 'static> ToTaskObserver<T> {
117    fn dispose_registration(&self) {
118        let registration = self.registration.borrow_mut().take();
119        if let Some(registration) = registration {
120            registration.dispose();
121        }
122    }
123
124    /// `Cancel`.
125    fn cancel(&self) {
126        self.resource.dispose();
127        self.source.try_set_canceled();
128    }
129}
130
131impl<T: 'static> IObserver<T> for ToTaskObserver<T> {
132    fn on_next(&self, value: T) {
133        if self.is_stopped.get() {
134            return;
135        }
136        *self.last_value.borrow_mut() = Some(value);
137    }
138
139    fn on_error(&self, error: ObservableError) {
140        if self.is_stopped.replace(true) {
141            return;
142        }
143        self.source.try_set_exception(error);
144        self.dispose_registration();
145        self.resource.dispose();
146    }
147
148    fn on_completed(&self) {
149        if self.is_stopped.replace(true) {
150            return;
151        }
152        let last = self.last_value.borrow_mut().take();
153        match last {
154            Some(value) => {
155                self.source.try_set_result(value);
156            }
157            None => {
158                self.source.try_set_exception(Rc::new(SequenceError::NoElements));
159            }
160        }
161        self.dispose_registration();
162        self.resource.dispose();
163    }
164}
165
166/// `ToTask(IObservable<TResult>, CancellationToken)`.
167pub(crate) fn to_task<T: 'static>(source: Rc<dyn IObservable<T>>, token: CancellationToken) -> Task<T> {
168    let completion = TaskCompletionSource::new();
169    let observer = Rc::new(ToTaskObserver {
170        source: completion.clone(),
171        registration: RefCell::new(None),
172        last_value: RefCell::new(None),
173        resource: SingleAssignmentDisposableValue::new(),
174        is_stopped: Cell::new(false),
175    });
176    if token.can_be_canceled() {
177        // The registration holds the observer, as upstream's does: a source that keeps no
178        // reference to its observer (`Never`) still ends the task when the token is canceled. The
179        // end of the sequence disposes the registration.
180        let target = observer.clone();
181        let registration = token.register_local(move || target.cancel());
182        *observer.registration.borrow_mut() = Some(registration);
183    }
184    let subscription = source.subscribe(observer.clone());
185    observer.resource.set_disposable(Some(subscription));
186    completion.task()
187}
188
189/// `ContinueOnScheduler`: a task that ends as `task` does, on `scheduler`.
190pub(crate) fn continue_on_scheduler<T: Clone + 'static>(task: Task<T>, scheduler: Rc<dyn IScheduler>) -> Task<T> {
191    let completion = TaskCompletionSource::new();
192    let result = completion.task();
193    task.continue_with(move |task| {
194        let task = task.clone();
195        scheduler.schedule_action(move || {
196            if task.is_canceled() {
197                completion.try_set_canceled();
198            } else {
199                match task.result() {
200                    Ok(value) => {
201                        completion.try_set_result(value);
202                    }
203                    Err(error) => {
204                        completion.try_set_exception(error);
205                    }
206                }
207            }
208        });
209    });
210    result
211}
212
213/// `StartAsyncImpl(Func<Task<TSource>>, options)`.
214pub(crate) fn start_async<T: Clone + 'static, Fut: Future<Output = Result<T, ObservableError>> + 'static>(
215    future: Fut,
216    scheduler: Option<Rc<dyn IScheduler>>,
217) -> Rc<dyn IObservable<T>> {
218    task_to_observable(Task::spawn_local(future), scheduler)
219}
220
221/// `StartAsyncImpl(Func<CancellationToken, Task<TSource>>, options)`: the subscription also cancels
222/// the token given to the function (`StartAsyncObservable<TSource>`).
223pub(crate) fn start_async_with_token<T: Clone + 'static, Fut: Future<Output = Result<T, ObservableError>> + 'static>(
224    function_async: impl FnOnce(CancellationToken) -> Fut,
225    scheduler: Option<Rc<dyn IScheduler>>,
226) -> Rc<dyn IObservable<T>> {
227    let cancellable = CancellationDisposable::new();
228    let result = task_to_observable(Task::spawn_local(function_async(cancellable.token())), scheduler);
229    Rc::new(ObservableBase::from_fn(move |observer: Rc<dyn IObserver<T>>| {
230        let subscription = result.subscribe(observer);
231        let cancellable: Rc<dyn IDisposable> = cancellable.clone();
232        let composite: Rc<dyn IDisposable> = StableCompositeDisposable::create(cancellable, subscription);
233        composite
234    }))
235}
236
237impl Observable {
238    /// Runs the future the function gives at once, as a task, and gives a sequence of its outcome:
239    /// a hot sequence, shared by its subscriptions (`StartAsync(Func<Task<TResult>>)`,
240    /// `StartAsync(Func<Task>)` for `T = ()`).
241    pub fn start_async<T: Clone + 'static, Fut: Future<Output = Result<T, ObservableError>> + 'static>(
242        function_async: impl FnOnce() -> Fut,
243    ) -> Rc<dyn IObservable<T>> {
244        start_async(function_async(), None)
245    }
246
247    /// [`start_async`](Self::start_async) with the notifications on `scheduler`
248    /// (`StartAsync(Func<Task<TResult>>, TaskObservationOptions)`).
249    pub fn start_async_on<T: Clone + 'static, Fut: Future<Output = Result<T, ObservableError>> + 'static>(
250        function_async: impl FnOnce() -> Fut,
251        scheduler: Rc<dyn IScheduler>,
252    ) -> Rc<dyn IObservable<T>> {
253        start_async(function_async(), Some(scheduler))
254    }
255
256    /// [`start_async`](Self::start_async) with a cancellation token, which disposing a subscription
257    /// cancels (`StartAsync(Func<CancellationToken, Task<TResult>>)`).
258    pub fn start_async_with_token<T: Clone + 'static, Fut: Future<Output = Result<T, ObservableError>> + 'static>(
259        function_async: impl FnOnce(CancellationToken) -> Fut,
260    ) -> Rc<dyn IObservable<T>> {
261        start_async_with_token(function_async, None)
262    }
263
264    /// [`start_async_with_token`](Self::start_async_with_token) with the notifications on
265    /// `scheduler` (`StartAsync(Func<CancellationToken, Task<TResult>>, TaskObservationOptions)`).
266    pub fn start_async_with_token_on<T: Clone + 'static, Fut: Future<Output = Result<T, ObservableError>> + 'static>(
267        function_async: impl FnOnce(CancellationToken) -> Fut,
268        scheduler: Rc<dyn IScheduler>,
269    ) -> Rc<dyn IObservable<T>> {
270        start_async_with_token(function_async, Some(scheduler))
271    }
272
273    /// Runs the future the function gives for each subscription, as a task, and gives a sequence
274    /// of its outcome (`FromAsync(Func<Task<TResult>>)`, `FromAsync(Func<Task>)` for `T = ()`):
275    /// `Defer` of `StartAsync`.
276    pub fn from_async<T: Clone + 'static, Fut: Future<Output = Result<T, ObservableError>> + 'static>(
277        function_async: impl Fn() -> Fut + 'static,
278    ) -> Rc<dyn IObservable<T>> {
279        Observable::defer(move || start_async(function_async(), None))
280    }
281
282    /// [`from_async`](Self::from_async) with the notifications on `scheduler`
283    /// (`FromAsync(Func<Task<TResult>>, TaskObservationOptions)`).
284    pub fn from_async_on<T: Clone + 'static, Fut: Future<Output = Result<T, ObservableError>> + 'static>(
285        function_async: impl Fn() -> Fut + 'static,
286        scheduler: Rc<dyn IScheduler>,
287    ) -> Rc<dyn IObservable<T>> {
288        Observable::defer(move || start_async(function_async(), Some(scheduler.clone())))
289    }
290
291    /// [`from_async`](Self::from_async) with a cancellation token, which disposing the
292    /// subscription cancels (`FromAsync(Func<CancellationToken, Task<TResult>>)`).
293    pub fn from_async_with_token<T: Clone + 'static, Fut: Future<Output = Result<T, ObservableError>> + 'static>(
294        function_async: impl Fn(CancellationToken) -> Fut + 'static,
295    ) -> Rc<dyn IObservable<T>> {
296        Observable::defer(move || start_async_with_token(&function_async, None))
297    }
298
299    /// [`from_async_with_token`](Self::from_async_with_token) with the notifications on
300    /// `scheduler` (`FromAsync(Func<CancellationToken, Task<TResult>>, TaskObservationOptions)`).
301    pub fn from_async_with_token_on<T: Clone + 'static, Fut: Future<Output = Result<T, ObservableError>> + 'static>(
302        function_async: impl Fn(CancellationToken) -> Fut + 'static,
303        scheduler: Rc<dyn IScheduler>,
304    ) -> Rc<dyn IObservable<T>> {
305        Observable::defer(move || start_async_with_token(&function_async, Some(scheduler.clone())))
306    }
307}
308
309macro_rules! task_observable_methods {
310    () => {
311        /// A task that ends with the last value of the sequence, or its error; a sequence without
312        /// values faults it with `SequenceError::NoElements` (`ToTask(IObservable<TResult>)`).
313        fn to_task(&self) -> ferroui_system::threading::tasks::Task<T> {
314            crate::linq::task_observable::to_task(self.observable(), ferroui_system::threading::CancellationToken::none())
315        }
316
317        /// [`to_task`](Self::to_task) that the cancellation of `token` cancels, disposing the
318        /// subscription (`ToTask(IObservable<TResult>, CancellationToken)`).
319        fn to_task_with_token(&self, token: ferroui_system::threading::CancellationToken) -> ferroui_system::threading::tasks::Task<T> {
320            crate::linq::task_observable::to_task(self.observable(), token)
321        }
322
323        /// [`to_task`](Self::to_task) whose task ends on `scheduler`
324        /// (`ToTask(IObservable<TResult>, IScheduler)`).
325        fn to_task_on(&self, scheduler: std::rc::Rc<dyn crate::concurrency::IScheduler>) -> ferroui_system::threading::tasks::Task<T>
326        where
327            T: Clone,
328        {
329            let task = crate::linq::task_observable::to_task(self.observable(), ferroui_system::threading::CancellationToken::none());
330            crate::linq::task_observable::continue_on_scheduler(task, scheduler)
331        }
332
333        /// [`to_task_with_token`](Self::to_task_with_token) whose task ends on `scheduler`
334        /// (`ToTask(IObservable<TResult>, CancellationToken, IScheduler)`).
335        fn to_task_with_token_on(
336            &self,
337            token: ferroui_system::threading::CancellationToken,
338            scheduler: std::rc::Rc<dyn crate::concurrency::IScheduler>,
339        ) -> ferroui_system::threading::tasks::Task<T>
340        where
341            T: Clone,
342        {
343            let task = crate::linq::task_observable::to_task(self.observable(), token);
344            crate::linq::task_observable::continue_on_scheduler(task, scheduler)
345        }
346    };
347}
348pub(crate) use task_observable_methods;
ferroui_reactive/linq/task_observable.rs

ferroui_reactive/linq/
task_observable.rs
