Struct apalis_core::context::JobContext

pub struct JobContext { /* private fields */ }

The context for a job is represented here Used to provide a context when a job is defined through the [Job] trait

Implementations

impl JobContext

pub fn new(id: String) -> Self

Build a new context with defaults given an ID.

Examples found in repository

src/request.rs (line 46)

    pub fn new(job: T) -> Self {
        let id = uuid::Uuid::new_v4().to_string();
        let context = JobContext::new(id);
        Self { job, context }
    }

pub fn data_opt<D: Any + Send + Sync>(&self) -> Option<&D>

Get a reference to a type previously inserted on this JobContext.

Example

let mut ctx = JobContext::new(1.to_string());
assert!(ctx.data_opt::<i32>().is_none());
ctx.insert(5i32);

assert_eq!(ctx.data_opt::<i32>(), Some(&5i32));

pub fn insert<D: Any + Send + Sync>(&mut self, data: D) -> Option<D>

Insert a type into this JobContext.

Important for embedding data for a job. If a extension of this type already existed, it will be returned.

Example

let mut ctx = JobContext::new(1.to_string());
assert!(ctx.insert(5i32).is_none());
assert!(ctx.insert(4u8).is_none());
assert_eq!(ctx.insert(9i32), Some(5i32));

Examples found in repository

src/layers/extensions/mod.rs (line 81)

    fn call(&mut self, mut req: JobRequest<J>) -> Self::Future {
        req.context_mut().insert(self.value.clone());
        self.inner.call(req)
    }

pub fn set_max_attempts(&mut self, max_attempts: i32)

Set the number of attempts

pub fn max_attempts(&self) -> i32

Gets the maximum attempts for a job. Default 25

Examples found in repository

src/layers/retry/mod.rs (line 30)

    fn retry(&self, req: &Req<T>, result: Result<&Res, &Err>) -> Option<Self::Future> {
        match result {
            Ok(_) => {
                // Treat all `Response`s as success,
                // so don't retry...
                None
            }
            Err(_) if (req.max_attempts() - req.attempts() > 0) => {
                Some(future::ready(DefaultRetryPolicy))
            }
            Err(_) => None,
        }
    }

pub fn id(&self) -> String

Get the id for a job

Examples found in repository

src/layers/sentry/mod.rs (line 154)

    fn call(&mut self, request: JobRequest<J>) -> Self::Future {
        let op = J::NAME;
        let trx_ctx = sentry_core::TransactionContext::new(op, "apalis.job");
        let job_type = std::any::type_name::<J>().to_string();
        let job_details = JobDetails {
            job_id: request.id(),
            current_attempt: request.attempts(),
            job_type,
        };

        SentryHttpFuture {
            on_first_poll: Some((job_details, trx_ctx)),
            transaction: None,
            future: self.service.call(request),
        }
    }

