xmtp_api/preflight/
dispatch.rs

1use super::*;
2use futures::StreamExt;
3use xmtp_proto::{
4    api::{HasStats, IsConnectedCheck},
5    api_client::*,
6    types::*,
7};
8
9#[xmtp_common::async_trait]
10impl<A: XmtpBackendClient> XmtpBackendClient for GuardedApi<A> {
11    type Error = ApiError;
12    async fn publish(&self, request: wire::PublishRequest) -> Result<wire::PublishResponse> {
13        self.check_preflight().await?;
14        self.raw.publish(request).await.map_err(dyn_err)
15    }
16    async fn create_upload(
17        &self,
18        request: wire::CreateUploadRequest,
19    ) -> Result<wire::CreateUploadResponse> {
20        self.check_preflight().await?;
21        self.raw.create_upload(request).await.map_err(dyn_err)
22    }
23    async fn query(&self, request: wire::QueryRequest) -> Result<wire::QueryResponse> {
24        self.check_preflight().await?;
25        self.raw.query(request).await.map_err(dyn_err)
26    }
27    async fn query_raw_envelopes(
28        &self,
29        request: wire::QueryRequest,
30    ) -> Result<ReceivedQueryResponse> {
31        self.check_preflight().await?;
32        self.raw.query_raw_envelopes(request).await.map_err(dyn_err)
33    }
34    async fn query_newest(
35        &self,
36        request: wire::QueryNewestRequest,
37    ) -> Result<wire::QueryNewestResponse> {
38        self.check_preflight().await?;
39        self.raw.query_newest(request).await.map_err(dyn_err)
40    }
41    async fn get_inbox_ids(
42        &self,
43        request: wire::GetInboxIdsRequest,
44    ) -> Result<wire::GetInboxIdsResponse> {
45        self.check_preflight().await?;
46        self.raw.get_inbox_ids(request).await.map_err(dyn_err)
47    }
48    async fn get_configuration(
49        &self,
50        request: wire::GetConfigurationRequest,
51    ) -> Result<wire::GetConfigurationResponse> {
52        self.check_preflight().await?;
53        self.raw.get_configuration(request).await.map_err(dyn_err)
54    }
55    async fn verify_smart_contract_wallet_signatures(
56        &self,
57        request: wire::VerifySmartContractWalletSignaturesRequest,
58    ) -> Result<wire::VerifySmartContractWalletSignaturesResponse> {
59        self.check_preflight().await?;
60        self.raw
61            .verify_smart_contract_wallet_signatures(request)
62            .await
63            .map_err(dyn_err)
64    }
65    async fn register(&self, request: wire::RegisterRequest) -> Result<wire::RecipientState> {
66        self.check_preflight().await?;
67        self.raw.register(request).await.map_err(dyn_err)
68    }
69    async fn unregister(
70        &self,
71        request: wire::UnregisterRequest,
72    ) -> Result<wire::UnregisterResponse> {
73        self.check_preflight().await?;
74        self.raw.unregister(request).await.map_err(dyn_err)
75    }
76    async fn update_subscriptions(
77        &self,
78        request: wire::UpdateSubscriptionsRequest,
79    ) -> Result<wire::RecipientState> {
80        self.check_preflight().await?;
81        self.raw
82            .update_subscriptions(request)
83            .await
84            .map_err(dyn_err)
85    }
86    fn backend_url(&self) -> Option<&str> {
87        self.raw.backend_url()
88    }
89    fn has_credential_source(&self) -> bool {
90        self.raw.has_credential_source()
91    }
92    fn set_limits(&self, limits: Arc<xmtp_configuration::LimitsConfiguration>) {
93        self.raw.set_limits(limits);
94    }
95    fn register_client_event_writer(&self, writer: &Arc<dyn xmtp_events::EventWriter<()>>) {
96        self.raw.register_client_event_writer(writer);
97    }
98    fn unregister_client_event_writer(&self, writer: &Arc<dyn xmtp_events::EventWriter<()>>) {
99        self.raw.unregister_client_event_writer(writer);
100    }
101}
102
103#[xmtp_common::async_trait]
104impl<A> XmtpMlsStreams for GuardedApi<A>
105where
106    A: XmtpMlsStreams + XmtpBackendClient,
107    A::GroupMessageStream: 'static,
108    A::WelcomeMessageStream: 'static,
109{
110    type Error = ApiError;
111    type GroupMessageStream = xmtp_common::BoxDynStream<'static, Result<GroupMessage>>;
112    type WelcomeMessageStream = xmtp_common::BoxDynStream<'static, Result<WelcomeMessage>>;
113    async fn subscribe_envelopes_with_cursors(
114        &self,
115        cursors: &TopicCursor,
116        limits: IncomingBatchLimits,
117    ) -> Result<IncomingSubscription<ApiError>> {
118        self.check_preflight().await?;
119        self.raw
120            .subscribe_envelopes_with_cursors(cursors, limits)
121            .await
122            .map(|stream| stream.map_error(dyn_err))
123            .map_err(dyn_err)
124    }
125    async fn subscribe_group_messages(
126        &self,
127        group_ids: &[&GroupId],
128    ) -> Result<Self::GroupMessageStream> {
129        self.check_preflight().await?;
130        Ok(Box::pin(
131            self.raw
132                .subscribe_group_messages(group_ids)
133                .await
134                .map_err(dyn_err)?
135                .map(|item| item.map_err(dyn_err)),
136        ))
137    }
138    async fn subscribe_group_messages_with_cursors(
139        &self,
140        cursors: &TopicCursor,
141    ) -> Result<Self::GroupMessageStream> {
142        self.check_preflight().await?;
143        Ok(Box::pin(
144            self.raw
145                .subscribe_group_messages_with_cursors(cursors)
146                .await
147                .map_err(dyn_err)?
148                .map(|item| item.map_err(dyn_err)),
149        ))
150    }
151    async fn subscribe_welcome_messages(
152        &self,
153        installations: &[&InstallationId],
154    ) -> Result<Self::WelcomeMessageStream> {
155        self.check_preflight().await?;
156        Ok(Box::pin(
157            self.raw
158                .subscribe_welcome_messages(installations)
159                .await
160                .map_err(dyn_err)?
161                .map(|item| item.map_err(dyn_err)),
162        ))
163    }
164    async fn subscribe_welcome_messages_with_cursors(
165        &self,
166        cursors: &TopicCursor,
167    ) -> Result<Self::WelcomeMessageStream> {
168        self.check_preflight().await?;
169        Ok(Box::pin(
170            self.raw
171                .subscribe_welcome_messages_with_cursors(cursors)
172                .await
173                .map_err(dyn_err)?
174                .map(|item| item.map_err(dyn_err)),
175        ))
176    }
177}
178xmtp_common::if_native! {
179#[xmtp_common::async_trait]
180impl<A> XmtpMlsBidiStreams for GuardedApi<A>
181where A: XmtpMlsBidiStreams + XmtpBackendClient, A::SubscribeStream: 'static,
182{
183    type Error = ApiError;
184    type SubscribeStream = xmtp_common::BoxDynStream<'static, Result<ReceivedSubscribeResponse>>;
185    fn host(&self) -> &str { self.raw.host() }
186    fn bidi_limits(&self) -> Arc<xmtp_configuration::LimitsConfiguration> { self.raw.bidi_limits() }
187    async fn subscribe_bidi(&self, requests: futures::stream::BoxStream<'static, wire::SubscribeRequest>) -> Result<Self::SubscribeStream> {
188        self.check_preflight().await?;
189        Ok(Box::pin(self.raw.subscribe_bidi(requests).await.map_err(dyn_err)?.map(|item| item.map_err(dyn_err))))
190    }
191}
192}
193impl<A: HasStats> HasStats for GuardedApi<A> {
194    fn aggregate_stats(&self) -> AggregateStats {
195        self.raw.aggregate_stats()
196    }
197    fn mls_stats(&self) -> ApiStats {
198        self.raw.mls_stats()
199    }
200    fn identity_stats(&self) -> IdentityStats {
201        self.raw.identity_stats()
202    }
203}
204#[xmtp_common::async_trait]
205impl<A: IsConnectedCheck> IsConnectedCheck for GuardedApi<A> {
206    async fn is_connected(&self) -> bool {
207        self.raw.is_connected().await
208    }
209}
xmtp_api/preflight/dispatch.rs

xmtp_api/preflight/
dispatch.rs
