liminal-server 0.25.4

use std::io::{self, ErrorKind};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::{channel, sync_channel};

use super::Reader;
use crate::test_log::CapturedLog;

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

fn error_count(log: &CapturedLog) -> usize {
    log.contents()
        .lines()
        .filter(|line| line.contains("ERROR"))
        .count()
}

fn eof_reader() -> Result<(Reader, Arc<AtomicBool>), Box<dyn std::error::Error>> {
    let (sender, inbox) = sync_channel(1);
    let completed = Arc::new(AtomicBool::new(false));
    let worker_completed = Arc::clone(&completed);
    let worker = std::thread::Builder::new().spawn(move || {
        assert!(sender.send(Ok(None)).is_ok());
        worker_completed.store(true, Ordering::Release);
    })?;
    let reader = Reader {
        inbox: Some(inbox),
        worker: Some(worker),
    };
    assert!(reader.receive()?.is_none());
    Ok((reader, completed))
}

#[test]
fn closing_a_reader_whose_peer_already_hung_up_logs_no_error() -> TestResult {
    let (reader, completed) = eof_reader()?;
    let log = CapturedLog::default();
    log.capture(|| {
        reader.close_with(|| {
            Err(io::Error::new(
                ErrorKind::NotConnected,
                "peer socket is already gone",
            ))
        });
    });
    assert!(
        completed.load(Ordering::Acquire),
        "close joined the EOF reader"
    );
    assert_eq!(error_count(&log), 0, "{}", log.contents());
    Ok(())
}

#[test]
fn a_reader_shutdown_failure_other_than_not_connected_logs_its_error() -> TestResult {
    let (reader, completed) = eof_reader()?;
    let log = CapturedLog::default();
    log.capture(|| {
        reader.close_with(|| {
            Err(io::Error::new(
                ErrorKind::PermissionDenied,
                "reader shutdown refused",
            ))
        });
    });
    assert!(completed.load(Ordering::Acquire));
    assert_eq!(error_count(&log), 1, "{}", log.contents());
    assert!(
        log.contents()
            .contains("wire link reader interruption failed")
    );
    assert!(log.contents().contains("reader shutdown refused"));
    Ok(())
}

#[test]
fn closing_a_reader_releases_its_full_inbox_before_joining() -> TestResult {
    let (sender, inbox) = sync_channel(1);
    let (filled, ready) = channel();
    let completed = Arc::new(AtomicBool::new(false));
    let worker_completed = Arc::clone(&completed);
    let worker = std::thread::Builder::new().spawn(move || {
        assert!(sender.send(Ok(None)).is_ok());
        assert!(filled.send(()).is_ok());
        assert!(sender.send(Ok(None)).is_err());
        worker_completed.store(true, Ordering::Release);
    })?;
    ready.recv()?;
    let reader = Reader {
        inbox: Some(inbox),
        worker: Some(worker),
    };
    let log = CapturedLog::default();
    log.capture(|| reader.close_with(|| Ok(())));
    assert!(completed.load(Ordering::Acquire));
    assert_eq!(error_count(&log), 0, "{}", log.contents());
    Ok(())
}

#[test]
fn an_already_interrupted_reader_still_reports_its_worker_panic() -> TestResult {
    let worker = std::thread::Builder::new().spawn(|| {
        std::panic::resume_unwind(Box::new("reader worker panic fixture"));
    })?;
    let reader = Reader {
        inbox: None,
        worker: Some(worker),
    };
    let log = CapturedLog::default();
    log.capture(|| {
        reader.close_with(|| {
            Err(io::Error::new(
                ErrorKind::NotConnected,
                "peer socket is already gone",
            ))
        });
    });
    assert_eq!(error_count(&log), 1, "{}", log.contents());
    assert!(log.contents().contains("wire link reader panicked"));
    assert!(
        !log.contents()
            .contains("wire link reader interruption failed")
    );
    Ok(())
}