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

xmtp_proto/
traits.rs
