liminal-cli 0.3.1

//! The publish connection's one link, chosen by the seat's server address
//! (AGT-006 R2): the raw wire, or the exchange's WebSocket door, where each
//! binary message is exactly one liminal frame. No fallback between them.

use std::io::{Read, Write};
use std::net::TcpStream;
use std::time::Duration;

use tungstenite::protocol::{WebSocket, WebSocketConfig};
use tungstenite::stream::MaybeTlsStream;
use tungstenite::{HandshakeError, Message};

pub(super) enum Link {
    Tcp(TcpStream),
    Ws {
        socket: Box<WebSocket<MaybeTlsStream<TcpStream>>>,
        /// The binary message being read and how much of it is consumed.
        message: Vec<u8>,
        at: usize,
    },
}

/// Opens the link `address` names, every read and write bounded by `timeout`
/// and every WebSocket message by `bound`.
pub(super) fn open(address: &str, timeout: Duration, bound: usize) -> Result<Link, String> {
    let websocket = crate::rail_link::is_websocket(address);
    let authority = if websocket {
        authority(address)?
    } else {
        address.to_owned()
    };
    let stream = TcpStream::connect(&authority).map_err(|e| format!("{authority}: {e}"))?;
    stream
        .set_read_timeout(Some(timeout))
        .and_then(|()| stream.set_write_timeout(Some(timeout)))
        .map_err(|e| e.to_string())?;
    if !websocket {
        return Ok(Link::Tcp(stream));
    }
    let config = WebSocketConfig::default()
        .max_message_size(Some(bound))
        .max_frame_size(Some(bound));
    match tungstenite::client_tls_with_config(address, stream, Some(config), None) {
        Ok((socket, _)) => Ok(Link::Ws {
            socket: Box::new(socket),
            message: Vec::new(),
            at: 0,
        }),
        Err(HandshakeError::Failure(error)) => Err(format!(
            "websocket handshake with {address} failed: {error}"
        )),
        Err(HandshakeError::Interrupted(_)) => Err(format!(
            "websocket handshake with {address} was interrupted on a blocking socket"
        )),
    }
}

/// The `host:port` a `ws://` or `wss://` address dials.
fn authority(address: &str) -> Result<String, String> {
    let uri: tungstenite::http::Uri = address
        .parse()
        .map_err(|e| format!("invalid websocket address {address}: {e}"))?;
    let host = uri
        .host()
        .ok_or_else(|| format!("websocket address {address} has no host"))?;
    let scheme_port = if uri.scheme_str() == Some("wss") {
        443
    } else {
        80
    };
    let port = uri.port_u16().unwrap_or(scheme_port);
    Ok(format!("{host}:{port}"))
}

impl Link {
    /// Sends one whole frame.
    pub(super) fn send(&mut self, frame: &[u8]) -> Result<(), String> {
        match self {
            Self::Tcp(socket) => socket
                .write_all(frame)
                .and_then(|()| socket.flush())
                .map_err(|e| e.to_string()),
            Self::Ws { socket, .. } => socket
                .send(Message::Binary(frame.to_vec().into()))
                .map_err(|e| e.to_string()),
        }
    }

    /// Fills `buf` from the frame being read.
    pub(super) fn read_exact(&mut self, buf: &mut [u8]) -> Result<(), String> {
        match self {
            Self::Tcp(socket) => socket.read_exact(buf).map_err(|e| e.to_string()),
            Self::Ws {
                socket,
                message,
                at,
            } => {
                if *at == message.len() {
                    *message = next_binary(socket)?;
                    *at = 0;
                }
                let end = at
                    .checked_add(buf.len())
                    .filter(|end| *end <= message.len())
                    .ok_or("a websocket message ended inside a liminal frame")?;
                buf.copy_from_slice(&message[*at..end]);
                *at = end;
                Ok(())
            }
        }
    }

    /// Checks a whole frame was one whole WebSocket message.
    pub(super) fn frame_ended(&mut self) -> Result<(), String> {
        match self {
            Self::Tcp(_) => Ok(()),
            Self::Ws { message, at, .. } if *at == message.len() => Ok(()),
            Self::Ws { .. } => {
                Err("one websocket message carried more than one liminal frame".to_owned())
            }
        }
    }
}

/// The next binary message; the library answers pings itself.
fn next_binary(socket: &mut WebSocket<MaybeTlsStream<TcpStream>>) -> Result<Vec<u8>, String> {
    loop {
        match socket.read().map_err(|e| e.to_string())? {
            Message::Binary(bytes) => return Ok(bytes.to_vec()),
            Message::Ping(_) | Message::Pong(_) => {}
            Message::Close(_) => return Err("the exchange closed the publish door".to_owned()),
            Message::Text(_) | Message::Frame(_) => {
                return Err("the publish door sent a message that is not a frame".to_owned());
            }
        }
    }
}