cranpose_services/
inbox.rs

1//! A process-wide queue between platform hosts and compositions.
2//!
3//! A host publishes from whatever thread the platform called it on, and
4//! every registered observer gets the item. An item published while no one
5//! observes waits in a backlog, which the next observer to register
6//! receives, so what arrives during launch reaches the first screen that
7//! listens.
8
9use std::{
10    collections::VecDeque,
11    sync::{Arc, Mutex, PoisonError},
12};
13
14type Observer<T> = Arc<dyn Fn(T) + Send + Sync>;
15
16struct State<T> {
17    observers: Vec<(u64, Observer<T>)>,
18    backlog: VecDeque<T>,
19    next_id: u64,
20}
21
22/// One kind of item that platform hosts hand to the application.
23pub(crate) struct Inbox<T> {
24    /// What the log calls the items.
25    name: &'static str,
26    state: Mutex<State<T>>,
27}
28
29impl<T: Clone + Send + 'static> Inbox<T> {
30    pub(crate) const fn new(name: &'static str) -> Self {
31        Self {
32            name,
33            state: Mutex::new(State {
34                observers: Vec::new(),
35                backlog: VecDeque::new(),
36                next_id: 0,
37            }),
38        }
39    }
40
41    fn state(&self) -> std::sync::MutexGuard<'_, State<T>> {
42        self.state.lock().unwrap_or_else(PoisonError::into_inner)
43    }
44
45    /// Registers `observer` and hands it the backlog.
46    pub(crate) fn observe(
47        &'static self,
48        observer: impl Fn(T) + Send + Sync + 'static,
49    ) -> InboxObserver {
50        let observer: Observer<T> = Arc::new(observer);
51        let (id, replay) = {
52            let mut state = self.state();
53            let id = state.next_id;
54            state.next_id += 1;
55            state.observers.push((id, Arc::clone(&observer)));
56            (id, std::mem::take(&mut state.backlog))
57        };
58        log::info!(
59            "{}: observer {id} registered, replays {} item(s)",
60            self.name,
61            replay.len()
62        );
63        for item in replay {
64            observer(item);
65        }
66        InboxObserver { id, inbox: self }
67    }
68
69    /// Hands `item` to every observer, or keeps it for the next one.
70    pub(crate) fn publish(&self, item: T) {
71        let observers: Vec<Observer<T>> = {
72            let mut state = self.state();
73            if state.observers.is_empty() {
74                state.backlog.push_back(item);
75                log::info!("{}: no observer yet, the item waits", self.name);
76                return;
77            }
78            state
79                .observers
80                .iter()
81                .map(|(_, observer)| Arc::clone(observer))
82                .collect()
83        };
84        log::info!(
85            "{}: an item goes to {} observer(s)",
86            self.name,
87            observers.len()
88        );
89        for observer in observers {
90            observer(item.clone());
91        }
92    }
93
94    /// Drops the backlog.
95    pub(crate) fn clear(&self) {
96        self.state().backlog.clear();
97    }
98}
99
100trait Unobserve: Sync {
101    fn unobserve(&self, id: u64);
102}
103
104impl<T: Send> Unobserve for Inbox<T> {
105    fn unobserve(&self, id: u64) {
106        self.state
107            .lock()
108            .unwrap_or_else(PoisonError::into_inner)
109            .observers
110            .retain(|(existing, _)| *existing != id);
111    }
112}
113
114/// Keeps an observer of a platform inbox, such as incoming content or opened
115/// links, registered until it is dropped.
116pub struct InboxObserver {
117    id: u64,
118    inbox: &'static dyn Unobserve,
119}
120
121impl Drop for InboxObserver {
122    fn drop(&mut self) {
123        self.inbox.unobserve(self.id);
124    }
125}
cranpose_services/inbox.rs

cranpose_services/
inbox.rs
