ferroui_remote_protocol/
task.rs

1//! The completion of a send: the port's counterpart of the `Task` the
2//! upstream `Send` returns and of the `TaskCompletionSource` behind it.
3//!
4//! The library has no asynchronous runtime (see the threading contract in
5//! `i_transport.rs`): a task is a result that another thread may still be
6//! producing, and waiting for it blocks the calling thread.
7
8use std::sync::{Arc, Condvar, Mutex};
9use std::time::Duration;
10
11use crate::error::Error;
12
13struct TaskState {
14    result: Mutex<Option<Result<(), Error>>>,
15    completed: Condvar,
16}
17
18/// The result of an operation that completes on another thread. A clone is
19/// another handle of the same operation.
20#[derive(Clone)]
21pub struct Task {
22    state: Arc<TaskState>,
23}
24
25impl Task {
26    /// `Task.CompletedTask`.
27    pub fn completed() -> Task {
28        Task::from_result(Ok(()))
29    }
30
31    /// A task that is already completed with the given result.
32    pub fn from_result(result: Result<(), Error>) -> Task {
33        Task { state: Arc::new(TaskState { result: Mutex::new(Some(result)), completed: Condvar::new() }) }
34    }
35
36    /// `Task.IsCompleted`.
37    pub fn is_completed(&self) -> bool {
38        self.state.result.lock().unwrap().is_some()
39    }
40
41    /// `Task.Wait()`: blocks until the operation is finished and returns its
42    /// result.
43    pub fn wait(&self) -> Result<(), Error> {
44        let mut result = self.state.result.lock().unwrap();
45        loop {
46            if let Some(result) = result.as_ref() {
47                return result.clone();
48            }
49            result = self.state.completed.wait(result).unwrap();
50        }
51    }
52
53    /// `Task.Wait(timeout)`: `None` when the operation is not finished after
54    /// the given time.
55    pub fn wait_timeout(&self, timeout: Duration) -> Option<Result<(), Error>> {
56        let result = self.state.result.lock().unwrap();
57        let (result, _) = self.state.completed.wait_timeout_while(result, timeout, |result| result.is_none()).unwrap();
58        result.clone()
59    }
60}
61
62/// `TaskCompletionSource<int>` of the upstream code: the producing side of a
63/// [`Task`].
64#[derive(Clone)]
65pub struct TaskCompletionSource {
66    task: Task,
67}
68
69impl TaskCompletionSource {
70    pub fn new() -> TaskCompletionSource {
71        TaskCompletionSource {
72            task: Task { state: Arc::new(TaskState { result: Mutex::new(None), completed: Condvar::new() }) },
73        }
74    }
75
76    /// `TaskCompletionSource.Task`.
77    pub fn task(&self) -> Task {
78        self.task.clone()
79    }
80
81    /// `TrySetResult`: false when the task was already completed.
82    pub fn try_set_result(&self) -> bool {
83        self.try_complete(Ok(()))
84    }
85
86    /// `TrySetException`: false when the task was already completed.
87    pub fn try_set_exception(&self, error: Error) -> bool {
88        self.try_complete(Err(error))
89    }
90
91    fn try_complete(&self, value: Result<(), Error>) -> bool {
92        let mut result = self.task.state.result.lock().unwrap();
93        if result.is_some() {
94            return false;
95        }
96        *result = Some(value);
97        self.task.state.completed.notify_all();
98        true
99    }
100}
101
102impl Default for TaskCompletionSource {
103    fn default() -> Self {
104        TaskCompletionSource::new()
105    }
106}
107
108// Tests of the port: the upstream types are the ones of the runtime library.
109#[cfg(test)]
110mod tests {
111    use super::*;
112
113    #[test]
114    fn a_completed_task_returns_its_result() {
115        assert!(Task::completed().is_completed());
116        assert!(Task::completed().wait().is_ok());
117        assert!(matches!(Task::from_result(Err(Error::EndOfStream)).wait(), Err(Error::EndOfStream)));
118    }
119
120    #[test]
121    fn a_task_completes_once() {
122        let source = TaskCompletionSource::new();
123        let task = source.task();
124        assert!(!task.is_completed());
125        assert!(task.wait_timeout(Duration::from_millis(10)).is_none());
126        assert!(source.try_set_result());
127        assert!(!source.try_set_exception(Error::EndOfStream));
128        assert!(task.wait().is_ok());
129    }
130
131    #[test]
132    fn waiting_ends_when_another_thread_completes_the_task() {
133        let source = TaskCompletionSource::new();
134        let task = source.task();
135        let thread = std::thread::spawn(move || {
136            std::thread::sleep(Duration::from_millis(20));
137            source.try_set_exception(Error::EndOfStream);
138        });
139        assert!(matches!(task.wait(), Err(Error::EndOfStream)));
140        thread.join().unwrap();
141    }
142}
ferroui_remote_protocol/task.rs

ferroui_remote_protocol/
task.rs
