bitwarden_ipc/
ipc_client_ext.rs

1use bitwarden_threading::cancellation_token::CancellationToken;
2use serde::{Serialize, de::DeserializeOwned};
3
4use crate::{
5    RpcHandler,
6    endpoint::Endpoint,
7    error::{RequestError, SubscribeError},
8    ipc_client::IpcClientTypedSubscription,
9    ipc_client_trait::IpcClient,
10    message::{OutgoingMessage, PayloadTypeName, TypedOutgoingMessage},
11    rpc::{
12        error::RpcError,
13        request::RpcRequest,
14        request_message::{RPC_REQUEST_PAYLOAD_TYPE_NAME, RpcRequestMessage},
15        response_message::IncomingRpcResponseMessage,
16    },
17    serde_utils,
18};
19
20/// Extension trait providing generic convenience methods on any [`IpcClient`].
21///
22/// This trait is automatically implemented for all types that implement [`IpcClient`],
23/// including `dyn IpcClient`. It provides typed subscriptions, handler registration,
24/// and RPC request functionality with full static type safety.
25pub trait IpcClientExt: IpcClient {
26    /// Register a new RPC handler for processing incoming RPC requests.
27    /// The handler will be executed by the IPC client when an RPC request is received and
28    /// the response will be sent back over IPC.
29    fn register_rpc_handler<H>(&self, handler: H) -> impl std::future::Future<Output = ()> + Send
30    where
31        H: RpcHandler + Send + Sync + 'static,
32    {
33        async move {
34            self.register_rpc_handler_erased(H::Request::NAME, Box::new(handler))
35                .await;
36        }
37    }
38
39    /// Send a message with a payload of any serializable type to the specified destination.
40    fn send_typed<Payload>(
41        &self,
42        payload: Payload,
43        destination: Endpoint,
44    ) -> impl std::future::Future<Output = Result<(), RequestError>> + Send
45    where
46        Payload: Serialize + PayloadTypeName + Send,
47    {
48        async move {
49            let message = TypedOutgoingMessage {
50                payload,
51                destination,
52            }
53            .try_into()
54            .map_err(|e: serde_utils::DeserializeError| {
55                RequestError::Rpc(RpcError::RequestSerialization(e.to_string()))
56            })?;
57
58            self.send(message).await.map_err(RequestError::from)
59        }
60    }
61
62    /// Create a subscription to receive messages that can be deserialized into the provided
63    /// payload type.
64    fn subscribe_typed<Payload>(
65        &self,
66    ) -> impl std::future::Future<
67        Output = Result<IpcClientTypedSubscription<Payload>, SubscribeError>,
68    > + Send
69    where
70        Payload: DeserializeOwned + PayloadTypeName,
71    {
72        async move {
73            Ok(IpcClientTypedSubscription::new(
74                self.subscribe(Some(Payload::PAYLOAD_TYPE_NAME.to_owned()))
75                    .await?,
76            ))
77        }
78    }
79
80    /// Send a request to the specified destination and wait for a response.
81    /// The destination must have a registered RPC handler for the request type, otherwise
82    /// an error will be returned by the remote endpoint.
83    fn request<Request>(
84        &self,
85        request: Request,
86        destination: Endpoint,
87        cancellation_token: Option<CancellationToken>,
88    ) -> impl std::future::Future<Output = Result<Request::Response, RequestError>> + Send
89    where
90        Request: RpcRequest + Send,
91        Request::Response: Send,
92    {
93        async move {
94            let request_payload = RpcRequestMessage::new(request);
95
96            // Each request gets its own response topic, subscribed to before sending. The handler
97            // publishes its reply there, so this subscription receives exactly one message: the
98            // response to this request. A deserialization failure is therefore unambiguously a
99            // malformed response to this request.
100            let mut response_subscription = self
101                .subscribe(Some(request_payload.response_topic.clone()))
102                .await?;
103
104            // Requests are dispatched by a single fixed topic that the receiver matches on to route
105            // them to the handler registry.
106            let payload = serde_utils::to_vec(&request_payload)
107                .map_err(|e| RequestError::Rpc(RpcError::RequestSerialization(e.to_string())))?;
108            let message = OutgoingMessage {
109                payload,
110                destination,
111                topic: Some(RPC_REQUEST_PAYLOAD_TYPE_NAME.to_owned()),
112            };
113
114            self.send(message).await.map_err(RequestError::from)?;
115
116            let received = response_subscription
117                .receive(cancellation_token)
118                .await
119                .map_err(|e| RequestError::Receive(e.into()))?;
120
121            let response: IncomingRpcResponseMessage<Request::Response> =
122                serde_utils::from_slice(&received.payload).map_err(|e| {
123                    RequestError::Rpc(RpcError::ResponseDeserialization(e.to_string()))
124                })?;
125
126            Ok(response.result?)
127        }
128    }
129}
130
131/// Blanket implementation: every [`IpcClient`] gets the extension methods for free.
132impl<T: IpcClient + ?Sized> IpcClientExt for T {}
bitwarden_ipc/ipc_client_ext.rs

bitwarden_ipc/
ipc_client_ext.rs
