liminal_cli/
rail_link.rs

1//! AGT-006 R2: a seat speaks one transport, chosen by its server address
2//! alone.
3//!
4//! A `ws://` or `wss://` server is the exchange's WebSocket door
5//! (`wss://` through the named TLS-terminating proxy); any other address is
6//! the raw wire. One never falls back to the other: a fallback would hide a
7//! refused door.
8
9use liminal_sdk::{
10    DeliveredMessage, RemoteConfig, SdkError, SubscriptionStream, WebSocketSubscriptionStream,
11};
12
13/// A close signal that releases a blocked consumer and its socket reader.
14pub type Close = Box<dyn Fn() -> Result<(), SdkError> + Send + Sync>;
15
16/// Whether `address` names the exchange's WebSocket door.
17#[must_use]
18pub fn is_websocket(address: &str) -> bool {
19    address.starts_with("ws://") || address.starts_with("wss://")
20}
21
22/// Installs on `config` the transport `address` (its server) names,
23/// carrying `credential`.
24///
25/// # Errors
26/// The SDK's refusal: an unusable address, a refused dial, handshake or pass.
27pub fn connect(
28    address: &str,
29    config: RemoteConfig,
30    credential: &[u8],
31) -> Result<RemoteConfig, SdkError> {
32    if is_websocket(address) {
33        config.connect_websocket_with_auth(credential)
34    } else {
35        config.connect_tcp_with_auth(credential)
36    }
37}
38
39/// One channel subscription on the seat's transport.
40#[derive(Debug)]
41pub enum Subscription {
42    /// The raw wire.
43    Tcp(SubscriptionStream),
44    /// The WebSocket door.
45    Ws(WebSocketSubscriptionStream),
46}
47
48impl Subscription {
49    /// Subscribes to `channel` at `address` carrying `credential`.
50    ///
51    /// # Errors
52    /// The SDK's refusal of the dial, the pass or the subscribe.
53    pub fn open(address: &str, channel: &str, credential: &[u8]) -> Result<Self, SdkError> {
54        if is_websocket(address) {
55            WebSocketSubscriptionStream::open_with_auth(address, channel, Vec::new(), credential)
56                .map(Self::Ws)
57        } else {
58            SubscriptionStream::open_with_auth(address, channel, Vec::new(), credential)
59                .map(Self::Tcp)
60        }
61    }
62
63    /// Blocks on a delivery or the subscription's end, without a timer.
64    ///
65    /// # Errors
66    /// The server's terminal, a reader failure or a local close.
67    pub fn recv(&self) -> Result<DeliveredMessage, SdkError> {
68        match self {
69            Self::Tcp(stream) => stream.recv(),
70            Self::Ws(stream) => stream.recv().map(DeliveredMessage::from),
71        }
72    }
73
74    /// Blocks up to `timeout` for the next delivery.
75    ///
76    /// # Errors
77    /// No delivery within `timeout`, or the subscription's end.
78    pub fn recv_timeout(&self, timeout: std::time::Duration) -> Result<DeliveredMessage, SdkError> {
79        match self {
80            Self::Tcp(stream) => stream.recv_timeout(timeout),
81            Self::Ws(stream) => stream.recv_timeout(timeout).map(DeliveredMessage::from),
82        }
83    }
84
85    /// A close signal held apart from the subscription.
86    ///
87    /// # Errors
88    /// The socket-clone failure.
89    pub fn close_handle(&self) -> Result<Close, SdkError> {
90        Ok(match self {
91            Self::Tcp(stream) => Box::new(stream.close_handle()?),
92            Self::Ws(stream) => Box::new(stream.close_handle()?),
93        })
94    }
95}
96
97#[cfg(test)]
98#[path = "rail_link_tests.rs"]
99mod tests;
liminal_cli/rail_link.rs

liminal_cli/
rail_link.rs
