liminal-sdk 0.16.0

use super::*;
use std::error::Error;
use std::net::TcpListener;
use std::sync::mpsc;

type TestResult = Result<(), Box<dyn Error + Send + Sync>>;

struct Fixture {
    server: TcpStream,
    client: PushClient,
}

impl Fixture {
    fn new() -> Result<Self, Box<dyn Error + Send + Sync>> {
        let listener = TcpListener::bind("127.0.0.1:0")?;
        let socket = TcpStream::connect(listener.local_addr()?)?;
        let (server, _) = listener.accept()?;
        let client = PushClient::start_reader(socket)?;
        Ok(Self { server, client })
    }

    fn push(&mut self) -> Result<(), SdkError> {
        write_frame(
            &mut self.server,
            &Frame::Push {
                flags: 0,
                stream_id: APPLICATION_STREAM_ID,
                correlation_id: 7,
                payload: vec![1, 2, 3],
            },
        )
    }
}

fn assert_frame(outcome: PushReceiveOutcome) -> TestResult {
    let PushReceiveOutcome::Frame(frame) = outcome else {
        return Err("receive ended before the pushed frame".into());
    };
    assert_eq!(frame.correlation_id, 7);
    assert_eq!(frame.payload, [1, 2, 3]);
    Ok(())
}

#[test]
fn receiver_claims_once_and_direct_receives_leave_the_push_untouched() -> TestResult {
    let mut fixture = Fixture::new()?;
    let receiver = fixture.client.receiver_handle()?;
    assert!(matches!(
        fixture.client.receiver_handle(),
        Err(SdkError::ReceiverAlreadyClaimed)
    ));
    assert_eq!(
        fixture.client.recv().err(),
        Some(SdkError::ReceiverAlreadyClaimed)
    );
    assert_eq!(
        fixture.client.recv_timeout(Duration::ZERO).err(),
        Some(SdkError::ReceiverAlreadyClaimed)
    );
    fixture.push()?;
    assert_frame(receiver.recv()?)?;
    Ok(())
}

#[test]
fn idle_receiver_has_no_wakes_or_timers_then_one_push_wakes_it_once() -> TestResult {
    let mut fixture = Fixture::new()?;
    let receiver = fixture.client.receiver_handle()?;
    let (waiting, entered) = mpsc::channel();
    fixture.client.mailbox.observe_blocking_wait(waiting)?;
    let task = std::thread::spawn(move || receiver.recv());
    entered.recv()?;
    assert_eq!(fixture.client.mailbox.receive_wait_counts()?, (0, 0));
    fixture.push()?;
    assert_frame(task.join().map_err(|_| "receive task panicked")??)?;
    assert_eq!(fixture.client.mailbox.receive_wait_counts()?, (1, 0));
    Ok(())
}

#[test]
fn owner_drop_wakes_the_receiver_once_with_owner_close() -> TestResult {
    let Fixture { mut server, client } = Fixture::new()?;
    let receiver = client.receiver_handle()?;
    let mailbox = Arc::clone(&client.mailbox);
    let (waiting, entered) = mpsc::channel();
    mailbox.observe_blocking_wait(waiting)?;
    let task = std::thread::spawn(move || receiver.recv());
    entered.recv()?;
    let peer = std::thread::spawn(move || -> std::io::Result<()> {
        let mut bytes = [0; 32];
        while server.read(&mut bytes)? != 0 {}
        Ok(())
    });
    drop(client);
    assert!(matches!(
        task.join().map_err(|_| "receive task panicked")??,
        PushReceiveOutcome::Closed {
            cause: PushCloseCause::OwnerClose
        }
    ));
    peer.join().map_err(|_| "peer task panicked")??;
    assert_eq!(mailbox.receive_wait_counts()?, (1, 0));
    Ok(())
}

#[test]
fn owner_close_reports_owner_close_to_an_existing_receiver() -> TestResult {
    let Fixture { mut server, client } = Fixture::new()?;
    let receiver = client.receiver_handle()?;
    let peer = std::thread::spawn(move || -> std::io::Result<()> {
        let mut bytes = [0; 32];
        while server.read(&mut bytes)? != 0 {}
        Ok(())
    });
    client.close()?;
    assert!(matches!(
        receiver.recv()?,
        PushReceiveOutcome::Closed {
            cause: PushCloseCause::OwnerClose
        }
    ));
    peer.join().map_err(|_| "peer task panicked")??;
    Ok(())
}

#[test]
fn peer_eof_is_distinct_from_owner_close() -> TestResult {
    let fixture = Fixture::new()?;
    let receiver = fixture.client.receiver_handle()?;
    fixture.server.shutdown(Shutdown::Write)?;
    assert!(matches!(
        receiver.recv()?,
        PushReceiveOutcome::Closed {
            cause: PushCloseCause::PeerClose
        }
    ));
    Ok(())
}

#[test]
fn partial_frame_eof_retains_the_reader_failure() -> TestResult {
    let mut fixture = Fixture::new()?;
    let receiver = fixture.client.receiver_handle()?;
    fixture.server.write_all(&[0xFF])?;
    fixture.server.shutdown(Shutdown::Write)?;
    assert!(matches!(receiver.recv()?,
        PushReceiveOutcome::Closed {
            cause: PushCloseCause::ReaderFailure {
                error: SdkError::Connection { description }
            }
        } if description == "server closed the push connection"));
    Ok(())
}

#[test]
fn an_active_direct_receive_refuses_a_claim_until_it_finishes() -> TestResult {
    let mut fixture = Fixture::new()?;
    let (waiting, entered) = mpsc::channel();
    fixture.client.mailbox.observe_blocking_wait(waiting)?;
    std::thread::scope(|scope| -> TestResult {
        let client = &fixture.client;
        let task = scope.spawn(move || client.recv());
        entered.recv()?;
        assert!(matches!(
            fixture.client.receiver_handle(),
            Err(SdkError::ReceiverAlreadyClaimed)
        ));
        write_frame(
            &mut fixture.server,
            &Frame::Push {
                flags: 0,
                stream_id: APPLICATION_STREAM_ID,
                correlation_id: 7,
                payload: vec![1, 2, 3],
            },
        )?;
        assert_frame(task.join().map_err(|_| "direct receive task panicked")??)?;
        Ok(())
    })?;
    let receiver = fixture.client.receiver_handle()?;
    fixture.push()?;
    assert_frame(receiver.recv()?)?;
    Ok(())
}