liminal-sdk 0.19.7

//! SDK-024 R2 and R3: the WebSocket client socket carries the server's
//! keepalive, and a read that owes a reply waits with no socket window.

use super::*;
use std::net::TcpListener;

type TestResult<T = ()> = Result<T, Box<dyn std::error::Error>>;

/// Longer than one 5 s socket window: the acceptor's hold, the test's
/// measuring span, not a wait the transport makes.
const HELD_REPLY: Duration = Duration::from_secs(6);

fn frame_bytes(frame: &Frame) -> TestResult<Vec<u8>> {
    Ok(encode_frame(frame)?)
}

/// An in-test acceptor: answers the Connect, then answers the next request
/// with a Pong after `hold`.
fn acceptor(listener: TcpListener, hold: Duration) -> std::thread::JoinHandle<Result<(), String>> {
    std::thread::spawn(move || -> Result<(), String> {
        let (stream, _) = listener.accept().map_err(|error| error.to_string())?;
        let mut socket = tungstenite::accept(stream).map_err(|error| error.to_string())?;
        let mut answer = |reply: Frame, hold: Duration| -> Result<(), String> {
            loop {
                let message = socket.read().map_err(|error| error.to_string())?;
                if message.is_binary() {
                    break;
                }
            }
            std::thread::sleep(hold);
            let bytes = frame_bytes(&reply).map_err(|error| error.to_string())?;
            socket
                .send(tungstenite::Message::binary(bytes))
                .map_err(|error| error.to_string())
        };
        answer(
            Frame::ConnectAck {
                flags: 0,
                selected_version: CLIENT_MAX_VERSION,
                capabilities: 0,
            },
            Duration::ZERO,
        )?;
        answer(Frame::Pong { flags: 0 }, hold)?;
        let _closed = socket.read();
        Ok(())
    })
}

fn opened(address: &str) -> TestResult<WsLink> {
    let mut driver = WebSocketFrameDriver::new();
    if driver
        .command_open()
        .map_err(|refusal| format!("{refusal:?}"))?
        != SocketCommand::Open
    {
        return Err("the driver's first command was not an open".into());
    }
    let socket = WsSocket::connect(
        &format!("ws://{address}/"),
        crate::remote::websocket::liminal_ws_message_bound()?,
    )?;
    if driver.handle_event(SocketEvent::Opened).output != DriverOutput::Opened {
        return Err("the driver refused the opened socket".into());
    }
    let mut link = WsLink {
        socket,
        driver,
        open_conversations: BTreeSet::new(),
        pump_fences_pending: 0,
        pump_diagnostics: ParticipantPumpDiagnostics::default(),
        negotiated_version: None,
        windows_closed: 0,
    };
    handshake(&mut link, b"").map_err(|error| described(&error))?;
    Ok(link)
}

fn described(error: &ExchangeError) -> String {
    match error {
        ExchangeError::Local(error) => error.to_string(),
        ExchangeError::Loss(loss) => loss.description.clone(),
    }
}

#[test]
fn the_websocket_socket_carries_the_servers_keepalive() -> TestResult {
    let listener = TcpListener::bind("127.0.0.1:0")?;
    let address = listener.local_addr()?.to_string();
    let server = acceptor(listener, Duration::ZERO);
    let mut link = opened(&address)?;
    let socket = socket2::SockRef::from(link.socket.tcp_for_tests());
    let read: (bool, Duration, Duration, u32) = (
        socket.keepalive()?,
        socket.tcp_keepalive_time()?,
        socket.tcp_keepalive_interval()?,
        socket.tcp_keepalive_retries()?,
    );
    let ping = frame_bytes(&Frame::Ping { flags: 0 })?;
    exchange_correlated(&mut link, ping).map_err(|error| described(&error))?;
    link.socket.execute_close();
    server.join().map_err(|_| "the acceptor panicked")??;
    assert_eq!(
        read,
        (true, Duration::from_secs(30), Duration::from_secs(10), 5),
        "SO_KEEPALIVE, idle, interval and retries"
    );
    Ok(())
}

#[test]
fn a_websocket_reply_held_past_one_window_closes_none() -> TestResult {
    let listener = TcpListener::bind("127.0.0.1:0")?;
    let address = listener.local_addr()?.to_string();
    let server = acceptor(listener, HELD_REPLY);
    let mut link = opened(&address)?;
    let ping = frame_bytes(&Frame::Ping { flags: 0 })?;
    let answered = exchange_correlated(&mut link, ping).map_err(|error| described(&error))?;
    link.socket.execute_close();
    server.join().map_err(|_| "the acceptor panicked")??;
    assert!(matches!(answered, Frame::Pong { .. }));
    assert_eq!(
        link.windows_closed, 0,
        "a read that owes a reply must wait with no 5 s socket window"
    );
    Ok(())
}