Change Cron worker API and update all examples

oxanus-macros/src/queue.rs

@@ -80,8 +80,8 @@ pub fn expand_derive_queue(input: DeriveInput) -> TokenStream {
     quote! {
         #[automatically_derived]
         impl oxanus::Queue for #struct_ident {
-            fn to_config() -> QueueConfig {
-                QueueConfig::#kind(#key)
+            fn to_config() -> oxanus::QueueConfig {
+                oxanus::QueueConfig::#kind(#key)
                     #concurrency
                     #throttle
             }

oxanus-macros/src/worker.rs

@@ -12,8 +12,10 @@ struct OxanusArgs {
     context: Option<Path>,
     error: Option<Path>,
     max_retries: Option<u32>,
+    retry_delay: Option<RetryDelay>,
     unique_id: Option<UniqueIdSpec>,
     on_conflict: Option<Ident>,
+    cron: Option<Cron>,
 }
 
 #[derive(Debug)]
@@ -31,6 +33,41 @@ enum UniqueIdSpec {
     CustomFunc(Path),
 }
 
+#[derive(Debug)]
+enum RetryDelay {
+    /// #[retry_delay = 3]
+    Value(u64),
+    /// #[retry_delay = mymod::func]
+    CustomFunc(Path),
+}
+
+#[derive(Debug, FromMeta)]
+struct Cron {
+    schedule: String,
+    queue: Option<Path>,
+}
+
+impl FromMeta for RetryDelay {
+    fn from_meta(meta: &Meta) -> darling::Result<Self> {
+        match meta {
+            Meta::NameValue(nv) => match &nv.value {
+                Expr::Lit(syn::ExprLit {
+                    lit: syn::Lit::Int(lit),
+                    ..
+                }) => {
+                    let value = lit.base10_parse::<u64>()?;
+                    Ok(RetryDelay::Value(value))
+                }
+                Expr::Path(expr_path) => Ok(RetryDelay::CustomFunc(expr_path.path.clone())),
+                other => Err(Error::custom(format!(
+                    "Unsupported retry_delay value: {other:?}",
+                ))),
+            },
+            _ => Err(Error::custom("retry_delay must be a name-value attribute")),
+        }
+    }
+}
+
 impl FromMeta for UniqueIdSpec {
     fn from_meta(meta: &Meta) -> darling::Result<Self> {
         match meta {
@@ -74,14 +111,14 @@ impl FromMeta for UniqueIdSpec {
                             args.push((ident, nv.value));
                         }
 
-                        _ => return Err(Error::custom("unsupported unique_id syntax")),
+                        _ => return Err(Error::custom("Unsupported unique_id syntax")),
                     }
                 }
 
                 let fmt = fmt.ok_or_else(|| Error::custom("missing fmt = \"...\""))?;
                 Ok(UniqueIdSpec::NamedFormatter { fmt, args })
             }
-            _ => Err(Error::custom("invalid unique_id attribute")),
+            _ => Err(Error::custom("Invalid unique_id attribute")),
         }
     }
 }
@@ -116,6 +153,11 @@ pub fn expand_derive_worker(input: DeriveInput) -> TokenStream {
         None => quote!(),
     };
 
