xmtp_api_backend/queries/
boxed_streams.rs

1use xmtp_proto::{
2    api_client::{XmtpBackendClient, XmtpMlsStreams},
3    backend_v1::*,
4    types::{GroupId, InstallationId, TopicCursor},
5};
6#[derive(Clone, Debug)]
7pub struct BoxedStreamsClient<C> {
8    inner: C,
9}
10impl<C> BoxedStreamsClient<C> {
11    pub fn new(inner: C) -> Self {
12        Self { inner }
13    }
14    pub fn inner(&self) -> &C {
15        &self.inner
16    }
17}
18#[xmtp_common::async_trait]
19impl<C: XmtpBackendClient> XmtpBackendClient for BoxedStreamsClient<C> {
20    type Error = C::Error;
21    async fn publish(&self, request: PublishRequest) -> Result<PublishResponse, Self::Error> {
22        self.inner.publish(request).await
23    }
24    async fn query(&self, request: QueryRequest) -> Result<QueryResponse, Self::Error> {
25        self.inner.query(request).await
26    }
27    async fn query_newest(
28        &self,
29        request: QueryNewestRequest,
30    ) -> Result<QueryNewestResponse, Self::Error> {
31        self.inner.query_newest(request).await
32    }
33    async fn get_inbox_ids(
34        &self,
35        request: GetInboxIdsRequest,
36    ) -> Result<GetInboxIdsResponse, Self::Error> {
37        self.inner.get_inbox_ids(request).await
38    }
39    async fn get_configuration(
40        &self,
41        request: GetConfigurationRequest,
42    ) -> Result<GetConfigurationResponse, Self::Error> {
43        self.inner.get_configuration(request).await
44    }
45    fn backend_url(&self) -> Option<&str> {
46        self.inner.backend_url()
47    }
48
49    fn has_credential_source(&self) -> bool {
50        self.inner.has_credential_source()
51    }
52
53    fn set_limits(&self, limits: std::sync::Arc<xmtp_configuration::LimitsConfiguration>) {
54        self.inner.set_limits(limits)
55    }
56    async fn verify_smart_contract_wallet_signatures(
57        &self,
58        request: VerifySmartContractWalletSignaturesRequest,
59    ) -> Result<VerifySmartContractWalletSignaturesResponse, Self::Error> {
60        self.inner
61            .verify_smart_contract_wallet_signatures(request)
62            .await
63    }
64    async fn register(&self, request: RegisterRequest) -> Result<RecipientState, Self::Error> {
65        self.inner.register(request).await
66    }
67    async fn unregister(
68        &self,
69        request: UnregisterRequest,
70    ) -> Result<UnregisterResponse, Self::Error> {
71        self.inner.unregister(request).await
72    }
73    async fn update_subscriptions(
74        &self,
75        request: UpdateSubscriptionsRequest,
76    ) -> Result<RecipientState, Self::Error> {
77        self.inner.update_subscriptions(request).await
78    }
79}
80#[xmtp_common::async_trait]
81impl<C: XmtpMlsStreams> XmtpMlsStreams for BoxedStreamsClient<C>
82where
83    C::GroupMessageStream: 'static,
84    C::WelcomeMessageStream: 'static,
85{
86    type Error = C::Error;
87    type GroupMessageStream = xmtp_proto::api_client::BoxedGroupS<C::Error>;
88    type WelcomeMessageStream = xmtp_proto::api_client::BoxedWelcomeS<C::Error>;
89    async fn subscribe_envelopes_with_cursors(
90        &self,
91        cursors: &TopicCursor,
92        limits: xmtp_proto::types::IncomingBatchLimits,
93    ) -> Result<xmtp_proto::types::IncomingSubscription<Self::Error>, Self::Error> {
94        self.inner
95            .subscribe_envelopes_with_cursors(cursors, limits)
96            .await
97    }
98    async fn subscribe_group_messages(
99        &self,
100        groups: &[&GroupId],
101    ) -> Result<Self::GroupMessageStream, Self::Error> {
102        Ok(Box::pin(self.inner.subscribe_group_messages(groups).await?))
103    }
104    async fn subscribe_group_messages_with_cursors(
105        &self,
106        groups: &TopicCursor,
107    ) -> Result<Self::GroupMessageStream, Self::Error> {
108        Ok(Box::pin(
109            self.inner
110                .subscribe_group_messages_with_cursors(groups)
111                .await?,
112        ))
113    }
114    async fn subscribe_welcome_messages(
115        &self,
116        installations: &[&InstallationId],
117    ) -> Result<Self::WelcomeMessageStream, Self::Error> {
118        Ok(Box::pin(
119            self.inner.subscribe_welcome_messages(installations).await?,
120        ))
121    }
122    async fn subscribe_welcome_messages_with_cursors(
123        &self,
124        installations: &TopicCursor,
125    ) -> Result<Self::WelcomeMessageStream, Self::Error> {
126        Ok(Box::pin(
127            self.inner
128                .subscribe_welcome_messages_with_cursors(installations)
129                .await?,
130        ))
131    }
132}
133xmtp_common::if_native! {
134    use xmtp_proto::api_client::XmtpMlsBidiStreams;
135
136    // `XmtpMlsBidiStreams` is native-only, so this forward is gated like the
137    // trait. It carries no per-call stat (the bidi stream is opened once and
138    // mutated in place, not counted per RPC like the unary/stream calls above);
139    // it exists so the stats wrapper is bidi-capable, letting the standard
140    // feature-switched test client open a `BidiConnection`.
141    #[xmtp_common::async_trait]
142    impl<C> XmtpMlsBidiStreams for BoxedStreamsClient<C>
143    where
144        C: XmtpMlsBidiStreams,
145    {
146        type SubscribeStream = <C as XmtpMlsBidiStreams>::SubscribeStream;
147        type Error = <C as XmtpMlsBidiStreams>::Error;
148
149        fn host(&self) -> &str {
150            self.inner.host()
151        }
152
153        fn bidi_limits(&self) -> std::sync::Arc<xmtp_configuration::LimitsConfiguration> {
154            self.inner.bidi_limits()
155        }
156
157        async fn subscribe_bidi(
158            &self,
159            requests: futures::stream::BoxStream<'static, xmtp_proto::backend_v1::SubscribeRequest>,
160        ) -> Result<Self::SubscribeStream, Self::Error> {
161
162            self.inner.subscribe_bidi(requests).await
163        }
164    }
165}
166
167#[xmtp_common::async_trait]
168impl<C: xmtp_proto::api::IsConnectedCheck> xmtp_proto::api::IsConnectedCheck
169    for BoxedStreamsClient<C>
170{
171    async fn is_connected(&self) -> bool {
172        self.inner.is_connected().await
173    }
174}
175
176impl<C: xmtp_proto::api::HasStats> xmtp_proto::api::HasStats for BoxedStreamsClient<C> {
177    fn aggregate_stats(&self) -> xmtp_proto::api_client::AggregateStats {
178        self.inner.aggregate_stats()
179    }
180    fn mls_stats(&self) -> xmtp_proto::api_client::ApiStats {
181        self.inner.mls_stats()
182    }
183    fn identity_stats(&self) -> xmtp_proto::api_client::IdentityStats {
184        self.inner.identity_stats()
185    }
186}
xmtp_api_backend/queries/boxed_streams.rs

xmtp_api_backend/queries/
boxed_streams.rs
