use swactor::{ actor::{ActorAddress, ActorInterface}, runtime::{Ctx, Inbox, Runtime, RuntimeConfig}, }; // --------------------------------------------------------------------------- // Shared test fixtures // --------------------------------------------------------------------------- #[derive(Clone)] struct EchoMessage { payload: usize, reply_to: ActorAddress, } #[derive(Clone, Debug, PartialEq)] struct EchoResponse(usize); struct EchoActor; impl ActorInterface for EchoActor { type Incoming = EchoMessage; type Response = EchoResponse; fn handle(&mut self, ctx: &Ctx, msg: EchoMessage) { let _ = ctx.send(msg.reply_to, EchoResponse(msg.payload)); } } /// Child actor that doubles the payload and replies. struct DoubleActor; #[derive(Clone)] struct DoubleRequest { value: usize, reply_to: ActorAddress, } #[derive(Clone, Debug, PartialEq)] struct DoubleResponse(usize); impl ActorInterface for DoubleActor { type Incoming = DoubleRequest; type Response = DoubleResponse; fn handle(&mut self, ctx: &Ctx, msg: DoubleRequest) { let _ = ctx.send(msg.reply_to, DoubleResponse(msg.value * 2)); } } /// Parent actor that spawns a DoubleActor child and delegates work. struct DelegateActor; #[derive(Clone)] struct DelegateRequest { value: usize, reply_to: ActorAddress, } impl ActorInterface for DelegateActor { type Incoming = DelegateRequest; type Response = (); fn handle(&mut self, ctx: &Ctx, msg: DelegateRequest) { let child = ctx.spawn(DoubleActor).expect("spawn child"); let _ = ctx.send( child, DoubleRequest { value: msg.value, reply_to: msg.reply_to, }, ); } } // --------------------------------------------------------------------------- // Tests // --------------------------------------------------------------------------- #[test] fn test_single_thread_spawn_actor_and_inbox() { let rt = Runtime::new(RuntimeConfig::default()); let actor_addr = rt.spawn(EchoActor).expect("spawn echo actor"); let inbox: Inbox = rt.new_inbox().unwrap(); rt.send_to( actor_addr, EchoMessage { payload: 42, reply_to: *inbox.addr(), }, ) .unwrap(); for _ in 0..10 { rt.tick(); if let Some(response) = inbox.try_recv() { assert_eq!(response, EchoResponse(42)); return; } } panic!("Did not receive EchoResponse"); } #[test] fn test_multi_thread_spawn_actor_and_inbox() { let config = RuntimeConfig { num_threads: 4, ..Default::default() }; let rt = Runtime::new(config); let actor_addr = rt.spawn(EchoActor).expect("spawn echo actor"); let inbox: Inbox = rt.new_inbox().unwrap(); rt.send_to( actor_addr, EchoMessage { payload: 99, reply_to: *inbox.addr(), }, ) .unwrap(); let handle = rt.run().unwrap(); let check = std::thread::spawn(move || { for _ in 0..100 { std::thread::sleep(std::time::Duration::from_millis(10)); if let Some(response) = inbox.try_recv() { handle.shutdown(); return Some(response); } } handle.shutdown(); None }); let result = check.join().unwrap(); assert_eq!(result, Some(EchoResponse(99))); } // --------------------------------------------------------------------------- // Behavioral story tests // --------------------------------------------------------------------------- #[test] fn send_to_unknown_address_fails() { let rt = Runtime::new(RuntimeConfig::default()); let bogus = ActorAddress::new_random(); let result = rt.send_to(bogus, 42u64); assert!(result.is_err()); } #[test] fn actor_spawns_child_and_delegates() { // Uses 2 workers so parent and child land on different workers, // avoiding the single-worker timing issue where pending_local // delivery precedes spawn-queue draining. let config = RuntimeConfig { num_threads: 2, ..Default::default() }; let rt = Runtime::new(config); let parent = rt.spawn(DelegateActor).expect("spawn parent"); let inbox: Inbox = rt.new_inbox().unwrap(); rt.send_to( parent, DelegateRequest { value: 7, reply_to: *inbox.addr(), }, ) .unwrap(); let handle = rt.run().unwrap(); let check = std::thread::spawn(move || { for _ in 0..100 { std::thread::sleep(std::time::Duration::from_millis(10)); if let Some(resp) = inbox.try_recv() { handle.shutdown(); return Some(resp); } } handle.shutdown(); None }); let result = check.join().unwrap(); assert_eq!(result, Some(DoubleResponse(14))); } #[test] fn multiple_actors_independent_mailboxes() { let rt = Runtime::new(RuntimeConfig::default()); let inbox_a: Inbox = rt.new_inbox().unwrap(); let inbox_b: Inbox = rt.new_inbox().unwrap(); let inbox_c: Inbox = rt.new_inbox().unwrap(); let actor_a = rt.spawn(EchoActor).unwrap(); let actor_b = rt.spawn(EchoActor).unwrap(); let actor_c = rt.spawn(EchoActor).unwrap(); rt.send_to(actor_a, EchoMessage { payload: 10, reply_to: *inbox_a.addr() }).unwrap(); rt.send_to(actor_b, EchoMessage { payload: 20, reply_to: *inbox_b.addr() }).unwrap(); rt.send_to(actor_c, EchoMessage { payload: 30, reply_to: *inbox_c.addr() }).unwrap(); for _ in 0..10 { rt.tick(); } assert_eq!(inbox_a.try_recv(), Some(EchoResponse(10))); assert_eq!(inbox_b.try_recv(), Some(EchoResponse(20))); assert_eq!(inbox_c.try_recv(), Some(EchoResponse(30))); // No cross-contamination assert_eq!(inbox_a.try_recv(), None); assert_eq!(inbox_b.try_recv(), None); assert_eq!(inbox_c.try_recv(), None); } #[test] fn round_robin_distributes_across_workers() { let config = RuntimeConfig { num_threads: 3, ..Default::default() }; let rt = Runtime::new(config); // Counter that just counts messages struct Noop; impl ActorInterface for Noop { type Incoming = (); type Response = (); fn handle(&mut self, _ctx: &Ctx, _msg: ()) {} } for _ in 0..6 { rt.spawn(Noop).unwrap(); } let s = rt.stats(); assert_eq!(s.num_workers, 3); // Count actors per worker from the address map snapshot let mut per_worker = [0usize; 3]; for (_addr, wid) in &s.actors { per_worker[*wid] += 1; } // Round-robin should place exactly 2 actors on each of the 3 workers for (wid, &count) in per_worker.iter().enumerate() { assert_eq!(count, 2, "worker {} should have 2 actors", wid); } }