xmtp_api_grpc/grpc_client/
client.rs

1//! The generic gRPC Client
2//! Generic over a inner "Channel".
3//! The  inner channel must implement a tower service to implicitly
4//! implement the gRPC Service
5
6use crate::{
7    error::{GrpcBuilderError, GrpcError},
8    streams::EscapableTonicStream,
9};
10use futures::Stream;
11use http::{request, uri::PathAndQuery};
12use pin_project::pin_project;
13use prost::bytes::Bytes;
14use std::{
15    pin::Pin,
16    sync::Arc,
17    task::{Context, Poll, ready},
18};
19use tonic::{
20    Status,
21    client::Grpc,
22    metadata::{self, MetadataMap, MetadataValue},
23};
24use url::Url;
25use xmtp_common::Retry;
26use xmtp_configuration::GRPC_PAYLOAD_LIMIT;
27use xmtp_proto::{
28    api::{ApiClientError, BytesStream, Client, IsConnectedCheck},
29    api_client::{ApiBuilder, NetConnectConfig},
30    codec::TransparentCodec,
31    types::AppVersion,
32};
33
34impl From<GrpcError> for ApiClientError {
35    fn from(source: GrpcError) -> ApiClientError {
36        ApiClientError::client(source)
37    }
38}
39
40/// Private trait to convert type to an HTTP Response
41trait ToHttp {
42    type Body;
43    fn to_http(self) -> http::Response<Self::Body>;
44}
45
46/// Convert a tonic Response to a generic HTTP response
47impl<T> ToHttp for tonic::Response<T> {
48    type Body = T;
49
50    fn to_http(self) -> http::Response<Self::Body> {
51        let (metadata, body, extensions) = self.into_parts();
52        let mut response = http::Response::new(body);
53        if cfg!(target_arch = "wasm32") {
54            *response.version_mut() = http::version::Version::HTTP_11;
55        } else {
56            *response.version_mut() = http::version::Version::HTTP_2;
57        }
58        *response.headers_mut() = metadata.into_headers();
59        *response.extensions_mut() = extensions;
60        response
61    }
62}
63
64#[derive(Clone, Debug)]
65pub struct GrpcClient {
66    inner: tonic::client::Grpc<crate::GrpcService>,
67    /// The URL this client dials ([`Client::host`]), as `Url` serializes it —
68    /// one normalization for every spelling of the same destination.
69    host: Arc<str>,
70    app_version: MetadataValue<metadata::Ascii>,
71    libxmtp_version: MetadataValue<metadata::Ascii>,
72}
73
74impl GrpcClient {
75    /// Builds a tonic request from a body and a generic HTTP Request.
76    ///
77    /// Generic over the body type `B` so the same path serves both the
78    /// unary / server-streaming transports (where `B` is `Bytes`) and the
79    /// XIP-83 bidirectional transport (where `B` is a `BoxDynStream` of
80    /// outbound frames). `tonic::Request<B>` is happy with either.
81    fn build_tonic_request<B>(
82        &self,
83        request: http::request::Builder,
84        body: B,
85    ) -> Result<tonic::Request<B>, Status> {
86        let request = request
87            .body(body)
88            .map_err(|e| tonic::Status::from_error(Box::new(e)))?;
89        let (mut parts, body) = request.into_parts();
90        xmtp_logging::propagation::inject(&tracing::Span::current(), &mut parts.headers);
91        let mut tonic_request = tonic::Request::from_parts(
92            MetadataMap::from_headers(parts.headers),
93            parts.extensions,
94            body,
95        );
96        let metadata = tonic_request.metadata_mut();
97        // must be lowercase otherwise panics
98        metadata.append("x-app-version", self.app_version.clone());
99        metadata.append("x-libxmtp-version", self.libxmtp_version.clone());
100        Ok(tonic_request)
101    }
102
103    async fn wait_for_ready(&self, client: &mut Grpc<crate::GrpcService>) -> Result<(), Status> {
104        client.ready().await.map_err(|e| {
105            tonic::Status::new(tonic::Code::Unknown, format!("Service was not ready: {e}"))
106        })?;
107        Ok(())
108    }
109}
110
111// just a more convenient way to map the stream type to
112// something more customized to the trait, without playing around with getting the
113// generics right on nested futures combinators.
114#[pin_project]
115/// A stream of bytes from a GRPC Network Source
116pub struct GrpcStream {
117    #[pin]
118    inner: crate::streams::NonBlocking,
119}
120
121impl From<crate::streams::NonBlocking> for GrpcStream {
122    fn from(value: crate::streams::NonBlocking) -> GrpcStream {
123        GrpcStream { inner: value }
124    }
125}
126
127// just a more convenient way to map the stream type to
128// something more customized to the trait, without playing around with getting the
129// generics right on nested futures combinators.
130impl Stream for GrpcStream {
131    type Item = Result<Bytes, ApiClientError>;
132
133    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
134        let this = self.project();
135        let item = ready!(this.inner.poll_next(cx));
136        Poll::Ready(item.map(|i| i.map_err(|e| ApiClientError::client(GrpcError::from(e)))))
137    }
138}
139
140#[xmtp_common::async_trait]
141impl Client for GrpcClient {
142    fn host(&self) -> &str {
143        &self.host
144    }
145
146    // Manual form: #[rpc_span] can't produce the rpc.grpc.* sub-namespace nor
147    // carry the bounded `path` field (it hard-codes skip_all + rpc.<fn_name>).
148    #[tracing::instrument(err, skip_all, fields(operation = "rpc.grpc.request", otel.kind = "client", otel.name = "rpc.grpc.request", path = %path))]
149    async fn request(
150        &self,
151        request: http::request::Builder,
152        path: http::uri::PathAndQuery,
153        body: Bytes,
154    ) -> Result<http::Response<Bytes>, ApiClientError> {
155        let client = &mut self.inner.clone();
156        self.wait_for_ready(client).await.map_err(GrpcError::from)?;
157        let request = self
158            .build_tonic_request(request, body)
159            .map_err(GrpcError::from)?;
160        let codec = TransparentCodec::default();
161        let response = client
162            .unary(request, path, codec)
163            .await
164            .map_err(GrpcError::from)?;
165
166        Ok(response.to_http())
167    }
168
169    // The span covers stream establishment only — the returned stream
170    // outlives it.
171    #[tracing::instrument(err, skip_all, fields(operation = "rpc.grpc.stream", otel.kind = "client", otel.name = "rpc.grpc.stream", path = %path))]
172    async fn stream(
173        &self,
174        request: request::Builder,
175        path: PathAndQuery,
176        body: Bytes,
177    ) -> Result<http::Response<BytesStream>, ApiClientError> {
178        let this = self.clone();
179        // client requires to be moved so it lives long enough for streaming response future.
180        let response = async move {
181            let mut client = this.inner.clone();
182            this.wait_for_ready(&mut client).await?;
183            let request = this.build_tonic_request(request, body)?;
184            let codec = TransparentCodec::default();
185            client.server_streaming(request, path, codec).await
186        };
187        let req = crate::streams::NonBlockingStreamRequest::new(
188            Box::pin(response) as crate::streams::ResponseFuture
189        );
190        let response = crate::streams::send(req).await.map_err(GrpcError::from)?;
191        let response = response.map(|body| {
192            BytesStream::new(GrpcStream {
193                inner: EscapableTonicStream::new(body),
194            })
195        });
196        Ok(response.to_http().map(Into::into))
197    }
198
199    // Full-duplex needs a real HTTP/2 transport; the gRPC-Web service used on
200    // wasm cannot carry it, so the browser keeps the trait's default error.
201    #[cfg(not(target_arch = "wasm32"))]
202    #[tracing::instrument(err, skip_all, fields(operation = "rpc.grpc.bidi_stream", otel.kind = "client", otel.name = "rpc.grpc.bidi_stream", path = %path))]
203    async fn bidi_stream(
204        &self,
205        request: request::Builder,
206        path: PathAndQuery,
207        body: xmtp_common::BoxDynStream<'static, Bytes>,
208    ) -> Result<http::Response<BytesStream>, ApiClientError> {
209        let this = self.clone();
210        // client requires to be moved so it lives long enough for streaming response future.
211        let response = async move {
212            let mut client = this.inner.clone();
213            this.wait_for_ready(&mut client).await?;
214            let request = this.build_tonic_request(request, body)?;
215            let codec = TransparentCodec::default();
216            client.streaming(request, path, codec).await
217        };
218        let req = crate::streams::NonBlockingStreamRequest::new(
219            Box::pin(response) as crate::streams::ResponseFuture
220        );
221        let response = crate::streams::send(req).await.map_err(GrpcError::from)?;
222        let response = response.map(|body| {
223            BytesStream::new(GrpcStream {
224                inner: EscapableTonicStream::new(body),
225            })
226        });
227        Ok(response.to_http().map(Into::into))
228    }
229}
230
231#[xmtp_common::async_trait]
232impl IsConnectedCheck for GrpcClient {
233    async fn is_connected(&self) -> bool {
234        self.inner.clone().ready().await.is_ok()
235    }
236}
237
238impl GrpcClient {
239    pub fn builder() -> ClientBuilder {
240        ClientBuilder::default()
241    }
242}
243
244#[derive(Default, Clone)]
245pub struct ClientBuilder {
246    pub host: Option<Url>,
247    /// version of the app
248    pub app_version: Option<MetadataValue<metadata::Ascii>>,
249    /// Version of the libxmtp core library
250    pub libxmtp_version: Option<MetadataValue<metadata::Ascii>>,
251    /// Rate per minute
252    pub limit: Option<u64>,
253    /// retry strategy for this client
254    pub retry: Option<Retry>,
255}
256
257impl NetConnectConfig for ClientBuilder {
258    fn set_libxmtp_version(&mut self, version: String) -> Result<(), Self::Error> {
259        self.libxmtp_version = Some(MetadataValue::try_from(&version)?);
260        Ok(())
261    }
262
263    fn set_app_version(&mut self, version: AppVersion) -> Result<(), Self::Error> {
264        self.app_version = Some(MetadataValue::try_from(&version)?);
265        Ok(())
266    }
267
268    fn set_host(&mut self, host: Url) {
269        self.host = Some(host);
270    }
271
272    fn rate_per_minute(&mut self, limit: u32) {
273        self.limit = Some(limit.into());
274    }
275
276    fn port(&self) -> Result<Option<String>, Self::Error> {
277        if let Some(h) = &self.host {
278            Ok(h.port().map(|u| u.to_string()))
279        } else {
280            Err(GrpcBuilderError::MissingHostUrl)
281        }
282    }
283
284    fn host(&self) -> Option<&str> {
285        self.host.as_ref().map(|s| s.as_str())
286    }
287
288    fn set_retry(&mut self, retry: xmtp_common::Retry) {
289        self.retry = Some(retry);
290    }
291}
292
293impl ApiBuilder for ClientBuilder {
294    type Output = crate::GrpcClient;
295    type Error = GrpcBuilderError;
296
297    fn build(self) -> Result<Self::Output, Self::Error> {
298        let host = self.host.ok_or(GrpcBuilderError::MissingHostUrl)?;
299        let host_str: Arc<str> = host.as_str().into();
300        let channel = crate::GrpcService::new(host, self.limit)?;
301        Ok(GrpcClient {
302            inner: tonic::client::Grpc::new(channel)
303                .max_decoding_message_size(GRPC_PAYLOAD_LIMIT)
304                .max_encoding_message_size(GRPC_PAYLOAD_LIMIT),
305            host: host_str,
306            app_version: self
307                .app_version
308                .unwrap_or(MetadataValue::try_from("0.0.0")?),
309            libxmtp_version: self.libxmtp_version.unwrap_or(MetadataValue::try_from(
310                env!("CARGO_PKG_VERSION").to_string(),
311            )?),
312        })
313    }
314}
315
316impl GrpcClient {
317    pub fn create(host: Url) -> Result<Self, GrpcBuilderError> {
318        let mut builder = Self::builder();
319        builder.set_host(host);
320        builder.build()
321    }
322
323    /// Create a grpc client with `app_version` attached
324    pub fn create_with_version(
325        host: Url,
326        app_version: AppVersion,
327    ) -> Result<Self, GrpcBuilderError> {
328        let mut builder = Self::builder();
329        builder.set_host(host);
330        builder.set_app_version(app_version)?;
331        builder.build()
332    }
333}
334
335#[cfg(test)]
336pub mod tests {
337    use crate::grpc_client::test::BackendTestClient;
338    use prost::Message;
339    use xmtp_proto::api_client::ApiBuilder;
340    use xmtp_proto::backend_v1::PublishRequest;
341    use xmtp_proto::prelude::{NetConnectConfig, XmtpTestClient};
342    use xmtp_proto::types::AppVersion;
343
344    #[xmtp_common::test(unwrap_try = true)]
345    async fn metadata_test() {
346        let mut client = BackendTestClient::create();
347        let app_version = AppVersion::from("test/1.0.0");
348        let libxmtp_version = "0.0.1".to_string();
349        client.set_app_version(app_version.clone())?;
350        client.set_libxmtp_version(libxmtp_version.clone())?;
351        let client = client.build()?;
352        let check = |enabled: bool| {
353            let span = tracing::info_span!(
354                "client",
355                operation = "rpc.grpc.request",
356                otel.kind = "client",
357                otel.name = "rpc.grpc.request"
358            );
359            let _entered = span.enter();
360            let mut expected = http::HeaderMap::new();
361            xmtp_logging::propagation::inject(&span, &mut expected);
362            let request = client
363                .build_tonic_request(
364                    Default::default(),
365                    prost::bytes::Bytes::from(PublishRequest { envelopes: vec![] }.encode_to_vec()),
366                )
367                .unwrap();
368            let headers = request.metadata();
369            assert_eq!(
370                headers.get("x-app-version").unwrap().to_str().unwrap(),
371                app_version.to_string()
372            );
373            assert_eq!(
374                headers.get("x-libxmtp-version").unwrap().to_str().unwrap(),
375                libxmtp_version
376            );
377            assert_eq!(headers.get("traceparent").is_some(), enabled);
378            assert_eq!(
379                headers.get("traceparent").map(|v| v.to_str().unwrap()),
380                expected.get("traceparent").map(|v| v.to_str().unwrap())
381            );
382        };
383        xmtp_common::if_native! { @
384            for enabled in [false, true] {
385                xmtp_logging::test_logging::with_trace_layer(enabled, || check(enabled));
386            }
387        }
388        xmtp_common::if_wasm! { @ check(false); }
389    }
390}
391
392xmtp_common::if_native! {
393    #[cfg(test)]
394    mod native_tests;
395}
xmtp_api_grpc/grpc_client/client.rs

xmtp_api_grpc/grpc_client/
client.rs
