liminal_server/link_route_call/
client.rs

1//! A synchronous client proves the same sealed call protocol over stream carriers.
2
3use super::{CallError, CallPacket, CallRequest, Frame, LinkTransport};
4use crate::link_carrier::{CarrierReceive, LinkCarrier, WireCarrier};
5use crate::link_dial::DialedLink;
6
7/// Owns its carrier and Noise counters. Debug exposes counters, never frame bytes.
8pub struct CallClient<Carrier: LinkCarrier> {
9    carrier: Carrier,
10    transport: LinkTransport,
11    bound: usize,
12    frames_sent: u64,
13}
14impl<Carrier: LinkCarrier> std::fmt::Debug for CallClient<Carrier> {
15    fn fmt(&self, out: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
16        out.debug_struct("CallClient")
17            .field("frames_sent", &self.frames_sent)
18            .finish_non_exhaustive()
19    }
20}
21impl CallClient<WireCarrier> {
22    /// Uses the already admitted and proved wire link without another connection.
23    #[must_use]
24    pub fn wire(link: DialedLink, bound: usize) -> Self {
25        Self::new(link.carrier, link.transport, bound)
26    }
27}
28impl<Carrier: LinkCarrier> CallClient<Carrier> {
29    /// Uses an endpoint-keyed wire or relay carrier and its proved transport.
30    #[must_use]
31    pub const fn new(carrier: Carrier, transport: LinkTransport, bound: usize) -> Self {
32        Self {
33            carrier,
34            transport,
35            bound,
36            frames_sent: 0,
37        }
38    }
39    /// Sends an original caller request and returns its far conversation answer.
40    /// The far full connection owns its configured deadline and all pending slots.
41    ///
42    /// # Errors
43    /// Encoding/cap/correlation refusal, carrier end, or a named crossing refusal.
44    pub fn call(&mut self, request: CallRequest) -> Result<Frame, CallError> {
45        let id = request.id();
46        let correlation = request.correlation()?;
47        let subject = request.subject().to_owned();
48        let sealed = CallPacket::Request(request).seal(&mut self.transport, self.bound)?;
49        self.carrier
50            .send(&subject, &sealed)
51            .map_err(|error| self.peer_gone("send", &error))?;
52        self.frames_sent = self
53            .frames_sent
54            .checked_add(1)
55            .ok_or(CallError::Unavailable)?;
56        match self
57            .carrier
58            .receive()
59            .map_err(|error| self.peer_gone("receive", &error))?
60        {
61            CarrierReceive::Sealed {
62                subject: answered,
63                sealed,
64            } if answered == subject => {
65                let CallPacket::Reply(reply) =
66                    CallPacket::open(&mut self.transport, &sealed, self.bound)?
67                else {
68                    return Err(CallError::Correlation);
69                };
70                if reply.id() != id {
71                    return Err(CallError::Correlation);
72                }
73                let frame = reply.into_frame();
74                match &frame {
75                    Frame::ConversationMessage {
76                        stream_id,
77                        conversation_id,
78                        ..
79                    }
80                    | Frame::ConversationError {
81                        stream_id,
82                        conversation_id,
83                        ..
84                    } if (*stream_id, *conversation_id) == correlation => Ok(frame),
85                    _ => Err(CallError::Correlation),
86                }
87            }
88            CarrierReceive::Refused {
89                reason_code,
90                reason,
91            } => Ok(Frame::ConversationError {
92                flags: 0,
93                stream_id: correlation.0,
94                conversation_id: correlation.1,
95                reason_code,
96                message: reason,
97            }),
98            CarrierReceive::End { reason } => {
99                tracing::warn!(reason, "link call ended by its peer");
100                Err(CallError::PeerGone)
101            }
102            _ => Err(CallError::Correlation),
103        }
104    }
105    /// The carrier failed at `step`; the failure and the carrier's named end
106    /// are logged here, where they are known, before the one-word refusal.
107    fn peer_gone(&self, step: &str, error: &crate::link_carrier::CarrierError) -> CallError {
108        tracing::warn!(step, %error, end = ?self.carrier.ended(), "link call lost its carrier");
109        CallError::PeerGone
110    }
111    /// The carrier's named end behind a `PeerGone`: the peer's close reason,
112    /// `wire_eof`, or the failure that ended the carrier; `None` while live.
113    #[must_use]
114    pub fn end_reason(&self) -> Option<String> {
115        self.carrier.ended()
116    }
117    /// Actual application sends; constructing or holding an idle client sends none.
118    #[must_use]
119    pub const fn frames_sent(&self) -> u64 {
120        self.frames_sent
121    }
122}
liminal_server/link_route_call/client.rs

liminal_server/link_route_call/
client.rs
