remove async channel

blockworks-foundation · Aug 12, 2024 · b83f81d · b83f81d
1 parent b36582f
commit b83f81d
Show file tree

Hide file tree

Showing 3 changed files with 4 additions and 5 deletions.
diff --git a/chaindata_standalone/Cargo.toml b/chaindata_standalone/Cargo.toml
@@ -26,7 +26,6 @@ base64 = "0.21.7"
 
 
 bincode = "1.3.3"
-async-channel = { workspace = true }
 async-trait = { workspace = true }
 
 mango-feeds-connector = { path = "../connector" }

diff --git a/chaindata_standalone/src/main.rs b/chaindata_standalone/src/main.rs
@@ -61,7 +61,7 @@ pub async fn main() {
 
 
     let (account_write_sender, account_write_receiver) = mpsc::channel::<AccountOrSnapshotUpdate>(100_000);
-    let (slot_sender, slot_receiver) = async_channel::unbounded::<SlotUpdate>();
+    let (slot_sender, slot_receiver) = mpsc::channel::<SlotUpdate>(10_000);
     let (account_update_sender, _) = broadcast::channel(524288); // 524288
 
     start_plumbing_task(grpc_accounts_rx, account_write_sender.clone(), slot_sender.clone());
@@ -178,7 +178,7 @@ fn debug_chaindata(chain_data: Arc<RwLock<ChainData>>, mut exit: broadcast::Rece
 fn start_plumbing_task(
     mut grpc_source_rx: mpsc::Receiver<Message>,
     account_write_sender: mpsc::Sender<AccountOrSnapshotUpdate>,
-    slot_sender: async_channel::Sender<SlotUpdate>) {
+    slot_sender: mpsc::Sender<SlotUpdate>) {
     tokio::spawn(async move {
         info!("starting plumbing task");
         loop {

diff --git a/chaindata_standalone/src/router_impl.rs b/chaindata_standalone/src/router_impl.rs
@@ -26,7 +26,7 @@ pub fn start_chaindata_updating(
     chain_data: ChainDataArcRw,
     // = account_write_receiver
     mut account_writes: mpsc::Receiver<AccountOrSnapshotUpdate>,
-    slot_updates: async_channel::Receiver<SlotUpdate>,
+    mut slot_updates: mpsc::Receiver<SlotUpdate>,
     account_update_sender: broadcast::Sender<(Pubkey, u64)>,
     mut exit: broadcast::Receiver<()>,
 ) -> JoinHandle<()> {
@@ -74,7 +74,7 @@ pub fn start_chaindata_updating(
                     }
                 }
                 res = slot_updates.recv() => {
-                    let Ok(slot_update) = res
+                    let Some(slot_update) = res
                     else {
                         warn!("slot channel err {res:?}");
                         continue;