Struct apalis_core::worker::Addr

pub struct Addr<A: Actor> { /* private fields */ }

Available on crate feature worker only.

Address of actor. Can be used to send messages to it.

Implementations

impl<W: Actor> Addr<W>

pub async fn send<M>(&self, message: M) -> Result<M::Result, RecvError>where
    M: Message + Send + 'static,
    M::Result: Send,
    W: ContextHandler<M>,

Send a message and wait for an answer.

Errors

Fails if noone will send message

Panics

If queue is full

Examples found in repository

src/worker/mod.rs (line 225)

    pub async fn send<M>(&self, message: M) -> M::Result
    where
        M: Message + Send + 'static,
        M::Result: Send,
        A: ContextHandler<M>,
    {
        #[allow(clippy::expect_used)]
        self.deref()
            .send(message)
            .await
            .expect("Failed to get response from actor. It should have never failed!")
    }

pub async fn do_send<M>(&self, message: M) -> Option<M::Result>where
    M: Message + Send + 'static,
    M::Result: Send,
    W: ContextHandler<M>,

Send a message and wait for an answer.

Errors

Fails if queue is full or actor is disconnected

Examples found in repository

src/worker/mod.rs (line 278)

    async fn send(&self, m: M) -> Option<M::Result> {
        self.do_send(m).await
    }
}

// #[async_trait::async_trait]
// impl<M> Sender<M> for mpsc::Sender<M>
// where
//     M: Message + Send + 'static + Debug,
//     M::Result: Send,
// {
//     async fn send(&self, m: M) {
//         let _result = self.send(m).await;
//     }
// }

#[async_trait::async_trait]
impl<T: 'static, W: 'static> Actor for W
where
    W: Worker<Job = T> + Handler<JobRequestWrapper<T>>,
    T: Send,
{
    /// At start hook of actor
    async fn on_start(&mut self, ctx: &mut Context<Self>) {
        <W as Worker>::on_start(self, ctx).await;
        let jobs = self
            .consume()
            // errors are silenced for now
            .then(|r| async move {
                match r {
                    Ok(Some(job)) => Ok(Some(job)),
                    _ => Ok(None),
                }
            });
        let stream = jobs.map(|c| JobRequestWrapper(c));
        ctx.notify_with(Box::pin(stream));
    }

    /// At stop hook of actor
    async fn on_stop(&mut self, ctx: &mut Context<Self>) {
        <W as Worker>::on_stop(self, ctx).await;
    }
}

impl Message for WorkerManagement {
    type Result = Result<WorkerStatus, WorkerError>;
}

#[async_trait::async_trait]

impl<W> Handler<WorkerManagement> for W
where
    W: Worker + Actor,
{
    type Result = Result<WorkerStatus, WorkerError>;

    async fn handle(&mut self, msg: WorkerManagement) -> Result<WorkerStatus, WorkerError> {
        self.manage(msg).await
    }
}
impl<T> Message for JobRequestWrapper<T> {
    type Result = anyhow::Result<()>;
}

#[async_trait::async_trait]
impl<T, W> Handler<JobRequestWrapper<T>> for W
where
    W: Worker<Job = T> + 'static,
    T: Job + Serialize + Debug + DeserializeOwned + Send + 'static,
{
    type Result = anyhow::Result<()>;
    async fn handle(&mut self, job: JobRequestWrapper<T>) -> Self::Result {
        match job.0 {
            Ok(Some(job)) => {
                self.handle_job(job).await?;
                Ok(())
            }
            Ok(None) => Ok(()),
            Err(e) => Err(e.into()),
        }
    }
}

