Use redis ConnectionManager

Cargo.toml

@@ -10,7 +10,7 @@ async-trait = "0.1.80"
 tokio = { version = "1.37.0", features = ["macros", "rt-multi-thread"] }
 log = "0.4.21"
 simple_logger = "5.0.0"
-redis = { version = "0.25.3", features = [ "tokio-rustls-comp" ] }
+redis = { version = "0.25.3", features = [ "tokio-rustls-comp", "connection-manager" ] }
 serde = { version = "1.0.199", features = [ "derive" ] }
 serde_json = "1.0.116"
 dotenv = "0.15.0"

src/main.rs

@@ -18,6 +18,7 @@ use inindexer::{
     run_indexer, AutoContinue, BlockIterator, CompleteTransaction, Indexer, IndexerOptions,
     PreprocessTransactionsSettings, TransactionReceipt,
 };
+use redis::aio::ConnectionManager;
 use redis_handler::PushToRedisStream;
 use serde::{Deserialize, Serialize};
 
@@ -191,7 +192,7 @@ async fn main() {
         std::env::var("REDIS_URL").expect("No $REDIS_URL environment variable set"),
     )
     .unwrap();
-    let connection = client.get_multiplexed_tokio_connection().await.unwrap();
+    let connection = ConnectionManager::new(client).await.unwrap();
 
     let mut indexer = NftIndexer(PushToRedisStream::new(connection, 10_000));
 

src/redis_handler.rs

@@ -1,16 +1,16 @@
 use async_trait::async_trait;
 use inindexer::near_utils::{NftBurnEvent, NftMintEvent};
-use redis::{aio::MultiplexedConnection, streams::StreamMaxlen, AsyncCommands};
+use redis::{streams::StreamMaxlen, AsyncCommands};
 
 use crate::{EventContext, ExtendedNftTransferEvent, NftEventHandler};
 
-pub struct PushToRedisStream {
-    connection: MultiplexedConnection,
+pub struct PushToRedisStream<C: AsyncCommands + Sync> {
+    connection: C,
     max_stream_size: usize,
 }
 
-impl PushToRedisStream {
-    pub fn new(connection: MultiplexedConnection, max_stream_size: usize) -> Self {
+impl<C: AsyncCommands + Sync> PushToRedisStream<C> {
+    pub fn new(connection: C, max_stream_size: usize) -> Self {
         Self {
             connection,
             max_stream_size,
@@ -19,7 +19,7 @@ impl PushToRedisStream {
 }
 
 #[async_trait]
-impl NftEventHandler for PushToRedisStream {
+impl<C: AsyncCommands + Sync> NftEventHandler for PushToRedisStream<C> {
     async fn handle_mint(&mut self, mint: NftMintEvent, context: EventContext) {
         let response: String = self
             .connection