xmtp_proto/
traits.rs

1//! Api Client Traits
2
3use crate::{
4    api::{FakeEmptyStream, RetryQuery, XmtpStream, combinators::Ignore},
5    api_client::{AggregateStats, ApiStats, IdentityStats},
6};
7use futures::Stream;
8use http::{request, uri::PathAndQuery};
9use prost::bytes::Bytes;
10use std::{borrow::Cow, pin::Pin, sync::Arc};
11use xmtp_common::{BoxDynStream, MaybeSend, MaybeSync, Retry};
12
13xmtp_common::if_test! {
14    pub mod mock;
15}
16
17mod boxed_client;
18pub(super) mod combinators;
19mod error;
20mod query;
21pub mod short_hex;
22pub mod stream;
23pub use boxed_client::*;
24pub use error::*;
25
26pub trait HasStats {
27    fn aggregate_stats(&self) -> AggregateStats;
28    fn mls_stats(&self) -> ApiStats;
29    fn identity_stats(&self) -> IdentityStats;
30}
31
32impl<T: HasStats + ?Sized> HasStats for Arc<T> {
33    fn aggregate_stats(&self) -> AggregateStats {
34        (**self).aggregate_stats()
35    }
36    fn mls_stats(&self) -> ApiStats {
37        (**self).mls_stats()
38    }
39    fn identity_stats(&self) -> IdentityStats {
40        (**self).identity_stats()
41    }
42}
43
44/// provides the necessary information for a backend API call.
45/// Indicates the Output type
46pub trait Endpoint<Specialized = ()>: MaybeSend + MaybeSync {
47    type Output: MaybeSend + MaybeSync;
48    fn grpc_endpoint(&self) -> Cow<'static, str>;
49
50    fn body(&self) -> Result<Bytes, BodyError>;
51}
52
53pub trait EndpointExt<S>: Endpoint<S> {
54    fn ignore_response(self) -> Ignore<Self>
55    where
56        Self: Sized + Endpoint<S>,
57    {
58        combinators::ignore(self)
59    }
60
61    fn retry(self) -> RetryQuery<Self>
62    where
63        Self: Sized + Endpoint<S>,
64    {
65        combinators::retry(self)
66    }
67
68    fn retry_with_strategy<St>(self, strategy: Retry<St>) -> RetryQuery<Self, St>
69    where
70        Self: Sized + Endpoint<S>,
71    {
72        combinators::retry_with_strategy(self, strategy)
73    }
74}
75
76impl<S, E> EndpointExt<S> for E where E: Endpoint<S> {}
77
78// choosing not to use the #[pin] macro here
79// because the manual structural pinning implementation is easy enough, and the
80// implementation is small & easy to verify
81/// concrete bytes stream type
82pub struct BytesStream {
83    stream: BoxDynStream<'static, Result<Bytes, ApiClientError>>,
84}
85
86impl BytesStream {
87    pub fn new(
88        stream: impl Stream<Item = Result<Bytes, ApiClientError>> + MaybeSend + 'static,
89    ) -> Self {
90        Self {
91            stream: Box::pin(stream),
92        }
93    }
94}
95
96impl BytesStream {
97    fn stream(
98        self: Pin<&mut Self>,
99    ) -> Pin<&mut BoxDynStream<'static, Result<Bytes, ApiClientError>>> {
100        // this is safe because 'stream' is pinned when 'self' is
101        // https://doc.rust-lang.org/std/pin/index.html#choosing-pinning-to-be-structural-for-field
102        unsafe { self.map_unchecked_mut(|s| &mut s.stream) }
103    }
104}
105
106impl Stream for BytesStream {
107    type Item = Result<Bytes, ApiClientError>;
108
109    fn poll_next(
110        self: std::pin::Pin<&mut Self>,
111        cx: &mut std::task::Context<'_>,
112    ) -> std::task::Poll<Option<Self::Item>> {
113        self.stream().poll_next(cx)
114    }
115}
116
117/// A client represents how a request body is formed and sent into
118/// a backend. The client is protocol agnostic, a Client may
119/// communicate with a backend over gRPC, JSON-RPC, HTTP-REST, etc.
120/// `http::Response`'s are used in order to maintain a
121/// common data format compatible with a wide variety of backends.
122/// an http response is easily derived from a grpc, jsonrpc or rest api.
123#[xmtp_common::async_trait]
124pub trait Client: MaybeSend + MaybeSync {
125    /// The URL this transport dials — its connection identity. Two clients
126    /// share process-level connection state (the XIP-83 shared bidi wire)
127    /// exactly when their hosts match, so this must name where the bytes
128    /// actually go: a client behind a proxy reports the proxy's URL, keeping
129    /// it a separate failure domain from a direct client to the same backend.
130    fn host(&self) -> &str;
131
132    /// Whether this transport stack was given a way to obtain a credential —
133    /// an auth callback or an auth handle. A stack with no auth
134    /// middleware in it reports `false`, and `build` refuses a deployment that
135    /// requires authentication.
136    fn has_credential_source(&self) -> bool {
137        false
138    }
139
140    async fn request(
141        &self,
142        request: request::Builder,
143        path: PathAndQuery,
144        body: Bytes,
145    ) -> Result<http::Response<Bytes>, ApiClientError>;
146
147    async fn stream(
148        &self,
149        request: request::Builder,
150        path: http::uri::PathAndQuery,
151        body: Bytes,
152    ) -> Result<http::Response<BytesStream>, ApiClientError>;
153
154    /// Open a bidirectional stream (XIP-83). `body` is the outbound stream of
155    /// encoded protobuf messages (one `Bytes` item per message); the response
156    /// carries the inbound message stream. Transports without full-duplex
157    /// support (e.g. gRPC-Web in the browser) keep this default and error.
158    async fn bidi_stream(
159        &self,
160        request: request::Builder,
161        path: http::uri::PathAndQuery,
162        body: BoxDynStream<'static, Bytes>,
163    ) -> Result<http::Response<BytesStream>, ApiClientError> {
164        let _ = (request, path, body);
165        Err(ApiClientError::OtherUnretryable(
166            "bidirectional streaming is not supported by this transport".into(),
167        ))
168    }
169
170    /// start a "fake" stream that does not create a TCP connection and will always be pending
171    fn fake_stream(&self) -> http::Response<BytesStream> {
172        let fake = FakeEmptyStream::new();
173        let mut response = http::Response::new(BytesStream::new(fake));
174        if cfg!(target_arch = "wasm32") {
175            *response.version_mut() = http::version::Version::HTTP_11;
176        } else {
177            *response.version_mut() = http::version::Version::HTTP_2;
178        }
179
180        response
181    }
182}
183
184#[xmtp_common::async_trait]
185impl<T: MaybeSend + MaybeSync + ?Sized> Client for &T
186where
187    T: Client,
188{
189    fn host(&self) -> &str {
190        (**self).host()
191    }
192
193    fn has_credential_source(&self) -> bool {
194        (**self).has_credential_source()
195    }
196
197    async fn request(
198        &self,
199        request: request::Builder,
200        path: PathAndQuery,
201        body: Bytes,
202    ) -> Result<http::Response<Bytes>, ApiClientError> {
203        (**self).request(request, path, body).await
204    }
205
206    async fn stream(
207        &self,
208        request: request::Builder,
209        path: http::uri::PathAndQuery,
210        body: Bytes,
211    ) -> Result<http::Response<BytesStream>, ApiClientError> {
212        (**self).stream(request, path, body).await
213    }
214
215    async fn bidi_stream(
216        &self,
217        request: request::Builder,
218        path: http::uri::PathAndQuery,
219        body: BoxDynStream<'static, Bytes>,
220    ) -> Result<http::Response<BytesStream>, ApiClientError> {
221        (**self).bidi_stream(request, path, body).await
222    }
223}
224
225#[xmtp_common::async_trait]
226impl<T: MaybeSend + MaybeSync + ?Sized> Client for Box<T>
227where
228    T: Client,
229{
230    fn host(&self) -> &str {
231        (**self).host()
232    }
233
234    fn has_credential_source(&self) -> bool {
235        (**self).has_credential_source()
236    }
237
238    async fn request(
239        &self,
240        request: request::Builder,
241        path: PathAndQuery,
242        body: Bytes,
243    ) -> Result<http::Response<Bytes>, ApiClientError> {
244        (**self).request(request, path, body).await
245    }
246
247    async fn stream(
248        &self,
249        request: request::Builder,
250        path: http::uri::PathAndQuery,
251        body: Bytes,
252    ) -> Result<http::Response<BytesStream>, ApiClientError> {
253        (**self).stream(request, path, body).await
254    }
255
256    async fn bidi_stream(
257        &self,
258        request: request::Builder,
259        path: http::uri::PathAndQuery,
260        body: BoxDynStream<'static, Bytes>,
261    ) -> Result<http::Response<BytesStream>, ApiClientError> {
262        (**self).bidi_stream(request, path, body).await
263    }
264}
265
266#[xmtp_common::async_trait]
267impl<T: MaybeSend + MaybeSync + ?Sized> Client for Arc<T>
268where
269    T: Client,
270{
271    fn host(&self) -> &str {
272        (**self).host()
273    }
274
275    fn has_credential_source(&self) -> bool {
276        (**self).has_credential_source()
277    }
278
279    async fn request(
280        &self,
281        request: request::Builder,
282        path: PathAndQuery,
283        body: Bytes,
284    ) -> Result<http::Response<Bytes>, ApiClientError> {
285        (**self).request(request, path, body).await
286    }
287
288    async fn stream(
289        &self,
290        request: request::Builder,
291        path: PathAndQuery,
292        body: Bytes,
293    ) -> Result<http::Response<BytesStream>, ApiClientError> {
294        (**self).stream(request, path, body).await
295    }
296
297    async fn bidi_stream(
298        &self,
299        request: request::Builder,
300        path: PathAndQuery,
301        body: BoxDynStream<'static, Bytes>,
302    ) -> Result<http::Response<BytesStream>, ApiClientError> {
303        (**self).bidi_stream(request, path, body).await
304    }
305}
306
307#[xmtp_common::async_trait]
308pub trait IsConnectedCheck: MaybeSend + MaybeSync {
309    /// Check if a client is connected
310    async fn is_connected(&self) -> bool;
311}
312
313#[xmtp_common::async_trait]
314impl<T: MaybeSend + MaybeSync + ?Sized> IsConnectedCheck for Arc<T>
315where
316    T: IsConnectedCheck,
317{
318    async fn is_connected(&self) -> bool {
319        (**self).is_connected().await
320    }
321}
322
323#[xmtp_common::async_trait]
324impl<T: MaybeSend + MaybeSync + ?Sized> IsConnectedCheck for Box<T>
325where
326    T: IsConnectedCheck,
327{
328    async fn is_connected(&self) -> bool {
329        (**self).is_connected().await
330    }
331}
332
333/// Queries describe the way an endpoint is called.
334/// these are extensions to the behavior of specific endpoints.
335#[xmtp_common::async_trait]
336pub trait Query<C: Client>: MaybeSend + MaybeSync {
337    type Output: MaybeSend + MaybeSync;
338    async fn query(&mut self, client: &C) -> Result<Self::Output, ApiClientError>;
339}
340
341#[xmtp_common::async_trait]
342pub trait QueryRaw<C: Client>: MaybeSend + MaybeSync {
343    async fn query_raw(&mut self, client: &C) -> Result<bytes::Bytes, ApiClientError>;
344}
345
346/// a companion to the [`Query`] trait, except for streaming calls.
347/// Not every query combinator/extension will apply to both
348/// steams and one-off calls (how do you 'page' a streaming api?),
349/// so these traits are separated.
350#[xmtp_common::async_trait]
351pub trait QueryStream<T, C>
352where
353    C: Client,
354{
355    /// Stream items from an endpoint. Use [`QueryStreamExt::subscribe`] to set
356    /// the type of item in the stream.
357    async fn stream(&mut self, client: &C) -> Result<XmtpStream<T>, ApiClientError>;
358
359    fn fake_stream(&mut self, client: &C) -> XmtpStream<T>;
360}
361
362#[xmtp_common::async_trait]
363pub trait QueryStreamExt<T, C: Client> {
364    /// Subscribe to the endpoint, indicating the type of stream item with `R`
365    async fn subscribe(&mut self, client: &C) -> Result<XmtpStream<T>, ApiClientError>
366    where
367        T: Default + prost::Message + 'static;
368}
369
370#[xmtp_common::async_trait]
371impl<T, C, E> QueryStreamExt<T, C> for E
372where
373    C: Client,
374    E: Endpoint<Output = T>,
375{
376    async fn subscribe(&mut self, client: &C) -> Result<XmtpStream<T>, ApiClientError>
377    where
378        T: Default + prost::Message + 'static,
379    {
380        self.stream(client).await
381    }
382}
383
384#[cfg(test)]
385mod test {
386    use crate::api::{
387        EndpointExt, Query,
388        mock::{MockNetworkClient, TestEndpoint},
389    };
390
391    // test ensures these combinations can compile
392    #[xmtp_common::test]
393    async fn endpoints_can_be_chained() {
394        let client = MockNetworkClient::new();
395        std::mem::drop(TestEndpoint.ignore_response().retry().query(&client));
396        std::mem::drop(TestEndpoint.retry().ignore_response().query(&client));
397    }
398}
xmtp_proto/traits.rs

xmtp_proto/
traits.rs
