ferroui_base/data/core/plugins/
task_stream_plugin.rs

1use super::IStreamPlugin;
2use crate::data::core::WeakValue;
3use crate::data::{BindingError, BindingErrorType, BindingNotification};
4use crate::reactive::{IObservable, LightweightSubject, Observable};
5use crate::threading::{Dispatcher, DispatcherPriority, FerroSynchronizationContext};
6use crate::{BoxedValue, PropertyValue};
7use std::cell::RefCell;
8use std::future::Future;
9use std::rc::Rc;
10
11#[derive(Clone)]
12enum TaskStatus {
13    Running,
14    RanToCompletion(Option<BoxedValue>),
15    Faulted(BindingError),
16}
17
18struct TaskShared {
19    status: RefCell<TaskStatus>,
20    continuations: RefCell<Vec<Box<dyn FnOnce()>>>,
21}
22
23impl TaskShared {
24    fn new(status: TaskStatus) -> Rc<Self> {
25        Rc::new(Self { status: RefCell::new(status), continuations: RefCell::new(Vec::new()) })
26    }
27
28    fn try_complete(&self, status: TaskStatus) -> bool {
29        if !matches!(*self.status.borrow(), TaskStatus::Running) {
30            return false;
31        }
32        self.status.replace(status);
33        let continuations = std::mem::take(&mut *self.continuations.borrow_mut());
34        for continuation in continuations {
35            continuation();
36        }
37        true
38    }
39}
40
41/// A task as a value that can be stored in a property and streamed by a
42/// binding with the `^` operator: the handle to the eventual outcome of an
43/// asynchronous operation, which is an untyped result or an error. It has
44/// identity equality.
45///
46/// The outcome is produced by a [`TaskValueSource`], by a future running on
47/// the dispatcher ([`TaskValue::run`]) or is known up front
48/// ([`TaskValue::from_result`], [`TaskValue::from_exception`]).
49#[derive(Clone)]
50pub struct TaskValue(Rc<TaskShared>);
51
52impl TaskValue {
53    /// A task that has completed with `value`.
54    pub fn from_result<T: PropertyValue>(value: T) -> Self {
55        Self::from_untyped_result(Some(Rc::new(value)))
56    }
57
58    /// A task that has completed with an untyped value: null, a boxed value
59    /// or a shared model object.
60    pub fn from_untyped_result(value: Option<BoxedValue>) -> Self {
61        Self(TaskShared::new(TaskStatus::RanToCompletion(value)))
62    }
63
64    /// A task that has failed with `error`.
65    pub fn from_exception(error: BindingError) -> Self {
66        Self(TaskShared::new(TaskStatus::Faulted(error)))
67    }
68
69    /// Runs `future` on the dispatcher of the calling thread and returns the
70    /// task for its outcome. A future that is dropped before it completes
71    /// (the dispatcher shut down, the future panicked) leaves the task
72    /// running forever.
73    pub fn run<F>(future: F) -> Self
74    where
75        F: Future<Output = Result<Option<BoxedValue>, BindingError>> + 'static,
76    {
77        let task = Self(TaskShared::new(TaskStatus::Running));
78        let shared = task.0.clone();
79        drop(Dispatcher::current_dispatcher().invoke_async_task_local(move || async move {
80            let status = match future.await {
81                Ok(value) => TaskStatus::RanToCompletion(value),
82                Err(error) => TaskStatus::Faulted(error),
83            };
84            shared.try_complete(status);
85        }));
86        task
87    }
88
89    /// Whether the task has finished: completed or failed.
90    pub fn is_completed(&self) -> bool {
91        !matches!(*self.0.status.borrow(), TaskStatus::Running)
92    }
93
94    /// Whether the task ran to completion.
95    pub fn is_completed_successfully(&self) -> bool {
96        matches!(*self.0.status.borrow(), TaskStatus::RanToCompletion(_))
97    }
98
99    /// Whether the task failed.
100    pub fn is_faulted(&self) -> bool {
101        matches!(*self.0.status.borrow(), TaskStatus::Faulted(_))
102    }
103
104    /// The outcome of the task, if it has finished.
105    pub fn result(&self) -> Option<Result<Option<BoxedValue>, BindingError>> {
106        match self.0.status.borrow().clone() {
107            TaskStatus::Running => None,
108            TaskStatus::RanToCompletion(value) => Some(Ok(value)),
109            TaskStatus::Faulted(error) => Some(Err(error)),
110        }
111    }
112
113    /// Runs `continuation` when the task finishes; right away if it has
114    /// finished already. The continuation runs where the task is completed:
115    /// a caller that needs it on the dispatcher posts from it.
116    pub fn continue_with(&self, continuation: impl FnOnce() + 'static) {
117        if self.is_completed() {
118            continuation();
119        } else {
120            self.0.continuations.borrow_mut().push(Box::new(continuation));
121        }
122    }
123}
124
125impl PartialEq for TaskValue {
126    fn eq(&self, other: &Self) -> bool {
127        Rc::ptr_eq(&self.0, &other.0)
128    }
129}
130
131/// The producer side of a [`TaskValue`] that is not bound to a future: the
132/// equivalent of a task completion source.
133pub struct TaskValueSource {
134    task: TaskValue,
135}
136
137impl Default for TaskValueSource {
138    fn default() -> Self {
139        Self::new()
140    }
141}
142
143impl TaskValueSource {
144    pub fn new() -> Self {
145        Self { task: TaskValue(TaskShared::new(TaskStatus::Running)) }
146    }
147
148    /// The task completed by this source.
149    pub fn task(&self) -> TaskValue {
150        self.task.clone()
151    }
152
153    /// Completes the task with `value`. Panics if the task has already
154    /// finished.
155    pub fn set_result<T: PropertyValue>(&self, value: T) {
156        self.set_untyped_result(Some(Rc::new(value)));
157    }
158
159    /// Completes the task with an untyped value. Panics if the task has
160    /// already finished.
161    pub fn set_untyped_result(&self, value: Option<BoxedValue>) {
162        if !self.try_set_untyped_result(value) {
163            panic!("An attempt was made to transition a task to a final state when it had already completed.");
164        }
165    }
166
167    /// Fails the task with `error`. Panics if the task has already finished.
168    pub fn set_exception(&self, error: BindingError) {
169        if !self.try_set_exception(error) {
170            panic!("An attempt was made to transition a task to a final state when it had already completed.");
171        }
172    }
173
174    /// Completes the task with an untyped value unless it has already
175    /// finished.
176    pub fn try_set_untyped_result(&self, value: Option<BoxedValue>) -> bool {
177        self.task.0.try_complete(TaskStatus::RanToCompletion(value))
178    }
179
180    /// Fails the task with `error` unless it has already finished.
181    pub fn try_set_exception(&self, error: BindingError) -> bool {
182        self.task.0.try_complete(TaskStatus::Faulted(error))
183    }
184}
185
186/// Handles binding to [`TaskValue`]s for the `^` stream binding operator.
187///
188/// A completed task produces its result, a failed one a binding error
189/// notification. The outcome of a task that is still running is delivered
190/// from a dispatcher job queued when the task finishes.
191pub struct TaskStreamPlugin;
192
193impl TaskStreamPlugin {
194    /// The stream of a finished task: its result, or a binding error
195    /// notification for a failed one. Panics if the task is still running.
196    pub fn handle_completed(task: &TaskValue) -> Rc<dyn IObservable<Option<BoxedValue>>> {
197        match task.result() {
198            Some(Ok(value)) => Observable::return_(value),
199            Some(Err(error)) => Observable::return_(Some(
200                Rc::new(BindingNotification::with_error(error, BindingErrorType::Error)) as BoxedValue,
201            )),
202            None => panic!("handle_completed called for non-completed task."),
203        }
204    }
205}
206
207impl IStreamPlugin for TaskStreamPlugin {
208    fn match_(&self, reference: &WeakValue) -> bool {
209        reference.upgrade().is_some_and(|t| t.is::<TaskValue>())
210    }
211
212    fn start(&self, reference: &WeakValue) -> Option<Rc<dyn IObservable<Option<BoxedValue>>>> {
213        let target = reference.upgrade();
214        let Some(task) = target.as_ref().and_then(|t| t.downcast_ref::<TaskValue>()) else {
215            return Some(Observable::empty());
216        };
217        Some(Self::observe(task))
218    }
219}
220
221impl TaskStreamPlugin {
222    /// The stream of a task: the outcome right away if the task has
223    /// finished, otherwise delivered from a dispatcher job queued when the
224    /// task finishes.
225    pub fn observe(task: &TaskValue) -> Rc<dyn IObservable<Option<BoxedValue>>> {
226        if task.is_completed() {
227            return Self::handle_completed(task);
228        }
229
230        // The continuation runs where the current synchronization context
231        // posts to: a job of this thread's dispatcher.
232        let subject = LightweightSubject::<Option<BoxedValue>>::new();
233        let dispatcher = Dispatcher::current_dispatcher();
234        let priority = FerroSynchronizationContext::current().map_or(DispatcherPriority::NORMAL, |c| c.priority());
235        let completed = task.clone();
236        let observer = subject.clone();
237        task.continue_with(move || {
238            dispatcher.clone().post_local(
239                move || {
240                    Self::handle_completed(&completed).subscribe(Rc::new(observer));
241                },
242                priority,
243            );
244        });
245        Rc::new(subject)
246    }
247}
ferroui_base/data/core/plugins/task_stream_plugin.rs

ferroui_base/data/core/plugins/
task_stream_plugin.rs
