liminal-cli 0.3.2

//! AGT-006 R2: a seat speaks one transport, chosen by its server address
//! alone.
//!
//! A `ws://` or `wss://` server is the exchange's WebSocket door
//! (`wss://` through the named TLS-terminating proxy); any other address is
//! the raw wire. One never falls back to the other: a fallback would hide a
//! refused door.

use liminal_sdk::{
    DeliveredMessage, RemoteConfig, SdkError, SubscriptionStream, WebSocketSubscriptionStream,
};

/// A close signal that releases a blocked consumer and its socket reader.
pub type Close = Box<dyn Fn() -> Result<(), SdkError> + Send + Sync>;

/// Whether `address` names the exchange's WebSocket door.
#[must_use]
pub fn is_websocket(address: &str) -> bool {
    address.starts_with("ws://") || address.starts_with("wss://")
}

/// Installs on `config` the transport `address` (its server) names,
/// carrying `credential`.
///
/// # Errors
/// The SDK's refusal: an unusable address, a refused dial, handshake or pass.
pub fn connect(
    address: &str,
    config: RemoteConfig,
    credential: &[u8],
) -> Result<RemoteConfig, SdkError> {
    if is_websocket(address) {
        config.connect_websocket_with_auth(credential)
    } else {
        config.connect_tcp_with_auth(credential)
    }
}

/// One channel subscription on the seat's transport.
#[derive(Debug)]
pub enum Subscription {
    /// The raw wire.
    Tcp(SubscriptionStream),
    /// The WebSocket door.
    Ws(WebSocketSubscriptionStream),
}

impl Subscription {
    /// Subscribes to `channel` at `address` carrying `credential`.
    ///
    /// # Errors
    /// The SDK's refusal of the dial, the pass or the subscribe.
    pub fn open(address: &str, channel: &str, credential: &[u8]) -> Result<Self, SdkError> {
        if is_websocket(address) {
            WebSocketSubscriptionStream::open_with_auth(address, channel, Vec::new(), credential)
                .map(Self::Ws)
        } else {
            SubscriptionStream::open_with_auth(address, channel, Vec::new(), credential)
                .map(Self::Tcp)
        }
    }

    /// Blocks on a delivery or the subscription's end, without a timer.
    ///
    /// # Errors
    /// The server's terminal, a reader failure or a local close.
    pub fn recv(&self) -> Result<DeliveredMessage, SdkError> {
        match self {
            Self::Tcp(stream) => stream.recv(),
            Self::Ws(stream) => stream.recv().map(DeliveredMessage::from),
        }
    }

    /// Blocks up to `timeout` for the next delivery.
    ///
    /// # Errors
    /// No delivery within `timeout`, or the subscription's end.
    pub fn recv_timeout(&self, timeout: std::time::Duration) -> Result<DeliveredMessage, SdkError> {
        match self {
            Self::Tcp(stream) => stream.recv_timeout(timeout),
            Self::Ws(stream) => stream.recv_timeout(timeout).map(DeliveredMessage::from),
        }
    }

    /// A close signal held apart from the subscription.
    ///
    /// # Errors
    /// The socket-clone failure.
    pub fn close_handle(&self) -> Result<Close, SdkError> {
        Ok(match self {
            Self::Tcp(stream) => Box::new(stream.close_handle()?),
            Self::Ws(stream) => Box::new(stream.close_handle()?),
        })
    }
}

#[cfg(test)]
#[path = "rail_link_tests.rs"]
mod tests;