protocol_droid/application/bridge/
transport_peer.rs

1//! A bridge peer over any [`ClientTransport`].
2//!
3//! The HTTP and legacy-SSE peers carry the caller's envelope verbatim, because
4//! those transports have a connection that already speaks JSON-RPC. A transport
5//! that is built from `(method, params)` — one that writes its own envelope,
6//! like the `2026-07-28` stateless client, which adds the revision and the
7//! client's capabilities to every request's `_meta` — cannot take a finished
8//! envelope. This peer takes the envelope apart, hands the pieces to the
9//! transport, and puts the answer back together under **the caller's id**.
10//!
11//! The id is the caller's for the reason it is on every other peer: a proxy
12//! that renumbers it leaves the downstream client waiting for a reply carrying
13//! an id it never sent. The transport mints its own id for the upstream hop and
14//! it never reaches the downstream side.
15//!
16//! A peer that declines is not a failure. [`TransportError::Refused`] comes
17//! back as an envelope carrying an `error` member, exactly as the HTTP peer's
18//! does, because the exchange succeeded and the answer was no.
19//!
20//! This transport has no standing stream, so there are no unsolicited frames
21//! to hold: the trait's defaults are the true answer.
22
23use std::sync::Arc;
24
25use serde_json::{Map, Value, json};
26
27use crate::application::bridge::{BridgeError, BridgePeer, SendFuture};
28use crate::application::client::{ClientTransport, TransportError};
29use crate::domain::ir::JsonObject;
30
31/// Forwards envelopes through a [`ClientTransport`].
32#[derive(Clone)]
33pub struct TransportPeer {
34    transport: Arc<dyn ClientTransport>,
35}
36
37impl core::fmt::Debug for TransportPeer {
38    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
39        f.debug_struct("TransportPeer").finish_non_exhaustive()
40    }
41}
42
43impl TransportPeer {
44    /// A peer sending through `transport`.
45    #[must_use]
46    pub fn new(transport: Arc<dyn ClientTransport>) -> Self {
47        Self { transport }
48    }
49}
50
51/// An envelope's method and params, or why it has none.
52fn parts(envelope: &Map<String, Value>) -> Result<(&str, JsonObject), BridgeError> {
53    let method = envelope
54        .get("method")
55        .and_then(Value::as_str)
56        .ok_or_else(|| BridgeError::Unreachable {
57            reason: "the envelope has no method to forward".to_owned(),
58        })?;
59    let params = match envelope.get("params") {
60        None | Some(Value::Null) => json!({}),
61        Some(other) => other.clone(),
62    };
63    let params = JsonObject::from_value(params).map_err(|error| BridgeError::Unreachable {
64        reason: format!("the params are not a JSON object: {error}"),
65    })?;
66    Ok((method, params))
67}
68
69fn unreachable(error: impl core::fmt::Display) -> BridgeError {
70    BridgeError::Unreachable {
71        reason: error.to_string(),
72    }
73}
74
75impl BridgePeer for TransportPeer {
76    /// Delivers one request and yields the answer, under the caller's id.
77    ///
78    /// # Errors
79    ///
80    /// [`BridgeError::Unreachable`] when the request could not be delivered or
81    /// its answer could not be read. A refusal from the peer is an `Ok`
82    /// envelope with an `error` member.
83    fn send(
84        &self,
85        envelope: Map<String, Value>,
86    ) -> SendFuture<'_, Result<Map<String, Value>, BridgeError>> {
87        Box::pin(async move {
88            let (method, params) = parts(&envelope)?;
89            let id = envelope.get("id").cloned().unwrap_or(Value::Null);
90            let mut reply = Map::new();
91            reply.insert("jsonrpc".to_owned(), Value::from("2.0"));
92            reply.insert("id".to_owned(), id);
93            match self.transport.send(method, &params).await {
94                Ok(result) => {
95                    let result: Value =
96                        serde_json::from_str(result.as_str()).map_err(unreachable)?;
97                    reply.insert("result".to_owned(), result);
98                }
99                Err(TransportError::Refused {
100                    code,
101                    message,
102                    data,
103                }) => {
104                    let mut error = Map::new();
105                    error.insert("code".to_owned(), Value::from(code));
106                    error.insert("message".to_owned(), Value::from(message));
107                    if let Some(data) = data
108                        && let Ok(data) = serde_json::from_str::<Value>(&data)
109                    {
110                        error.insert("data".to_owned(), data);
111                    }
112                    reply.insert("error".to_owned(), Value::Object(error));
113                }
114                Err(other) => return Err(unreachable(other)),
115            }
116            Ok(reply)
117        })
118    }
119
120    /// Delivers a notification.
121    ///
122    /// # Errors
123    ///
124    /// [`BridgeError::Unreachable`] when it could not be delivered.
125    fn notify(&self, envelope: Map<String, Value>) -> SendFuture<'_, Result<(), BridgeError>> {
126        Box::pin(async move {
127            let (method, params) = parts(&envelope)?;
128            self.transport
129                .notify(method, &params)
130                .await
131                .map_err(unreachable)
132        })
133    }
134
135    /// There is no session to end.
136    fn close(&self) -> SendFuture<'_, Result<(), BridgeError>> {
137        Box::pin(async { Ok(()) })
138    }
139}
140
141#[cfg(test)]
142mod tests;
protocol_droid/application/bridge/transport_peer.rs

protocol_droid/application/bridge/
transport_peer.rs
