standardize backend for storage and mq

README.md

@@ -93,7 +93,7 @@ async fn main() -> {
         .register_with_count(2, {
             WorkerBuilder::new(format!("email-worker"))
                 .data(0usize)
-                .with_storage(storage)
+                .backend(storage)
                 .build_fn(send_email)
         })
         .run()

benches/storages.rs

@@ -37,7 +37,7 @@ macro_rules! define_bench {
                                     let worker =
                                         WorkerBuilder::new(format!("{}-bench", $name))
                                             .data(c)
-                                            .source(storage)
+                                            .backend(storage)
                                             .build_fn(handle_test_job);
                                     worker
                                 })

examples/actix-web/src/main.rs

@@ -45,7 +45,7 @@ async fn main() -> Result<()> {
         .register_with_count(2, {
             WorkerBuilder::new("tasty-avocado")
                 .layer(TraceLayer::new())
-                .with_storage(storage)
+                .backend(storage)
                 .build_fn(send_email)
         })
         .run_with_signal(signal::ctrl_c());

examples/axum/src/main.rs

@@ -77,7 +77,7 @@ async fn main() -> Result<()> {
             .register_with_count(2, {
                 WorkerBuilder::new("tasty-pear")
                     .layer(TraceLayer::new())
-                    .with_storage(storage.clone())
+                    .backend(storage.clone())
                     .build_fn(send_email)
             })
             .run()

examples/basics/src/main.rs

@@ -101,7 +101,7 @@ async fn main() -> Result<(), std::io::Error> {
                 // Add shared context to all jobs executed by this worker
                 .data(EmailService::new())
                 .data(ValidEmailCache::new())
-                .with_storage(sqlite)
+                .backend(sqlite)
                 .build_fn(send_email)
         })
         .shutdown_timeout(Duration::from_secs(5))

examples/mysql/src/main.rs

@@ -37,7 +37,7 @@ async fn main() -> Result<()> {
         .register_with_count(1, {
             WorkerBuilder::new("tasty-avocado")
                 .layer(TraceLayer::new())
-                .with_storage(mysql)
+                .backend(mysql)
                 .build_fn(send_email)
         })
         .run()

examples/postgres/src/main.rs

@@ -49,7 +49,7 @@ async fn main() -> Result<()> {
             WorkerBuilder::new("tasty-orange")
                 .layer(TraceLayer::new())
                 .layer(RetryLayer::new(RetryPolicy::retries(5)))
-                .with_storage(pg)
+                .backend(pg)
                 .build_fn(send_email)
         })
         .on_event(|e| debug!("{e:?}"))

examples/prometheus/src/main.rs

@@ -52,7 +52,7 @@ async fn main() -> Result<()> {
             .register_with_count(2, {
                 WorkerBuilder::new("tasty-banana")
                     .layer(PrometheusLayer)
-                    .with_storage(storage.clone())
+                    .backend(storage.clone())
                     .build_fn(send_email)
             })
             .run()

examples/redis-deadpool/src/main.rs

@@ -26,7 +26,7 @@ async fn main() -> Result<()> {
     produce_jobs(&mut storage).await?;
 
     let worker = WorkerBuilder::new("rango-tango")
-        .with_storage(storage)
+        .backend(storage)
         .data(pool)
         .build_fn(send_email);
 

examples/redis-mq-example/src/main.rs

@@ -138,7 +138,7 @@ async fn main() -> Result<()> {
 
     let worker = WorkerBuilder::new("rango-tango")
         .layer(TraceLayer::new())
-        .with_mq(mq)
+        .backend(mq)
         .build_fn(send_email);
 
     Monitor::<TokioExecutor>::new()

examples/redis-with-msg-pack/src/main.rs

@@ -36,7 +36,7 @@ async fn main() -> Result<()> {
     produce_jobs(storage.clone()).await?;
 
     let worker = WorkerBuilder::new("rango-tango")
-        .with_storage(storage)
+        .backend(storage)
         .build_fn(send_email);
 
     Monitor::<TokioExecutor>::new()

examples/redis/src/main.rs

@@ -51,7 +51,7 @@ async fn main() -> Result<()> {
         .layer(RateLimitLayer::new(5, Duration::from_secs(1)))
         .layer(TimeoutLayer::new(Duration::from_millis(500)))
         .data(Count::default())
-        .with_storage(storage)
+        .backend(storage)
         .build_fn(send_email);
 
     Monitor::<TokioExecutor>::new()

examples/rest-api/src/main.rs

@@ -318,28 +318,28 @@ async fn main() -> anyhow::Result<()> {
             WorkerBuilder::new("tasty-apple")
                 .layer(SentryJobLayer)
                 .layer(TraceLayer::new())
-                .with_storage(worker_storage.clone())
+                .backend(worker_storage.clone())
                 .build_fn(send_email)
         })
         .register_with_count(4, move |c| {
             WorkerBuilder::new(format!("tasty-avocado-{c}"))
                 .layer(SentryJobLayer)
                 .layer(TraceLayer::new())
-                .with_storage(sqlite_storage.clone())
+                .backend(sqlite_storage.clone())
                 .build_fn(notification_service)
         })
         .register_with_count(2, move |c| {
             WorkerBuilder::new(format!("tasty-banana-{c}"))
                 .layer(SentryJobLayer)
                 .layer(TraceLayer::new())
-                .with_storage(pg_storage.clone())
+                .backend(pg_storage.clone())
                 .build_fn(document_service)
         })
         .register_with_count(2, move |c| {
             WorkerBuilder::new(format!("tasty-pear-{c}"))
                 .layer(SentryJobLayer::new())
                 .layer(TraceLayer::new())
-                .with_storage(mysql_storage.clone())
+                .backend(mysql_storage.clone())
                 .build_fn(upload_service)
         })
         .run();

examples/sentry/src/main.rs

@@ -138,7 +138,7 @@ async fn main() -> Result<()> {
                 .layer(NewSentryLayer::new_from_top())
                 .layer(SentryLayer::new())
                 .layer(TraceLayer::new())
-                .with_storage(storage.clone())
+                .backend(storage.clone())
                 .build_fn(email_service)
         })
         .run()

examples/sqlite/src/main.rs

@@ -62,13 +62,13 @@ async fn main() -> Result<()> {
         .register_with_count(2, {
             WorkerBuilder::new("tasty-banana")
                 .layer(TraceLayer::new())
-                .with_storage(email_storage)
+                .backend(email_storage)
                 .build_fn(send_email)
         })
         .register_with_count(10, {
             WorkerBuilder::new("tasty-mango")
                 .layer(TraceLayer::new())
-                .with_storage(notification_storage)
+                .backend(notification_storage)
                 .build_fn(job::notify)
         })
         .run()

examples/tracing/src/main.rs

@@ -73,7 +73,7 @@ async fn main() -> Result<()> {
         .register(
             WorkerBuilder::new("tasty-avocado")
                 .chain(|srv| srv.layer(TraceLayer::new()))
-                .with_storage(storage)
+                .backend(storage)
                 .build_fn(email_service),
         )
         .run()

packages/apalis-core/src/builder.rs

@@ -9,11 +9,9 @@ use tower::{
 use crate::{
     error::Error,
     layers::extensions::Data,
-    mq::MessageQueue,
     request::Request,
     service_fn::service_fn,
     service_fn::ServiceFn,
-    storage::Storage,
     worker::{Ready, Worker, WorkerId},
     Backend,
 };
@@ -55,7 +53,7 @@ impl<Serv> WorkerBuilder<(), (), Identity, Serv> {
     }
 }
 
-impl<J, S, M, Serv> WorkerBuilder<J, S, M, Serv> {
+impl<J, M, Serv> WorkerBuilder<J, (), M, Serv> {
     /// Consume a stream directly
     pub fn stream<NS: Stream<Item = Result<Option<Request<NJ>>, Error>> + Send + 'static, NJ>(
         self,
@@ -70,36 +68,8 @@ impl<J, S, M, Serv> WorkerBuilder<J, S, M, Serv> {
         }
     }
 
-    /// Set the source to a [Storage]
-    pub fn with_storage<NS: Storage<Job = NJ>, NJ>(
-        self,
-        storage: NS,
-    ) -> WorkerBuilder<NJ, NS, M, Serv> {
-        WorkerBuilder {
-            request: PhantomData,
-            layer: self.layer,
-            source: storage,
-            id: self.id,
-            service: self.service,
-        }
-    }
-
-    /// Set the source to a [MessageQueue]
-    pub fn with_mq<NS: MessageQueue<NJ>, NJ>(
-        self,
-        message_queue: NS,
-    ) -> WorkerBuilder<NJ, NS, M, Serv> {
-        WorkerBuilder {
-            request: PhantomData,
-            layer: self.layer,
-            source: message_queue,
-            id: self.id,
-            service: self.service,
-        }
-    }
-
-    /// Set the source to a generic backend that implements only [Backend]
-    pub fn source<NS: Backend<Request<NJ>>, NJ>(
+    /// Set the source to a backend that implements [Backend]
+    pub fn backend<NS: Backend<Request<NJ>>, NJ>(
         self,
         backend: NS,
     ) -> WorkerBuilder<NJ, NS, M, Serv> {
@@ -168,16 +138,12 @@ where
 
     S::Response: 'static,
     M: Layer<S>,
-    // P::Layer: Layer<S>,
-    // M: Layer<<P::Layer as Layer<S>>::Service>,
 {
     type Source = P;
 
     type Service = M::Service;
-    /// Build a worker, given a tower service
     fn build(self, service: S) -> Worker<Ready<Self::Service, P>> {
         let worker_id = self.id;
-        // let common_layer = self.source.common_layer(worker_id.clone());
         let poller = self.source;
         let middleware = self.layer;
         let service = middleware.service(service);

packages/apalis-core/src/layers.rs

@@ -91,7 +91,7 @@ pub mod extensions {
     ///
     /// let worker = WorkerBuilder::new("tasty-avocado")
     ///     .data(state)
-    ///     .source(MemoryStorage::new())
+    ///     .backend(MemoryStorage::new())
     ///     .build(service_fn(email_service));
     /// ```
 

packages/apalis-core/src/monitor/mod.rs

@@ -325,7 +325,7 @@ mod tests {
             Ok::<_, io::Error>(request)
         });
         let worker = WorkerBuilder::new("rango-tango")
-            .source(backend)
+            .backend(backend)
             .build(service);
         let monitor: Monitor<TestExecutor> = Monitor::new();
         let monitor = monitor.register(worker);
@@ -351,7 +351,7 @@ mod tests {
             Ok::<_, io::Error>(request)
         });
         let worker = WorkerBuilder::new("rango-tango")
-            .source(backend)
+            .backend(backend)
             .build(service);
         let monitor: Monitor<TestExecutor> = Monitor::new();
         let monitor = monitor.on_event(|e| {

packages/apalis-core/src/worker/mod.rs

@@ -720,8 +720,8 @@ mod tests {
 
     #[tokio::test]
     async fn it_works() {
-        let backend = MemoryStorage::new();
-        let mut handle = backend.clone();
+        let in_memory = MemoryStorage::new();
+        let mut handle = in_memory.clone();
 
         tokio::spawn(async move {
             for i in 0..ITEMS {
@@ -749,7 +749,7 @@ mod tests {
         let worker = WorkerBuilder::new("rango-tango")
             // .chain(|svc| svc.timeout(Duration::from_millis(500)))
             .data(Count::default())
-            .source(backend);
+            .backend(in_memory);
         let worker = worker.build_fn(task);
         let worker = worker.with_executor(TokioTestExecutor);
         let w = worker.clone();

packages/apalis-redis/src/lib.rs

@@ -18,7 +18,7 @@
 //!     Monitor::<TokioExecutor>::new()
 //!        .register(
 //!            WorkerBuilder::new("tasty-pear")
-//!                .source(storage.clone())
+//!                .backend(storage.clone())
 //!                .build_fn(send_email),
 //!        )
 //!        .run()

packages/apalis-sql/src/postgres.rs

@@ -31,7 +31,7 @@
 //!          .register_with_count(4, {
 //!              WorkerBuilder::new(&format!("tasty-avocado"))
 //!                  .data(0usize)
-//!                  .source(pg)
+//!                  .backend(pg)
 //!                  .build_fn(send_email)
 //!          })
 //!          .run()

src/lib.rs

@@ -39,7 +39,7 @@
 //!         .register_with_count(2, {
 //!             WorkerBuilder::new(&format!("quick-sand"))
 //!                 .data(0usize)
-//!                 .source(storage.clone())
+//!                 .backend(storage.clone())
 //!                 .build_fn(send_email)
 //!         })
 //!         .run()