src/storage/worker.rs (line 144)

    async fn handle_job(&mut self, mut job: JobRequest<Self::Job>) -> Result<JobResult, JobError> {
        let instant = Instant::now();
        let mut storage = self.storage.clone();
        let worker_id = self.id.to_string();
        let handle = self.service().ready().await?;
        let job_id = job.id();
        job.set_status(JobState::Running);
        job.set_lock_at(Some(Utc::now()));
        job.record_attempt();
        job.set_lock_by(Some(worker_id.clone()));
        if let Err(e) = storage.update_by_id(job_id.clone(), &job).await {
            #[cfg(feature = "broker")]
            Broker::global()
                .issue_send(WorkerMessage::new(
                    worker_id.clone(),
                    WorkerEvent::Error(format!("{}", e)),
                ))
                .await;
            T::on_worker_error(job.inner(), &job, &WorkerError::Storage(e));
        };
        T::on_service_ready(job.inner(), &job, instant.elapsed());
        let res = handle.call(job).await;

        if let Ok(Some(mut job)) = storage.fetch_by_id(job_id.clone()).await {
            job.set_done_at(Some(Utc::now()));
            let finalize = match res {
                Ok(ref r) => match r {
                    JobResult::Success => {
                        job.set_status(JobState::Done);
                        storage.ack(worker_id.clone(), job_id.clone()).await
                    }
                    JobResult::Retry => {
                        job.set_status(JobState::Retry);
                        storage.retry(worker_id.clone(), job_id.clone()).await
                    }
                    JobResult::Kill => {
                        job.set_status(JobState::Killed);
                        storage.kill(worker_id.clone(), job_id.clone()).await
                    }

                    JobResult::Reschedule(wait) => {
                        job.set_status(JobState::Retry);
                        storage.reschedule(&job, *wait).await
                    }
                },
                Err(ref e) => {
                    job.set_status(JobState::Failed);
                    job.set_last_error(format!("{}", e));

                    #[cfg(feature = "broker")]
                    Broker::global()
                        .issue_send(WorkerMessage::new(
                            worker_id.clone(),
                            WorkerEvent::Error(format!("{}", e)),
                        ))
                        .await;
                    // let base: i32 = 2; // an explicit type is required
                    // let millis = base.pow(job.attempts());
                    storage.reschedule(&job, Duration::from_millis(10000)).await
                }
            };
            if let Err(e) = finalize {
                #[cfg(feature = "broker")]
                Broker::global()
                    .issue_send(WorkerMessage::new(
                        worker_id.clone(),
                        WorkerEvent::Error(format!("{}", e)),
                    ))
                    .await;
                T::on_worker_error(job.inner(), &job, &WorkerError::Storage(e));
            }
            if let Err(e) = storage.update_by_id(job_id.clone(), &job).await {
                #[cfg(feature = "broker")]
                Broker::global()
                    .issue_send(WorkerMessage::new(
                        worker_id.clone(),
                        WorkerEvent::Error(format!("{}", e)),
                    ))
                    .await;
                T::on_worker_error(job.inner(), &job, &WorkerError::Storage(e));
            };
        }

        res
    }

pub fn attempts(&self) -> i32

Gets the current attempts for a job. Default 0

Examples found in repository

src/request.rs (line 72)

    pub fn record_attempt(&mut self) {
        self.context.set_attempts(self.context.attempts() + 1);
    }

src/layers/retry/mod.rs (line 30)

    fn retry(&self, req: &Req<T>, result: Result<&Res, &Err>) -> Option<Self::Future> {
        match result {
            Ok(_) => {
                // Treat all `Response`s as success,
                // so don't retry...
                None
            }
            Err(_) if (req.max_attempts() - req.attempts() > 0) => {
                Some(future::ready(DefaultRetryPolicy))
            }
            Err(_) => None,
        }
    }

src/layers/sentry/mod.rs (line 155)

    fn call(&mut self, request: JobRequest<J>) -> Self::Future {
        let op = J::NAME;
        let trx_ctx = sentry_core::TransactionContext::new(op, "apalis.job");
        let job_type = std::any::type_name::<J>().to_string();
        let job_details = JobDetails {
            job_id: request.id(),
            current_attempt: request.attempts(),
            job_type,
        };

        SentryHttpFuture {
            on_first_poll: Some((job_details, trx_ctx)),
            transaction: None,
            future: self.service.call(request),
        }
    }

pub fn set_attempts(&mut self, attempts: i32)

Set the number of attempts

Examples found in repository

src/request.rs (line 72)

    pub fn record_attempt(&mut self) {
        self.context.set_attempts(self.context.attempts() + 1);
    }

pub fn done_at(&self) -> &Option<DateTime<Utc>>

Get the time a job was done

pub fn set_done_at(&mut self, done_at: Option<DateTime<Utc>>)

Set the time a job was done

Examples found in repository

