use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
use std::sync::Arc;
use std::time::Duration;
use acton_reactive::prelude::*;
use acton_test::prelude::*;
#[acton_test]
async fn test_creating_child_actors() -> anyhow::Result<()> {
#[acton_actor]
struct Parent;
#[acton_actor]
struct Child {
received_task: bool,
}
#[acton_message]
struct Task;
let child_ran = Arc::new(AtomicBool::new(false));
let child_ran_clone = child_ran.clone();
let mut runtime = ActonApp::launch_async().await;
let parent = runtime.new_actor::<Parent>();
let parent_handle = parent.start().await;
let mut child = runtime.new_actor::<Child>();
child
.mutate_on::<Task>(|actor, _ctx| {
actor.model.received_task = true;
Reply::ready()
})
.after_stop(move |actor| {
child_ran_clone.store(actor.model.received_task, Ordering::SeqCst);
Reply::ready()
});
let child_handle = parent_handle.supervise(child).await?;
assert_eq!(parent_handle.children().len(), 1);
child_handle.send(Task).await;
tokio::time::sleep(Duration::from_millis(100)).await;
runtime.shutdown_all().await?;
assert!(child_ran.load(Ordering::SeqCst));
Ok(())
}
#[acton_test]
async fn test_ern_hierarchy() -> anyhow::Result<()> {
#[acton_actor]
struct ServiceState;
#[acton_actor]
struct WorkerState;
let mut runtime = ActonApp::launch_async().await;
let service = runtime.new_actor_with_name::<ServiceState>("payment-service".to_string());
let child_config = ActorConfig::new(Ern::with_root("worker-1").unwrap(), None, None)?;
let child = runtime.new_actor_with_config::<WorkerState>(child_config);
let service_handle = service.start().await;
let child_handle = service_handle.supervise(child).await?;
assert!(service_handle.name().starts_with("paymentservice"));
assert!(child_handle.name().starts_with("worker"));
runtime.shutdown_all().await?;
Ok(())
}
#[acton_test]
async fn test_cascading_shutdown() -> anyhow::Result<()> {
#[acton_actor]
struct Parent;
#[acton_actor]
struct Child;
let child_stopped = Arc::new(AtomicBool::new(false));
let child_clone = child_stopped.clone();
let mut runtime = ActonApp::launch_async().await;
let parent = runtime.new_actor::<Parent>();
let parent_handle = parent.start().await;
let mut child = runtime.new_actor::<Child>();
child.after_stop(move |_actor| {
child_clone.store(true, Ordering::SeqCst);
Reply::ready()
});
let _child_handle = parent_handle.supervise(child).await?;
assert_eq!(parent_handle.children().len(), 1);
tokio::time::sleep(Duration::from_millis(50)).await;
runtime.shutdown_all().await?;
assert!(child_stopped.load(Ordering::SeqCst));
Ok(())
}
#[acton_test]
async fn test_worker_pool_pattern() -> anyhow::Result<()> {
#[acton_actor]
struct Supervisor;
#[acton_actor]
struct Worker {
task_count: u32,
}
#[acton_message]
struct Task;
let total_tasks = Arc::new(AtomicU32::new(0));
let mut runtime = ActonApp::launch_async().await;
let supervisor = runtime.new_actor::<Supervisor>();
let supervisor_handle = supervisor.start().await;
let mut worker_handles = Vec::new();
for i in 0..3 {
let counter = total_tasks.clone();
let config =
ActorConfig::new(Ern::with_root(format!("worker-{i}")).unwrap(), None, None)?;
let mut worker = runtime.new_actor_with_config::<Worker>(config);
worker
.mutate_on::<Task>(|actor, _ctx| {
actor.model.task_count += 1;
Reply::ready()
})
.after_stop(move |actor| {
counter.fetch_add(actor.model.task_count, Ordering::SeqCst);
Reply::ready()
});
let handle = supervisor_handle.supervise(worker).await?;
worker_handles.push(handle);
}
for i in 0..9 {
let worker = &worker_handles[i % 3];
worker.send(Task).await;
}
tokio::time::sleep(Duration::from_millis(100)).await;
runtime.shutdown_all().await?;
assert_eq!(total_tasks.load(Ordering::SeqCst), 9);
Ok(())
}
#[acton_test]
async fn test_child_lifecycle_hooks() -> anyhow::Result<()> {
#[acton_actor]
struct Parent;
#[acton_actor]
struct Child;
let child_started = Arc::new(AtomicBool::new(false));
let child_started_clone = child_started.clone();
let child_stopped = Arc::new(AtomicBool::new(false));
let child_stopped_clone = child_stopped.clone();
let mut runtime = ActonApp::launch_async().await;
let parent = runtime.new_actor::<Parent>();
let parent_handle = parent.start().await;
let mut child = runtime.new_actor::<Child>();
child
.after_start(move |_actor| {
child_started_clone.store(true, Ordering::SeqCst);
Reply::ready()
})
.after_stop(move |_actor| {
child_stopped_clone.store(true, Ordering::SeqCst);
Reply::ready()
});
let _child_handle = parent_handle.supervise(child).await?;
tokio::time::sleep(Duration::from_millis(50)).await;
assert!(child_started.load(Ordering::SeqCst));
runtime.shutdown_all().await?;
assert!(child_stopped.load(Ordering::SeqCst));
Ok(())
}
#[acton_test]
async fn test_find_child() -> anyhow::Result<()> {
#[acton_actor]
struct Parent;
#[acton_actor]
struct Child;
let mut runtime = ActonApp::launch_async().await;
let parent = runtime.new_actor::<Parent>();
let parent_handle = parent.start().await;
let config = ActorConfig::new(Ern::with_root("my-child").unwrap(), None, None)?;
let child = runtime.new_actor_with_config::<Child>(config);
let child_id = child.id().clone();
let _child_handle = parent_handle.supervise(child).await?;
let found = parent_handle.find_child(&child_id);
assert!(found.is_some());
runtime.shutdown_all().await?;
Ok(())
}