+    let retry_delay = match args.retry_delay {
+        Some(retry_delay) => expand_retry_delay(retry_delay),
+        None => quote!(),
+    };
+
     let on_conflict = match args.on_conflict {
         Some(on_conflict) => quote! {
             fn on_conflict(&self) -> oxanus::JobConflictStrategy {
@@ -136,6 +178,11 @@ pub fn expand_derive_worker(input: DeriveInput) -> TokenStream {
         None => quote!(),
     };
 
+    let cron = match args.cron {
+        Some(cron) => expand_cron(cron),
+        None => quote!(),
+    };
+
     quote! {
         #[automatically_derived]
         #[async_trait::async_trait]
@@ -151,7 +198,30 @@ pub fn expand_derive_worker(input: DeriveInput) -> TokenStream {
 
             #max_retries
 
+            #retry_delay
+
             #on_conflict
+
+            #cron
+        }
+    }
+}
+
+fn expand_retry_delay(retry_delay: RetryDelay) -> TokenStream {
+    match retry_delay {
+        RetryDelay::Value(value) => {
+            quote! {
+                fn retry_delay(&self, _retries: u32) -> u64 {
+                    #value
+                }
+            }
+        }
+        RetryDelay::CustomFunc(func) => {
+            quote! {
+                fn retry_delay(&self, retries: u32) -> u64 {
+                    #func(self, retries)
+                }
+            }
         }
     }
 }
@@ -183,7 +253,7 @@ fn expand_unique_id(spec: UniqueIdSpec, fields: &Fields) -> TokenStream {
             }
         }
 
-        UniqueIdSpec::CustomFunc(path) => quote!(#path(self)),
+        UniqueIdSpec::CustomFunc(func) => quote!(#func(self)),
     };
 
     quote! {
@@ -192,3 +262,30 @@ fn expand_unique_id(spec: UniqueIdSpec, fields: &Fields) -> TokenStream {
         }
     }
 }
+
+fn expand_cron(cron: Cron) -> TokenStream {
+    let cron_schedule = cron.schedule;
+    let cron_queue_config = match cron.queue {
+        Some(queue) => quote! {
+            fn cron_queue_config() -> Option<oxanus::QueueConfig>
+            where
+                Self: Sized,
+            {
+                use oxanus::Queue;
+                Some(#queue::to_config())
+            }
+        },
+        None => quote!(),
+    };
+
+    quote! {
+        fn cron_schedule() -> Option<String>
+        where
+            Self: Sized,
+        {
+            Some(#cron_schedule.to_string())
+        }
+
+        #cron_queue_config
+    }
+}

oxanus/examples/cron.rs

@@ -7,32 +7,22 @@ enum WorkerError {}
 #[derive(Debug, Serialize, Deserialize, Clone)]
 struct WorkerState {}
 
-#[derive(Debug, Serialize, Deserialize, Default)]
+#[derive(Debug, Serialize, Deserialize, oxanus::Worker)]
+#[oxanus(context = WorkerState)]
+#[oxanus(cron(schedule = "*/5 * * * * *", queue = QueueOne))]
 struct TestWorker {}
 
-#[async_trait::async_trait]
-impl oxanus::Worker for TestWorker {
-    type Context = WorkerState;
-    type Error = WorkerError;
-
-    async fn process(
-        &self,
-        oxanus::Context { .. }: &oxanus::Context<WorkerState>,
-    ) -> Result<(), WorkerError> {
+impl TestWorker {
+    async fn process(&self, _: &oxanus::Context<WorkerState>) -> Result<(), WorkerError> {
         tokio::time::sleep(std::time::Duration::from_secs(1)).await;
         Ok(())
     }
 }
 
-#[derive(Serialize)]
+#[derive(Serialize, oxanus::Queue)]
+#[oxanus(key = "one")]
 struct QueueOne;
 
-impl oxanus::Queue for QueueOne {
-    fn to_config() -> oxanus::QueueConfig {
-        oxanus::QueueConfig::as_static("one")
-    }
-}
-
 #[tokio::main]
 pub async fn main() -> Result<(), oxanus::OxanusError> {
     tracing_subscriber::registry()
@@ -43,7 +33,7 @@ pub async fn main() -> Result<(), oxanus::OxanusError> {
     let ctx = oxanus::Context::value(WorkerState {});
     let storage = oxanus::Storage::builder().build_from_env()?;
     let config = oxanus::Config::new(&storage)
-        .register_cron_worker::<TestWorker>("*/5 * * * * *", QueueOne)
+        .register_worker::<TestWorker>()
         .with_graceful_shutdown(tokio::signal::ctrl_c());
 
     oxanus::run(config, ctx).await?;

oxanus/examples/cron_w_err.rs

@@ -9,44 +9,26 @@ enum WorkerError {
 }
 
 #[derive(Debug, Serialize, Deserialize, Clone)]
-struct WorkerState {}
+struct WorkerContext {}
 
-#[derive(Debug, Serialize, Deserialize, Default)]
+#[derive(Debug, Serialize, Deserialize, oxanus::Worker)]
+#[oxanus(max_retries = 3, retry_delay = 0)]
+#[oxanus(cron(schedule = "*/10 * * * * *"))]
 struct TestWorker {}
 
-#[async_trait::async_trait]
-impl oxanus::Worker for TestWorker {
-    type Context = WorkerState;
-    type Error = WorkerError;
-
-    async fn process(
-        &self,
-        oxanus::Context { .. }: &oxanus::Context<WorkerState>,
-    ) -> Result<(), WorkerError> {
+impl TestWorker {
+    async fn process(&self, _: &oxanus::Context<WorkerContext>) -> Result<(), WorkerError> {
         if rand::rng().random_bool(0.5) {
             Err(WorkerError::GenericError("foo".to_string()))
         } else {
             Ok(())
         }
     }
-
-    fn max_retries(&self) -> u32 {
-        3
-    }
-
-    fn retry_delay(&self, _retries: u32) -> u64 {
-        0
-    }
 }
 
-#[derive(Serialize)]
-struct QueueOne;
-
-impl oxanus::Queue for QueueOne {
-    fn to_config() -> oxanus::QueueConfig {
-        oxanus::QueueConfig::as_static("one")
-    }
-}
+#[derive(Serialize, oxanus::Queue)]
+#[oxanus(prefix = "two")]
+struct QueueDynamic(i32);
 
 #[tokio::main]
 pub async fn main() -> Result<(), oxanus::OxanusError> {
@@ -55,10 +37,10 @@ pub async fn main() -> Result<(), oxanus::OxanusError> {
         .with(EnvFilter::from_default_env())
         .init();
 
-    let ctx = oxanus::Context::value(WorkerState {});
+    let ctx = oxanus::Context::value(WorkerContext {});
     let storage = oxanus::Storage::builder().build_from_env()?;
     let config = oxanus::Config::new(&storage)
-        .register_cron_worker::<TestWorker>("*/10 * * * * *", QueueOne)
+        .register_cron_worker::<TestWorker>(QueueDynamic(2))
         .with_graceful_shutdown(tokio::signal::ctrl_c());
 
     oxanus::run(config, ctx).await?;

oxanus/examples/dynamic.rs

@@ -7,24 +7,19 @@ enum WorkerError {}
 #[derive(Debug, Clone)]
 struct WorkerState {}
 
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug, Serialize, Deserialize, oxanus::Worker)]
+#[oxanus(context = WorkerState)]
 struct Worker2Sec {}
 
-#[async_trait::async_trait]
-impl oxanus::Worker for Worker2Sec {
-    type Context = WorkerState;
-    type Error = WorkerError;
-
-    async fn process(
-        &self,
-        oxanus::Context { .. }: &oxanus::Context<WorkerState>,
-    ) -> Result<(), WorkerError> {
-        tokio::time::sleep(std::time::Duration::from_millis(2000)).await;
+impl Worker2Sec {
+    async fn process(&self, _: &oxanus::Context<WorkerState>) -> Result<(), WorkerError> {
+        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
         Ok(())
     }
 }
 
-#[derive(Serialize)]
+#[derive(Serialize, oxanus::Queue)]
+#[oxanus(prefix = "two")]
 struct QueueDynamic(Animal, i32);
 
 #[derive(Debug, Serialize)]
@@ -34,12 +29,6 @@ enum Animal {
     Bird,
 }
 
-impl oxanus::Queue for QueueDynamic {
-    fn to_config() -> oxanus::QueueConfig {
-        oxanus::QueueConfig::as_dynamic("two")
-    }
-}
-
 #[tokio::main]
 pub async fn main() -> Result<(), oxanus::OxanusError> {
     tracing_subscriber::registry()