Struct apalis_core::request::JobRequest

pub struct JobRequest<T> { /* private fields */ }

Represents a job which can be serialized and executed

Implementations

impl<T> JobRequest<T>

pub fn new(job: T) -> Self

Creates a new JobRequest

pub fn new_with_context(job: T, ctx: JobContext) -> Self

Creates a Job request with context provided

pub fn inner(&self) -> &T

Get the underlying reference of the [Job]

Examples found in repository

src/storage/worker.rs (line 157)

    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 context_mut(&mut self) -> &mut JobContext

Gets a mutable reference to the job context.

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 context(&self) -> &JobContext

Gets a reference to the job context.

pub fn record_attempt(&mut self)

Records a job attempt

Examples found in repository

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

    fn clone_request(&self, req: &Req<T>) -> Option<Req<T>> {
        let mut req = req.clone();
        req.record_attempt();
        Some(req)
    }

src/storage/worker.rs (line 147)

    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
    }

Methods from Deref<Target = JobContext>

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<T: Clone> Clone for JobRequest<T>

fn clone(&self) -> JobRequest<T>

Returns a copy of the value.

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

Performs copy-assignment from source.

impl<T: Debug> Debug for JobRequest<T>

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

Formats the value using the given formatter.

impl<T> Deref for JobRequest<T>

type Target = JobContext

The resulting type after dereferencing.

fn deref(&self) -> &Self::Target

Dereferences the value.

impl<T> DerefMut for JobRequest<T>

fn deref_mut(&mut self) -> &mut Self::Target

Mutably dereferences the value.

impl<'de, T> Deserialize<'de> for JobRequest<T>where
    T: Deserialize<'de>,

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

Deserialize this value from the given Serde deserializer.

impl<T> Policy<JobRequest<T>, JobResult, JobError> for DefaultRetryPolicywhere
    T: Clone,

Available on crate feature retry only.

type Future = Ready<DefaultRetryPolicy>

The Future type returned by Policy::retry.

fn retry(
    &self,
    req: &JobRequest<T>,
    result: Result<&JobResult, &JobError>
) -> Option<Self::Future>

Check the policy if a certain request should be retried.

fn clone_request(&self, req: &JobRequest<T>) -> Option<JobRequest<T>>

Tries to clone a request before being passed to the inner service.

impl<T> Serialize for JobRequest<T>where
    T: Serialize,

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

Serialize this value into the given Serde serializer.

impl<J, S, T> Service<JobRequest<J>> for AddExtension<S, T>where
    S: Service<JobRequest<J>>,
    T: Clone + Send + Sync + 'static,

Available on crate feature extensions only.

type Response = <S as Service<JobRequest<J>>>::Response

Responses given by the service.

type Error = <S as Service<JobRequest<J>>>::Error

Errors produced by the service.

type Future = <S as Service<JobRequest<J>>>::Future

The future response value.

fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>>

Returns Poll::Ready(Ok(())) when the service is able to process requests.

fn call(&mut self, req: JobRequest<J>) -> Self::Future

Process the request and return the response asynchronously.

impl<S, J, F> Service<JobRequest<J>> for PrometheusService<S>where
    S: Service<JobRequest<J>, Response = JobResult, Error = JobError, Future = F>,
    F: Future<Output = Result<JobResult, JobError>> + 'static,
    J: Job,

Available on crate feature prometheus only.

type Response = <S as Service<JobRequest<J>>>::Response

Responses given by the service.

type Error = <S as Service<JobRequest<J>>>::Error

Errors produced by the service.

type Future = ResponseFuture<F>

The future response value.

fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>>

Returns Poll::Ready(Ok(())) when the service is able to process requests.

fn call(&mut self, request: JobRequest<J>) -> Self::Future

Process the request and return the response asynchronously.

impl<S, J, F> Service<JobRequest<J>> for SentryJobService<S>where
    S: Service<JobRequest<J>, Response = JobResult, Error = JobError, Future = F>,
    F: Future<Output = Result<JobResult, JobError>> + 'static,
    J: Job,

Available on crate feature sentry only.

type Response = <S as Service<JobRequest<J>>>::Response

Responses given by the service.

type Error = <S as Service<JobRequest<J>>>::Error

Errors produced by the service.

type Future = SentryHttpFuture<<S as Service<JobRequest<J>>>::Future>

The future response value.

fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>>

Returns Poll::Ready(Ok(())) when the service is able to process requests.

fn call(&mut self, request: JobRequest<J>) -> Self::Future

Process the request and return the response asynchronously.

impl<J, S, OnRequestT, OnResponseT, OnFailureT, MakeSpanT, F> Service<JobRequest<J>> for Trace<S, MakeSpanT, OnRequestT, OnResponseT, OnFailureT>where
    S: Service<JobRequest<J>, Response = JobResult, Error = JobError, Future = F> + Unpin + Send + 'static,
    S::Error: Display + 'static,
    MakeSpanT: MakeSpan<J>,
    OnRequestT: OnRequest<J>,
    OnResponseT: OnResponse + Clone + 'static,
    F: Future<Output = Result<JobResult, JobError>> + 'static,
    OnFailureT: OnFailure + Clone + 'static,

Available on crate feature trace only.

type Response = JobResult

Responses given by the service.

type Error = JobError

Errors produced by the service.

type Future = ResponseFuture<F, OnResponseT, OnFailureT>

The future response value.

fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>>

Returns Poll::Ready(Ok(())) when the service is able to process requests.

fn call(&mut self, req: JobRequest<J>) -> Self::Future

Process the request and return the response asynchronously.

impl<T, F, Res, Request> Service<JobRequest<Request>> for JobFn<T>where
    Request: Debug + 'static + Job,
    T: Fn(Request, JobContext) -> F,
    Res: IntoJobResponse,
    F: Future<Output = Res> + 'static + Send,

type Response = JobResult

Responses given by the service.

type Error = JobError

Errors produced by the service.

type Future = JobFnHttpFuture<Pin<Box<dyn Future<Output = Result<JobResult, JobError>> + Send + 'static, Global>>>

The future response value.

fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>>

Returns Poll::Ready(Ok(())) when the service is able to process requests.

fn call(&mut self, job: JobRequest<Request>) -> Self::Future

Process the request and return the response asynchronously.