behavior_actors/atomic/worker/

activation.rs

//! Worker activation authorized by successful initialization.

use std::sync::Arc;

use core::marker::PhantomData;

use behavior::{
    ActionItem, Behavior, BehaviorAddr, EndpointAddress, EstablishedRecipient, Here,
    InterpretationProgress, InterpreterRequest, ItemSettlement, Never, ReturnsToEmitter,
    finish_item, prepare_item,
};

use super::initialization::{ActivationPermit, InitializationAttempt};
use super::{ActivationPlan, WorkerAttempt};

/// The original worker/plan-typed activation correlation issued with a permit.
///
/// Readiness and rejection are absent from this owner. Its original private
/// issuer remains inside BeginActivation. A caller cannot fabricate a grant,
/// but into_parts returns the original permit and supports reissuing a request.
/// This owner does not impose a one-mint-ever policy on that affine permit.
pub struct WorkerActivationGrant<W, P> {
    worker: WorkerAttempt,
    token: Arc<()>,
    worker_plan: PhantomData<fn() -> (W, P)>,
}

impl<W, P> Clone for WorkerActivationGrant<W, P> {
    fn clone(&self) -> Self {
        Self {
            worker: self.worker.clone(),
            token: Arc::clone(&self.token),
            worker_plan: PhantomData,
        }
    }
}

impl<W, P> WorkerActivationGrant<W, P> {
    fn issued(worker: &WorkerAttempt) -> Self {
        Self {
            worker: worker.clone(),
            token: Arc::new(()),
            worker_plan: PhantomData,
        }
    }

    /// Original worker correlation carried by this grant.
    #[must_use]
    pub fn worker(&self) -> WorkerAttempt {
        self.worker.clone()
    }
}

impl<W, P> PartialEq for WorkerActivationGrant<W, P> {
    fn eq(&self, other: &Self) -> bool {
        self.worker == other.worker && Arc::ptr_eq(&self.token, &other.token)
    }
}

impl<W, P> Eq for WorkerActivationGrant<W, P> {}

impl<W, P> WorkerActivationGrant<W, P>
where
    W: Behavior,
    BehaviorAddr<W>: EndpointAddress,
    P: ActivationPlan,
{
    /// Form the existing input from the original concrete plan's acquired reply.
    ///
    /// The independent correlation stays owned, as when started() produces a
    /// correlated input. Acceptance remains the actor's existing policy.
    /// The receiving actor applies its existing correlation policy. Resolving a
    /// reply does not consume the initialization permit.
    #[must_use]
    pub fn resolve(&self, reply: Result<P::Ready, P::Rejection>) -> WorkerActivation<W, P> {
        WorkerActivation {
            worker: self.worker(),
            activation: self.clone(),
            outcome: match reply {
                Ok(readiness) => WorkerActivationOutcome::Ready(readiness),
                Err(rejection) => WorkerActivationOutcome::Rejected(rejection),
            },
            worker_type: PhantomData,
        }
    }
}

/// Exact reason activation work was not admitted by its owner.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ActivationStartRejection {
    OwnerStopped,
}

/// One activation request consuming an initialized worker permit and plan.
///
/// ```compile_fail,E0382
/// fn duplicate<W, P>(request: behavior_actors::atomic::BeginActivation<W, P>)
/// where
///     W: behavior::Behavior,
///     behavior::BehaviorAddr<W>: behavior::EndpointAddress,
///     P: behavior_actors::atomic::ActivationPlan,
/// {
///     let _accepted = request;
///     let _duplicate = request;
/// }
/// ```
#[must_use = "activation work must be admitted or returned whole"]
pub struct BeginActivation<W, P>
where
    W: Behavior,
    BehaviorAddr<W>: EndpointAddress,
    P: ActivationPlan,
{
    activation: WorkerActivationGrant<W, P>,
    permit: ActivationPermit<W>,
    plan: P,
}

impl<W, P> BeginActivation<W, P>
where
    W: Behavior,
    BehaviorAddr<W>: EndpointAddress,
    P: ActivationPlan,
{
    /// Consume the successful initialization authority and concrete plan.
    #[must_use]
    pub fn new(plan: P, permit: ActivationPermit<W>) -> Self {
        Self {
            activation: WorkerActivationGrant::issued(permit.worker_evidence()),
            permit,
            plan,
        }
    }

    /// Transfer original permit/plan and its exact typed grant independently.
    #[must_use]
    pub fn into_parts(self) -> (ActivationPermit<W>, P, WorkerActivationGrant<W, P>) {
        let Self {
            activation,
            permit,
            plan,
        } = self;
        (permit, plan, activation)
    }

    /// Exact installed worker target carried by the permit.
    #[must_use]
    pub fn target(&self) -> EstablishedRecipient<W::Protocol> {
        self.permit.target()
    }

    /// Worker creation correlation authorized by the consumed permit.
    #[must_use]
    pub fn worker(&self) -> WorkerAttempt {
        self.permit.worker()
    }

    /// Initialization correlation authorized by the consumed permit.
    #[must_use]
    pub fn initialization(&self) -> InitializationAttempt {
        self.permit.initialization()
    }

    pub(in super::super) fn attempt(&self) -> WorkerActivationGrant<W, P> {
        self.activation.clone()
    }

    /// Produce the input Bombay must admit before polling the plan.
    #[must_use]
    pub fn started(&self) -> WorkerActivation<W, P> {
        WorkerActivation {
            worker: self.worker(),
            activation: self.activation.clone(),
            outcome: WorkerActivationOutcome::Started,
            worker_type: PhantomData,
        }
    }

    /// Return a request that its owner could not admit.
    #[must_use]
    pub fn start_rejected(self, reason: ActivationStartRejection) -> WorkerActivation<W, P> {
        let worker = self.worker();
        let activation = self.activation.clone();
        WorkerActivation {
            worker,
            activation,
            outcome: WorkerActivationOutcome::StartRejected {
                request: self,
                reason,
            },
            worker_type: PhantomData,
        }
    }

    /// Consume the plan and produce its exact ready or rejected input.
    pub async fn activate(self) -> WorkerActivation<W, P> {
        let Self {
            activation,
            permit,
            plan,
        } = self;
        let worker = permit.worker();
        match plan.activate().await {
            Ok(readiness) => WorkerActivation {
                worker,
                activation,
                outcome: WorkerActivationOutcome::Ready(readiness),
                worker_type: PhantomData,
            },
            Err(rejection) => WorkerActivation {
                worker,
                activation,
                outcome: WorkerActivationOutcome::Rejected(rejection),
                worker_type: PhantomData,
            },
        }
    }
}

