use super::super::fan;
use super::{echo, engine_at, flapping_at, refusal, set};
use crate::channel::entries;
use crate::run::{Notice, Pause};
use crate::test_support::{Notices, Scratch};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::{Receiver, SyncSender};
use std::sync::{Arc, Mutex, PoisonError};
use std::time::Duration;
const ENDED: &str = "south: south stopped";
const BOUND: Duration = Duration::from_secs(10);
#[test]
fn a_terminal_sentence_is_said_while_a_sibling_is_still_serving() {
let scratch = Scratch::new();
let _north = flapping_at(
&scratch.join("north"),
vec![None, Some(refusal("north stopped"))],
);
let _south = engine_at(&scratch.join("south"), vec![refusal("south stopped")]);
let (notices, into) = Notices::new();
let (release, awaited) = std::sync::mpsc::sync_channel(1);
let sink: Notice = Arc::new(move |line: &str| watch(&into, &release, line));
let released = Arc::new(AtomicBool::new(false));
let flag = Arc::clone(&released);
let held = Mutex::new(awaited);
let pause: Pause = Arc::new(move |_| wait_out(&held, &flag));
let said = fan(
entries::read_dir(scratch.path()),
set(),
echo,
&sink,
&pause,
);
assert!(
released.load(Ordering::Relaxed),
"north waited out {BOUND:?} for a sentence south had already earned: {:?}",
notices.heard()
);
assert_eq!(said, ["north: north stopped", "south: south stopped"]);
assert!(notices.heard().contains(&ENDED.to_owned()), "{said:?}");
}
fn watch(into: &Notice, release: &SyncSender<String>, line: &str) {
into(line);
if line == ENDED {
let _ = release.try_send(line.to_owned());
}
}
fn wait_out(held: &Mutex<Receiver<String>>, flag: &AtomicBool) {
let got = held
.lock()
.unwrap_or_else(PoisonError::into_inner)
.recv_timeout(BOUND);
flag.store(got.is_ok(), Ordering::Relaxed);
}