liminal-server 0.25.4

Standalone server for the liminal messaging bus
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122

//! A synchronous client proves the same sealed call protocol over stream carriers.

use super::{CallError, CallPacket, CallRequest, Frame, LinkTransport};
use crate::link_carrier::{CarrierReceive, LinkCarrier, WireCarrier};
use crate::link_dial::DialedLink;

/// Owns its carrier and Noise counters. Debug exposes counters, never frame bytes.
pub struct CallClient<Carrier: LinkCarrier> {
    carrier: Carrier,
    transport: LinkTransport,
    bound: usize,
    frames_sent: u64,
}
impl<Carrier: LinkCarrier> std::fmt::Debug for CallClient<Carrier> {
    fn fmt(&self, out: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        out.debug_struct("CallClient")
            .field("frames_sent", &self.frames_sent)
            .finish_non_exhaustive()
    }
}
impl CallClient<WireCarrier> {
    /// Uses the already admitted and proved wire link without another connection.
    #[must_use]
    pub fn wire(link: DialedLink, bound: usize) -> Self {
        Self::new(link.carrier, link.transport, bound)
    }
}
impl<Carrier: LinkCarrier> CallClient<Carrier> {
    /// Uses an endpoint-keyed wire or relay carrier and its proved transport.
    #[must_use]
    pub const fn new(carrier: Carrier, transport: LinkTransport, bound: usize) -> Self {
        Self {
            carrier,
            transport,
            bound,
            frames_sent: 0,
        }
    }
    /// Sends an original caller request and returns its far conversation answer.
    /// The far full connection owns its configured deadline and all pending slots.
    ///
    /// # Errors
    /// Encoding/cap/correlation refusal, carrier end, or a named crossing refusal.
    pub fn call(&mut self, request: CallRequest) -> Result<Frame, CallError> {
        let id = request.id();
        let correlation = request.correlation()?;
        let subject = request.subject().to_owned();
        let sealed = CallPacket::Request(request).seal(&mut self.transport, self.bound)?;
        self.carrier
            .send(&subject, &sealed)
            .map_err(|error| self.peer_gone("send", &error))?;
        self.frames_sent = self
            .frames_sent
            .checked_add(1)
            .ok_or(CallError::Unavailable)?;
        match self
            .carrier
            .receive()
            .map_err(|error| self.peer_gone("receive", &error))?
        {
            CarrierReceive::Sealed {
                subject: answered,
                sealed,
            } if answered == subject => {
                let CallPacket::Reply(reply) =
                    CallPacket::open(&mut self.transport, &sealed, self.bound)?
                else {
                    return Err(CallError::Correlation);
                };
                if reply.id() != id {
                    return Err(CallError::Correlation);
                }
                let frame = reply.into_frame();
                match &frame {
                    Frame::ConversationMessage {
                        stream_id,
                        conversation_id,
                        ..
                    }
                    | Frame::ConversationError {
                        stream_id,
                        conversation_id,
                        ..
                    } if (*stream_id, *conversation_id) == correlation => Ok(frame),
                    _ => Err(CallError::Correlation),
                }
            }
            CarrierReceive::Refused {
                reason_code,
                reason,
            } => Ok(Frame::ConversationError {
                flags: 0,
                stream_id: correlation.0,
                conversation_id: correlation.1,
                reason_code,
                message: reason,
            }),
            CarrierReceive::End { reason } => {
                tracing::warn!(reason, "link call ended by its peer");
                Err(CallError::PeerGone)
            }
            _ => Err(CallError::Correlation),
        }
    }
    /// The carrier failed at `step`; the failure and the carrier's named end
    /// are logged here, where they are known, before the one-word refusal.
    fn peer_gone(&self, step: &str, error: &crate::link_carrier::CarrierError) -> CallError {
        tracing::warn!(step, %error, end = ?self.carrier.ended(), "link call lost its carrier");
        CallError::PeerGone
    }
    /// The carrier's named end behind a `PeerGone`: the peer's close reason,
    /// `wire_eof`, or the failure that ended the carrier; `None` while live.
    #[must_use]
    pub fn end_reason(&self) -> Option<String> {
        self.carrier.ended()
    }
    /// Actual application sends; constructing or holding an idle client sends none.
    #[must_use]
    pub const fn frames_sent(&self) -> u64 {
        self.frames_sent
    }
}