use tokio unbounded channel

holaplex · Oct 4, 2023 · 6f7d4c2 · 6f7d4c2
1 parent 092b385
commit 6f7d4c2
Show file tree

Hide file tree

Showing 3 changed files with 20 additions and 36 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/indexer/Cargo.toml b/indexer/Cargo.toml
@@ -29,7 +29,6 @@ dashmap = "5.4.0"
 spl-token = "=3.5.0"
 solana-client = "1.14"
 backoff = { version = "0.4.0", features = ["tokio"] }
-crossbeam = "0.8.2"
 
 [dependencies.hub-core]
 package = "holaplex-hub-core"

diff --git a/indexer/src/handler.rs b/indexer/src/handler.rs
@@ -1,6 +1,5 @@
 use std::sync::Arc;
 
-use crossbeam::channel::{self, Receiver, Sender};
 use dashmap::DashMap;
 use futures::{sink::SinkExt, stream::StreamExt};
 use holaplex_hub_nfts_solana_core::{db::Connection, proto::SolanaNftEvents};
@@ -10,6 +9,10 @@ use hub_core::{
     producer::Producer,
     tokio::{
         self,
+        sync::{
+            mpsc::{self, UnboundedReceiver, UnboundedSender},
+            Mutex,
+        },
         task::{self, JoinSet},
     },
 };
@@ -25,9 +28,8 @@ use crate::{processor::Processor, Args, GeyserGrpcConnector};
 pub struct MessageHandler {
     connector: GeyserGrpcConnector,
     processor: Processor,
-    dashmap: DashMap<u64, Vec<SubscribeUpdateTransaction>>,
-    tx: Sender<SubscribeUpdateTransaction>,
-    rx: Receiver<SubscribeUpdateTransaction>,
+    tx: UnboundedSender<SubscribeUpdateTransaction>,
+    rx: Arc<Mutex<UnboundedReceiver<SubscribeUpdateTransaction>>>,
     parallelism: usize,
 }
 
@@ -47,23 +49,22 @@ impl MessageHandler {
 
         let rpc = Arc::new(RpcClient::new(solana_endpoint));
         let connector = GeyserGrpcConnector::new(dragon_mouth_endpoint, dragon_mouth_x_token);
-        let (tx, rx) = channel::unbounded();
+        let (tx, rx) = mpsc::unbounded_channel();
         let processor = Processor::new(db, rpc, producer);
 
         Ok(Self {
             connector,
             processor,
-            dashmap: DashMap::new(),
             tx,
-            rx,
+            rx: Arc::new(Mutex::new(rx)),
             parallelism,
         })
     }
 
     async fn connect(&self, request: SubscribeRequest) -> Result<()> {
         (|| async {
             let (mut subscribe_tx, mut stream) = self.connector.subscribe().await?;
-
+            let dashmap = DashMap::new();
             subscribe_tx
                 .send(request.clone())
                 .await
@@ -73,10 +74,10 @@ impl MessageHandler {
                 match message {
                     Ok(msg) => match msg.update_oneof {
                         Some(UpdateOneof::Transaction(tx)) => {
-                            self.dashmap.entry(tx.slot).or_insert(Vec::new()).push(tx);
+                            dashmap.entry(tx.slot).or_insert(Vec::new()).push(tx);
                         },
                         Some(UpdateOneof::Slot(slot)) => {
-                            if let Some((_, transactions)) = self.dashmap.remove(&slot.slot) {
+                            if let Some((_, transactions)) = dashmap.remove(&slot.slot) {
                                 for tx in transactions {
                                     self.tx.send(tx)?;
                                 }
@@ -119,30 +120,29 @@ impl MessageHandler {
             }
         });
 
-        let rx = self.rx.clone();
         let processor = self.processor;
 
         let process_task = task::spawn(async move {
             let mut set = JoinSet::new();
 
             loop {
                 let processor = processor.clone();
+                let mut rx = self.rx.lock().await;
 
                 while set.len() >= self.parallelism {
                     match set.join_next().await {
-                        Some(Err(e)) => bail!("failed to join task {:?}", e),
+                        Some(Err(e)) => {
+                            return Result::<(), Error>::Err(anyhow!(
+                                "failed to join task {:?}",
+                                e
+                            ));
+                        },
                         Some(Ok(_)) | None => (),
                     }
                 }
 
-                match rx.recv() {
-                    Ok(tx) => {
-                        set.spawn(processor.process_transaction(tx));
-                    },
-                    Err(e) => {
-                        error!("{:?}", e);
-                        return Result::<(), Error>::Err(e.into());
-                    },
+                if let Some(tx) = rx.recv().await {
+                    set.spawn(processor.process_transaction(tx));
                 }
             }
         });