src/storage/worker.rs (line 163)

    async fn handle_job(&mut self, mut job: JobRequest<Self::Job>) -> Result<JobResult, JobError> {
        let instant = Instant::now();
        let mut storage = self.storage.clone();
        let worker_id = self.id.to_string();
        let handle = self.service().ready().await?;
        let job_id = job.id();
        job.set_status(JobState::Running);
        job.set_lock_at(Some(Utc::now()));
        job.record_attempt();
        job.set_lock_by(Some(worker_id.clone()));
        if let Err(e) = storage.update_by_id(job_id.clone(), &job).await {
            #[cfg(feature = "broker")]
            Broker::global()
                .issue_send(WorkerMessage::new(
                    worker_id.clone(),
                    WorkerEvent::Error(format!("{}", e)),
                ))
                .await;
            T::on_worker_error(job.inner(), &job, &WorkerError::Storage(e));
        };
        T::on_service_ready(job.inner(), &job, instant.elapsed());
        let res = handle.call(job).await;

        if let Ok(Some(mut job)) = storage.fetch_by_id(job_id.clone()).await {
            job.set_done_at(Some(Utc::now()));
            let finalize = match res {
                Ok(ref r) => match r {
                    JobResult::Success => {
                        job.set_status(JobState::Done);
                        storage.ack(worker_id.clone(), job_id.clone()).await
                    }
                    JobResult::Retry => {
                        job.set_status(JobState::Retry);
                        storage.retry(worker_id.clone(), job_id.clone()).await
                    }
                    JobResult::Kill => {
                        job.set_status(JobState::Killed);
                        storage.kill(worker_id.clone(), job_id.clone()).await
                    }

                    JobResult::Reschedule(wait) => {
                        job.set_status(JobState::Retry);
                        storage.reschedule(&job, *wait).await
                    }
                },
                Err(ref e) => {
                    job.set_status(JobState::Failed);
                    job.set_last_error(format!("{}", e));

                    #[cfg(feature = "broker")]
                    Broker::global()
                        .issue_send(WorkerMessage::new(
                            worker_id.clone(),
                            WorkerEvent::Error(format!("{}", e)),
                        ))
                        .await;
                    // let base: i32 = 2; // an explicit type is required
                    // let millis = base.pow(job.attempts());
                    storage.reschedule(&job, Duration::from_millis(10000)).await
                }
            };
            if let Err(e) = finalize {
                #[cfg(feature = "broker")]
                Broker::global()
                    .issue_send(WorkerMessage::new(
                        worker_id.clone(),
                        WorkerEvent::Error(format!("{}", e)),
                    ))
                    .await;
                T::on_worker_error(job.inner(), &job, &WorkerError::Storage(e));
            }
            if let Err(e) = storage.update_by_id(job_id.clone(), &job).await {
                #[cfg(feature = "broker")]
                Broker::global()
                    .issue_send(WorkerMessage::new(
                        worker_id.clone(),
                        WorkerEvent::Error(format!("{}", e)),
                    ))
                    .await;
                T::on_worker_error(job.inner(), &job, &WorkerError::Storage(e));
            };
        }

        res
    }

pub fn run_at(&self) -> &DateTime<Utc>

Get the time a job is supposed to start

pub fn set_run_at(&mut self, run_at: DateTime<Utc>)

Set the time a job should run

pub fn lock_at(&self) -> &Option<DateTime<Utc>>

Get the time a job was locked

pub fn set_lock_at(&mut self, lock_at: Option<DateTime<Utc>>)

Set the lock_at value

Examples found in repository

