xmtp_proto/
traits.rs

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

xmtp_proto/
traits.rs
