diff --git a/crates/xmtp_api_grpc/src/grpc_client/client.rs b/crates/xmtp_api_grpc/src/grpc_client/client.rs index 5b65cd90a7..1c4c4b2141 100644 --- a/crates/xmtp_api_grpc/src/grpc_client/client.rs +++ b/crates/xmtp_api_grpc/src/grpc_client/client.rs @@ -80,12 +80,17 @@ impl GrpcClient { } } - /// Builds a tonic request from a body and a generic HTTP Request - fn build_tonic_request( + /// Builds a tonic request from a body and a generic HTTP Request. + /// + /// Generic over the body type `B` so the same path serves both the + /// unary / server-streaming transports (where `B` is `Bytes`) and the + /// XIP-83 bidirectional transport (where `B` is a `BoxDynStream` of + /// outbound frames). `tonic::Request` is happy with either. + fn build_tonic_request( &self, request: http::request::Builder, - body: Bytes, - ) -> Result, Status> { + body: B, + ) -> Result, Status> { let request = request .body(body) .map_err(|e| tonic::Status::from_error(Box::new(e)))?; @@ -187,6 +192,36 @@ impl Client for GrpcClient { }); Ok(response.to_http().map(Into::into)) } + + // Full-duplex needs a real HTTP/2 transport; the gRPC-Web service used on + // wasm cannot carry it, so the browser keeps the trait's default error. + #[cfg(not(target_arch = "wasm32"))] + async fn bidi_stream( + &self, + request: request::Builder, + path: PathAndQuery, + body: xmtp_common::BoxDynStream<'static, Bytes>, + ) -> Result, ApiClientError> { + let this = self.clone(); + // client requires to be moved so it lives long enough for streaming response future. + let response = async move { + let mut client = this.inner.clone(); + this.wait_for_ready(&mut client).await?; + let request = this.build_tonic_request(request, body)?; + let codec = TransparentCodec::default(); + client.streaming(request, path, codec).await + }; + let req = crate::streams::NonBlockingStreamRequest::new( + Box::pin(response) as crate::streams::ResponseFuture + ); + let response = crate::streams::send(req).await.map_err(GrpcError::from)?; + let response = response.map(|body| { + BytesStream::new(GrpcStream { + inner: EscapableTonicStream::new(body), + }) + }); + Ok(response.to_http().map(Into::into)) + } } #[xmtp_common::async_trait] @@ -311,7 +346,7 @@ pub mod tests { let request = client .build_tonic_request( Default::default(), - PublishRequest { envelopes: vec![] }.encode_to_vec().into(), + prost::bytes::Bytes::from(PublishRequest { envelopes: vec![] }.encode_to_vec()), ) .unwrap(); diff --git a/crates/xmtp_proto/src/traits.rs b/crates/xmtp_proto/src/traits.rs index a46bba0eb5..470464c6d6 100644 --- a/crates/xmtp_proto/src/traits.rs +++ b/crates/xmtp_proto/src/traits.rs @@ -143,6 +143,22 @@ pub trait Client: MaybeSend + MaybeSync { body: Bytes, ) -> Result, ApiClientError>; + /// Open a bidirectional stream (XIP-83). `body` is the outbound stream of + /// encoded protobuf messages (one `Bytes` item per message); the response + /// carries the inbound message stream. Transports without full-duplex + /// support (e.g. gRPC-Web in the browser) keep this default and error. + async fn bidi_stream( + &self, + request: request::Builder, + path: http::uri::PathAndQuery, + body: BoxDynStream<'static, Bytes>, + ) -> Result, ApiClientError> { + let _ = (request, path, body); + Err(ApiClientError::OtherUnretryable( + "bidirectional streaming is not supported by this transport".into(), + )) + } + /// start a "fake" stream that does not create a TCP connection and will always be pending fn fake_stream(&self) -> http::Response { let fake = FakeEmptyStream::new(); @@ -179,6 +195,15 @@ where ) -> Result, ApiClientError> { (**self).stream(request, path, body).await } + + async fn bidi_stream( + &self, + request: request::Builder, + path: http::uri::PathAndQuery, + body: BoxDynStream<'static, Bytes>, + ) -> Result, ApiClientError> { + (**self).bidi_stream(request, path, body).await + } } #[xmtp_common::async_trait] @@ -203,6 +228,15 @@ where ) -> Result, ApiClientError> { (**self).stream(request, path, body).await } + + async fn bidi_stream( + &self, + request: request::Builder, + path: http::uri::PathAndQuery, + body: BoxDynStream<'static, Bytes>, + ) -> Result, ApiClientError> { + (**self).bidi_stream(request, path, body).await + } } #[xmtp_common::async_trait] @@ -227,6 +261,15 @@ where ) -> Result, ApiClientError> { (**self).stream(request, path, body).await } + + async fn bidi_stream( + &self, + request: request::Builder, + path: PathAndQuery, + body: BoxDynStream<'static, Bytes>, + ) -> Result, ApiClientError> { + (**self).bidi_stream(request, path, body).await + } } #[xmtp_common::async_trait] diff --git a/crates/xmtp_proto/src/traits/boxed_client.rs b/crates/xmtp_proto/src/traits/boxed_client.rs index 9507b0de39..9fe9bf1bd2 100644 --- a/crates/xmtp_proto/src/traits/boxed_client.rs +++ b/crates/xmtp_proto/src/traits/boxed_client.rs @@ -48,6 +48,15 @@ where ) -> Result, ApiClientError> { self.inner.stream(request, path, body).await } + + async fn bidi_stream( + &self, + request: request::Builder, + path: http::uri::PathAndQuery, + body: xmtp_common::BoxDynStream<'static, Bytes>, + ) -> Result, ApiClientError> { + self.inner.bidi_stream(request, path, body).await + } } pub trait ToBoxedClient {