src/storage/worker.rs (line 146)

    async fn handle_job(&mut self, mut job: JobRequest<Self::Job>) -> Result<JobResult, JobError> {
        let instant = Instant::now();
        let mut storage = self.storage.clone();
        let worker_id = self.id.to_string();
        let handle = self.service().ready().await?;
        let job_id = job.id();
        job.set_status(JobState::Running);
        job.set_lock_at(Some(Utc::now()));
        job.record_attempt();
        job.set_lock_by(Some(worker_id.clone()));
        if let Err(e) = storage.update_by_id(job_id.clone(), &job).await {
            #[cfg(feature = "broker")]
            Broker::global()
                .issue_send(WorkerMessage::new(
                    worker_id.clone(),
                    WorkerEvent::Error(format!("{}", e)),
                ))
                .await;
            T::on_worker_error(job.inner(), &job, &WorkerError::Storage(e));
        };
        T::on_service_ready(job.inner(), &job, instant.elapsed());
        let res = handle.call(job).await;

        if let Ok(Some(mut job)) = storage.fetch_by_id(job_id.clone()).await {
            job.set_done_at(Some(Utc::now()));
            let finalize = match res {
                Ok(ref r) => match r {
                    JobResult::Success => {
                        job.set_status(JobState::Done);
                        storage.ack(worker_id.clone(), job_id.clone()).await
                    }
                    JobResult::Retry => {
                        job.set_status(JobState::Retry);
                        storage.retry(worker_id.clone(), job_id.clone()).await
                    }
                    JobResult::Kill => {
                        job.set_status(JobState::Killed);
                        storage.kill(worker_id.clone(), job_id.clone()).await
                    }

                    JobResult::Reschedule(wait) => {
                        job.set_status(JobState::Retry);
                        storage.reschedule(&job, *wait).await
                    }
                },
                Err(ref e) => {
                    job.set_status(JobState::Failed);
                    job.set_last_error(format!("{}", e));

                    #[cfg(feature = "broker")]
                    Broker::global()
                        .issue_send(WorkerMessage::new(
                            worker_id.clone(),
                            WorkerEvent::Error(format!("{}", e)),
                        ))
                        .await;
                    // let base: i32 = 2; // an explicit type is required
                    // let millis = base.pow(job.attempts());
                    storage.reschedule(&job, Duration::from_millis(10000)).await
                }
            };
            if let Err(e) = finalize {
                #[cfg(feature = "broker")]
                Broker::global()
                    .issue_send(WorkerMessage::new(
                        worker_id.clone(),
                        WorkerEvent::Error(format!("{}", e)),
                    ))
                    .await;
                T::on_worker_error(job.inner(), &job, &WorkerError::Storage(e));
            }
            if let Err(e) = storage.update_by_id(job_id.clone(), &job).await {
                #[cfg(feature = "broker")]
                Broker::global()
                    .issue_send(WorkerMessage::new(
                        worker_id.clone(),
                        WorkerEvent::Error(format!("{}", e)),
                    ))
                    .await;
                T::on_worker_error(job.inner(), &job, &WorkerError::Storage(e));
            };
        }

        res
    }

pub fn status(&self) -> &JobState

Get the job status

pub fn set_status(&mut self, status: JobState)

Set the job status

Examples found in repository

src/storage/worker.rs (line 145)

    async fn handle_job(&mut self, mut job: JobRequest<Self::Job>) -> Result<JobResult, JobError> {
        let instant = Instant::now();
        let mut storage = self.storage.clone();
        let worker_id = self.id.to_string();
        let handle = self.service().ready().await?;
        let job_id = job.id();
        job.set_status(JobState::Running);
        job.set_lock_at(Some(Utc::now()));
        job.record_attempt();
        job.set_lock_by(Some(worker_id.clone()));
        if let Err(e) = storage.update_by_id(job_id.clone(), &job).await {
            #[cfg(feature = "broker")]
            Broker::global()
                .issue_send(WorkerMessage::new(
                    worker_id.clone(),
                    WorkerEvent::Error(format!("{}", e)),
                ))
                .await;
            T::on_worker_error(job.inner(), &job, &WorkerError::Storage(e));
        };
        T::on_service_ready(job.inner(), &job, instant.elapsed());
        let res = handle.call(job).await;

        if let Ok(Some(mut job)) = storage.fetch_by_id(job_id.clone()).await {
            job.set_done_at(Some(Utc::now()));
            let finalize = match res {
                Ok(ref r) => match r {
                    JobResult::Success => {
                        job.set_status(JobState::Done);
                        storage.ack(worker_id.clone(), job_id.clone()).await
                    }
                    JobResult::Retry => {
                        job.set_status(JobState::Retry);
                        storage.retry(worker_id.clone(), job_id.clone()).await
                    }
                    JobResult::Kill => {
                        job.set_status(JobState::Killed);
                        storage.kill(worker_id.clone(), job_id.clone()).await
                    }

                    JobResult::Reschedule(wait) => {
                        job.set_status(JobState::Retry);
                        storage.reschedule(&job, *wait).await
                    }
                },
                Err(ref e) => {
                    job.set_status(JobState::Failed);
                    job.set_last_error(format!("{}", e));

                    #[cfg(feature = "broker")]
                    Broker::global()
                        .issue_send(WorkerMessage::new(
                            worker_id.clone(),
                            WorkerEvent::Error(format!("{}", e)),
                        ))
                        .await;
                    // let base: i32 = 2; // an explicit type is required
                    // let millis = base.pow(job.attempts());
                    storage.reschedule(&job, Duration::from_millis(10000)).await
                }
            };
            if let Err(e) = finalize {
                #[cfg(feature = "broker")]
                Broker::global()
                    .issue_send(WorkerMessage::new(
                        worker_id.clone(),
                        WorkerEvent::Error(format!("{}", e)),
                    ))
                    .await;
                T::on_worker_error(job.inner(), &job, &WorkerError::Storage(e));
            }
            if let Err(e) = storage.update_by_id(job_id.clone(), &job).await {
                #[cfg(feature = "broker")]
                Broker::global()
                    .issue_send(WorkerMessage::new(
                        worker_id.clone(),
                        WorkerEvent::Error(format!("{}", e)),
                    ))
                    .await;
                T::on_worker_error(job.inner(), &job, &WorkerError::Storage(e));
            };
        }

        res
    }

