ravenrs_app/
job.rs

1//! Work an action does away from the screen: it runs on a thread of its own, says how far it got,
2//! can be stopped, and hands back the records it changed, which the frame keeps on the screen's
3//! thread.
4
5use std::fmt;
6use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
7use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
8
9use crate::{ActionInput, MessageKind, Object};
10
11/// What a background action does: it reads what it was given, does its work and tells as it goes.
12/// An error ends it, in words for people.
13pub type JobWork = Arc<dyn Fn(&JobContext) -> Result<(), String> + Send + Sync>;
14
15/// Something that wakes the screen, from any thread.
16pub type JobWaker = Arc<dyn Fn() + Send + Sync>;
17
18/// A change a job hands back, kept by the frame on the screen's thread.
19#[derive(Clone, Debug, PartialEq)]
20#[non_exhaustive]
21#[cfg_attr(feature = "live", ravenrs_live::expose(part = "variants"))]
22pub enum JobChange {
23    /// Keep this record, new or changed.
24    Save(Object),
25    /// Remove this record.
26    Delete(Object),
27    /// Tell people this.
28    Tell(String, MessageKind),
29}
30
31#[derive(Default)]
32struct Shared {
33    /// How far it got, from 0 to 1, and what it is doing; nothing while it cannot tell.
34    progress: Mutex<(Option<f32>, String)>,
35    cancel: AtomicBool,
36    changes: Mutex<Vec<JobChange>>,
37    /// How it ended, once it did.
38    end: Mutex<Option<Result<(), String>>>,
39    waker: Mutex<Option<JobWaker>>,
40}
41
42fn locked<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
43    // A job that panicked still leaves what it handed over readable.
44    mutex.lock().unwrap_or_else(PoisonError::into_inner)
45}
46
47#[cfg_attr(feature = "live", ravenrs_live::expose)]
48impl Shared {
49    fn wake(&self) {
50        let waker = locked(&self.waker).clone();
51        if let Some(waker) = waker {
52            waker();
53        }
54    }
55}
56
57/// What a job sees while it runs, on its own thread.
58///
59/// ```
60/// use ravenrs_app::{JobContext, Object};
61///
62/// fn count(job: &JobContext) -> Result<(), String> {
63///     for (done, record) in job.selected().iter().enumerate() {
64///         if job.cancelled() {
65///             return Ok(());
66///         }
67///         job.progress((done + 1) as f32 / job.selected().len() as f32, record.entity());
68///         let mut record = record.clone();
69///         record.set("Counted", true);
70///         job.save(record);
71///     }
72///     Ok(())
73/// }
74/// # let _ = count;
75/// ```
76#[cfg_attr(feature = "live", ravenrs_live::expose(part = "fields"))]
77pub struct JobContext {
78    selected: Vec<Object>,
79    input: ActionInput,
80    shared: Arc<Shared>,
81}
82
83impl fmt::Debug for JobContext {
84    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
85        f.debug_struct("JobContext").field("selected", &self.selected.len()).finish_non_exhaustive()
86    }
87}
88
89#[cfg_attr(feature = "live", ravenrs_live::expose)]
90impl JobContext {
91    /// The records selected when it started, as they were then.
92    pub fn selected(&self) -> &[Object] {
93        &self.selected
94    }
95
96    /// What people gave the action.
97    pub fn input(&self) -> &ActionInput {
98        &self.input
99    }
100
101    /// How far it got, from 0 to 1, and what it is doing now.
102    pub fn progress(&self, done: f32, doing: impl Into<String>) {
103        *locked(&self.shared.progress) = (Some(done.clamp(0.0, 1.0)), doing.into());
104        self.shared.wake();
105    }
106
107    /// What it is doing, when it cannot tell how far it got.
108    pub fn doing(&self, doing: impl Into<String>) {
109        *locked(&self.shared.progress) = (None, doing.into());
110        self.shared.wake();
111    }
112
113    /// Whether someone asked it to stop. It stops by returning; what it handed over stays.
114    pub fn cancelled(&self) -> bool {
115        self.shared.cancel.load(Ordering::Relaxed)
116    }
117
118    /// Keeps this record, new or changed.
119    pub fn save(&self, record: Object) {
120        self.hand(JobChange::Save(record));
121    }
122
123    /// Removes this record.
124    pub fn delete(&self, record: Object) {
125        self.hand(JobChange::Delete(record));
126    }
127
128    /// Tells people something.
129    pub fn tell(&self, text: impl Into<String>, kind: MessageKind) {
130        self.hand(JobChange::Tell(text.into(), kind));
131    }
132
133    fn hand(&self, change: JobChange) {
134        locked(&self.shared.changes).push(change);
135        self.shared.wake();
136    }
137}
138
139static NEXT_JOB: AtomicU64 = AtomicU64::new(1);
140
141/// A job that started, as the screen sees it. A handle: the frame keeps one while it runs.
142#[derive(Clone)]
143#[cfg_attr(feature = "live", ravenrs_live::expose(part = "fields"))]
144pub struct Job {
145    id: u64,
146    action: String,
147    caption: String,
148    shared: Arc<Shared>,
149}
150
151impl fmt::Debug for Job {
152    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
153        f.debug_struct("Job").field("id", &self.id).field("action", &self.action).finish_non_exhaustive()
154    }
155}
156
157#[cfg_attr(feature = "live", ravenrs_live::expose)]
158impl Job {
159    /// Starts this work on a thread of its own (or right here, where there are no threads) and
160    /// wakes the screen with `waker` each time it has something to show.
161    pub(crate) fn start(
162        action: &str,
163        caption: &str,
164        work: JobWork,
165        selected: Vec<Object>,
166        input: ActionInput,
167        waker: Option<JobWaker>,
168    ) -> Self {
169        let shared = Arc::new(Shared { waker: Mutex::new(waker), ..Shared::default() });
170        let job = Self {
171            id: NEXT_JOB.fetch_add(1, Ordering::Relaxed),
172            action: action.to_owned(),
173            caption: caption.to_owned(),
174            shared: shared.clone(),
175        };
176        let cx = JobContext { selected, input, shared };
177        let run = move || {
178            // Ends it even when the work panics, so the screen never waits for nothing.
179            struct Ending(Arc<Shared>, Option<Result<(), String>>);
180            impl Drop for Ending {
181                fn drop(&mut self) {
182                    let end =
183                        self.1.take().unwrap_or_else(|| Err("the work stopped unexpectedly".to_owned()));
184                    *locked(&self.0.end) = Some(end);
185                    self.0.wake();
186                }
187            }
188            let mut ending = Ending(cx.shared.clone(), None);
189            ending.1 = Some(work(&cx));
190        };
191        #[cfg(not(target_arch = "wasm32"))]
192        {
193            let name = format!("job {}", job.action);
194            if let Err(error) = std::thread::Builder::new().name(name).spawn(run) {
195                *locked(&job.shared.end) = Some(Err(error.to_string()));
196            }
197        }
198        #[cfg(target_arch = "wasm32")]
199        run();
200        job
201    }
202
203    /// What it is known by, while the application runs.
204    pub fn id(&self) -> u64 {
205        self.id
206    }
207
208    /// The id of the action that started it.
209    pub fn action(&self) -> &str {
210        &self.action
211    }
212
213    /// What it says, as the action does.
214    pub fn caption(&self) -> &str {
215        &self.caption
216    }
217
218    /// How far it got, from 0 to 1 (nothing while it cannot tell), and what it is doing.
219    pub fn progress(&self) -> (Option<f32>, String) {
220        locked(&self.shared.progress).clone()
221    }
222
223    /// Asks it to stop.
224    pub fn cancel(&self) {
225        self.shared.cancel.store(true, Ordering::Relaxed);
226    }
227
228    /// Whether someone asked it to stop.
229    pub fn is_cancelled(&self) -> bool {
230        self.shared.cancel.load(Ordering::Relaxed)
231    }
232
233    /// Changes the waker, for the screen that shows it now.
234    pub(crate) fn set_waker(&self, waker: Option<JobWaker>) {
235        *locked(&self.shared.waker) = waker;
236    }
237
238    /// What it handed over since last asked.
239    pub(crate) fn take_changes(&self) -> Vec<JobChange> {
240        std::mem::take(&mut *locked(&self.shared.changes))
241    }
242
243    /// How it ended, once it did and everything it handed over was taken.
244    pub(crate) fn take_end(&self) -> Option<Result<(), String>> {
245        let end = locked(&self.shared.end).take()?;
246        // What it handed over just before ending comes first.
247        if !locked(&self.shared.changes).is_empty() {
248            *locked(&self.shared.end) = Some(end);
249            return None;
250        }
251        Some(end)
252    }
253}
ravenrs_app/job.rs

ravenrs_app/
job.rs
