bitwarden_ipc/traits/
communication_backend.rs

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

bitwarden_ipc/traits/
communication_backend.rs