pub fn lock_by(&self) -> &Option<String>

Get the time a job was locked

pub fn set_lock_by(&mut self, lock_by: Option<String>)

Set lock_by

Examples found in repository

src/storage/worker.rs (line 148)

    async fn handle_job(&mut self, mut job: JobRequest<Self::Job>) -> Result<JobResult, JobError> {
        let instant = Instant::now();
        let mut storage = self.storage.clone();
        let worker_id = self.id.to_string();
        let handle = self.service().ready().await?;
        let job_id = job.id();
        job.set_status(JobState::Running);
        job.set_lock_at(Some(Utc::now()));
        job.record_attempt();
        job.set_lock_by(Some(worker_id.clone()));
        if let Err(e) = storage.update_by_id(job_id.clone(), &job).await {
            #[cfg(feature = "broker")]
            Broker::global()
                .issue_send(WorkerMessage::new(
                    worker_id.clone(),
                    WorkerEvent::Error(format!("{}", e)),
                ))
                .await;
            T::on_worker_error(job.inner(), &job, &WorkerError::Storage(e));
        };
        T::on_service_ready(job.inner(), &job, instant.elapsed());
        let res = handle.call(job).await;

        if let Ok(Some(mut job)) = storage.fetch_by_id(job_id.clone()).await {
            job.set_done_at(Some(Utc::now()));
            let finalize = match res {
                Ok(ref r) => match r {
                    JobResult::Success => {
                        job.set_status(JobState::Done);
                        storage.ack(worker_id.clone(), job_id.clone()).await
                    }
                    JobResult::Retry => {
                        job.set_status(JobState::Retry);
                        storage.retry(worker_id.clone(), job_id.clone()).await
                    }
                    JobResult::Kill => {
                        job.set_status(JobState::Killed);
                        storage.kill(worker_id.clone(), job_id.clone()).await
                    }

                    JobResult::Reschedule(wait) => {
                        job.set_status(JobState::Retry);
                        storage.reschedule(&job, *wait).await
                    }
                },
                Err(ref e) => {
                    job.set_status(JobState::Failed);
                    job.set_last_error(format!("{}", e));

                    #[cfg(feature = "broker")]
                    Broker::global()
                        .issue_send(WorkerMessage::new(
                            worker_id.clone(),
                            WorkerEvent::Error(format!("{}", e)),
                        ))
                        .await;
                    // let base: i32 = 2; // an explicit type is required
                    // let millis = base.pow(job.attempts());
                    storage.reschedule(&job, Duration::from_millis(10000)).await
                }
            };
            if let Err(e) = finalize {
                #[cfg(feature = "broker")]
                Broker::global()
                    .issue_send(WorkerMessage::new(
                        worker_id.clone(),
                        WorkerEvent::Error(format!("{}", e)),
                    ))
                    .await;
                T::on_worker_error(job.inner(), &job, &WorkerError::Storage(e));
            }
            if let Err(e) = storage.update_by_id(job_id.clone(), &job).await {
                #[cfg(feature = "broker")]
                Broker::global()
                    .issue_send(WorkerMessage::new(
                        worker_id.clone(),
                        WorkerEvent::Error(format!("{}", e)),
                    ))
                    .await;
                T::on_worker_error(job.inner(), &job, &WorkerError::Storage(e));
            };
        }

        res
    }

