server: Upgrade omniqueue

svix · Feb 29, 2024 · 09bdb29 · 09bdb29
1 parent 1d1698a
commit 09bdb29
Show file tree

Hide file tree

Showing 4 changed files with 15 additions and 23 deletions.
diff --git a/server/Cargo.lock b/server/Cargo.lock
diff --git a/server/svix-server/Cargo.toml b/server/svix-server/Cargo.toml
@@ -72,7 +72,7 @@ urlencoding = "2.1.2"
 form_urlencoded = "1.1.0"
 lapin = "2.1.1"
 sentry = { version = "0.32.2", features = ["tracing"] }
-omniqueue = { git = "https://github.com/svix/omniqueue-rs.git", rev = "648f43f0778d02115df2d91b1a961519a62f42a7", default-features = false, features = ["memory_queue", "rabbitmq"] }
+omniqueue = { git = "https://github.com/svix/omniqueue-rs.git", rev = "044fc688670759859f193bb646fac043bbfd08ba", default-features = false, features = ["in_memory", "rabbitmq"] }
 
 [target.'cfg(not(target_env = "msvc"))'.dependencies]
 tikv-jemallocator = { version = "0.5", optional = true }

diff --git a/server/svix-server/src/queue/mod.rs b/server/svix-server/src/queue/mod.rs
@@ -3,13 +3,7 @@ use std::{sync::Arc, time::Duration};
 use axum::async_trait;
 use chrono::{DateTime, Utc};
 use omniqueue::{
-    backends::memory_queue::MemoryQueueBackend,
-    queue::{
-        consumer::{DynConsumer, QueueConsumer},
-        producer::QueueProducer,
-        Delivery, QueueBackend as _,
-    },
-    scheduled::ScheduledProducer,
+    backends::InMemoryBackend, Delivery, DynConsumer, QueueConsumer, ScheduledQueueProducer,
 };
 use serde::{Deserialize, Serialize};
 use svix_ksuid::*;
@@ -53,14 +47,14 @@ pub async fn new_pair(
             redis::new_pair(pool, prefix).await
         }
         QueueBackend::Memory => {
-            let (producer, consumer) = MemoryQueueBackend::builder(())
+            let (producer, consumer) = InMemoryBackend::builder()
                 .build_pair()
                 .await
                 .expect("building in-memory queue can't fail");
 
             (
-                TaskQueueProducer::Omni(Arc::new(producer.into_dyn_scheduled(Default::default()))),
-                TaskQueueConsumer::Omni(consumer.into_dyn(Default::default())),
+                TaskQueueProducer::Omni(Arc::new(producer.into_dyn_scheduled())),
+                TaskQueueConsumer::Omni(consumer.into_dyn()),
             )
         }
         QueueBackend::RabbitMq(dsn) => {
@@ -150,7 +144,7 @@ impl QueueTask {
 #[derive(Clone)]
 pub enum TaskQueueProducer {
     Redis(RedisQueueProducer),
-    Omni(Arc<omniqueue::scheduled::DynScheduledProducer>),
+    Omni(Arc<omniqueue::DynScheduledQueueProducer>),
 }
 
 impl TaskQueueProducer {

diff --git a/server/svix-server/src/queue/rabbitmq.rs b/server/svix-server/src/queue/rabbitmq.rs
@@ -4,9 +4,8 @@ use lapin::{
     ConnectionProperties,
 };
 use omniqueue::{
-    backends::rabbitmq::{RabbitMqBackend, RabbitMqConfig},
-    queue::{consumer::QueueConsumer, QueueBackend},
-    scheduled::ScheduledProducer,
+    backends::{RabbitMqBackend, RabbitMqConfig},
+    QueueConsumer, ScheduledQueueProducer,
 };
 use svix_ksuid::{KsuidLike, KsuidMs};
 
@@ -86,9 +85,8 @@ pub async fn new_pair(
     .await
     .expect("Error initializing rabbitmq queue");
 
-    let producer =
-        TaskQueueProducer::Omni(Arc::new(producer.into_dyn_scheduled(Default::default())));
-    let consumer = TaskQueueConsumer::Omni(consumer.into_dyn(Default::default()));
+    let producer = TaskQueueProducer::Omni(Arc::new(producer.into_dyn_scheduled()));
+    let consumer = TaskQueueConsumer::Omni(consumer.into_dyn());
 
     Ok((producer, consumer))
 }