Struct apalis_core::worker::Context

pub struct Context<W: Actor> { /* private fields */ }

Available on crate feature worker only.

Context for execution of actor

Implementations

impl<W: Actor> Context<W>

pub fn new(addr: Addr<W>, handle: JoinHandle<()>) -> Self

Default constructor

Examples found in repository

src/worker/mod.rs (line 435)

    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
    }

pub fn addr(&self) -> Addr<W>

Gets an address of current worker

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()
    }

    /// 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(),
        ));
    }

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

Gets an recipient for current worker with specified message type

Examples found in repository

src/worker/broker.rs (line 66)

    pub fn subscribe<M: BrokerMessage, A: Actor + ContextHandler<M>>(&self, ctx: &mut Context<A>) {
        self.subscribe_recipient(ctx.recipient::<M>())
    }

src/worker/deadlock.rs (line 145)

    async fn on_start(&mut self, ctx: &mut Context<Self>) {
        let recipient = ctx.recipient::<Reminder>();
        drop(task::spawn(async move {
            loop {
                recipient.send(Reminder).await;
                time::sleep(Duration::from_millis(100)).await
            }
        }));
    }

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

Sends worker specified message

pub fn notify_later<M>(&self, message: M, later: Duration)where
    M: Message + Send + 'static,
    W: Handler<M>,
    M::Result: Send,

Sends actor specified message in some time

pub fn notify_every<M>(&self, every: Duration)where
    M: Message + Default + Send + 'static,
    W: Handler<M>,
    M::Result: Send,

Sends actor specified message in a loop with specified duration

pub fn notify_with<M, S>(&self, stream: S)where
    M: Message + Send + 'static,
    S: Stream<Item = M> + Unpin + Send + 'static,
    W: Handler<M>,
    M::Result: Send,

Notifies actor with items from stream

Examples found in repository

src/worker/mod.rs (line 312)

    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));
    }

src/storage/worker.rs (line 114)

    async fn on_start(&mut self, ctx: &mut Context<Self>) {
        // To change this just modify the controller then restart.
        ctx.notify_with(KeepAliveStream::new(interval(self.config.keep_alive)));
        // Sets up reactivate orphaned jobs
        // Setup scheduling for non_sql storages eg Redis
        for (pulse, duration) in self.config.heartbeats.iter() {
            let start = Instant::now() + Duration::from_millis(17);
            ctx.notify_with(HeartbeatStream::new(
                pulse.clone(),
                interval_at(start, *duration),
            ));
        }
    }

Trait Implementations

impl<W: Debug + Actor> Debug for Context<W>

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

Formats the value using the given formatter.