use core::time::Duration;
use std::sync::Arc;
use shep_client::shep_core::values::UpDuration;
use tokio::task::JoinHandle;
use crate::{
bot::{
channel,
monitor::{Board, Monitor},
},
config::Config,
error::Error,
shepherd::Live,
stop::{self, Stop},
};
pub(super) struct Running {
pub(super) handle: Option<JoinHandle<()>>,
pub(super) request: stop::Request,
}
impl Running {
pub(super) fn active(&self) -> bool {
self.handle
.as_ref()
.is_none_or(|handle| !handle.is_finished())
}
}
pub struct Refresh<B> {
pub board: Arc<B>,
pub live: Arc<Live>,
pub ignore_dogs: bool,
pub interval: Duration,
}
impl<B: Board> Refresh<B> {
#[must_use]
pub fn new(board: &Arc<B>, config: &Config, interval: UpDuration, live: &Arc<Live>) -> Self {
Self {
board: Arc::clone(board),
live: Arc::clone(live),
ignore_dogs: config.ignore_dogs,
interval: Duration::from_millis(interval.as_millis()),
}
}
}
pub fn start<B: Board + Send + Sync + 'static>(
monitor: &Arc<Monitor>,
refresh: Refresh<B>,
) -> bool {
let mut task = monitor.task.lock().expect("not poisoned");
if task.as_ref().is_some_and(Running::active) {
return false;
}
let (mut stop, request) = Stop::new();
let monitor = Arc::clone(monitor);
let handle = tokio::spawn(async move {
let Refresh {
board,
live,
ignore_dogs,
interval,
} = refresh;
let board: &B = &board;
let wanted: Vec<u32> = match live.flock().await {
Ok(roll) => roll
.iter()
.filter(|info| !ignore_dogs || info.dog.is_none())
.map(|info| info.id)
.collect(),
Err(err) => {
eprintln!("shep-discord: {err}");
Vec::new()
}
};
match board.me().await {
Ok(me) => match channel::rediscover(board, me, &wanted).await {
Ok(found) => monitor.adopt(found),
Err(err) => eprintln!("shep-discord: {err}"),
},
Err(err) => eprintln!("shep-discord: {err}"),
}
let mut ticker = tokio::time::interval(interval);
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
loop {
tokio::select! {
biased;
() = stop.wait() => return,
_ = ticker.tick() => {
if let Err(err) = monitor.update_all(board, &live, ignore_dogs).await {
eprintln!("shep-discord: {err}");
}
}
}
}
});
*task = Some(Running {
handle: Some(handle),
request,
});
true
}
#[must_use]
pub fn is_running(monitor: &Monitor) -> bool {
monitor
.task
.lock()
.expect("not poisoned")
.as_ref()
.is_some_and(Running::active)
}
pub async fn stop(monitor: &Monitor) -> bool {
let handle = {
let mut task = monitor.task.lock().expect("not poisoned");
let Some(running) = task.as_mut() else {
return false;
};
running.request.request();
running.handle.take()
};
let Some(handle) = handle else {
return true;
};
if let Err(err) = handle.await {
eprintln!("{}", task_ended_badly_message(&err));
}
*monitor.task.lock().expect("not poisoned") = None;
true
}
pub async fn refresh_now<B: Board>(
monitor: &Monitor,
board: &B,
live: &Live,
ignore_dogs: bool,
) -> Result<Option<usize>, Error> {
if !is_running(monitor) {
return Ok(None);
}
monitor.update_all(board, live, ignore_dogs).await.map(Some)
}
pub fn start_from_config<B: Board + Send + Sync + 'static>(
monitor: &Arc<Monitor>,
config: &Config,
board: Option<&Arc<B>>,
live: &Arc<Live>,
) -> bool {
let (Some(interval), Some(board)) = (config.monitor_interval, board) else {
return false;
};
start(monitor, Refresh::new(board, config, interval, live))
}
pub(super) fn task_ended_badly_message(err: &tokio::task::JoinError) -> String {
if err.is_panic() {
format!(
"shep-discord: the monitor refresh task panicked: {err}. The messages it drew stay in the channel, and the next start adopts them."
)
} else {
format!("shep-discord: the monitor refresh task was cancelled: {err}.")
}
}
#[cfg(test)]
pub(super) fn mark_running_with_a_stand_in_task(monitor: &Monitor) {
let (mut stop, request) = Stop::new();
let handle = tokio::spawn(async move { stop.wait().await });
*monitor.task.lock().expect("not poisoned") = Some(Running {
handle: Some(handle),
request,
});
}
#[cfg(test)]
mod tests {
use core::sync::atomic::{AtomicBool, Ordering};
use shep_client::shep_core::{
protocol::{ProcessInfo, Request, Response},
status::ProcStatus,
};
use crate::test_support::{CountingChannel, test_live};
use super::*;
fn info(id: u32, name: &str) -> ProcessInfo {
ProcessInfo::builder(id, name, ProcStatus::Online).build()
}
#[tokio::test]
async fn a_redraw_is_refused_while_the_monitor_is_off() {
let (live, mut fake) = test_live().await;
let monitor = Monitor::new();
let sink = CountingChannel::new();
fake.expect(Request::ListFlock)
.answer(Response::Flock(vec![info(1, "web")]));
let redrawn = refresh_now(&monitor, &sink, &live, false)
.await
.expect("ok");
assert_eq!(redrawn, None, "there is no monitor to redraw");
assert_eq!((sink.sends(), sink.edits(), sink.deletes()), (0, 0, 0));
}
#[tokio::test]
async fn a_redraw_while_running_refreshes_the_whole_flock() {
let (live, mut fake) = test_live().await;
let monitor = Monitor::new();
let sink = CountingChannel::new();
mark_running_with_a_stand_in_task(&monitor);
fake.expect(Request::ListFlock)
.answer(Response::Flock(vec![info(1, "web"), info(2, "api")]));
let redrawn = refresh_now(&monitor, &sink, &live, false)
.await
.expect("ok");
assert_eq!(redrawn, Some(2), "both sheep were drawn");
assert_eq!(sink.sends(), 2);
}
#[tokio::test]
async fn a_config_that_does_not_ask_for_a_boot_monitor_starts_nothing() {
let (live, _fake) = test_live().await;
let live = Arc::new(live);
let monitor = Arc::new(Monitor::new());
for toml in [
"token = \"t\"\nguild_id = 1\n",
"token = \"t\"\nguild_id = 1\nmonitor_channel = 7\n",
"token = \"t\"\nguild_id = 1\nmonitor_interval = \"1m\"\n",
] {
let config = Config::from_toml(toml).expect("parsed");
let board = config
.monitor_channel
.map(|_channel| Arc::new(CountingChannel::new()));
assert!(
!start_from_config(&monitor, &config, board.as_ref(), &live),
"{toml:?} does not ask for a monitor from boot"
);
assert!(!is_running(&monitor));
}
}
#[tokio::test]
async fn a_config_that_asks_for_a_boot_monitor_starts_one() {
let (live, mut fake) = test_live().await;
let live = Arc::new(live);
let monitor = Arc::new(Monitor::new());
let config = Config::from_toml(
"token = \"t\"\nguild_id = 1\nmonitor_channel = 7\nmonitor_interval = \"1m\"\n",
)
.expect("parsed");
fake.expect(Request::ListFlock)
.answer(Response::Flock(Vec::new()));
let board = Arc::new(CountingChannel::new());
let started = start_from_config(&monitor, &config, Some(&board), &live);
assert!(started, "this config does ask for a monitor from boot");
assert!(is_running(&monitor));
assert!(stop(&monitor).await, "the task it started is there to stop");
}
#[tokio::test]
async fn nothing_printed_for_a_person_carries_a_dash() {
let panicked = tokio::spawn(async { panic!("deliberate") })
.await
.expect_err("the task panicked");
crate::test_support::assert_no_dashes(&task_ended_badly_message(&panicked));
}
#[tokio::test]
async fn stopping_waits_for_its_task_rather_than_abandoning_it() {
let monitor = Monitor::new();
let ended = Arc::new(AtomicBool::new(false));
let (mut signal, request) = Stop::new();
let flag = Arc::clone(&ended);
let handle = tokio::spawn(async move {
signal.wait().await;
tokio::task::yield_now().await;
flag.store(true, Ordering::SeqCst);
});
*monitor.task.lock().expect("not poisoned") = Some(Running {
handle: Some(handle),
request,
});
assert!(is_running(&monitor));
assert!(stop(&monitor).await);
assert!(
ended.load(Ordering::SeqCst),
"stop returned while its task was still running, so a start could spawn a second one \
beside it"
);
assert!(!is_running(&monitor));
assert!(!stop(&monitor).await, "there is nothing left to stop");
}
#[tokio::test]
async fn a_monitor_counts_as_running_while_its_task_is_draining() {
let monitor = Monitor::new();
let draining = Arc::new(AtomicBool::new(false));
let (release, released) = tokio::sync::oneshot::channel::<()>();
let (mut signal, request) = Stop::new();
let flag = Arc::clone(&draining);
let handle = tokio::spawn(async move {
signal.wait().await;
flag.store(true, Ordering::SeqCst);
let _ = released.await;
});
*monitor.task.lock().expect("not poisoned") = Some(Running {
handle: Some(handle),
request,
});
let watch = async {
while !draining.load(Ordering::SeqCst) {
tokio::task::yield_now().await;
}
let running = is_running(&monitor);
let _ = release.send(());
running
};
let (stopped, running_mid_drain) = tokio::join!(stop(&monitor), watch);
assert!(stopped);
assert!(
running_mid_drain,
"the monitor reported itself idle while its task was still draining, so a start \
racing this stop would have spawned a second refresh loop"
);
assert!(!is_running(&monitor), "and idle once the task is gone");
}
}