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

bitwarden_ipc/
ipc_client_ext.rs
