liminal_server/link_carrier/
relay.rs

1//! The relay carrier: an ordinary wire connection to the relay on its
2//! admitted relay stream, answering receipts and refusals on it.
3use super::{Carrier, CarrierError, CarrierReceive, LinkCarrier, WireCarrier, write_frame};
4use liminal::protocol::Frame;
5use std::net::TcpStream;
6
7/// An ordinary wire connection to the relay on its admitted relay stream.
8#[derive(Debug)]
9pub struct RelayCarrier {
10    wire: WireCarrier,
11    relay: crate::link_handshake::MachineIdentity,
12}
13
14impl RelayCarrier {
15    /// Takes a connection whose relay stream has been granted and marked.
16    ///
17    /// # Errors
18    /// Stream zero is reserved for connection control.
19    pub fn new(
20        socket: TcpStream,
21        stream_id: u32,
22        relay: crate::link_handshake::MachineIdentity,
23    ) -> Result<Self, CarrierError> {
24        Ok(Self {
25            wire: WireCarrier::new(socket, stream_id)?,
26            relay,
27        })
28    }
29
30    /// Interrupts a blocking receive through the same owned socket.
31    ///
32    /// # Errors
33    /// The socket cannot be duplicated.
34    pub fn interrupt_handle(&self) -> Result<TcpStream, CarrierError> {
35        self.wire.interrupt_handle()
36    }
37
38    /// Answers an imported frame; the relay carries the full receipt back.
39    ///
40    /// # Errors
41    /// The carrier has ended or the ordinary connection cannot send the ack.
42    pub fn acknowledge(
43        &mut self,
44        message_id: u64,
45        durable_position: Option<u64>,
46    ) -> Result<(), CarrierError> {
47        self.reply(&Frame::PublishAck {
48            flags: 0,
49            stream_id: self.wire.stream_id,
50            message_id,
51            durable_position,
52        })
53    }
54
55    /// Answers a named endpoint refusal on the relay stream.
56    ///
57    /// # Errors
58    /// The carrier has ended or the ordinary connection cannot send the refusal.
59    pub fn refuse(&mut self, reason_code: u16, reason: &str) -> Result<(), CarrierError> {
60        self.reply(&Frame::PublishError {
61            flags: 0,
62            stream_id: self.wire.stream_id,
63            reason_code,
64            message: Some(reason.to_owned()),
65        })
66    }
67
68    /// The topology record for the endpoint's observed relayed link.
69    #[must_use]
70    pub fn topology(
71        &self,
72        local: crate::link_handshake::MachineIdentity,
73        peer: crate::link_handshake::MachineIdentity,
74        counters: crate::console::records::LinkCounters,
75    ) -> crate::console::records::Link {
76        crate::console::records::Link::federated(
77            &hex::encode(local.ed25519_public()),
78            &hex::encode(peer.ed25519_public()),
79            Some(crate::console::records::Carrier::Relay(hex::encode(
80                self.relay.ed25519_public(),
81            ))),
82            crate::console::records::LinkState::Up,
83            None,
84            counters,
85        )
86    }
87
88    fn reply(&mut self, frame: &Frame) -> Result<(), CarrierError> {
89        if self.wire.ended.is_some() {
90            return Err(CarrierError::Ended);
91        }
92        write_frame(&mut self.wire.socket, frame)
93    }
94}
95
96impl LinkCarrier for RelayCarrier {
97    fn refuse_unsolicited(&self) -> Result<(), CarrierError> {
98        self.wire.refuse_unsolicited()
99    }
100    fn carrier(&self) -> Carrier {
101        Carrier::Relay
102    }
103    fn ended(&self) -> Option<String> {
104        self.wire.ended()
105    }
106    fn send(&mut self, subject: &str, sealed: &[u8]) -> Result<(), CarrierError> {
107        self.wire.send(subject, sealed)
108    }
109    fn receive(&mut self) -> Result<CarrierReceive, CarrierError> {
110        self.wire.receive()
111    }
112}
liminal_server/link_carrier/relay.rs

liminal_server/link_carrier/
relay.rs
