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

xmtp_api/preflight/
dispatch.rs
