Merge 2f14fc6 into 8bc2899

benches/storages.rs

@@ -6,7 +6,6 @@ use apalis::{
     postgres::{PgPool, PostgresStorage},
     sqlite::{SqlitePool, SqliteStorage},
 };
-use apalis_redis::Config;
 use criterion::*;
 use futures::Future;
 use paste::paste;

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

@@ -9,13 +9,12 @@ use apalis::{
 
 use apalis_core::{
     codec::json::JsonCodec,
-    layers::{Ack, AckLayer},
+    layers::{Ack, AckLayer, AckResponse},
 };
 use email_service::{send_email, Email};
 use futures::{channel::mpsc, SinkExt};
 use rsmq_async::{Rsmq, RsmqConnection, RsmqError};
 use tokio::time::sleep;
-use tower::layer::util::Identity;
 use tracing::{error, info};
 
 struct RedisMq<T> {
@@ -71,13 +70,9 @@ impl<T: Send> Ack<T> for RedisMq<T> {
 
     type Error = RsmqError;
 
-    async fn ack(
-        &mut self,
-        worker_id: &WorkerId,
-        data: &Self::Acknowledger,
-    ) -> Result<(), Self::Error> {
-        println!("Attempting to ACK {}", data);
-        self.conn.delete_message("email", data).await?;
+    async fn ack(&mut self, ack: AckResponse<String>) -> Result<(), Self::Error> {
+        println!("Attempting to ACK {}", ack.acknowledger);
+        self.conn.delete_message("email", &ack.acknowledger).await?;
         Ok(())
     }
 }
@@ -141,8 +136,7 @@ async fn main() -> Result<()> {
         codec: RedisCodec::new(Box::new(JsonCodec)),
         config: Config::default(),
     };
-    // This can be in another part of the program
-    // produce_jobs(&mut mq).await?;
+    produce_jobs(&mut mq).await?;
 
     let worker = WorkerBuilder::new("rango-tango")
         .layer(TraceLayer::new())

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

@@ -12,7 +12,8 @@ pub trait MessageQueue<Message>: Backend<Request<Message>> {
     type Error;
 
     /// Enqueues a message to the queue.
-    fn enqueue(&mut self, message: Message) -> impl Future<Output = Result<(), Self::Error>> + Send;
+    fn enqueue(&mut self, message: Message)
+        -> impl Future<Output = Result<(), Self::Error>> + Send;
 
     /// Attempts to dequeue a message from the queue.
     /// Returns `None` if the queue is empty.