fix: standardize backend for storage and mq

geofmureithi · Jul 10, 2024 · a76dc1f · a76dc1f
1 parent d5496ff
commit a76dc1f
Show file tree

Hide file tree

Showing 18 changed files with 26 additions and 45 deletions.
diff --git a/README.md b/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()

diff --git a/examples/actix-web/src/main.rs b/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());

diff --git a/examples/axum/src/main.rs b/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()

diff --git a/examples/basics/src/main.rs b/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))

diff --git a/examples/mysql/src/main.rs b/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()

diff --git a/examples/postgres/src/main.rs b/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:?}"))

diff --git a/examples/prometheus/src/main.rs b/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()

diff --git a/examples/redis-deadpool/src/main.rs b/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);
 

diff --git a/examples/redis-mq-example/src/main.rs b/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()

diff --git a/examples/redis-with-msg-pack/src/main.rs b/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()

diff --git a/examples/redis/src/main.rs b/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()

diff --git a/examples/rest-api/src/main.rs b/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();

diff --git a/examples/sentry/src/main.rs b/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()

diff --git a/examples/sqlite/src/main.rs b/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()

diff --git a/examples/tracing/src/main.rs b/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()

diff --git a/packages/apalis-core/src/builder.rs b/packages/apalis-core/src/builder.rs
@@ -9,7 +9,6 @@ use tower::{
 use crate::{
     error::Error,
     layers::extensions::Data,
-    mq::MessageQueue,
     request::Request,
     service_fn::service_fn,
     service_fn::ServiceFn,
@@ -84,22 +83,8 @@ impl<J, S, M, Serv> WorkerBuilder<J, S, M, Serv> {
         }
     }
 
-    /// 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 only [Backend]
+    pub fn backend<NS: Backend<Request<NJ>>, NJ>(
         self,
         backend: NS,
     ) -> WorkerBuilder<NJ, NS, M, Serv> {
@@ -168,16 +153,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);

diff --git a/packages/apalis-core/src/monitor/mod.rs b/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| {

diff --git a/packages/apalis-core/src/worker/mod.rs b/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();