liminal-server 0.25.4

//! A carrier names its end: the peer's close reason, `wire_eof`, or the
//! failure that ended it, so a caller's `PeerGone` can say what was seen.
use super::{CarrierError, CarrierReceive, LinkCarrier, WireCarrier, write_frame};
use crate::link::LinkLife;
use crate::link_handshake::MachineIdentity;
use liminal::protocol::Frame;
use std::net::{TcpListener, TcpStream};
use std::sync::{Arc, mpsc};

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

fn pair() -> Result<(WireCarrier, TcpStream), Box<dyn std::error::Error>> {
    let listener = TcpListener::bind("127.0.0.1:0")?;
    let near = TcpStream::connect(listener.local_addr()?)?;
    let (far, _) = listener.accept()?;
    Ok((WireCarrier::new(near, 1)?, far))
}

#[test]
fn a_live_carrier_names_no_end() -> TestResult {
    let (carrier, _far) = pair()?;
    assert_eq!(carrier.ended(), None);
    Ok(())
}

#[test]
fn a_peer_that_hangs_up_leaves_wire_eof_named_on_the_carrier() -> TestResult {
    let (mut carrier, far) = pair()?;
    drop(far);
    assert!(matches!(
        carrier.receive()?,
        CarrierReceive::End { reason } if reason == "wire_eof"
    ));
    assert_eq!(carrier.ended().as_deref(), Some("wire_eof"));
    Ok(())
}

#[test]
fn a_peers_disconnect_leaves_its_name_on_the_carrier() -> TestResult {
    let (mut carrier, mut far) = pair()?;
    write_frame(&mut far, &Frame::Disconnect { flags: 0 })?;
    assert!(matches!(
        carrier.receive()?,
        CarrierReceive::End { reason } if reason == "peer_disconnect"
    ));
    assert_eq!(carrier.ended().as_deref(), Some("peer_disconnect"));
    Ok(())
}

#[test]
fn an_unexpected_frame_leaves_the_failure_named_on_the_carrier() -> TestResult {
    let (mut carrier, mut far) = pair()?;
    write_frame(&mut far, &Frame::Ping { flags: 0 })?;
    let Err(error) = carrier.receive() else {
        return Err("an unexpected frame was received as a message".into());
    };
    assert!(matches!(error, CarrierError::UnexpectedFrame));
    assert_eq!(
        carrier.ended(),
        Some(CarrierError::UnexpectedFrame.to_string())
    );
    Ok(())
}

/// A carrier whose reader thread is running, and a channel the link's life
/// signals on every publication (its end among them).
fn reading(carrier: &mut WireCarrier) -> Result<mpsc::Receiver<()>, Box<dyn std::error::Error>> {
    let life = LinkLife::new(MachineIdentity::new([7; 32]));
    let (signal, signalled) = mpsc::channel();
    life.notify_with(Arc::new(move || {
        if signal.send(()).is_err() {
            tracing::debug!("the test departed before the life's signal");
        }
    }))?;
    carrier.install_life(life)?;
    Ok(signalled)
}

#[test]
fn a_peer_that_hangs_up_is_named_through_the_reader() -> TestResult {
    let (mut carrier, far) = pair()?;
    let _signalled = reading(&mut carrier)?;
    drop(far);
    assert!(matches!(
        carrier.receive()?,
        CarrierReceive::End { reason } if reason == "wire_eof"
    ));
    assert_eq!(carrier.ended().as_deref(), Some("wire_eof"));
    Ok(())
}

#[test]
fn a_send_refused_after_the_reader_saw_the_end_names_that_end() -> TestResult {
    let (mut carrier, far) = pair()?;
    let signalled = reading(&mut carrier)?;
    drop(far);
    signalled.recv()?;
    let Err(error) = carrier.send("embed.bge", b"sealed") else {
        return Err("a send after the peer hung up was admitted".into());
    };
    assert!(matches!(error, CarrierError::Ended), "{error}");
    assert_eq!(carrier.ended().as_deref(), Some("wire_eof"));
    Ok(())
}