xmtp_api_backend/queries/backend/
transport.rs

1//! Backend subscription transport for the retained client wrapper (native-only).
2
3use crate::BackendClient;
4use futures::StreamExt;
5use prost::Message;
6use prost::bytes::Bytes;
7use xmtp_proto::ApiEndpoint;
8use xmtp_proto::api::{ApiClientError, Client, XmtpStream};
9use xmtp_proto::api_client::XmtpMlsBidiStreams;
10use xmtp_proto::backend_v1::{SubscribeRequest, SubscribeResponse};
11
12const SUBSCRIBE_PATH: &str = "/xmtp.backend.v1.SubscriptionService/Subscribe";
13
14#[xmtp_common::async_trait]
15impl<C> XmtpMlsBidiStreams for BackendClient<C>
16where
17    C: Client,
18{
19    type SubscribeStream = XmtpStream<SubscribeResponse>;
20
21    type Error = ApiClientError;
22
23    fn host(&self) -> &str {
24        self.client.host()
25    }
26
27    fn bidi_limits(&self) -> std::sync::Arc<xmtp_configuration::LimitsConfiguration> {
28        self.limits.load_full()
29    }
30
31    // Spans the open handshake (not the stream's lifetime) as `rpc.subscribe_bidi`.
32    // Bidi is consumed directly by `xmtp_mls` with no `xmtp_api` wrapper in front,
33    // so this transport impl is the RPC boundary — the same layer the other
34    // `rpc_span`s live at for unary calls. Move it up if a wrapper is ever added.
35    #[xmtp_common::rpc_span]
36    async fn subscribe_bidi(
37        &self,
38        requests: futures::stream::BoxStream<'static, SubscribeRequest>,
39    ) -> Result<Self::SubscribeStream, Self::Error> {
40        tracing::debug!("opening bidirectional subscription");
41        let outbound = requests.map(|frame| Bytes::from(frame.encode_to_vec()));
42        let response = self
43            .client
44            .bidi_stream(
45                http::Request::builder(),
46                http::uri::PathAndQuery::from_static(SUBSCRIBE_PATH),
47                Box::pin(outbound),
48            )
49            .await
50            .map_err(|e| e.endpoint(SUBSCRIBE_PATH.to_string()))?;
51        Ok(XmtpStream::new(
52            response.into_body(),
53            ApiEndpoint::Path(SUBSCRIBE_PATH.to_string()),
54        ))
55    }
56}
57
58#[cfg(test)]
59mod tests {
60    #![allow(clippy::unwrap_used)]
61    use super::*;
62    use futures::stream;
63    use xmtp_common::BoxDynStream;
64    use xmtp_proto::api::BytesStream;
65    use xmtp_proto::api::mock::MockNetworkClient;
66    use xmtp_proto::backend_v1::subscribe_request::Update;
67    use xmtp_proto::backend_v1::{Ping, Pong, subscribe_request, subscribe_response};
68
69    fn req(request: subscribe_request::Request) -> SubscribeRequest {
70        SubscribeRequest {
71            request: Some(request),
72        }
73    }
74
75    fn resp(response: subscribe_response::Response) -> SubscribeResponse {
76        SubscribeResponse {
77            response: Some(response),
78        }
79    }
80
81    fn ping_req(nonce: u64) -> SubscribeRequest {
82        req(subscribe_request::Request::Ping(Ping { nonce }))
83    }
84
85    /// `subscribe_bidi` must prost-encode each outbound `SubscribeRequest` in
86    /// order, dial the `Subscribe` path, and decode the inbound byte frames back
87    /// into `SubscribeResponse`s through `XmtpStream`.
88    #[xmtp_common::test(unwrap_try = true)]
89    async fn encodes_outbound_and_decodes_inbound() {
90        let captured: std::sync::Arc<std::sync::Mutex<Option<BoxDynStream<'static, Bytes>>>> =
91            Default::default();
92        let sink = captured.clone();
93
94        let mut mock = MockNetworkClient::new();
95        mock.expect_bidi_stream()
96            .return_once(move |_req, path, body| {
97                assert_eq!(
98                    path.path(),
99                    "/xmtp.backend.v1.SubscriptionService/Subscribe"
100                );
101                *sink.lock().unwrap() = Some(body);
102                let frames: Vec<Result<Bytes, ApiClientError>> = vec![
103                    Ok(Bytes::from(
104                        resp(subscribe_response::Response::Ping(Ping { nonce: 7 })).encode_to_vec(),
105                    )),
106                    Ok(Bytes::from(
107                        resp(subscribe_response::Response::Pong(Pong { nonce: 9 })).encode_to_vec(),
108                    )),
109                ];
110                Ok(http::Response::new(BytesStream::new(stream::iter(frames))))
111            });
112
113        let client = BackendClient::new(mock);
114        let outbound = stream::iter(vec![
115            req(subscribe_request::Request::Update(Update {
116                id: 1,
117                ..Default::default()
118            })),
119            ping_req(3),
120        ])
121        .boxed();
122
123        let inbound = client.subscribe_bidi(outbound).await?;
124        let decoded: Vec<SubscribeResponse> = inbound.map(|r| r.unwrap()).collect().await;
125        assert_eq!(
126            decoded,
127            vec![
128                resp(subscribe_response::Response::Ping(Ping { nonce: 7 })),
129                resp(subscribe_response::Response::Pong(Pong { nonce: 9 })),
130            ],
131        );
132
133        // The outbound stream handed to the transport carries the same requests,
134        // prost-encoded, in order. Take it out of the mutex before awaiting so no
135        // lock is held across the `.await`.
136        let captured_outbound = captured.lock().unwrap().take().unwrap();
137        let sent_bytes: Vec<Bytes> = captured_outbound.collect().await;
138        let sent: Vec<SubscribeRequest> = sent_bytes
139            .iter()
140            .map(|b| SubscribeRequest::decode(b.clone()).unwrap())
141            .collect();
142        assert_eq!(
143            sent,
144            vec![
145                req(subscribe_request::Request::Update(Update {
146                    id: 1,
147                    ..Default::default()
148                })),
149                ping_req(3),
150            ],
151        );
152    }
153
154    /// A transport error opening the stream is surfaced as a `ClientWithEndpoint`
155    /// tagged with the `Subscribe` path, not a bare client error.
156    #[xmtp_common::test(unwrap_try = true)]
157    async fn tags_open_error_with_subscribe_endpoint() {
158        #[derive(Debug, thiserror::Error)]
159        #[error("boom")]
160        struct Boom;
161        impl xmtp_common::RetryableError for Boom {
162            fn is_retryable(&self) -> bool {
163                false
164            }
165        }
166
167        let mut mock = MockNetworkClient::new();
168        mock.expect_bidi_stream()
169            .return_once(|_req, _path, _body| Err(ApiClientError::client(Boom)));
170
171        let client = BackendClient::new(mock);
172        let outbound = stream::iter(vec![ping_req(1)]).boxed();
173
174        // `XmtpStream` isn't `Debug`, so match instead of `unwrap_err`.
175        match client.subscribe_bidi(outbound).await {
176            Err(ApiClientError::ClientWithEndpoint { endpoint, .. }) => {
177                assert_eq!(endpoint, "/xmtp.backend.v1.SubscriptionService/Subscribe");
178            }
179            Err(other) => panic!("expected ClientWithEndpoint, got {other:?}"),
180            Ok(_) => panic!("subscribe_bidi should error when the transport fails to open"),
181        }
182    }
183}
xmtp_api_backend/queries/backend/transport.rs

xmtp_api_backend/queries/backend/
transport.rs
