use behavior::{Address, Behavior, BirthMode, Never, ShutdownEvent, TimeEvent};
use bombay_engine::{Driver, Environment, RunError, RunExit};
use observe::ObservationSpace;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use super::lifecycle::{IncarnationReporter, LifecycleFactory};
use super::{
CreationFailure, LaunchMode, Lifecycle, NoLifecycle, PreparedIncarnation,
ProvisionalIncarnation,
};
use super::{IncarnationEffects, NoParent, ParentReporter};
use crate::{
ActorEnvironment, ActorRef, ChildLease, ChildRuntime, EndpointRegistry, EventSender, Handle,
IncarnationEndpoint, MailboxAnchor, MailboxConfig, MailboxDeliveryClosed, MailboxReceiver,
MailboxSender, ObservesCreations, RejectedDelivery, RouteSends, RuntimeBirthMode,
ShutdownRequestError, SystemChildren, TaskOutcome,
};
#[doc(hidden)]
pub(crate) type ActorResult<B, E> = Result<
RunExit<behavior::Exit<<B as Behavior>::Addr>>,
RunError<<B as Behavior>::Error, <E as Environment>::Error>,
>;
pub struct System<R, L = NoLifecycle> {
mailbox: MailboxConfig,
router: R,
lifecycle: L,
}
impl<R: Clone, L: Clone> Clone for System<R, L> {
fn clone(&self) -> Self {
Self {
mailbox: self.mailbox,
router: self.router.clone(),
lifecycle: self.lifecycle.clone(),
}
}
}
impl<R> System<R, NoLifecycle> {
pub const fn new(mailbox: MailboxConfig, router: R) -> Self {
Self {
mailbox,
router,
lifecycle: NoLifecycle,
}
}
pub const fn with_lifecycle<S>(
mailbox: MailboxConfig,
router: R,
lifecycle: S,
) -> System<R, Lifecycle<S>> {
System {
mailbox,
router,
lifecycle: Lifecycle(lifecycle),
}
}
}
type AddrOf<B> = <B as Behavior>::Addr;
type EventOf<B> = <B as Behavior>::Event;
type NonceOf<B> = <AddrOf<B> as Address>::Nonce;
type AnchorOf<B> = MailboxAnchor<EventOf<B>>;
type ChildOf<B> = <<B as Behavior>::Birth as BirthMode>::Child;
type RuntimeOf<B, Y> =
<<B as Behavior>::Birth as RuntimeBirthMode<AddrOf<B>, Y, AnchorOf<B>>>::Runtime;
type AnchorEndpoint<B> = IncarnationEndpoint<AddrOf<B>, ActorRef<AddrOf<B>, AnchorOf<B>>>;
type EffectsOf<R, B, Y, P> = IncarnationEffects<
R,
NonceOf<B>,
<RuntimeOf<B, Y> as ChildRuntime<AddrOf<B>, ChildOf<B>, AnchorOf<B>>>::Lease,
AnchorOf<B>,
P,
AddrOf<B>,
>;
#[doc(hidden)]
pub(crate) type BehaviorRef<B, L = NoLifecycle> = ActorRef<AddrOf<B>, MailboxSender<EventOf<B>>, L>;
#[doc(hidden)]
pub(crate) type BehaviorChildren<R, B, L = NoLifecycle> = RuntimeOf<B, System<R, L>>;
#[doc(hidden)]
pub(crate) type BehaviorEnvironment<R, B, P = NoParent, L = NoLifecycle> =
ActorEnvironment<B, MailboxReceiver<EventOf<B>>, R, BehaviorChildren<R, B, L>, AnchorOf<B>, P>;
#[doc(hidden)]
type BehaviorRegistration<R, B> =
<R as EndpointRegistry<AddrOf<B>, <B as Behavior>::Msg, AnchorEndpoint<B>>>::Registration;
type BehaviorReporter<R, B, L> =
<L as LifecycleFactory<AddrOf<B>, BehaviorRegistration<R, B>>>::Reporter;
#[doc(hidden)]
pub type RootOutcome<R, B, L = NoLifecycle> =
ActorResult<B, BehaviorEnvironment<R, B, NoParent, L>>;
pub struct RootEndpoint<B: Behavior<Ph = Never>> {
inner: ActorRef<AddrOf<B>, AnchorOf<B>>,
}
impl<B> Clone for RootEndpoint<B>
where
B: Behavior<Ph = Never>,
AddrOf<B>: Clone,
AnchorOf<B>: Clone,
{
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
}
}
}
impl<B> crate::DeliveryEndpoint<AddrOf<B>, B::Msg> for RootEndpoint<B>
where
B: Behavior<Ph = Never>,
AddrOf<B>: Send + Sync,
B::Event: behavior::UserEvent<Addr = AddrOf<B>> + Send,
B::Msg: Send,
{
type Error = MailboxDeliveryClosed;
async fn deliver(
&self,
from: AddrOf<B>,
message: B::Msg,
) -> Result<(), RejectedDelivery<B::Msg, Self::Error>> {
crate::DeliveryEndpoint::deliver(&self.inner, from, message).await
}
}
pub struct RootRetirement<R, T> {
handle: Handle<R, T>,
}
impl<R, T> RootRetirement<R, T> {
pub fn abort(&self) {
self.handle.abort();
}
pub async fn outcome(self) -> TaskOutcome<T> {
self.handle.outcome().await
}
}
impl<A, E, L, T> RootRetirement<ActorRef<A, MailboxSender<E>, L>, T>
where
E: ShutdownEvent,
L: IncarnationReporter,
{
pub fn request_shutdown(&self) -> Result<(), ShutdownRequestError> {
self.handle.actor_ref().request_shutdown()
}
}
#[doc(hidden)]
pub type BehaviorRetirement<R, B, L = NoLifecycle> =
RootRetirement<BehaviorRef<B, BehaviorReporter<R, B, L>>, RootOutcome<R, B, L>>;
pub struct RootActivation<B: Behavior<Ph = Never>, R> {
pub endpoint: RootEndpoint<B>,
pub retirement: R,
}
#[doc(hidden)]
pub type BehaviorActivation<R, B, L = NoLifecycle> = RootActivation<B, BehaviorRetirement<R, B, L>>;
#[doc(hidden)]
pub type BehaviorSpawnResult<R, B, L = NoLifecycle> = Result<
Handle<
BehaviorRef<B, BehaviorReporter<R, B, L>>,
ActorResult<B, BehaviorEnvironment<R, B, NoParent, L>>,
>,
<R as EndpointRegistry<AddrOf<B>, <B as Behavior>::Msg, AnchorEndpoint<B>>>::Error,
>;
#[doc(hidden)]
type BehaviorPreparation<R, B, P, L> = PreparedIncarnation<
B,
BehaviorEnvironment<R, B, P, L>,
BehaviorRegistration<R, B>,
BehaviorRef<B, BehaviorReporter<R, B, L>>,
ActorResult<B, BehaviorEnvironment<R, B, P, L>>,
AddrOf<B>,
BehaviorReporter<R, B, L>,
>;
#[doc(hidden)]
type BehaviorProvisional<R, B, P, L> = ProvisionalIncarnation<
B,
BehaviorEnvironment<R, B, P, L>,
ActorResult<B, BehaviorEnvironment<R, B, P, L>>,
AddrOf<B>,
AnchorOf<B>,
MailboxSender<EventOf<B>>,
>;
#[doc(hidden)]
#[derive(Debug, thiserror::Error)]
pub enum SystemBirthError<R, I, F> {
#[error("child address registration failed: {0:?}")]
Registration(R),
#[error("child initialization failed: {0:?}")]
Initialization(I),
#[error("child initialization effect failed: {0:?}")]
Effects(F),
}
impl<R, I, F> From<RunError<I, F>> for SystemBirthError<R, I, F> {
fn from(error: RunError<I, F>) -> Self {
match error {
RunError::Behavior(initialization) => Self::Initialization(initialization),
RunError::Environment(effects) => Self::Effects(effects),
RunError::Poisoned => unreachable!("init cannot poison the executor"),
}
}
}
impl<R, I, F> CreationFailure for SystemBirthError<R, I, F> {
fn rejection(&self) -> behavior::CreationRejection {
match self {
Self::Registration(_) | Self::Effects(_) => {
behavior::CreationRejection::EnvironmentFailed
}
Self::Initialization(_) => behavior::CreationRejection::InitializationFailed,
}
}
}
impl<R: Clone, L: Clone> System<R, L> {
pub async fn activate<B>(
&self,
address: AddrOf<B>,
behavior: B,
) -> Result<
BehaviorActivation<R, B, L>,
SystemBirthError<
<R as EndpointRegistry<AddrOf<B>, B::Msg, AnchorEndpoint<B>>>::Error,
B::Error,
<BehaviorEnvironment<R, B, NoParent, L> as Environment>::Error,
>,
>
where
B: Behavior<Ph = Never> + Send + 'static,
AddrOf<B>: Send + Sync + 'static,
NonceOf<B>: Send + 'static,
B::Sends: RouteSends<AddrOf<B>, EffectsOf<R, B, Self, NoParent>>
+ ObservesCreations<NonceOf<B>>
+ Send
+ 'static,
<B::Sends as RouteSends<AddrOf<B>, EffectsOf<R, B, Self, NoParent>>>::Error:
Send + Sync + 'static,
B::Birth: RuntimeBirthMode<AddrOf<B>, Self, AnchorOf<B>> + 'static,
ChildOf<B>: Send + 'static,
RuntimeOf<B, Self>: Send + Sync + 'static,
<RuntimeOf<B, Self> as ChildRuntime<AddrOf<B>, ChildOf<B>, AnchorOf<B>>>::Error:
CreationFailure + Send + Sync + 'static,
B::Event: ShutdownEvent + TimeEvent + Send + 'static,
B::Msg: Send + 'static,
B::Error: Send + Sync + 'static,
R: Clone + EndpointRegistry<AddrOf<B>, B::Msg, AnchorEndpoint<B>> + Send + Sync + 'static,
BehaviorRegistration<R, B>: Send + 'static,
L: LifecycleFactory<AddrOf<B>, BehaviorRegistration<R, B>>,
{
let mut provisional = self.prepare_provisional(address, behavior, NoParent);
let pending_exit = match provisional.driver().run_init().await {
Ok(exit) => exit,
Err(error) => {
provisional.driver().retire().await;
return Err(SystemBirthError::from(error));
}
};
let prepared = provisional
.commit(&self.router, &self.lifecycle)
.map_err(SystemBirthError::Registration)?;
let actor_ref = prepared.actor_ref().clone();
let endpoint = RootEndpoint {
inner: ActorRef::new(address, actor_ref.sender_anchor()),
};
let handle = prepared.launch(LaunchMode::Initialized(pending_exit), false);
Ok(RootActivation {
endpoint,
retirement: RootRetirement { handle },
})
}
pub fn spawn<B>(&self, address: AddrOf<B>, behavior: B) -> BehaviorSpawnResult<R, B, L>
where
B: Behavior<Ph = Never> + Send + 'static,
AddrOf<B>: Send + Sync + 'static,
NonceOf<B>: Send + 'static,
B::Sends: RouteSends<AddrOf<B>, EffectsOf<R, B, Self, NoParent>>
+ ObservesCreations<NonceOf<B>>
+ Send
+ 'static,
<B::Sends as RouteSends<AddrOf<B>, EffectsOf<R, B, Self, NoParent>>>::Error:
Send + Sync + 'static,
B::Birth: RuntimeBirthMode<AddrOf<B>, Self, AnchorOf<B>> + 'static,
ChildOf<B>: Send + 'static,
RuntimeOf<B, Self>: Send + Sync + 'static,
<RuntimeOf<B, Self> as ChildRuntime<AddrOf<B>, ChildOf<B>, AnchorOf<B>>>::Error:
CreationFailure + Send + Sync + 'static,
B::Event: TimeEvent + Send + 'static,
B::Msg: Send + 'static,
B::Error: Send + Sync + 'static,
R: Clone + EndpointRegistry<AddrOf<B>, B::Msg, AnchorEndpoint<B>> + Send + Sync + 'static,
BehaviorRegistration<R, B>: Send + 'static,
L: LifecycleFactory<AddrOf<B>, BehaviorRegistration<R, B>>,
{
let prepared = self.prepare(address, behavior, NoParent)?;
Ok(prepared.launch(LaunchMode::Uninitialized, false))
}
#[allow(
clippy::type_complexity,
reason = "the private result retains the exact typed registration error"
)]
fn prepare<B, Parent>(
&self,
address: AddrOf<B>,
behavior: B,
parent: Parent,
) -> Result<
BehaviorPreparation<R, B, Parent, L>,
<R as EndpointRegistry<AddrOf<B>, B::Msg, AnchorEndpoint<B>>>::Error,
>
where
B: Behavior<Ph = Never> + Send + 'static,
AddrOf<B>: Send + Sync + 'static,
NonceOf<B>: Send + 'static,
B::Sends: RouteSends<AddrOf<B>, EffectsOf<R, B, Self, Parent>>
+ ObservesCreations<NonceOf<B>>
+ Send
+ 'static,
<B::Sends as RouteSends<AddrOf<B>, EffectsOf<R, B, Self, Parent>>>::Error:
Send + Sync + 'static,
B::Birth: RuntimeBirthMode<AddrOf<B>, Self, AnchorOf<B>> + 'static,
ChildOf<B>: Send + 'static,
RuntimeOf<B, Self>: Send + Sync + 'static,
<RuntimeOf<B, Self> as ChildRuntime<AddrOf<B>, ChildOf<B>, AnchorOf<B>>>::Error:
CreationFailure + Send + Sync + 'static,
B::Event: TimeEvent + Send + 'static,
B::Msg: Send + 'static,
B::Error: Send + Sync + 'static,
Parent: Send + 'static,
R: Clone + EndpointRegistry<AddrOf<B>, B::Msg, AnchorEndpoint<B>> + Send + Sync + 'static,
BehaviorRegistration<R, B>: Send + 'static,
L: LifecycleFactory<AddrOf<B>, BehaviorRegistration<R, B>>,
{
self.prepare_provisional(address, behavior, parent)
.commit(&self.router, &self.lifecycle)
}
fn prepare_provisional<B, Parent>(
&self,
address: AddrOf<B>,
behavior: B,
parent: Parent,
) -> BehaviorProvisional<R, B, Parent, L>
where
B: Behavior<Ph = Never> + Send + 'static,
AddrOf<B>: Send + Sync + 'static,
NonceOf<B>: Send + 'static,
B::Sends: RouteSends<AddrOf<B>, EffectsOf<R, B, Self, Parent>>
+ ObservesCreations<NonceOf<B>>
+ Send
+ 'static,
<B::Sends as RouteSends<AddrOf<B>, EffectsOf<R, B, Self, Parent>>>::Error:
Send + Sync + 'static,
B::Birth: RuntimeBirthMode<AddrOf<B>, Self, AnchorOf<B>> + 'static,
ChildOf<B>: Send + 'static,
RuntimeOf<B, Self>: Send + Sync + 'static,
<RuntimeOf<B, Self> as ChildRuntime<AddrOf<B>, ChildOf<B>, AnchorOf<B>>>::Error:
CreationFailure + Send + Sync + 'static,
B::Event: TimeEvent + Send + 'static,
B::Msg: Send + 'static,
B::Error: Send + Sync + 'static,
Parent: Send + 'static,
R: Clone + EndpointRegistry<AddrOf<B>, B::Msg, AnchorEndpoint<B>> + Send + Sync + 'static,
BehaviorRegistration<R, B>: Send + 'static,
L: LifecycleFactory<AddrOf<B>, BehaviorRegistration<R, B>>,
{
let observation_space = ObservationSpace::new();
let subject = observation_space
.subject(())
.expect("a fresh completion namespace must be vacant");
let observation = observation_space
.observe(&())
.expect("a newly registered completion subject must resolve");
let peer_space = ObservationSpace::new();
let peer_subject = peer_space
.subject(())
.expect("a fresh peer-completion namespace must be vacant");
let (sender, source) = self.mailbox.create::<B::Event>();
let cancellation_requested = Arc::new(AtomicBool::new(false));
let response = sender.anchor();
let endpoint =
IncarnationEndpoint::new(ActorRef::new(address, response.clone()), peer_space);
let children =
<B::Birth as RuntimeBirthMode<AddrOf<B>, Self, AnchorOf<B>>>::runtime(self.clone());
let environment = ActorEnvironment::new(
address,
source,
self.router.clone(),
children,
response,
parent,
);
ProvisionalIncarnation::new(
Driver::new(behavior, environment),
address,
endpoint,
sender,
subject,
observation,
peer_subject,
cancellation_requested,
)
}
}
impl<R, L, B, ParentSink> ChildRuntime<AddrOf<B>, B, ParentSink> for SystemChildren<System<R, L>>
where
B: Behavior<Ph = Never> + Send + 'static,
AddrOf<B>: Send + Sync + 'static,
NonceOf<B>: Send + 'static,
B::Sends: RouteSends<AddrOf<B>, EffectsOf<R, B, System<R, L>, ParentReporter<AddrOf<B>, ParentSink>>>
+ ObservesCreations<NonceOf<B>>
+ Send
+ 'static,
<B::Sends as RouteSends<
AddrOf<B>,
EffectsOf<R, B, System<R, L>, ParentReporter<AddrOf<B>, ParentSink>>,
>>::Error: Send + Sync + 'static,
B::Birth: RuntimeBirthMode<AddrOf<B>, System<R, L>, AnchorOf<B>> + 'static,
ChildOf<B>: Send + 'static,
RuntimeOf<B, System<R, L>>: Send + Sync + 'static,
<RuntimeOf<B, System<R, L>> as ChildRuntime<AddrOf<B>, ChildOf<B>, AnchorOf<B>>>::Error:
CreationFailure + Send + Sync + 'static,
B::Event: ShutdownEvent + TimeEvent + Send + 'static,
B::Msg: Send + 'static,
B::Error: Send + Sync + 'static,
R: Clone + EndpointRegistry<AddrOf<B>, B::Msg, AnchorEndpoint<B>> + Send + Sync + 'static,
BehaviorRegistration<R, B>: Send + 'static,
L: Clone + LifecycleFactory<AddrOf<B>, BehaviorRegistration<R, B>> + Send + Sync + 'static,
ParentSink: EventSender + Clone + Send + Sync + 'static,
ParentSink::Event: Send + 'static,
{
type Lease = ChildLease<
BehaviorRef<B, BehaviorReporter<R, B, L>>,
ActorResult<B, BehaviorEnvironment<R, B, ParentReporter<AddrOf<B>, ParentSink>, L>>,
>;
type Error = SystemBirthError<
<R as EndpointRegistry<AddrOf<B>, B::Msg, AnchorEndpoint<B>>>::Error,
B::Error,
<BehaviorEnvironment<R, B, ParentReporter<AddrOf<B>, ParentSink>, L> as Environment>::Error,
>;
async fn birth(
&self,
parent: AddrOf<B>,
child: behavior::Create<AddrOf<B>, B>,
response: ParentSink,
) -> Result<Self::Lease, Self::Error> {
let address = parent.birth(child.nonce);
let kind = child.kind;
let reporter = ParentReporter::new(child.nonce, response);
let mut provisional = self
.system
.prepare_provisional(address, child.child, reporter);
let pending_exit = match provisional.driver().run_init().await {
Ok(exit) => exit,
Err(error) => {
provisional.driver().retire().await;
return Err(SystemBirthError::from(error));
}
};
let prepared = provisional
.commit(&self.system.router, &self.system.lifecycle)
.map_err(SystemBirthError::Registration)?;
let restarted = matches!(kind, behavior::CreationKind::ReplacementIncarnation { .. });
let handle = prepared.launch(LaunchMode::Initialized(pending_exit), restarted);
Ok(handle.into_child_lease())
}
}
#[cfg(test)]
mod tests {
use std::convert::Infallible;
use crate::{
ActorRef, AddressRouter, DeliveryRouter, EndpointRegistry, IncarnationEndpoint,
MailboxConfig, RunExit, System, TaskOutcome,
};
use behavior::{
Actions, Address, Behavior, Births, Create, Delivery, Handler, MailAddr, Never, NoBirths,
Pure, Recipient, ServiceSends, UnwatchPeer, User, WatchEvent,
};
struct StopWithMessage;
struct CancelUnknownPeer;
impl Behavior for CancelUnknownPeer {
type Addr = MailAddr;
type Msg = u8;
type Event = WatchEvent<User<MailAddr, u8>>;
type Sends = ServiceSends<UnwatchPeer<MailAddr>>;
type Ph = Never;
type Error = Never;
type Birth = NoBirths;
fn init(&mut self) -> behavior::BehaviorActed<Self> {
Ok(Actions::new(
ServiceSends::one(UnwatchPeer::new(MailAddr(8))),
Vec::new(),
behavior::Step::Stop(behavior::Exit::Normal),
))
}
fn transition(&mut self, _event: Self::Event) -> behavior::BehaviorActed<Self> {
unreachable!("the initialization fold stops")
}
}
struct ForwardTo(MailAddr);
impl Handler for StopWithMessage {
type Addr = MailAddr;
type Msg = u8;
fn receive(
&mut self,
_from: MailAddr,
_message: u8,
) -> behavior::Acted<MailAddr, Never, Vec<Delivery<MailAddr, Never>>, NoBirths, Never>
{
Ok(Actions::stop(behavior::Exit::Normal))
}
}
impl Handler<u8> for ForwardTo {
type Addr = MailAddr;
type Msg = u8;
fn receive(
&mut self,
_from: MailAddr,
message: u8,
) -> behavior::Acted<MailAddr, Never, Vec<Delivery<MailAddr, u8>>, NoBirths, Never>
{
Ok(Actions {
sends: vec![Delivery::new(Recipient::global(self.0), message + 1)],
creates: Vec::new(),
become_: behavior::Step::Stop(behavior::Exit::Normal),
})
}
}
struct StopAfterReceiving;
struct BirthAndSend {
receiver: MailAddr,
}
struct ForwardChild {
receiver: MailAddr,
}
impl Handler<u8, Births<Pure<ForwardChild, u8>>> for BirthAndSend {
type Addr = MailAddr;
type Msg = u8;
fn receive(
&mut self,
_from: MailAddr,
message: u8,
) -> behavior::Acted<
MailAddr,
Never,
Vec<Delivery<MailAddr, u8>>,
Births<Pure<ForwardChild, u8>>,
Never,
> {
Ok(Actions {
sends: vec![Delivery::new(
Recipient::global(MailAddr(1).birth(7)),
message,
)],
creates: vec![Create::birth(
7,
Pure::new(ForwardChild {
receiver: self.receiver,
}),
)],
become_: behavior::Step::Stop(behavior::Exit::Normal),
})
}
}
impl Handler<u8> for ForwardChild {
type Addr = MailAddr;
type Msg = u8;
fn receive(
&mut self,
_from: MailAddr,
message: u8,
) -> behavior::Acted<MailAddr, Never, Vec<Delivery<MailAddr, u8>>, NoBirths, Never>
{
Ok(Actions {
sends: vec![Delivery::new(Recipient::global(self.receiver), message + 1)],
creates: Vec::new(),
become_: behavior::Step::Stop(behavior::Exit::Normal),
})
}
}
impl Handler<u8> for StopAfterReceiving {
type Addr = MailAddr;
type Msg = u8;
fn receive(
&mut self,
_from: MailAddr,
_message: u8,
) -> behavior::Acted<MailAddr, Never, Vec<Delivery<MailAddr, u8>>, NoBirths, Never>
{
Ok(Actions::stop(behavior::Exit::Normal))
}
}
#[derive(Clone, Copy)]
struct NoRouter;
impl DeliveryRouter<MailAddr, Never> for NoRouter {
type Error = Infallible;
async fn deliver(
&self,
_from: MailAddr,
delivery: Delivery<MailAddr, Never>,
) -> Result<(), Self::Error> {
match delivery.message {}
}
}
impl<S> EndpointRegistry<MailAddr, u8, IncarnationEndpoint<MailAddr, ActorRef<MailAddr, S>>>
for NoRouter
{
type Error = Infallible;
type Registration = ();
fn register(
&self,
_address: MailAddr,
_endpoint: IncarnationEndpoint<MailAddr, ActorRef<MailAddr, S>>,
) -> Result<Self::Registration, Self::Error> {
Ok(())
}
}
#[tokio::test]
async fn configured_system_constructs_mailbox_and_returns_actor_reference() {
let system = System::new(MailboxConfig::bounded(4), NoRouter);
let spawned = system
.spawn(MailAddr(9), Pure::new(StopWithMessage))
.unwrap();
spawned.actor_ref().send(MailAddr(7), 1).await.unwrap();
assert_eq!(
spawned.outcome().await,
TaskOutcome::Returned(Ok(RunExit::Stopped(behavior::Exit::Normal)))
);
}
#[tokio::test]
async fn unwatch_service_composes_through_system_without_a_creation_lane() {
let system = System::new(MailboxConfig::bounded(1), NoRouter);
let spawned = system.spawn(MailAddr(9), CancelUnknownPeer).unwrap();
assert_eq!(
spawned.outcome().await,
TaskOutcome::Returned(Ok(RunExit::Stopped(behavior::Exit::Normal)))
);
}
#[tokio::test]
async fn address_registration_lives_exactly_as_long_as_actor_task() {
let router = AddressRouter::default();
let system = System::new(MailboxConfig::bounded(4), router);
let first = system
.spawn(MailAddr(9), Pure::new(ForwardTo(MailAddr(88))))
.unwrap();
assert!(
system
.spawn(MailAddr(9), Pure::new(ForwardTo(MailAddr(88))))
.is_err(),
"a running actor owns its address generation"
);
first.actor_ref().send(MailAddr(7), 1).await.unwrap();
assert!(matches!(
first.outcome().await,
TaskOutcome::Returned(Err(_))
));
let replacement = system
.spawn(MailAddr(9), Pure::new(ForwardTo(MailAddr(88))))
.expect("task exit releases the old address generation");
replacement.actor_ref().send(MailAddr(7), 1).await.unwrap();
assert!(matches!(
replacement.outcome().await,
TaskOutcome::Returned(Err(_))
));
}
#[tokio::test]
async fn two_actors_exchange_a_typed_message_through_shared_routing() {
let router = AddressRouter::default();
let system = System::new(MailboxConfig::bounded(4), router);
let receiver = system
.spawn(MailAddr(2), Pure::new(StopAfterReceiving))
.unwrap();
let sender = system
.spawn(MailAddr(1), Pure::new(ForwardTo(MailAddr(2))))
.unwrap();
sender.actor_ref().send(MailAddr(0), 40).await.unwrap();
assert!(matches!(
sender.outcome().await,
TaskOutcome::Returned(Ok(RunExit::Stopped(behavior::Exit::Normal)))
));
assert!(matches!(
receiver.outcome().await,
TaskOutcome::Returned(Ok(RunExit::Stopped(behavior::Exit::Normal)))
));
}
#[tokio::test]
async fn mailbox_anchor_routes_without_pinning_mailbox_open() {
let router = AddressRouter::default();
let system = System::new(MailboxConfig::bounded(4), router);
let receiver = system
.spawn(MailAddr(2), Pure::new(StopAfterReceiving))
.unwrap();
let sender = system
.spawn(MailAddr(1), Pure::new(ForwardTo(MailAddr(2))))
.unwrap();
sender.actor_ref().send(MailAddr(0), 40).await.unwrap();
assert!(matches!(
sender.outcome().await,
TaskOutcome::Returned(Ok(RunExit::Stopped(behavior::Exit::Normal)))
));
assert!(matches!(
receiver.outcome().await,
TaskOutcome::Returned(Ok(RunExit::Stopped(behavior::Exit::Normal)))
));
let parked = system
.spawn(MailAddr(3), Pure::new(StopAfterReceiving))
.unwrap();
assert!(matches!(
parked.close().await,
TaskOutcome::Returned(Ok(RunExit::EnvironmentClosed))
));
assert!(
system
.spawn(MailAddr(3), Pure::new(StopAfterReceiving))
.is_ok(),
"environment closure releases the address generation"
);
}
#[tokio::test]
async fn typed_creator_spawns_child_before_routing_same_transition_send() {
let router = AddressRouter::default();
let system = System::new(MailboxConfig::bounded(4), router);
let receiver = system
.spawn(MailAddr(9), Pure::new(StopAfterReceiving))
.unwrap();
let parent = system
.spawn(
MailAddr(1),
Pure::new(BirthAndSend {
receiver: MailAddr(9),
}),
)
.unwrap();
parent.actor_ref().send(MailAddr(0), 40).await.unwrap();
assert!(matches!(
parent.outcome().await,
TaskOutcome::Returned(Ok(RunExit::Stopped(behavior::Exit::Normal)))
));
assert!(matches!(
receiver.outcome().await,
TaskOutcome::Returned(Ok(RunExit::Stopped(behavior::Exit::Normal)))
));
}
}