/// Actor Trait
#[async_trait::async_trait]
pub trait Actor: Send + Sized + 'static {
    /// Capacity of worker queue
    fn mailbox_capacity(&self) -> usize {
        100
    }

    /// At start hook of actor
    async fn on_start(&mut self, _ctx: &mut Context<Self>) {}

    /// At stop hook of actor
    async fn on_stop(&mut self, _ctx: &mut Context<Self>) {}

    /// Initilize actor with its address.
    fn preinit(self) -> InitializedActor<Self> {
        let mailbox_capacity = self.mailbox_capacity();
        InitializedActor::new(self, mailbox_capacity)
    }

    /// Initialize actor with default values
    fn preinit_default() -> InitializedActor<Self>
    where
        Self: Default,
    {
        Self::default().preinit()
    }

    /// Starts an actor and returns its address
    async fn start(self) -> Addr<Self> {
        self.preinit().start().await
    }

    /// Starts an actor with default values and returns its address
    async fn start_default() -> Addr<Self>
    where
        Self: Default,
    {
        Self::default().start().await
    }
}

/// Initialized actor. Mainly used to take address before starting it.
#[derive(Debug)]
pub struct InitializedActor<A: Actor> {
    /// Address of actor
    pub address: Addr<A>,
    /// Worker itself
    pub actor: A,
    receiver: Receiver<Envelope<A>>,
}

impl<A: Actor> InitializedActor<A> {
    /// Constructor.
    pub fn new(actor: A, mailbox_capacity: usize) -> Self {
        let (sender, receiver) = mpsc::channel(mailbox_capacity);
        InitializedActor {
            actor,
            address: Addr::new(sender),
            receiver,
        }
    }

    /// Start Actor
    pub async fn start(self) -> Addr<A> {
        let address = self.address;
        let mut receiver = self.receiver;
        let mut actor = self.actor;
        let (handle_sender, handle_receiver) = oneshot::channel();
        let move_addr = address.clone();
        let actor_future = async move {
            #[allow(clippy::expect_used)]
            let join_handle = handle_receiver
                .await
                .expect("Unreachable as the message is always sent.");
            let mut ctx = Context::new(move_addr.clone(), join_handle);
            actor.on_start(&mut ctx).await;
            while let Some(Envelope(mut message)) = receiver.recv().await {
                EnvelopeProxy::handle(&mut *message, &mut actor, &mut ctx).await;
            }
            tracing::error!(actor = std::any::type_name::<A>(), "actor stopped");
            actor.on_stop(&mut ctx).await;
            // tokio::time::sleep(Duration::from_secs(1)).await;
            // Restart
        }
        .in_current_span();
        #[cfg(not(feature = "broker"))]
        let join_handle = task::spawn(actor_future);
        #[cfg(feature = "broker")]
        let join_handle = deadlock::spawn_task_with_actor_id(address.actor_id, actor_future);
        // TODO: propagate the error.
        let _error = handle_sender.send(join_handle);
        address
    }
}

/// Message trait for setting result of message
pub trait Message {
    /// Result type of message
    type Result: 'static;
}

/// Trait for actor for handling specific message type
#[async_trait::async_trait]
pub trait ContextHandler<M: Message>: Actor {
    /// Result of handler
    type Result: MessageResponse<M>;

    /// Message handler
    async fn handle(&mut self, ctx: &mut Context<Self>, msg: M) -> Self::Result;
}

/// Trait for actor for handling specific message type without context
#[async_trait::async_trait]
pub trait Handler<M: Message>: Actor {
    /// Result of handler
    type Result: MessageResponse<M>;

    /// Message handler
    async fn handle(&mut self, msg: M) -> Self::Result;
}

#[async_trait::async_trait]
impl<M: Message + Send + 'static, S: Handler<M>> ContextHandler<M> for S {
    type Result = S::Result;

    async fn handle(&mut self, _: &mut Context<Self>, msg: M) -> Self::Result {
        Handler::handle(self, msg).await
    }
}

/// Dev trait for Message responding
#[async_trait::async_trait]
pub trait MessageResponse<M: Message>: Send {
    /// Handles message
    async fn handle(self, sender: oneshot::Sender<M::Result>);
}

#[async_trait::async_trait]
impl<M> MessageResponse<M> for M::Result
where
    M: Message,
    M::Result: Send,
{
    async fn handle(self, sender: oneshot::Sender<M::Result>) {
        drop(sender.send(self));
    }
}

