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

//! The relay carrier: an ordinary wire connection to the relay on its
//! admitted relay stream, answering receipts and refusals on it.
use super::{Carrier, CarrierError, CarrierReceive, LinkCarrier, WireCarrier, write_frame};
use liminal::protocol::Frame;
use std::net::TcpStream;

/// An ordinary wire connection to the relay on its admitted relay stream.
#[derive(Debug)]
pub struct RelayCarrier {
    wire: WireCarrier,
    relay: crate::link_handshake::MachineIdentity,
}

impl RelayCarrier {
    /// Takes a connection whose relay stream has been granted and marked.
    ///
    /// # Errors
    /// Stream zero is reserved for connection control.
    pub fn new(
        socket: TcpStream,
        stream_id: u32,
        relay: crate::link_handshake::MachineIdentity,
    ) -> Result<Self, CarrierError> {
        Ok(Self {
            wire: WireCarrier::new(socket, stream_id)?,
            relay,
        })
    }

    /// Interrupts a blocking receive through the same owned socket.
    ///
    /// # Errors
    /// The socket cannot be duplicated.
    pub fn interrupt_handle(&self) -> Result<TcpStream, CarrierError> {
        self.wire.interrupt_handle()
    }

    /// Answers an imported frame; the relay carries the full receipt back.
    ///
    /// # Errors
    /// The carrier has ended or the ordinary connection cannot send the ack.
    pub fn acknowledge(
        &mut self,
        message_id: u64,
        durable_position: Option<u64>,
    ) -> Result<(), CarrierError> {
        self.reply(&Frame::PublishAck {
            flags: 0,
            stream_id: self.wire.stream_id,
            message_id,
            durable_position,
        })
    }

    /// Answers a named endpoint refusal on the relay stream.
    ///
    /// # Errors
    /// The carrier has ended or the ordinary connection cannot send the refusal.
    pub fn refuse(&mut self, reason_code: u16, reason: &str) -> Result<(), CarrierError> {
        self.reply(&Frame::PublishError {
            flags: 0,
            stream_id: self.wire.stream_id,
            reason_code,
            message: Some(reason.to_owned()),
        })
    }

    /// The topology record for the endpoint's observed relayed link.
    #[must_use]
    pub fn topology(
        &self,
        local: crate::link_handshake::MachineIdentity,
        peer: crate::link_handshake::MachineIdentity,
        counters: crate::console::records::LinkCounters,
    ) -> crate::console::records::Link {
        crate::console::records::Link::federated(
            &hex::encode(local.ed25519_public()),
            &hex::encode(peer.ed25519_public()),
            Some(crate::console::records::Carrier::Relay(hex::encode(
                self.relay.ed25519_public(),
            ))),
            crate::console::records::LinkState::Up,
            None,
            counters,
        )
    }

    fn reply(&mut self, frame: &Frame) -> Result<(), CarrierError> {
        if self.wire.ended.is_some() {
            return Err(CarrierError::Ended);
        }
        write_frame(&mut self.wire.socket, frame)
    }
}

impl LinkCarrier for RelayCarrier {
    fn refuse_unsolicited(&self) -> Result<(), CarrierError> {
        self.wire.refuse_unsolicited()
    }
    fn carrier(&self) -> Carrier {
        Carrier::Relay
    }
    fn ended(&self) -> Option<String> {
        self.wire.ended()
    }
    fn send(&mut self, subject: &str, sealed: &[u8]) -> Result<(), CarrierError> {
        self.wire.send(subject, sealed)
    }
    fn receive(&mut self) -> Result<CarrierReceive, CarrierError> {
        self.wire.receive()
    }
}