WIP delay support - SQS, Mem, and Rabbit work, Redis is _rough_.

_rabbit/enabled_plugins

@@ -0,0 +1 @@
+[rabbitmq_management, rabbitmq_delayed_message_exchange].

omniqueue/Cargo.toml

@@ -23,6 +23,7 @@ lapin = { version = "2", optional = true }
 redis = { version = "0.23", features = ["tokio-comp", "tokio-native-tls-comp", "streams"], optional = true }
 serde = { version = "1", features = ["derive", "rc"] }
 serde_json = "1"
+svix-ksuid = { version = "0.7.0", optional = true }
 thiserror = "1"
 tokio = { version = "1", features = ["full"] }
 tokio-util = { version = "0.7", optional = true }
@@ -43,6 +44,6 @@ gcp_pubsub = [
 ]
 memory_queue = []
 rabbitmq = ["dep:lapin"]
-redis = ["dep:bb8", "dep:bb8-redis", "dep:redis"]
+redis = ["dep:bb8", "dep:bb8-redis", "dep:redis", "dep:svix-ksuid"]
 redis_cluster = ["redis", "redis/cluster-async"]
 sqs = ["dep:aws-config", "dep:aws-sdk-sqs"]

omniqueue/src/backends/memory_queue.rs

@@ -430,11 +430,15 @@ mod tests {
         let delay = Duration::from_millis(100);
         let now = Instant::now();
         p.send_serde_json_scheduled(&payload1, delay).await.unwrap();
-        let delivery = c.receive().await.unwrap();
-
-        // `receive` will wait until the next message is available, so this assertion should be
-        // true if the delay was observed.
+        let delivery = c
+            .receive_all(1, delay * 2)
+            .await
+            .unwrap()
+            .into_iter()
+            .next()
+            .unwrap();
         assert!(now.elapsed() >= delay);
+        assert!(now.elapsed() < delay * 2);
         assert_eq!(Some(payload1), delivery.payload_serde_json().unwrap());
     }
 }

omniqueue/src/backends/rabbitmq.rs

@@ -4,6 +4,7 @@ use std::{any::TypeId, collections::HashMap};
 use async_trait::async_trait;
 use futures::StreamExt;
 use futures_util::FutureExt;
+use lapin::types::AMQPValue;
 pub use lapin::{
     acker::Acker as LapinAcker,
     options::{
@@ -18,6 +19,7 @@ use crate::{
     decoding::DecoderRegistry,
     encoding::{CustomEncoder, EncoderRegistry},
     queue::{consumer::QueueConsumer, producer::QueueProducer, Acker, Delivery, QueueBackend},
+    scheduled::ScheduledProducer,
     QueueError,
 };
 
@@ -89,13 +91,13 @@ async fn producer(
 
 #[async_trait]
 impl QueueBackend for RabbitMqBackend {
-    type Config = RabbitMqConfig;
-
     type PayloadIn = Vec<u8>;
+
     type PayloadOut = Vec<u8>;
+    type Producer = RabbitMqProducer;
 
     type Consumer = RabbitMqConsumer;
-    type Producer = RabbitMqProducer;
+    type Config = RabbitMqConfig;
 
     async fn new_pair(
         cfg: RabbitMqConfig,
@@ -168,6 +170,36 @@ impl QueueProducer for RabbitMqProducer {
     }
 }
 
+#[async_trait]
+impl ScheduledProducer for RabbitMqProducer {
+    async fn send_raw_scheduled(
+        &self,
+        payload: &Self::Payload,
+        delay: Duration,
+    ) -> Result<(), QueueError> {
+        let mut headers = FieldTable::default();
+
+        let delay_ms: u32 = delay
+            .as_millis()
+            .try_into()
+            .map_err(|_| QueueError::Generic("delay is too large".into()))?;
+        headers.insert("x-delay".into(), AMQPValue::LongUInt(delay_ms));
+
+        self.channel
+            .basic_publish(
+                &self.exchange,
+                &self.routing_key,
+                self.options,
+                payload,
+                self.properties.clone().with_headers(headers),
+            )
+            .await
+            .map_err(QueueError::generic)?;
+
+        Ok(())
+    }
+}
+
 pub struct RabbitMqConsumer {
     registry: DecoderRegistry<Vec<u8>>,
     consumer: Consumer,