orinium_browser/engine/port/
transport.rs

1//! Where a channel's messages physically travel.
2//!
3//! [`InProcess`] is the only implementation today. A shared-memory or socket
4//! one would have to satisfy the same protocol tests — see the tests in
5//! [`crate::engine::port::mailbox`].
6
7use super::mailbox::{Consume, Disconnected, Mailbox, Outbox, Produce, Undeliverable};
8
9/// Creates the two ends of a one-way message channel.
10pub trait Transport: std::fmt::Debug + Send + Sync {
11    /// Pairs a producer with a consumer.
12    ///
13    /// The channel is unbounded, matching the queues it replaces: backpressure
14    /// would change the caller's timing, and a transport that needs it has to
15    /// apply it inside the message type instead.
16    fn channel<T: Send + 'static>(&self) -> (Outbox<T>, Mailbox<T>);
17}
18
19/// Carries messages between threads of this process, over `std::sync::mpsc`.
20///
21/// Zero-sized on purpose: the placement is a property of the type, not a field
22/// someone can swap out at runtime. A remote peer gets its own transport type
23/// with a connection in it, which keeps "where does this run" visible in the
24/// type rather than in a value that could be wrong at a call site.
25#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
26pub struct InProcess;
27
28impl Transport for InProcess {
29    fn channel<T: Send + 'static>(&self) -> (Outbox<T>, Mailbox<T>) {
30        let (sender, receiver) = std::sync::mpsc::channel();
31        (
32            Outbox::of(sender),
33            Mailbox::of(Receiver(std::sync::Mutex::new(receiver))),
34        )
35    }
36}
37
38impl<T: Send + 'static> Produce<T> for std::sync::mpsc::Sender<T> {
39    // Fully qualified: an unqualified `self.send(..)` would resolve back to
40    // this impl and recurse forever.
41    fn send(&self, message: T) -> Result<(), Undeliverable<T>> {
42        std::sync::mpsc::Sender::send(self, message).map_err(|error| Undeliverable(error.0))
43    }
44}
45
46/// Wraps `mpsc::Receiver` so it can sit behind a `Sync` trait object.
47///
48/// `Receiver` is `Send` but not `Sync`, and a mailbox has to be shareable for
49/// the owner to stay movable between threads.
50struct Receiver<T>(std::sync::Mutex<std::sync::mpsc::Receiver<T>>);
51
52impl<T: Send + 'static> Consume<T> for Receiver<T> {
53    fn recv(&self) -> Result<T, Disconnected> {
54        // Held across the park. A second thread reading this same mailbox would
55        // block on the lock rather than spin, which is the intended answer:
56        // one consumer per mailbox.
57        self.0
58            .lock()
59            .map_err(|_| Disconnected)?
60            .recv()
61            .map_err(|_| Disconnected)
62    }
63
64    fn try_recv(&self) -> Option<T> {
65        self.0.lock().ok()?.try_recv().ok()
66    }
67}
68
69#[cfg(test)]
70mod tests {
71    use super::*;
72    use crate::test_support::{assert_send, assert_sync};
73
74    /// A transport must be usable from another thread, otherwise it cannot
75    /// describe a channel between the UI thread and a worker.
76    #[test]
77    fn transport_is_shareable_across_threads() {
78        assert_send::<InProcess>();
79        assert_sync::<InProcess>();
80    }
81
82    /// A transport is chosen once, where the two ends are paired; both ends
83    /// keep working after it is gone.
84    #[test]
85    fn a_channel_outlives_the_transport_that_made_it() {
86        let (outbox, mailbox) = {
87            let transport = InProcess;
88            transport.channel::<&'static str>()
89        };
90        outbox.send("hello").expect("consumer is present");
91        assert_eq!(mailbox.recv().expect("producer is alive"), "hello");
92    }
93}
orinium_browser/engine/port/transport.rs

orinium_browser/engine/port/
transport.rs
