Revert "Update redis/bb8"

We are seeing potential issues with this version of redis
elsewhere. Let's revert for now until we can further
troubleshoot.

This reverts commit df8cd28.

omniqueue/Cargo.toml

@@ -14,12 +14,12 @@ async-trait = "0.1"
 aws-config = { version = "1.1.5", features = ["behavior-version-latest"], optional = true }
 aws-sdk-sqs = { version = "1.13.0", optional = true }
 bb8 = { version = "0.8", optional = true }
-bb8-redis = { version = "0.15.0", optional = true }
+bb8-redis = { version = "0.14.0", optional = true }
 futures-util = { version = "0.3.28", default-features = false, features = ["async-await", "std"], optional = true }
 google-cloud-googleapis = { version = "0.12.0", optional = true }
 google-cloud-pubsub = { version = "0.23.0", optional = true }
 lapin = { version = "2", optional = true }
-redis = { version = "0.25.3", features = ["tokio-comp", "tokio-native-tls-comp", "streams"], optional = true }
+redis = { version = "0.24.0", features = ["tokio-comp", "tokio-native-tls-comp", "streams"], optional = true }
 serde = "1.0.196"
 serde_json = "1"
 svix-ksuid = { version = "0.8.0", optional = true }

omniqueue/src/backends/redis/mod.rs

@@ -31,7 +31,7 @@ use std::{marker::PhantomData, sync::Arc, time::Duration};
 
 use async_trait::async_trait;
 use bb8::ManageConnection;
-pub use bb8_redis::RedisConnectionManager;
+pub use bb8_redis::RedisMultiplexedConnectionManager;
 use redis::{
     streams::{StreamClaimReply, StreamId, StreamReadOptions, StreamReadReply},
     FromRedisValue, RedisResult,
@@ -64,7 +64,7 @@ pub trait RedisConnection:
     fn from_dsn(dsn: &str) -> Result<Self>;
 }
 
-impl RedisConnection for RedisConnectionManager {
+impl RedisConnection for RedisMultiplexedConnectionManager {
     type Connection = <Self as ManageConnection>::Connection;
     type Error = <Self as ManageConnection>::Error;
 
@@ -96,7 +96,7 @@ pub struct RedisConfig {
     pub ack_deadline_ms: i64,
 }
 
-pub struct RedisBackend<R = RedisConnectionManager>(PhantomData<R>);
+pub struct RedisBackend<R = RedisMultiplexedConnectionManager>(PhantomData<R>);
 #[cfg(feature = "redis_cluster")]
 pub type RedisClusterBackend = RedisBackend<RedisClusterConnectionManager>;
 

omniqueue/tests/it/redis.rs

@@ -33,7 +33,7 @@ async fn make_test_queue() -> (QueueBuilder<RedisBackend>, RedisStreamDrop) {
         .collect();
 
     let client = Client::open(ROOT_URL).unwrap();
-    let mut conn = client.get_multiplexed_async_connection().await.unwrap();
+    let mut conn = client.get_async_connection().await.unwrap();
 
     let _: () = conn
         .xgroup_create_mkstream(&stream_name, "test_cg", 0i8)