impl<W, P> InterpreterRequest for BeginActivation<W, P>
where
    W: Behavior,
    BehaviorAddr<W>: EndpointAddress,
    P: ActivationPlan,
{
    type ReturnToEmitter = ReturnsToEmitter<WorkerActivation<W, P>, Here>;
    type LogicalProtocols = behavior::NoBirthProtocols;
}

impl<W, P> ActionItem for BeginActivation<W, P>
where
    W: Behavior,
    BehaviorAddr<W>: EndpointAddress,
    EstablishedRecipient<W::Protocol>: Send,
    P: ActivationPlan,
{
    type Custody = (Option<Self>, Option<Self::Reply>);
    type Input<'a>
        = &'a mut Option<Self>
    where
        Self: 'a;
    type Reply = ItemSettlement<Self, Self::Accepted, Self::Rejection, Self::Prerequisite>;

    fn prepare_interpretation(
        progress: &mut Option<InterpretationProgress<Self, Self::Custody, Self::Reply>>,
    ) {
        prepare_item::<Self>(progress);
    }

    fn interpretation_input<'a>(
        custody: &'a mut Self::Custody,
    ) -> Option<(Self::Input<'a>, &'a mut Option<Self::Reply>)>
    where
        Self: 'a,
    {
        match custody {
            (input @ Some(_), received @ None) => Some((input, received)),
            _ => None,
        }
    }

    fn finish_interpretation(
        progress: &mut Option<InterpretationProgress<Self, Self::Custody, Self::Reply>>,
    ) {
        finish_item::<Self>(progress);
    }

    type Accepted = ();
    type Rejection = ActivationStartRejection;
    type Prerequisite = Never;
}

pub(in crate::atomic) enum WorkerActivationOutcome<W, P>
where
    W: Behavior,
    BehaviorAddr<W>: EndpointAddress,
    P: ActivationPlan,
{
    Started,
    StartRejected {
        request: BeginActivation<W, P>,
        reason: ActivationStartRejection,
    },
    Ready(P::Ready),
    Rejected(P::Rejection),
}

/// Exact input produced by one accepted worker activation request.
#[must_use = "worker activation must be admitted or transferred outward"]
pub struct WorkerActivation<W, P>
where
    W: Behavior,
    BehaviorAddr<W>: EndpointAddress,
    P: ActivationPlan,
{
    worker: WorkerAttempt,
    activation: WorkerActivationGrant<W, P>,
    outcome: WorkerActivationOutcome<W, P>,
    worker_type: PhantomData<fn() -> W>,
}

impl<W, P> WorkerActivation<W, P>
where
    W: Behavior,
    BehaviorAddr<W>: EndpointAddress,
    P: ActivationPlan,
{
    /// Worker creation correlation carried by this input.
    #[must_use]
    pub fn worker(&self) -> WorkerAttempt {
        self.worker.clone()
    }

    pub(in super::super) const fn attempt(&self) -> &WorkerActivationGrant<W, P> {
        &self.activation
    }

    pub(in crate::atomic) fn into_parts(
        self,
    ) -> (
        WorkerAttempt,
        WorkerActivationGrant<W, P>,
        WorkerActivationOutcome<W, P>,
    ) {
        (self.worker, self.activation, self.outcome)
    }

    pub(in crate::atomic) fn into_started(self) -> Result<(), Self> {
        let Self {
            worker,
            activation,
            outcome,
            worker_type,
        } = self;
        match outcome {
            WorkerActivationOutcome::Started => Ok(()),
            outcome => Err(Self {
                worker,
                activation,
                outcome,
                worker_type,
            }),
        }
    }

    pub(in super::super) fn into_start_rejection(
        self,
    ) -> Result<(BeginActivation<W, P>, ActivationStartRejection), Self> {
        let Self {
            worker,
            activation,
            outcome,
            worker_type,
        } = self;
        match outcome {
            WorkerActivationOutcome::StartRejected { request, reason } => Ok((request, reason)),
            outcome => Err(Self {
                worker,
                activation,
                outcome,
                worker_type,
            }),
        }
    }

    /// Consume readiness only when this is the ready alternative.
    pub fn into_ready(self) -> Result<P::Ready, Self> {
        let Self {
            worker,
            activation,
            outcome,
            worker_type,
        } = self;
        match outcome {
            WorkerActivationOutcome::Ready(readiness) => Ok(readiness),
            outcome => Err(Self {
                worker,
                activation,
                outcome,
                worker_type,
            }),
        }
    }

    /// Consume the application rejection only when this is the rejected alternative.
    pub fn into_rejection(self) -> Result<P::Rejection, Self> {
        let Self {
            worker,
            activation,
            outcome,
            worker_type,
        } = self;
        match outcome {
            WorkerActivationOutcome::Rejected(rejection) => Ok(rejection),
            outcome => Err(Self {
                worker,
                activation,
                outcome,
                worker_type,
            }),
        }
    }
}