From fa395efbae67190de3dba50c2c50f0fde230a401 Mon Sep 17 00:00:00 2001 From: asonix Date: Wed, 22 Mar 2023 22:15:38 -0500 Subject: [PATCH] Add metrics example --- Cargo.toml | 1 + examples/metrics-example/.gitignore | 1 + examples/metrics-example/Cargo.toml | 19 +++++ examples/metrics-example/src/main.rs | 119 +++++++++++++++++++++++++++ 4 files changed, 140 insertions(+) create mode 100644 examples/metrics-example/.gitignore create mode 100644 examples/metrics-example/Cargo.toml create mode 100644 examples/metrics-example/src/main.rs diff --git a/Cargo.toml b/Cargo.toml index 5030abf..502d90c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -17,6 +17,7 @@ members = [ "examples/basic-example", "examples/long-example", "examples/managed-example", + "examples/metrics-example", "examples/panic-example", ] diff --git a/examples/metrics-example/.gitignore b/examples/metrics-example/.gitignore new file mode 100644 index 0000000..23f168a --- /dev/null +++ b/examples/metrics-example/.gitignore @@ -0,0 +1 @@ +/my-sled-db diff --git a/examples/metrics-example/Cargo.toml b/examples/metrics-example/Cargo.toml new file mode 100644 index 0000000..1b0bce2 --- /dev/null +++ b/examples/metrics-example/Cargo.toml @@ -0,0 +1,19 @@ +[package] +name = "metrics-example" +version = "0.1.0" +authors = ["asonix "] +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.14.0", path = "../..", features = [ + "error-logging", +] } +background-jobs-sled-storage = { version = "0.10.0", path = "../../jobs-sled" } +tracing = "0.1" +tracing-subscriber = { version = "0.2", features = ["fmt"] } +serde = { version = "1.0", features = ["derive"] } +sled = "0.34" diff --git a/examples/metrics-example/src/main.rs b/examples/metrics-example/src/main.rs new file mode 100644 index 0000000..fe19174 --- /dev/null +++ b/examples/metrics-example/src/main.rs @@ -0,0 +1,119 @@ +use actix_rt::Arbiter; +use anyhow::Error; +use background_jobs::{ + // memory_storage::{ActixTimer, Storage}, + ActixJob as Job, + MaxRetries, + WorkerConfig, +}; +use background_jobs_sled_storage::Storage; +use std::{future::Future, pin::Pin, time::Duration}; +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, +} + +#[actix_rt::main] +async fn main() -> Result<(), Error> { + let env_filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("warn")); + + let stats_handle = background_jobs::metrics::install()?; + + 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 storage = Storage::new(ActixTimer); + + let arbiter = Arbiter::new(); + + // Configure and start our workers + let queue_handle = + WorkerConfig::new_in_arbiter(arbiter.handle(), storage, |_| MyState::new("My App")) + .register::() + .set_worker_count(DEFAULT_QUEUE, 4) + .start(); + + // Queue our jobs + for _ in 0..1000 { + queue_handle.queue(MyJob::new(0, 0)).await?; + } + + actix_rt::spawn(async move { + loop { + actix_rt::time::sleep(Duration::from_millis(500)).await; + println!("{:?}", stats_handle.get()); + } + }); + + // 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 Future = Pin> + 'static>>; + + // 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); + + Box::pin(async move { + actix_rt::time::sleep(Duration::from_millis(50)).await; + Ok(()) + }) + } +}