orinium_browser/engine/port/
mailbox.rs

1//! The two ends of a one-way message channel.
2//!
3//! [`Outbox`] is where messages are posted, [`Mailbox`] is where they are
4//! collected. Both hide how the bytes actually travel, so a caller written
5//! against them keeps working when the channel moves from this address space
6//! to another process — see [`crate::engine::port::transport`].
7
8use std::sync::Arc;
9
10/// A message that never reached the far end of a channel.
11///
12/// The message is handed back rather than dropped: the producer may be the only
13/// holder of it, and silently discarding work is worse than reporting that the
14/// consumer is gone.
15pub struct Undeliverable<T>(pub T);
16
17// Spelled out rather than derived: the payload is deliberately not required to
18// be `Debug`, since reporting the failure must not depend on the message type.
19impl<T> std::fmt::Debug for Undeliverable<T> {
20    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
21        f.write_str("Undeliverable(..)")
22    }
23}
24
25impl<T: PartialEq> PartialEq for Undeliverable<T> {
26    fn eq(&self, other: &Self) -> bool {
27        self.0 == other.0
28    }
29}
30
31impl<T: Eq> Eq for Undeliverable<T> {}
32
33impl<T> std::fmt::Display for Undeliverable<T> {
34    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
35        f.write_str("the receiving end of the channel is gone")
36    }
37}
38
39impl<T> std::error::Error for Undeliverable<T> {}
40
41/// The far end of a channel disappeared before it could be read.
42#[derive(Debug, Clone, Copy, PartialEq, Eq)]
43pub struct Disconnected;
44
45impl std::fmt::Display for Disconnected {
46    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
47        f.write_str("the sending end of the channel is gone")
48    }
49}
50
51impl std::error::Error for Disconnected {}
52
53/// Posts messages into a channel.
54///
55/// Cheap to share: several producers may hold a clone, and `send` takes `&self`
56/// so no producer needs unique access to post work.
57pub struct Outbox<T> {
58    inner: Arc<dyn Produce<T> + Send + Sync>,
59}
60
61impl<T: Send + 'static> Outbox<T> {
62    /// Wraps a transport-specific producer.
63    ///
64    /// Transports build ends through this; everyone else gets them from
65    /// [`Transport::channel`](crate::engine::port::transport::Transport::channel).
66    pub fn of<P: Produce<T> + Send + Sync + 'static>(producer: P) -> Self {
67        Self {
68            inner: Arc::new(producer),
69        }
70    }
71
72    /// Posts `message`, or hands it back if no consumer remains.
73    ///
74    /// Closing is reported by the send failing, so a producer learns about it
75    /// without a separate query.
76    pub fn send(&self, message: T) -> Result<(), Undeliverable<T>> {
77        self.inner.send(message)
78    }
79}
80
81impl<T> std::fmt::Debug for Outbox<T> {
82    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
83        f.write_str("Outbox")
84    }
85}
86
87impl<T> Clone for Outbox<T> {
88    fn clone(&self) -> Self {
89        Self {
90            inner: Arc::clone(&self.inner),
91        }
92    }
93}
94
95/// Collects messages from a channel.
96///
97/// `recv` and `try_recv` both take `&self`, so a mailbox can be held behind a
98/// shared reference and moved between threads. One consumer at a time is still
99/// the rule: a transport whose consumer parks in `recv` may hold an internal
100/// lock for the duration.
101pub struct Mailbox<T> {
102    inner: Arc<dyn Consume<T> + Send + Sync>,
103}
104
105impl<T: Send + 'static> Mailbox<T> {
106    /// Wraps a transport-specific consumer.
107    ///
108    /// Transports build ends through this; everyone else gets them from
109    /// [`Transport::channel`](crate::engine::port::transport::Transport::channel).
110    pub fn of<C: Consume<T> + Send + Sync + 'static>(consumer: C) -> Self {
111        Self {
112            inner: Arc::new(consumer),
113        }
114    }
115
116    /// Waits for the next message, parking rather than spinning.
117    ///
118    /// `Err(Disconnected)` means every producer is gone and the queue is
119    /// drained — the only unambiguous way for a consumer to learn the channel
120    /// is closed.
121    pub fn recv(&self) -> Result<T, Disconnected> {
122        self.inner.recv()
123    }
124
125    /// Takes the next message if one is already waiting, without blocking.
126    ///
127    /// The UI thread polls here rather than calling [`Mailbox::recv`], because
128    /// it has to keep servicing the event loop between messages.
129    pub fn try_recv(&self) -> Option<T> {
130        self.inner.try_recv()
131    }
132}
133
134impl<T> std::fmt::Debug for Mailbox<T> {
135    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
136        f.write_str("Mailbox")
137    }
138}
139
140impl<T> Clone for Mailbox<T> {
141    fn clone(&self) -> Self {
142        Self {
143            inner: Arc::clone(&self.inner),
144        }
145    }
146}
147
148/// A transport's producing half.
149pub trait Produce<T>: Send + Sync {
150    fn send(&self, message: T) -> Result<(), Undeliverable<T>>;
151}
152
153/// A transport's consuming half.
154///
155/// There is no `is_disconnected` here on purpose: a poll that finds an empty
156/// queue cannot tell "nothing yet" from "nothing ever", and answering wrongly
157/// either way would let a caller discard live work. A consumer learns the
158/// channel is closed from [`Consume::recv`] returning [`Disconnected`].
159pub trait Consume<T>: Send + Sync {
160    fn recv(&self) -> Result<T, Disconnected>;
161    fn try_recv(&self) -> Option<T>;
162}
163
164#[cfg(test)]
165mod tests {
166    use super::*;
167    use crate::engine::port::transport::{InProcess, Transport};
168    use crate::test_support::{assert_send, assert_sync};
169
170    /// The protocol tests run against the production transport, so they cover
171    /// the two ends against one another rather than a stand-in that could
172    /// quietly disagree with it.
173    fn pair() -> (Outbox<u32>, Mailbox<u32>) {
174        InProcess.channel()
175    }
176
177    #[test]
178    fn messages_arrive_in_the_order_they_were_posted() {
179        let (outbox, mailbox) = pair();
180        for value in 0..3 {
181            outbox.send(value).expect("consumer is present");
182        }
183        let received: Vec<u32> = std::iter::from_fn(|| mailbox.try_recv()).collect();
184        assert_eq!(received, [0, 1, 2]);
185    }
186
187    /// The poll path must never block: a UI thread that calls it on an idle
188    /// channel has to get `None` and carry on with the event loop.
189    #[test]
190    fn try_recv_on_an_idle_channel_reports_nothing() {
191        let (_outbox, mailbox) = pair();
192        assert!(mailbox.try_recv().is_none());
193    }
194
195    #[test]
196    fn recv_waits_for_a_message_that_has_not_been_posted_yet() {
197        let (outbox, mailbox) = pair();
198        let producer = std::thread::spawn(move || {
199            outbox.send(7).expect("consumer is present");
200        });
201        assert_eq!(mailbox.recv().expect("producer is alive"), 7);
202        producer.join().expect("producer thread finished");
203    }
204
205    #[test]
206    fn dropping_the_outbox_drains_before_reporting_disconnection() {
207        let (outbox, mailbox) = pair();
208        outbox.send(1).expect("consumer is present");
209        drop(outbox);
210        assert_eq!(mailbox.recv().expect("a message is still queued"), 1);
211        assert_eq!(mailbox.recv(), Err(Disconnected));
212    }
213
214    #[test]
215    fn sending_to_a_dropped_consumer_hands_the_message_back() {
216        let (outbox, mailbox) = pair();
217        drop(mailbox);
218        assert_eq!(outbox.send(3), Err(Undeliverable(3)));
219    }
220
221    /// Several producers may post into one mailbox, which is what lets the
222    /// network core, the immediate-scheme path and a worker all feed the same
223    /// queue without coordinating first.
224    #[test]
225    fn every_clone_of_an_outbox_reaches_the_same_mailbox() {
226        let (outbox, mailbox) = pair();
227        let mut producers = Vec::new();
228        for value in 0..4u32 {
229            let outbox = outbox.clone();
230            producers.push(std::thread::spawn(move || {
231                outbox.send(value).expect("consumer is present");
232            }));
233        }
234        for producer in producers {
235            producer.join().expect("producer thread finished");
236        }
237        let mut received: Vec<u32> = std::iter::from_fn(|| mailbox.try_recv()).collect();
238        received.sort_unstable();
239        assert_eq!(received, [0, 1, 2, 3]);
240    }
241
242    /// Both ends are shared across threads by design, so this bound is part of
243    /// the contract rather than an implementation detail.
244    #[test]
245    fn ends_are_shareable_across_threads() {
246        assert_send::<Outbox<u32>>();
247        assert_sync::<Outbox<u32>>();
248        assert_send::<Mailbox<u32>>();
249        assert_sync::<Mailbox<u32>>();
250    }
251}
orinium_browser/engine/port/mailbox.rs

orinium_browser/engine/port/
mailbox.rs
