xmtp_api_backend/
client.rs

1use crate::endpoints::backend;
2use arc_swap::ArcSwap;
3use std::sync::Arc;
4use xmtp_configuration::LimitsConfiguration;
5use xmtp_proto::{
6    api::{ApiClientError, Client, Query},
7    api_client::XmtpBackendClient,
8    backend_v1::*,
9};
10
11/// Sends backend requests through one transport.
12#[derive(Clone, Debug)]
13pub struct BackendClient<C> {
14    pub(crate) client: C,
15    pub(crate) auth_handle: Option<crate::AuthHandle>,
16    /// The shapes this deployment accepts. Swapped in once, by
17    /// `build`, after the configuration is read and before any stream opens.
18    /// The compiled defaults until then, which is what every transport built
19    /// without a client keeps.
20    pub(crate) limits: Arc<ArcSwap<LimitsConfiguration>>,
21}
22impl<C> BackendClient<C> {
23    pub fn new(client: C) -> Self {
24        Self {
25            client,
26            auth_handle: None,
27            limits: Arc::new(ArcSwap::from_pointee(LimitsConfiguration::default())),
28        }
29    }
30    pub fn inner(&self) -> &C {
31        &self.client
32    }
33    pub fn with_auth_handle(mut self, handle: Option<crate::AuthHandle>) -> Self {
34        self.auth_handle = handle;
35        self
36    }
37    pub fn auth_handle(&self) -> Option<crate::AuthHandle> {
38        self.auth_handle.clone()
39    }
40    /// What this transport chunks its streams and metadata reads to.
41    pub(crate) fn limits(&self) -> arc_swap::Guard<Arc<LimitsConfiguration>> {
42        self.limits.load()
43    }
44}
45#[xmtp_common::async_trait]
46impl<C: Client> XmtpBackendClient for BackendClient<C> {
47    type Error = ApiClientError;
48    fn register_client_event_writer(&self, writer: &Arc<dyn xmtp_events::EventWriter<()>>) {
49        if let Some(handle) = &self.auth_handle {
50            handle.register_event_writer(writer);
51        }
52    }
53    fn unregister_client_event_writer(&self, writer: &Arc<dyn xmtp_events::EventWriter<()>>) {
54        if let Some(handle) = &self.auth_handle {
55            handle.unregister_event_writer(writer);
56        }
57    }
58    async fn publish(&self, request: PublishRequest) -> Result<PublishResponse, Self::Error> {
59        backend::Publish(request).query(&self.client).await
60    }
61    async fn create_upload(
62        &self,
63        request: CreateUploadRequest,
64    ) -> Result<CreateUploadResponse, Self::Error> {
65        backend::CreateUpload(request).query(&self.client).await
66    }
67    async fn query(&self, request: QueryRequest) -> Result<QueryResponse, Self::Error> {
68        backend::Query(request).query(&self.client).await
69    }
70    async fn query_newest(
71        &self,
72        request: QueryNewestRequest,
73    ) -> Result<QueryNewestResponse, Self::Error> {
74        backend::QueryNewest(request).query(&self.client).await
75    }
76    async fn get_inbox_ids(
77        &self,
78        request: GetInboxIdsRequest,
79    ) -> Result<GetInboxIdsResponse, Self::Error> {
80        backend::GetInboxIds(request).query(&self.client).await
81    }
82    async fn get_configuration(
83        &self,
84        request: GetConfigurationRequest,
85    ) -> Result<GetConfigurationResponse, Self::Error> {
86        backend::GetConfiguration(request).query(&self.client).await
87    }
88    fn backend_url(&self) -> Option<&str> {
89        Some(self.client.host())
90    }
91
92    fn has_credential_source(&self) -> bool {
93        self.client.has_credential_source()
94    }
95
96    fn set_limits(&self, limits: Arc<LimitsConfiguration>) {
97        // A zero here would panic `chunks(0)` in `streams.rs`. Wire conversion
98        // replaces zeroes; this also does so for a snapshot an app built
99        // in Rust and supplied through a `ConfigProvider`.
100        self.limits.store(Arc::new(limits.without_zeroes()));
101    }
102    async fn verify_smart_contract_wallet_signatures(
103        &self,
104        request: VerifySmartContractWalletSignaturesRequest,
105    ) -> Result<VerifySmartContractWalletSignaturesResponse, Self::Error> {
106        backend::VerifySmartContractWalletSignatures(request)
107            .query(&self.client)
108            .await
109    }
110    async fn register(&self, request: RegisterRequest) -> Result<RecipientState, Self::Error> {
111        backend::Register(request).query(&self.client).await
112    }
113    async fn unregister(
114        &self,
115        request: UnregisterRequest,
116    ) -> Result<UnregisterResponse, Self::Error> {
117        backend::Unregister(request).query(&self.client).await
118    }
119    async fn update_subscriptions(
120        &self,
121        request: UpdateSubscriptionsRequest,
122    ) -> Result<RecipientState, Self::Error> {
123        backend::UpdateSubscriptions(request)
124            .query(&self.client)
125            .await
126    }
127}
128
129#[xmtp_common::async_trait]
130impl<C: xmtp_proto::api::IsConnectedCheck> xmtp_proto::api::IsConnectedCheck for BackendClient<C> {
131    async fn is_connected(&self) -> bool {
132        self.client.is_connected().await
133    }
134}
xmtp_api_backend/client.rs

xmtp_api_backend/
client.rs
