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

xmtp_api_backend/queries/
boxed_streams.rs
