//! gen_server tests: call round-trip, cast, lifecycle callbacks, and the two //! server-down detection paths (reply-channel close vs. inbox-send failure). use smarm::gen_server::{start, CallError, GenServer}; use smarm::run; use std::sync::{Arc, Mutex}; // --------------------------------------------------------------------------- // A trivial counter server: casts mutate, calls read (or blow up). // --------------------------------------------------------------------------- struct Counter { n: i64, } enum Req { Get, Boom, } enum Op { Add(i64), } impl GenServer for Counter { type Call = Req; type Reply = i64; type Cast = Op; fn handle_call(&mut self, req: Req) -> i64 { match req { Req::Get => self.n, Req::Boom => panic!("boom"), } } fn handle_cast(&mut self, op: Op) { match op { Op::Add(x) => self.n += x, } } } // Casts are applied in order and a later call observes the accumulated state. #[test] fn cast_then_call_roundtrip() { let got = Arc::new(Mutex::new(0i64)); let got2 = got.clone(); run(move || { let server = start(Counter { n: 0 }); server.cast(Op::Add(5)).unwrap(); server.cast(Op::Add(3)).unwrap(); let n = server.call(Req::Get).unwrap(); *got2.lock().unwrap() = n; }); assert_eq!(*got.lock().unwrap(), 8); } // --------------------------------------------------------------------------- // Lifecycle: init runs before the first message, terminate on graceful exit. // --------------------------------------------------------------------------- struct Lifecycle { log: Arc>>, } impl GenServer for Lifecycle { type Call = (); type Reply = (); type Cast = (); fn init(&mut self) { self.log.lock().unwrap().push("init"); } fn handle_call(&mut self, _req: ()) { self.log.lock().unwrap().push("call"); } fn handle_cast(&mut self, _req: ()) {} fn terminate(&mut self) { self.log.lock().unwrap().push("terminate"); } } // init -> handle_call -> (drop last ref closes inbox) -> terminate. #[test] fn init_and_terminate_run() { let log = Arc::new(Mutex::new(Vec::new())); let log2 = log.clone(); run(move || { let server = start(Lifecycle { log: log2 }); server.call(()).unwrap(); // Dropping the only ref closes the inbox; the server breaks out of its // recv loop and runs terminate. run() will not return until it has. drop(server); }); assert_eq!(*log.lock().unwrap(), vec!["init", "call", "terminate"]); } // --------------------------------------------------------------------------- // Server-down detection. // --------------------------------------------------------------------------- // The server dies *while* a call is in flight (handler panics): the reply // sender drops on unwind, closing the reply channel, so the parked caller's // recv returns Err -> ServerDown. #[test] fn call_to_panicking_handler_is_server_down() { let got = Arc::new(Mutex::new(None)); let got2 = got.clone(); run(move || { let server = start(Counter { n: 0 }); let r = server.call(Req::Boom); *got2.lock().unwrap() = Some(r); }); assert_eq!(*got.lock().unwrap(), Some(Err(CallError::ServerDown))); } // A call issued *after* the server is already gone: the inbox is closed, so the // send itself fails -> ServerDown (the other detection path). #[test] fn call_after_server_gone_is_server_down() { let got = Arc::new(Mutex::new(None)); let got2 = got.clone(); run(move || { let server = start(Counter { n: 0 }); let server2 = server.clone(); // This kills the server (and is itself ServerDown). assert_eq!(server.call(Req::Boom), Err(CallError::ServerDown)); // Inbox now closed; a fresh call can't even be enqueued. let r = server2.call(Req::Get); *got2.lock().unwrap() = Some(r); }); assert_eq!(*got.lock().unwrap(), Some(Err(CallError::ServerDown))); } // --------------------------------------------------------------------------- // call_timeout // --------------------------------------------------------------------------- use smarm::gen_server::CallTimeoutError; use std::time::{Duration, Instant}; /// Replies after sleeping `delay_ms` (parking the server actor, not the OS /// thread), so callers can race a deadline against the reply. struct Slow; impl GenServer for Slow { type Call = u64; // delay in ms type Reply = u64; type Cast = (); fn handle_call(&mut self, delay_ms: u64) -> u64 { if delay_ms > 0 { smarm::sleep(Duration::from_millis(delay_ms)); } delay_ms } fn handle_cast(&mut self, _: ()) {} } #[test] fn call_timeout_returns_reply_within_deadline() { run(|| { let srv = start(Slow); assert_eq!(srv.call_timeout(0, Duration::from_secs(10)), Ok(0)); }); } #[test] fn call_timeout_times_out_on_slow_handler() { run(|| { let srv = start(Slow); let start_t = Instant::now(); let r = srv.call_timeout(500, Duration::from_millis(50)); assert_eq!(r, Err(CallTimeoutError::Timeout)); let elapsed = start_t.elapsed(); // Gave up at the deadline, not at the reply. assert!(elapsed >= Duration::from_millis(50)); assert!(elapsed < Duration::from_millis(500)); }); } #[test] fn server_survives_an_abandoned_call_and_late_reply_is_discarded() { run(|| { let srv = start(Slow); assert_eq!( srv.call_timeout(100, Duration::from_millis(20)), Err(CallTimeoutError::Timeout) ); // The timed-out request is still handled; its reply send fails // harmlessly (receiver dropped). The server must keep serving, and // the late reply must not leak into THIS call's reply channel. assert_eq!(srv.call_timeout(0, Duration::from_secs(10)), Ok(0)); // Plain unbounded call still fine too. assert_eq!(srv.call(0), Ok(0)); }); } #[test] fn call_timeout_to_dead_server_is_server_down_not_timeout() { struct Bomb; impl GenServer for Bomb { type Call = (); type Reply = (); type Cast = (); fn handle_call(&mut self, _: ()) { panic!("kaboom"); } fn handle_cast(&mut self, _: ()) {} } run(|| { let srv = start(Bomb); // Dies mid-call: reply channel closes -> ServerDown (even though the // generous deadline never fires). assert_eq!( srv.call_timeout((), Duration::from_secs(10)), Err(CallTimeoutError::ServerDown) ); // Already gone: inbox send fails -> ServerDown. assert_eq!( srv.call_timeout((), Duration::from_secs(10)), Err(CallTimeoutError::ServerDown) ); }); }