use std::sync::{
Arc, Mutex,
atomic::{AtomicUsize, Ordering},
mpsc,
};
use std::thread::{self, JoinHandle};
use std::time::Duration;
use timerwheel::executor::{BoxTask, Executor, ExecutorMetrics, Pool, RejectedTask};
use timerwheel::{Error, Timer};
struct CountingExecutor {
accepted: AtomicUsize,
sender: Mutex<Option<mpsc::Sender<BoxTask>>>,
worker: Mutex<Option<JoinHandle<()>>>,
}
impl CountingExecutor {
fn new() -> Arc<Self> {
let (sender, receiver) = mpsc::channel::<BoxTask>();
let worker = thread::spawn(move || {
while let Ok(task) = receiver.recv() {
task.run();
}
});
Arc::new(Self {
accepted: AtomicUsize::new(0),
sender: Mutex::new(Some(sender)),
worker: Mutex::new(Some(worker)),
})
}
}
impl Executor for CountingExecutor {
fn try_execute(&self, task: BoxTask) -> std::result::Result<(), RejectedTask> {
self.accepted.fetch_add(1, Ordering::SeqCst);
let Some(sender) = self.sender.lock().expect("sender lock").as_ref().cloned() else {
return Err(RejectedTask::new(timerwheel::Error::Closed, task));
};
sender
.send(task)
.map_err(|error| RejectedTask::new(timerwheel::Error::Closed, error.0))
}
fn shutdown(&self) -> timerwheel::Result<()> {
self.sender.lock().expect("sender lock").take();
if let Some(worker) = self.worker.lock().expect("worker lock").take() {
let _ = worker.join();
}
Ok(())
}
fn metrics(&self) -> ExecutorMetrics {
ExecutorMetrics::default()
}
}
#[test]
fn timer_builder_rejects_zero_tick() {
assert_eq!(
Timer::builder()
.tick(Duration::ZERO)
.build()
.expect_err("zero tick should reject"),
Error::InvalidConfig("tick must be greater than zero")
);
}
#[test]
fn timer_builder_rejects_zero_bucket_count() {
assert_eq!(
Timer::builder()
.bucket_count(0)
.build()
.expect_err("zero bucket count should reject"),
Error::InvalidConfig("bucket_count must be greater than zero")
);
}
#[test]
fn timer_builder_rejects_zero_command_capacity() {
assert_eq!(
Timer::builder()
.command_capacity(0)
.build()
.expect_err("zero command capacity should reject"),
Error::InvalidConfig("command_capacity must be greater than zero")
);
}
#[test]
fn timer_builder_rejects_zero_max_pending() {
assert_eq!(
Timer::builder()
.max_pending(0)
.build()
.expect_err("zero max pending should reject"),
Error::InvalidConfig("max_pending must be greater than zero")
);
}
#[test]
fn pool_builder_rejects_zero_workers() {
assert_eq!(
Pool::builder()
.workers(0)
.build()
.expect_err("zero workers should reject"),
Error::InvalidConfig("workers must be greater than zero")
);
}
#[test]
fn pool_builder_rejects_zero_queue_capacity() {
assert_eq!(
Pool::builder()
.queue_capacity(0)
.build()
.expect_err("zero queue capacity should reject"),
Error::InvalidConfig("queue_capacity must be greater than zero")
);
}
#[test]
fn timer_uses_supplied_executor() {
let executor = CountingExecutor::new();
let timer_executor: Arc<dyn Executor> = executor.clone();
let timer = Timer::builder()
.tick(Duration::from_millis(1))
.executor(timer_executor)
.build()
.expect("timer builds");
timer
.schedule(Duration::from_millis(1), || {})
.expect("schedule succeeds");
std::thread::sleep(Duration::from_millis(20));
assert!(executor.accepted.load(Ordering::SeqCst) >= 1);
timer.shutdown().expect("timer shuts down");
executor.shutdown().expect("executor shuts down");
}
#[test]
fn timer_accepts_erased_executor() {
let executor: Arc<dyn Executor> = CountingExecutor::new();
let timer = Timer::builder()
.executor(Arc::clone(&executor))
.build()
.expect("timer builds");
timer.shutdown().expect("timer shuts down");
executor.shutdown().expect("executor shuts down");
}