pub fn last_error(&self) -> &Option<String>

Get the time a job was locked

pub fn set_last_error(&mut self, error: String)

Set the last error

Examples found in repository

src/storage/worker.rs (line 186)

    async fn handle_job(&mut self, mut job: JobRequest<Self::Job>) -> Result<JobResult, JobError> {
        let instant = Instant::now();
        let mut storage = self.storage.clone();
        let worker_id = self.id.to_string();
        let handle = self.service().ready().await?;
        let job_id = job.id();
        job.set_status(JobState::Running);
        job.set_lock_at(Some(Utc::now()));
        job.record_attempt();
        job.set_lock_by(Some(worker_id.clone()));
        if let Err(e) = storage.update_by_id(job_id.clone(), &job).await {
            #[cfg(feature = "broker")]
            Broker::global()
                .issue_send(WorkerMessage::new(
                    worker_id.clone(),
                    WorkerEvent::Error(format!("{}", e)),
                ))
                .await;
            T::on_worker_error(job.inner(), &job, &WorkerError::Storage(e));
        };
        T::on_service_ready(job.inner(), &job, instant.elapsed());
        let res = handle.call(job).await;

        if let Ok(Some(mut job)) = storage.fetch_by_id(job_id.clone()).await {
            job.set_done_at(Some(Utc::now()));
            let finalize = match res {
                Ok(ref r) => match r {
                    JobResult::Success => {
                        job.set_status(JobState::Done);
                        storage.ack(worker_id.clone(), job_id.clone()).await
                    }
                    JobResult::Retry => {
                        job.set_status(JobState::Retry);
                        storage.retry(worker_id.clone(), job_id.clone()).await
                    }
                    JobResult::Kill => {
                        job.set_status(JobState::Killed);
                        storage.kill(worker_id.clone(), job_id.clone()).await
                    }

                    JobResult::Reschedule(wait) => {
                        job.set_status(JobState::Retry);
                        storage.reschedule(&job, *wait).await
                    }
                },
                Err(ref e) => {
                    job.set_status(JobState::Failed);
                    job.set_last_error(format!("{}", e));

                    #[cfg(feature = "broker")]
                    Broker::global()
                        .issue_send(WorkerMessage::new(
                            worker_id.clone(),
                            WorkerEvent::Error(format!("{}", e)),
                        ))
                        .await;
                    // let base: i32 = 2; // an explicit type is required
                    // let millis = base.pow(job.attempts());
                    storage.reschedule(&job, Duration::from_millis(10000)).await
                }
            };
            if let Err(e) = finalize {
                #[cfg(feature = "broker")]
                Broker::global()
                    .issue_send(WorkerMessage::new(
                        worker_id.clone(),
                        WorkerEvent::Error(format!("{}", e)),
                    ))
                    .await;
                T::on_worker_error(job.inner(), &job, &WorkerError::Storage(e));
            }
            if let Err(e) = storage.update_by_id(job_id.clone(), &job).await {
                #[cfg(feature = "broker")]
                Broker::global()
                    .issue_send(WorkerMessage::new(
                        worker_id.clone(),
                        WorkerEvent::Error(format!("{}", e)),
                    ))
                    .await;
                T::on_worker_error(job.inner(), &job, &WorkerError::Storage(e));
            };
        }

        res
    }

Trait Implementations

impl Clone for JobContext

fn clone(&self) -> JobContext

Returns a copy of the value.

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

Performs copy-assignment from source.

impl Debug for JobContext

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

Formats the value using the given formatter.

impl<'de> Deserialize<'de> for JobContext

fn deserialize<__D>(__deserializer: __D) -> Result<Self, __D::Error>where
    __D: Deserializer<'de>,

Deserialize this value from the given Serde deserializer.

impl Serialize for JobContext

fn serialize<__S>(&self, __serializer: __S) -> Result<__S::Ok, __S::Error>where
    __S: Serializer,

Serialize this value into the given Serde serializer.