/// Context for execution of actor
#[derive(Debug)]
pub struct Context<W: Actor> {
    addr: Addr<W>,
    #[allow(dead_code)]
    handle: JoinHandle<()>,
}

impl<W: Actor> Context<W> {
    /// Default constructor
    pub fn new(addr: Addr<W>, handle: JoinHandle<()>) -> Self {
        Self { addr, handle }
    }

    /// Gets an address of current worker
    pub fn addr(&self) -> Addr<W> {
        self.addr.clone()
    }

    /// Gets an recipient for current worker with specified message type
    pub fn recipient<M>(&self) -> Recipient<M>
    where
        M: Message + Send + 'static,
        W: ContextHandler<M>,
        M::Result: Send,
    {
        self.addr().recipient()
    }

    /// Sends worker specified message
    pub fn notify<M>(&self, message: M)
    where
        M: Message + Send + 'static,
        W: ContextHandler<M>,
        M::Result: Send,
    {
        let addr = self.addr();
        drop(task::spawn(
            async move { addr.do_send(message).await }.in_current_span(),
        ));
    }

    /// Sends actor specified message in some time
    pub fn notify_later<M>(&self, message: M, later: Duration)
    where
        M: Message + Send + 'static,
        W: Handler<M>,
        M::Result: Send,
    {
        let addr = self.addr();
        drop(task::spawn(
            async move {
                time::sleep(later).await;
                addr.do_send(message).await
            }
            .in_current_span(),
        ));
    }

    /// Sends actor specified message in a loop with specified duration
    pub fn notify_every<M>(&self, every: Duration)
    where
        M: Message + Default + Send + 'static,
        W: Handler<M>,
        M::Result: Send,
    {
        let addr = self.addr();
        drop(task::spawn(
            async move {
                loop {
                    time::sleep(every).await;
                    let _res = addr.do_send(M::default()).await;
                }
            }
            .in_current_span(),
        ));
    }

    /// Notifies actor with items from stream
    pub fn notify_with<M, S>(&self, mut stream: S)
    where
        M: Message + Send + 'static,
        S: Stream<Item = M> + Unpin + Send + 'static,
        W: Handler<M>,
        M::Result: Send,
    {
        let addr = self.addr();
        drop(task::spawn(
            async move {
                while let Some(item) = stream.next().await {
                    addr.do_send(item).await;
                }
            }
            .in_current_span(),
        ));
    }

src/worker/deadlock.rs (line 193)

pub(crate) async fn r#in(to: WorkerId, from: WorkerId) {
    DEADLOCK_ACTOR.do_send(AddEdge { from, to }).await;
}

pub(crate) async fn out(to: WorkerId, from: WorkerId) {
    DEADLOCK_ACTOR.do_send(RemoveEdge { from, to }).await;
}

pub fn recipient<M>(&self) -> Recipient<M>where
    M: Message + Send + 'static,
    W: ContextHandler<M>,
    M::Result: Send,

Constructs recipient for sending only specific messages

Examples found in repository

src/worker/mod.rs (line 535)

    pub fn recipient<M>(&self) -> Recipient<M>
    where
        M: Message + Send + 'static,
        W: ContextHandler<M>,
        M::Result: Send,
    {
        self.addr().recipient()
    }

src/worker/monitor.rs (line 84)

    pub fn register<W>(mut self, worker: W) -> Self
    where
        W: Worker,
    {
        let addr = tokio::spawn(async {
            let addr = worker.start().await;
            addr.recipient()
        });
        self.workers.push(addr);
        self
    }

pub fn expect_running(self) -> AlwaysAddr<W>

Contstructs address which will never panic on sending.

Beware: You need to make sure that this actor will be always alive

Trait Implementations

impl<A: Actor> Clone for Addr<A>

fn clone(&self) -> Self

Returns a copy of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<A: Debug + Actor> Debug for Addr<A>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.