Compare commits
3 commits
9cb57a03f5
...
8217012237
Author | SHA1 | Date | |
---|---|---|---|
asonix | 8217012237 | ||
asonix | 424f792cb0 | ||
asonix | 048dc341bc |
|
@ -21,6 +21,7 @@ members = [
|
||||||
"jobs-sled",
|
"jobs-sled",
|
||||||
"jobs-tokio",
|
"jobs-tokio",
|
||||||
"examples/basic-example",
|
"examples/basic-example",
|
||||||
|
"examples/error-example",
|
||||||
"examples/long-example",
|
"examples/long-example",
|
||||||
"examples/managed-example",
|
"examples/managed-example",
|
||||||
"examples/metrics-example",
|
"examples/metrics-example",
|
||||||
|
|
|
@ -84,6 +84,7 @@ impl MyJob {
|
||||||
|
|
||||||
impl Job for MyJob {
|
impl Job for MyJob {
|
||||||
type State = MyState;
|
type State = MyState;
|
||||||
|
type Error = Error;
|
||||||
type Future = Ready<Result<(), Error>>;
|
type Future = Ready<Result<(), Error>>;
|
||||||
type Spawner = Spawner;
|
type Spawner = Spawner;
|
||||||
|
|
||||||
|
|
1
examples/error-example/.gitignore
vendored
Normal file
1
examples/error-example/.gitignore
vendored
Normal file
|
@ -0,0 +1 @@
|
||||||
|
/my-sled-db
|
20
examples/error-example/Cargo.toml
Normal file
20
examples/error-example/Cargo.toml
Normal file
|
@ -0,0 +1,20 @@
|
||||||
|
[package]
|
||||||
|
name = "error-example"
|
||||||
|
version = "0.1.0"
|
||||||
|
authors = ["asonix <asonix@asonix.dog>"]
|
||||||
|
edition = "2021"
|
||||||
|
|
||||||
|
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
|
||||||
|
|
||||||
|
[dependencies]
|
||||||
|
actix-rt = "2.0.0"
|
||||||
|
anyhow = "1.0"
|
||||||
|
background-jobs = { version = "0.17.0", path = "../..", features = [
|
||||||
|
"error-logging",
|
||||||
|
"sled",
|
||||||
|
] }
|
||||||
|
time = "0.3"
|
||||||
|
tracing = "0.1"
|
||||||
|
tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] }
|
||||||
|
serde = { version = "1.0", features = ["derive"] }
|
||||||
|
sled = "0.34"
|
132
examples/error-example/src/main.rs
Normal file
132
examples/error-example/src/main.rs
Normal file
|
@ -0,0 +1,132 @@
|
||||||
|
use actix_rt::Arbiter;
|
||||||
|
use anyhow::Error;
|
||||||
|
use background_jobs::{actix::WorkerConfig, sled::Storage, Job, MaxRetries};
|
||||||
|
use std::{
|
||||||
|
future::{ready, Ready},
|
||||||
|
time::{Duration, SystemTime},
|
||||||
|
};
|
||||||
|
use tracing::info;
|
||||||
|
use tracing_subscriber::EnvFilter;
|
||||||
|
|
||||||
|
const DEFAULT_QUEUE: &str = "default";
|
||||||
|
|
||||||
|
#[derive(Clone, Debug)]
|
||||||
|
pub struct MyState {
|
||||||
|
pub app_name: String,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
|
||||||
|
pub struct MyJob {
|
||||||
|
some_usize: usize,
|
||||||
|
other_usize: usize,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
|
||||||
|
pub struct ErroringJob;
|
||||||
|
|
||||||
|
#[actix_rt::main]
|
||||||
|
async fn main() -> Result<(), Error> {
|
||||||
|
let env_filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
|
||||||
|
|
||||||
|
tracing_subscriber::fmt::fmt()
|
||||||
|
.with_env_filter(env_filter)
|
||||||
|
.init();
|
||||||
|
|
||||||
|
// Set up our Storage
|
||||||
|
let db = sled::Config::new().temporary(true).open()?;
|
||||||
|
let storage = Storage::new(db)?;
|
||||||
|
|
||||||
|
let arbiter = Arbiter::new();
|
||||||
|
|
||||||
|
// Configure and start our workers
|
||||||
|
let queue_handle =
|
||||||
|
WorkerConfig::new_in_arbiter(arbiter.handle(), storage, |_| MyState::new("My App"))
|
||||||
|
.register::<ErroringJob>()
|
||||||
|
.register::<MyJob>()
|
||||||
|
.set_worker_count(DEFAULT_QUEUE, 16)
|
||||||
|
.start();
|
||||||
|
|
||||||
|
// Queue some panicking job
|
||||||
|
for _ in 0..32 {
|
||||||
|
queue_handle.queue(ErroringJob).await?;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Queue our jobs
|
||||||
|
queue_handle.queue(MyJob::new(1, 2)).await?;
|
||||||
|
queue_handle.queue(MyJob::new(3, 4)).await?;
|
||||||
|
queue_handle.queue(MyJob::new(5, 6)).await?;
|
||||||
|
queue_handle
|
||||||
|
.schedule(MyJob::new(7, 8), SystemTime::now() + Duration::from_secs(2))
|
||||||
|
.await?;
|
||||||
|
|
||||||
|
// Block on Actix
|
||||||
|
actix_rt::signal::ctrl_c().await?;
|
||||||
|
|
||||||
|
arbiter.stop();
|
||||||
|
let _ = arbiter.join();
|
||||||
|
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
impl MyState {
|
||||||
|
pub fn new(app_name: &str) -> Self {
|
||||||
|
MyState {
|
||||||
|
app_name: app_name.to_owned(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl MyJob {
|
||||||
|
pub fn new(some_usize: usize, other_usize: usize) -> Self {
|
||||||
|
MyJob {
|
||||||
|
some_usize,
|
||||||
|
other_usize,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Job for MyJob {
|
||||||
|
type State = MyState;
|
||||||
|
type Error = Error;
|
||||||
|
type Future = Ready<Result<(), Error>>;
|
||||||
|
|
||||||
|
// The name of the job. It is super important that each job has a unique name,
|
||||||
|
// because otherwise one job will overwrite another job when they're being
|
||||||
|
// registered.
|
||||||
|
const NAME: &'static str = "MyJob";
|
||||||
|
|
||||||
|
// The queue that this processor belongs to
|
||||||
|
//
|
||||||
|
// Workers have the option to subscribe to specific queues, so this is important to
|
||||||
|
// determine which worker will call the processor
|
||||||
|
//
|
||||||
|
// Jobs can optionally override the queue they're spawned on
|
||||||
|
const QUEUE: &'static str = DEFAULT_QUEUE;
|
||||||
|
|
||||||
|
// The number of times background-jobs should try to retry a job before giving up
|
||||||
|
//
|
||||||
|
// Jobs can optionally override this value
|
||||||
|
const MAX_RETRIES: MaxRetries = MaxRetries::Count(1);
|
||||||
|
|
||||||
|
fn run(self, state: MyState) -> Self::Future {
|
||||||
|
info!("{}: args, {:?}", state.app_name, self);
|
||||||
|
|
||||||
|
ready(Ok(()))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Job for ErroringJob {
|
||||||
|
type State = MyState;
|
||||||
|
type Error = Error;
|
||||||
|
type Future = Ready<Result<(), Error>>;
|
||||||
|
|
||||||
|
const NAME: &'static str = "ErroringJob";
|
||||||
|
|
||||||
|
const QUEUE: &'static str = DEFAULT_QUEUE;
|
||||||
|
|
||||||
|
const MAX_RETRIES: MaxRetries = MaxRetries::Count(0);
|
||||||
|
|
||||||
|
fn run(self, _: MyState) -> Self::Future {
|
||||||
|
ready(Err(anyhow::anyhow!("boom")))
|
||||||
|
}
|
||||||
|
}
|
|
@ -90,6 +90,7 @@ impl MyJob {
|
||||||
|
|
||||||
impl Job for MyJob {
|
impl Job for MyJob {
|
||||||
type State = MyState;
|
type State = MyState;
|
||||||
|
type Error = Error;
|
||||||
type Future = Ready<Result<(), Error>>;
|
type Future = Ready<Result<(), Error>>;
|
||||||
type Spawner = Spawner;
|
type Spawner = Spawner;
|
||||||
|
|
||||||
|
@ -120,6 +121,7 @@ impl Job for MyJob {
|
||||||
|
|
||||||
impl Job for LongJob {
|
impl Job for LongJob {
|
||||||
type State = MyState;
|
type State = MyState;
|
||||||
|
type Error = Error;
|
||||||
type Future = Pin<Box<dyn Future<Output = Result<(), Error>>>>;
|
type Future = Pin<Box<dyn Future<Output = Result<(), Error>>>>;
|
||||||
type Spawner = Spawner;
|
type Spawner = Spawner;
|
||||||
|
|
||||||
|
|
|
@ -102,6 +102,7 @@ impl MyJob {
|
||||||
|
|
||||||
impl Job for MyJob {
|
impl Job for MyJob {
|
||||||
type State = MyState;
|
type State = MyState;
|
||||||
|
type Error = Error;
|
||||||
type Future = Ready<Result<(), Error>>;
|
type Future = Ready<Result<(), Error>>;
|
||||||
type Spawner = Spawner;
|
type Spawner = Spawner;
|
||||||
|
|
||||||
|
@ -132,6 +133,7 @@ impl Job for MyJob {
|
||||||
|
|
||||||
impl Job for StopJob {
|
impl Job for StopJob {
|
||||||
type State = MyState;
|
type State = MyState;
|
||||||
|
type Error = Error;
|
||||||
type Future = Ready<Result<(), Error>>;
|
type Future = Ready<Result<(), Error>>;
|
||||||
type Spawner = Spawner;
|
type Spawner = Spawner;
|
||||||
|
|
||||||
|
|
|
@ -89,6 +89,7 @@ impl MyJob {
|
||||||
|
|
||||||
impl Job for MyJob {
|
impl Job for MyJob {
|
||||||
type State = MyState;
|
type State = MyState;
|
||||||
|
type Error = Error;
|
||||||
type Future = Pin<Box<dyn Future<Output = Result<(), Error>> + 'static>>;
|
type Future = Pin<Box<dyn Future<Output = Result<(), Error>> + 'static>>;
|
||||||
type Spawner = Spawner;
|
type Spawner = Spawner;
|
||||||
|
|
||||||
|
|
|
@ -83,6 +83,7 @@ impl MyJob {
|
||||||
|
|
||||||
impl Job for MyJob {
|
impl Job for MyJob {
|
||||||
type State = MyState;
|
type State = MyState;
|
||||||
|
type Error = Error;
|
||||||
type Future = Ready<Result<(), Error>>;
|
type Future = Ready<Result<(), Error>>;
|
||||||
|
|
||||||
// The name of the job. It is super important that each job has a unique name,
|
// The name of the job. It is super important that each job has a unique name,
|
||||||
|
@ -112,6 +113,7 @@ impl Job for MyJob {
|
||||||
|
|
||||||
impl Job for PanickingJob {
|
impl Job for PanickingJob {
|
||||||
type State = MyState;
|
type State = MyState;
|
||||||
|
type Error = Error;
|
||||||
type Future = Ready<Result<(), Error>>;
|
type Future = Ready<Result<(), Error>>;
|
||||||
|
|
||||||
const NAME: &'static str = "PanickingJob";
|
const NAME: &'static str = "PanickingJob";
|
||||||
|
|
|
@ -90,6 +90,7 @@ impl MyJob {
|
||||||
|
|
||||||
impl Job for MyJob {
|
impl Job for MyJob {
|
||||||
type State = MyState;
|
type State = MyState;
|
||||||
|
type Error = anyhow::Error;
|
||||||
type Future = Ready<anyhow::Result<()>>;
|
type Future = Ready<anyhow::Result<()>>;
|
||||||
type Spawner = Spawner;
|
type Spawner = Spawner;
|
||||||
|
|
||||||
|
|
|
@ -79,6 +79,7 @@ impl MyJob {
|
||||||
|
|
||||||
impl Job for MyJob {
|
impl Job for MyJob {
|
||||||
type State = MyState;
|
type State = MyState;
|
||||||
|
type Error = Error;
|
||||||
type Future = Ready<Result<(), Error>>;
|
type Future = Ready<Result<(), Error>>;
|
||||||
|
|
||||||
// The name of the job. It is super important that each job has a unique name,
|
// The name of the job. It is super important that each job has a unique name,
|
||||||
|
|
|
@ -1,7 +1,7 @@
|
||||||
[package]
|
[package]
|
||||||
name = "background-jobs-core"
|
name = "background-jobs-core"
|
||||||
description = "Core types for implementing an asynchronous jobs processor"
|
description = "Core types for implementing an asynchronous jobs processor"
|
||||||
version = "0.17.0"
|
version = "0.17.1"
|
||||||
license = "AGPL-3.0"
|
license = "AGPL-3.0"
|
||||||
authors = ["asonix <asonix@asonix.dog>"]
|
authors = ["asonix <asonix@asonix.dog>"]
|
||||||
repository = "https://git.asonix.dog/asonix/background-jobs"
|
repository = "https://git.asonix.dog/asonix/background-jobs"
|
||||||
|
|
|
@ -48,8 +48,11 @@ pub trait Job: Serialize + DeserializeOwned + 'static {
|
||||||
/// The application state provided to this job at runtime.
|
/// The application state provided to this job at runtime.
|
||||||
type State: Clone + 'static;
|
type State: Clone + 'static;
|
||||||
|
|
||||||
|
/// The error type this job returns
|
||||||
|
type Error: Into<Box<dyn std::error::Error>>;
|
||||||
|
|
||||||
/// The future returned by this job
|
/// The future returned by this job
|
||||||
type Future: Future<Output = Result<(), Error>> + Send;
|
type Future: Future<Output = Result<(), Self::Error>> + Send;
|
||||||
|
|
||||||
/// The name of the job
|
/// The name of the job
|
||||||
///
|
///
|
||||||
|
@ -176,9 +179,9 @@ where
|
||||||
let (fut, span) = res?;
|
let (fut, span) = res?;
|
||||||
|
|
||||||
if let Some(span) = span {
|
if let Some(span) = span {
|
||||||
fut.instrument(span).await?;
|
fut.instrument(span).await.map_err(Into::into)?;
|
||||||
} else {
|
} else {
|
||||||
fut.await?;
|
fut.await.map_err(Into::into)?;
|
||||||
}
|
}
|
||||||
|
|
||||||
Ok(())
|
Ok(())
|
||||||
|
|
|
@ -6,8 +6,6 @@
|
||||||
//! This crate shouldn't be depended on directly, except in the case of implementing a custom jobs
|
//! This crate shouldn't be depended on directly, except in the case of implementing a custom jobs
|
||||||
//! processor. For a default solution based on Actix and Sled, look at the `background-jobs` crate.
|
//! processor. For a default solution based on Actix and Sled, look at the `background-jobs` crate.
|
||||||
|
|
||||||
use anyhow::Error;
|
|
||||||
|
|
||||||
mod catch_unwind;
|
mod catch_unwind;
|
||||||
mod job;
|
mod job;
|
||||||
mod job_info;
|
mod job_info;
|
||||||
|
@ -28,8 +26,8 @@ pub use unsend_job::{JoinError, UnsendJob, UnsendSpawner};
|
||||||
/// The error type returned by the `process` method
|
/// The error type returned by the `process` method
|
||||||
pub enum JobError {
|
pub enum JobError {
|
||||||
/// Some error occurred while processing the job
|
/// Some error occurred while processing the job
|
||||||
#[error("Error performing job: {0}")]
|
#[error("{0}")]
|
||||||
Processing(#[from] Error),
|
Processing(#[from] Box<dyn std::error::Error>),
|
||||||
|
|
||||||
/// Creating a `Job` type from the provided `serde_json::Value` failed
|
/// Creating a `Job` type from the provided `serde_json::Value` failed
|
||||||
#[error("Could not make JSON value from arguments")]
|
#[error("Could not make JSON value from arguments")]
|
||||||
|
|
|
@ -1,5 +1,4 @@
|
||||||
use crate::{Backoff, Job, MaxRetries};
|
use crate::{Backoff, Job, MaxRetries};
|
||||||
use anyhow::Error;
|
|
||||||
use serde::{de::DeserializeOwned, ser::Serialize};
|
use serde::{de::DeserializeOwned, ser::Serialize};
|
||||||
use std::{
|
use std::{
|
||||||
fmt::Debug,
|
fmt::Debug,
|
||||||
|
@ -45,10 +44,13 @@ pub trait UnsendJob: Serialize + DeserializeOwned + 'static {
|
||||||
/// The application state provided to this job at runtime.
|
/// The application state provided to this job at runtime.
|
||||||
type State: Clone + 'static;
|
type State: Clone + 'static;
|
||||||
|
|
||||||
|
/// The error type this job returns
|
||||||
|
type Error: Into<Box<dyn std::error::Error>> + Send;
|
||||||
|
|
||||||
/// The future returned by this job
|
/// The future returned by this job
|
||||||
///
|
///
|
||||||
/// Importantly, this Future does not require Send
|
/// Importantly, this Future does not require Send
|
||||||
type Future: Future<Output = Result<(), Error>>;
|
type Future: Future<Output = Result<(), Self::Error>>;
|
||||||
|
|
||||||
/// The spawner type that will be used to spawn the unsend future
|
/// The spawner type that will be used to spawn the unsend future
|
||||||
type Spawner: UnsendSpawner;
|
type Spawner: UnsendSpawner;
|
||||||
|
@ -148,7 +150,8 @@ where
|
||||||
T: UnsendJob,
|
T: UnsendJob,
|
||||||
{
|
{
|
||||||
type State = T::State;
|
type State = T::State;
|
||||||
type Future = UnwrapFuture<<T::Spawner as UnsendSpawner>::Handle<Result<(), Error>>>;
|
type Error = T::Error;
|
||||||
|
type Future = UnwrapFuture<<T::Spawner as UnsendSpawner>::Handle<Result<(), Self::Error>>>;
|
||||||
|
|
||||||
const NAME: &'static str = <Self as UnsendJob>::NAME;
|
const NAME: &'static str = <Self as UnsendJob>::NAME;
|
||||||
const QUEUE: &'static str = <Self as UnsendJob>::QUEUE;
|
const QUEUE: &'static str = <Self as UnsendJob>::QUEUE;
|
||||||
|
|
Loading…
Reference in a new issue