bitwarden_ipc/traits/
communication_backend.rs

1use std::fmt::Debug;
2
3use crate::{
4    error::IpcErrorKind,
5    message::{IncomingMessage, OutgoingMessage},
6};
7
8/// This trait defines the interface that will be used to send and receive messages over IPC.
9/// It is up to the platform to implement this trait and any necessary thread synchronization and
10/// broadcasting.
11pub trait CommunicationBackend: Send + Sync + 'static {
12    /// Error returned by [`send`](Self::send), classified via [`IpcErrorKind`].
13    type SendError: Debug + Send + Sync + 'static + IpcErrorKind;
14    /// Receiver type returned by [`subscribe`](Self::subscribe).
15    type Receiver: CommunicationBackendReceiver;
16
17    /// Send a message to the destination specified in the message. This function may be called
18    /// from any thread at any time.
19    ///
20    /// Both recoverable and fatal errors may be returned, classified via
21    /// [`IpcErrorKind::kind()`]. A recoverable error (e.g. a transient transport failure) is
22    /// logged and the IPC client keeps running; a fatal error stops the client from processing any
23    /// further messages. Ambiguous cases should be classified as recoverable.
24    ///
25    /// The implementation of this trait needs to guarantee that:
26    ///     - Multiple concurrent receivers and senders can coexist.
27    fn send(
28        &self,
29        message: OutgoingMessage,
30    ) -> impl std::future::Future<Output = Result<(), Self::SendError>> + Send + Sync;
31
32    /// Subscribe to receive messages. This function will return a receiver that can be used to
33    /// receive messages asynchronously.
34    ///
35    /// The implementation of this trait needs to guarantee that:
36    ///     - Multiple concurrent receivers may be created.
37    ///     - All concurrent receivers will receive the same messages.
38    ///      - Multiple concurrent receivers and senders can coexist.
39    fn subscribe(&self) -> impl std::future::Future<Output = Self::Receiver> + Send + Sync;
40}
41
42/// This trait defines the interface for receiving messages from the communication backend.
43///
44/// The implementation of this trait needs to guarantee that:
45///     - The receiver buffers messages from the creation of the receiver until the first call to
46///       receive().
47///     - The receiver buffers messages between calls to receive().
48pub trait CommunicationBackendReceiver: Send + Sync + 'static {
49    /// Error returned by [`receive`](Self::receive), classified via [`IpcErrorKind`].
50    type ReceiveError: Debug + Send + Sync + 'static + IpcErrorKind;
51
52    /// Receive a message. This function will block asynchronously until a message is received.
53    ///
54    /// Both recoverable and fatal errors may be returned, classified via
55    /// [`IpcErrorKind::kind()`]. A recoverable error (e.g. the receiver lagging) is logged and
56    /// the IPC client's processing loop continues; a fatal error (e.g. the channel being closed)
57    /// stops the loop. Ambiguous cases should be classified as recoverable.
58    ///
59    /// Do not call this function from multiple threads at the same time. Use the subscribe function
60    /// to create one receiver per thread.
61    fn receive(
62        &self,
63    ) -> impl std::future::Future<Output = Result<IncomingMessage, Self::ReceiveError>> + Send + Sync;
64}
65
66pub(crate) mod noop {
67    use super::*;
68
69    /// A no-op implementation of the `CommunicationBackend` trait.
70    ///
71    /// Sending discards messages silently and receiving blocks forever (the future never resolves).
72    /// This is useful as a default backend for platforms that do not need IPC communication.
73    pub struct NoopCommunicationBackend;
74
75    /// Receiver for [`NoopCommunicationBackend`] that never yields a message.
76    pub struct NoopCommunicationBackendReceiver;
77
78    impl CommunicationBackend for NoopCommunicationBackend {
79        type SendError = std::convert::Infallible;
80        type Receiver = NoopCommunicationBackendReceiver;
81
82        async fn send(&self, _message: OutgoingMessage) -> Result<(), Self::SendError> {
83            Ok(())
84        }
85
86        async fn subscribe(&self) -> Self::Receiver {
87            NoopCommunicationBackendReceiver
88        }
89    }
90
91    impl CommunicationBackendReceiver for NoopCommunicationBackendReceiver {
92        type ReceiveError = std::convert::Infallible;
93
94        async fn receive(&self) -> Result<IncomingMessage, Self::ReceiveError> {
95            std::future::pending().await
96        }
97    }
98}
99
100#[cfg(any(test, feature = "test-support"))]
101pub(crate) mod test_support {
102    use std::sync::Arc;
103
104    use tokio::sync::{
105        Mutex, RwLock,
106        broadcast::{self, Receiver, Sender},
107    };
108
109    use super::*;
110
111    /// A test implementation of the `CommunicationBackend` trait. Provides methods to inject
112    /// incoming messages and inspect outgoing messages.
113    #[derive(Debug)]
114    pub struct TestCommunicationBackend {
115        outgoing_tx: Sender<OutgoingMessage>,
116        outgoing_rx: Receiver<OutgoingMessage>,
117        outgoing: Arc<RwLock<Vec<OutgoingMessage>>>,
118        incoming_tx: Sender<IncomingMessage>,
119        incoming_rx: Receiver<IncomingMessage>,
120    }
121
122    impl Clone for TestCommunicationBackend {
123        fn clone(&self) -> Self {
124            TestCommunicationBackend {
125                outgoing_tx: self.outgoing_tx.clone(),
126                outgoing_rx: self.outgoing_rx.resubscribe(),
127                outgoing: self.outgoing.clone(),
128                incoming_tx: self.incoming_tx.clone(),
129                incoming_rx: self.incoming_rx.resubscribe(),
130            }
131        }
132    }
133
134    impl Default for TestCommunicationBackend {
135        fn default() -> Self {
136            Self::new()
137        }
138    }
139
140    /// Receiver for [`TestCommunicationBackend`].
141    #[derive(Debug)]
142    pub struct TestCommunicationBackendReceiver(RwLock<Receiver<IncomingMessage>>);
143
144    impl TestCommunicationBackend {
145        /// Create a new test communication backend.
146        pub fn new() -> Self {
147            let (outgoing_tx, outgoing_rx) = broadcast::channel(10);
148            let (incoming_tx, incoming_rx) = broadcast::channel(10);
149            TestCommunicationBackend {
150                outgoing_tx,
151                outgoing_rx,
152                outgoing: Arc::new(RwLock::new(Vec::new())),
153                incoming_tx,
154                incoming_rx,
155            }
156        }
157
158        /// Inject a message as if it were received from a remote endpoint.
159        pub fn push_incoming(&self, message: IncomingMessage) {
160            self.incoming_tx
161                .send(message)
162                .expect("Failed to send incoming message");
163        }
164
165        /// Get a copy of all the outgoing messages that have been sent.
166        pub async fn outgoing(&self) -> Vec<OutgoingMessage> {
167            self.outgoing.read().await.clone()
168        }
169
170        /// Drain all outgoing messages, returning them and clearing the internal buffer.
171        pub async fn drain_outgoing(&self) -> Vec<OutgoingMessage> {
172            self.outgoing.write().await.drain(..).collect()
173        }
174    }
175
176    impl CommunicationBackend for TestCommunicationBackend {
177        type SendError = ();
178        type Receiver = TestCommunicationBackendReceiver;
179
180        async fn send(&self, message: OutgoingMessage) -> Result<(), Self::SendError> {
181            self.outgoing.write().await.push(message);
182            Ok(())
183        }
184
185        async fn subscribe(&self) -> Self::Receiver {
186            TestCommunicationBackendReceiver(RwLock::new(self.incoming_rx.resubscribe()))
187        }
188    }
189
190    impl CommunicationBackendReceiver for TestCommunicationBackendReceiver {
191        type ReceiveError = ();
192
193        async fn receive(&self) -> Result<IncomingMessage, Self::ReceiveError> {
194            Ok(self
195                .0
196                .write()
197                .await
198                .recv()
199                .await
200                .expect("Failed to receive incoming message"))
201        }
202    }
203
204    #[allow(unused)]
205    #[derive(Clone)]
206    pub struct TestTwoWayCommunicationBackend {
207        outgoing: broadcast::Sender<OutgoingMessage>,
208        receiver: TestTwoWayCommunicationBackendReceiver,
209    }
210
211    #[allow(unused)]
212    #[derive(Clone)]
213    pub struct TestTwoWayCommunicationBackendReceiver {
214        incoming: Arc<Mutex<broadcast::Receiver<OutgoingMessage>>>,
215    }
216
217    impl CommunicationBackendReceiver for TestTwoWayCommunicationBackendReceiver {
218        type ReceiveError = ();
219
220        async fn receive(&self) -> Result<IncomingMessage, Self::ReceiveError> {
221            let mut incoming = self.incoming.lock().await;
222            let message = incoming
223                .recv()
224                .await
225                .expect("Failed to receive incoming message");
226            Ok(IncomingMessage {
227                payload: message.payload,
228                destination: message.destination,
229                source: crate::endpoint::Source::DesktopMain,
230                topic: message.topic,
231            })
232        }
233    }
234
235    impl TestTwoWayCommunicationBackend {
236        #[allow(unused)]
237        pub fn new() -> (Self, Self) {
238            let (outgoing0, incoming0) = broadcast::channel(128);
239            let (outgoing1, incoming1) = broadcast::channel(128);
240            let one = TestTwoWayCommunicationBackend {
241                outgoing: outgoing0,
242                receiver: TestTwoWayCommunicationBackendReceiver {
243                    incoming: Arc::new(Mutex::new(incoming1)),
244                },
245            };
246            let two = TestTwoWayCommunicationBackend {
247                outgoing: outgoing1,
248                receiver: TestTwoWayCommunicationBackendReceiver {
249                    incoming: Arc::new(Mutex::new(incoming0)),
250                },
251            };
252            (one, two)
253        }
254    }
255
256    impl CommunicationBackend for TestTwoWayCommunicationBackend {
257        type SendError = ();
258        type Receiver = TestTwoWayCommunicationBackendReceiver;
259
260        async fn send(&self, message: OutgoingMessage) -> Result<(), Self::SendError> {
261            self.outgoing
262                .send(message)
263                .expect("Failed to send outgoing message");
264            Ok(())
265        }
266
267        async fn subscribe(&self) -> Self::Receiver {
268            TestTwoWayCommunicationBackendReceiver {
269                incoming: Arc::new(Mutex::new(
270                    self.receiver.incoming.lock().await.resubscribe(),
271                )),
272            }
273        }
274    }
275}
bitwarden_ipc/traits/communication_backend.rs

bitwarden_ipc/traits/
communication_backend.rs
