remove all occurance of async-channel

Shourya742 · Shourya742 · commit f65203497427 · 2025-01-29T13:15:39.000+05:30
diff --git a/roles/Cargo.lock b/roles/Cargo.lock
diff --git a/roles/pool/Cargo.toml b/roles/pool/Cargo.toml
@@ -18,7 +18,6 @@ path = "src/lib/mod.rs"
 
 [dependencies]
 stratum-common = { path = "../../common" }
-async-channel = "1.5.1"
 binary_sv2 = { path = "../../protocols/v2/binary-sv2/binary-sv2" }
 buffer_sv2 = { path = "../../utils/buffer" }
 codec_sv2 = { path = "../../protocols/v2/codec-sv2", features = ["noise_sv2"] }
diff --git a/roles/pool/src/lib/error.rs b/roles/pool/src/lib/error.rs
@@ -10,7 +10,6 @@ use roles_logic_sv2::parsers::Mining;
 pub enum PoolError {
     Io(std::io::Error),
     ChannelSend(Box<dyn std::marker::Send + Debug>),
-    ChannelRecv(async_channel::RecvError),
     BinarySv2(binary_sv2::Error),
     Codec(codec_sv2::Error),
     Noise(noise_sv2::Error),
@@ -21,7 +20,7 @@ pub enum PoolError {
     Custom(String),
     Sv2ProtocolError((u32, Mining<'static>)),
     TokioChannelRecv(Box<dyn std::marker::Send + Debug>),
-    TokioBroadcastChannelRecv(tokio::sync::broadcast::error::RecvError)
+    TokioBroadcastChannelRecv(tokio::sync::broadcast::error::RecvError),
 }
 
 impl std::fmt::Display for PoolError {
@@ -30,7 +29,6 @@ impl std::fmt::Display for PoolError {
         match self {
             Io(ref e) => write!(f, "I/O error: `{:?}", e),
             ChannelSend(ref e) => write!(f, "Channel send failed: `{:?}`", e),
-            ChannelRecv(ref e) => write!(f, "Channel recv failed: `{:?}`", e),
             BinarySv2(ref e) => write!(f, "Binary SV2 error: `{:?}`", e),
             Codec(ref e) => write!(f, "Codec SV2 error: `{:?}", e),
             Framing(ref e) => write!(f, "Framing SV2 error: `{:?}`", e),
@@ -41,9 +39,9 @@ impl std::fmt::Display for PoolError {
             Custom(ref e) => write!(f, "Custom SV2 error: `{:?}`", e),
             Sv2ProtocolError(ref e) => {
                 write!(f, "Received Sv2 Protocol Error from upstream: `{:?}`", e)
-            },
+            }
             TokioChannelRecv(ref e) => write!(f, "Channel recv failed: `{:?}`", e),
-            TokioBroadcastChannelRecv(ref e) => write!(f, "BroadCastChannel Recv failed: {:?}", e)
+            TokioBroadcastChannelRecv(ref e) => write!(f, "BroadCastChannel Recv failed: {:?}", e),
         }
     }
 }
@@ -62,12 +60,6 @@ impl From<std::io::Error> for PoolError {
     }
 }
 
-impl From<async_channel::RecvError> for PoolError {
-    fn from(e: async_channel::RecvError) -> PoolError {
-        PoolError::ChannelRecv(e)
-    }
-}
-
 impl From<binary_sv2::Error> for PoolError {
     fn from(e: binary_sv2::Error) -> PoolError {
         PoolError::BinarySv2(e)
@@ -92,19 +84,17 @@ impl From<roles_logic_sv2::Error> for PoolError {
     }
 }
 
-impl<'a, T: 'static + std::marker::Send + Debug> From<async_channel::SendError<T>> for PoolError {
-    fn from(e: async_channel::SendError<T>) -> PoolError {
-        PoolError::ChannelSend(Box::new(e))
-    }
-}
-
-impl<'a, T: 'static + std::marker::Send + Debug> From<tokio::sync::mpsc::error::SendError<T>> for PoolError {
+impl<T: 'static + std::marker::Send + Debug> From<tokio::sync::mpsc::error::SendError<T>>
+    for PoolError
+{
     fn from(e: tokio::sync::mpsc::error::SendError<T>) -> PoolError {
         PoolError::TokioChannelRecv(Box::new(e))
     }
 }
 
-impl<'a, T: 'static + std::marker::Send + Debug> From<tokio::sync::broadcast::error::SendError<T>> for PoolError {
+impl<T: 'static + std::marker::Send + Debug> From<tokio::sync::broadcast::error::SendError<T>>
+    for PoolError
+{
     fn from(e: tokio::sync::broadcast::error::SendError<T>) -> PoolError {
         PoolError::TokioChannelRecv(Box::new(e))
     }
diff --git a/roles/pool/src/lib/mining_pool/mod.rs b/roles/pool/src/lib/mining_pool/mod.rs
@@ -744,8 +744,8 @@ impl Pool {
 
 //     use super::Configuration;
 
-//     // this test is used to verify the `coinbase_tx_prefix` and `coinbase_tx_suffix` values tested
-//     // against in message generator
+//     // this test is used to verify the `coinbase_tx_prefix` and `coinbase_tx_suffix` values
+// tested     // against in message generator
 //     // `stratum/test/message-generator/test/pool-sri-test-extended.json`
 //     #[test]
 //     fn test_coinbase_outputs_from_config() {
@@ -778,9 +778,9 @@ impl Pool {
 //         let _coinbase_tx_value_remaining: u64 = 625000000;
 //         let _coinbase_tx_outputs_count = 0;
 //         let coinbase_tx_locktime = 0;
-//         let coinbase_tx_outputs: Vec<bitcoin::TxOut> = super::get_coinbase_output(&config).unwrap();
-//         // extranonce len set to max_extranonce_size in `ChannelFactory::new_extended_channel()`
-//         let extranonce_len = 32;
+//         let coinbase_tx_outputs: Vec<bitcoin::TxOut> =
+// super::get_coinbase_output(&config).unwrap();         // extranonce len set to
+// max_extranonce_size in `ChannelFactory::new_extended_channel()`         let extranonce_len = 32;
 
 //         // build coinbase TX from 'job_creator::coinbase()'
 
@@ -813,9 +813,9 @@ impl Pool {
 //             coinbase_tx_prefix
 //                 == [
 //                     2, 0, 0, 0, 0, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
-//                     0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 255, 255, 255, 255, 56, 3, 76, 163, 38,
-//                     0, 83, 116, 114, 97, 116, 117, 109, 32, 118, 50, 32, 83, 82, 73, 32, 80, 111,
-//                     111, 108
+//                     0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 255, 255, 255, 255, 56, 3, 76, 163,
+// 38,                     0, 83, 116, 114, 97, 116, 117, 109, 32, 118, 50, 32, 83, 82, 73, 32, 80,
+// 111,                     111, 108
 //                 ]
 //                 .to_vec()
 //                 .try_into()
@@ -827,8 +827,8 @@ impl Pool {
 //                 == [
 //                     255, 255, 255, 255, 1, 0, 0, 0, 0, 0, 0, 0, 0, 22, 0, 20, 235, 225, 183, 220,
 //                     194, 147, 204, 170, 14, 231, 67, 168, 111, 137, 223, 130, 88, 194, 8, 252, 1,
-//                     32, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
-//                     0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0
+//                     32, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
+// 0,                     0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0
 //                 ]
 //                 .to_vec()
 //                 .try_into()
@@ -840,8 +840,8 @@ impl Pool {
 //     // copied from roles-logic-sv2::job_creator
 //     fn coinbase_tx_prefix(coinbase: &Transaction, script_prefix_len: usize) -> B064K<'static> {
 //         let encoded = coinbase.serialize();
-//         // If script_prefix_len is not 0 we are not in a test enviornment and the coinbase have the
-//         // 0 witness
+//         // If script_prefix_len is not 0 we are not in a test enviornment and the coinbase have
+// the         // 0 witness
 //         let segwit_bytes = match script_prefix_len {
 //             0 => 0,
 //             _ => 2,
@@ -864,8 +864,8 @@ impl Pool {
 //         script_prefix_len: usize,
 //     ) -> B064K<'static> {
 //         let encoded = coinbase.serialize();
-//         // If script_prefix_len is not 0 we are not in a test enviornment and the coinbase have the
-//         // 0 witness
+//         // If script_prefix_len is not 0 we are not in a test enviornment and the coinbase have
+// the         // 0 witness
 //         let segwit_bytes = match script_prefix_len {
 //             0 => 0,
 //             _ => 2,
@@ -877,8 +877,7 @@ impl Pool {
 //         + 4  // index
 //         + 1  // bytes in script TODO can be also 3
 //         + script_prefix_len  // bip34_bytes
-//         + (extranonce_len as usize)..]
-//             .to_vec();
+//         + (extranonce_len as usize)..] .to_vec();
 //         r.try_into().unwrap()
 //     }
 
diff --git a/roles/pool/src/lib/mining_pool/setup_connection.rs b/roles/pool/src/lib/mining_pool/setup_connection.rs
@@ -53,7 +53,7 @@ impl SetupConnectionHandler {
             }
             Ok(EitherFrame::Shutdown) => {
                 debug!("Shutdown message received");
-                return Err(PoolError::RolesLogic(Error::Shutdown))
+                return Err(PoolError::RolesLogic(Error::Shutdown));
             }
             Err(e) => {
                 error!("Error receiving message: {:?}", e);
diff --git a/roles/pool/src/lib/mod.rs b/roles/pool/src/lib/mod.rs
@@ -28,7 +28,7 @@ impl PoolSv2 {
         // producers are clonable so no issue. but its unbounded.
         // tokio also provide unbounded mpsc.
         // let (status_tx, status_rx) = unbounded();
-        let (status_tx,mut  status_rx) = tokio::sync::mpsc::unbounded_channel();
+        let (status_tx, mut status_rx) = tokio::sync::mpsc::unbounded_channel();
         // r_new_t consumer is sent in pool::start,  s_new_t is sent in templateRx::connect
         // sender or producer I dont give a damn about. even the r_new_t is passed in only
         // start then to on_new_template, so mpsc makes sense here as well.
@@ -39,7 +39,7 @@ impl PoolSv2 {
         // sent to on_new_prevhash, so mpsc also works here.
         // let (s_prev_hash, r_prev_hash) = bounded(10);
         let (s_prev_hash, r_prev_hash) = tokio::sync::mpsc::channel(10);
-        // s_solution is sent to pool (no one give a damn about clonable), r_solution is sent 
+        // s_solution is sent to pool (no one give a damn about clonable), r_solution is sent
         // to templateRx and then to on_new_solution, so mpsc works.
         let (s_solution, r_solution) = tokio::sync::mpsc::channel(10);
         // This is spicy, as the r_message_recv_signal is cloning at few of the places, so, we can
diff --git a/roles/pool/src/lib/status.rs b/roles/pool/src/lib/status.rs
@@ -7,26 +7,15 @@ use super::error::PoolError;
 /// the main thread to know which component sent the message
 #[derive(Debug)]
 pub enum Sender {
-    Downstream(async_channel::Sender<Status>),
-    DownstreamListener(async_channel::Sender<Status>),
-    Upstream(async_channel::Sender<Status>),
     DownstreamTokio(tokio::sync::mpsc::UnboundedSender<Status>),
     DownstreamListenerTokio(tokio::sync::mpsc::UnboundedSender<Status>),
     UpstreamTokio(tokio::sync::mpsc::UnboundedSender<Status>),
-    
 }
 
 #[derive(Debug)]
 pub enum Error {
-    AsyncChannel(async_channel::SendError<Status>),
     TokioChannel(tokio::sync::mpsc::error::SendError<Status>),
-    TokioChannelUnbounded(tokio::sync::mpsc::error::SendError<Status>)
-}
-
-impl From<async_channel::SendError<Status>> for Error {
-    fn from(value: async_channel::SendError<Status>) -> Self {
-        Self::AsyncChannel(value)
-    }
+    TokioChannelUnbounded(tokio::sync::mpsc::error::SendError<Status>),
 }
 
 impl From<tokio::sync::mpsc::error::SendError<Status>> for Error {
@@ -41,33 +30,26 @@ impl Sender {
     pub fn listener_to_connection(&self) -> Self {
         match self {
             // should only be used to clone the DownstreamListener(Sender) into Downstream(Sender)s
-            Self::DownstreamListener(inner) => Self::Downstream(inner.clone()),
             Self::DownstreamListenerTokio(inner) => Self::DownstreamTokio(inner.clone()),
             _ => unreachable!(),
         }
     }
 
     pub async fn send(&self, status: Status) -> Result<(), Error> {
         match self {
-            Self::Downstream(inner) => inner.send(status).await.map_err(|e| Error::AsyncChannel(e)),
-            Self::DownstreamListener(inner) => inner.send(status).await.map_err(|e| Error::AsyncChannel(e)),
-            Self::Upstream(inner) => inner.send(status).await.map_err(|e| Error::AsyncChannel(e)),
-            Self::DownstreamListenerTokio(inner) => inner.send(status).map_err(|e| Error::TokioChannel(e)),
-            Self::DownstreamTokio(inner) => inner.send(status).map_err(|e| Error::TokioChannel(e)),
-            Self::UpstreamTokio(inner) => inner.send(status).map_err(|e| Error::TokioChannel(e))
+            Self::DownstreamListenerTokio(inner) => inner.send(status).map_err(Error::TokioChannel),
+            Self::DownstreamTokio(inner) => inner.send(status).map_err(Error::TokioChannel),
+            Self::UpstreamTokio(inner) => inner.send(status).map_err(Error::TokioChannel),
         }
     }
 }
 
 impl Clone for Sender {
     fn clone(&self) -> Self {
         match self {
-            Self::Downstream(inner) => Self::Downstream(inner.clone()),
-            Self::DownstreamListener(inner) => Self::DownstreamListener(inner.clone()),
-            Self::Upstream(inner) => Self::Upstream(inner.clone()),
             Self::DownstreamTokio(inner) => Self::DownstreamTokio(inner.clone()),
             Self::DownstreamListenerTokio(inner) => Self::DownstreamListenerTokio(inner.clone()),
-            Self::UpstreamTokio(inner) => Self::UpstreamTokio(inner.clone())
+            Self::UpstreamTokio(inner) => Self::UpstreamTokio(inner.clone()),
         }
     }
 }
@@ -95,46 +77,6 @@ async fn send_status(
     outcome: error_handling::ErrorBranch,
 ) -> error_handling::ErrorBranch {
     match sender {
-        Sender::Downstream(tx) => match e {
-            PoolError::Sv2ProtocolError((id, Mining::OpenMiningChannelError(_))) => {
-                tx.send(Status {
-                    state: State::DownstreamInstanceDropped(id),
-                })
-                .await
-                .unwrap_or(());
-            }
-            _ => {
-                let string_err = e.to_string();
-                tx.send(Status {
-                    state: State::Healthy(string_err),
-                })
-                .await
-                .unwrap_or(());
-            }
-        },
-        Sender::DownstreamListener(tx) => match e {
-            PoolError::RolesLogic(roles_logic_sv2::Error::NoDownstreamsConnected) => {
-                tx.send(Status {
-                    state: State::Healthy("No Downstreams Connected".to_string()),
-                })
-                .await
-                .unwrap_or(());
-            }
-            _ => {
-                tx.send(Status {
-                    state: State::DownstreamShutdown(e),
-                })
-                .await
-                .unwrap_or(());
-            }
-        },
-        Sender::Upstream(tx) => {
-            tx.send(Status {
-                state: State::TemplateProviderShutdown(e),
-            })
-            .await
-            .unwrap_or(());
-        },
         Sender::DownstreamTokio(tx) => match e {
             PoolError::Sv2ProtocolError((id, Mining::OpenMiningChannelError(_))) => {
                 tx.send(Status {
@@ -188,9 +130,6 @@ pub async fn handle_error(sender: &Sender, e: PoolError) -> error_handling::Erro
             // the map won't get send called on them
             send_status(sender, e, error_handling::ErrorBranch::Continue).await
         }
-        PoolError::ChannelRecv(_) => {
-            send_status(sender, e, error_handling::ErrorBranch::Break).await
-        }
         PoolError::BinarySv2(_) => send_status(sender, e, error_handling::ErrorBranch::Break).await,
         PoolError::Codec(_) => send_status(sender, e, error_handling::ErrorBranch::Break).await,
         PoolError::Noise(_) => send_status(sender, e, error_handling::ErrorBranch::Continue).await,
@@ -210,10 +149,10 @@ pub async fn handle_error(sender: &Sender, e: PoolError) -> error_handling::Erro
         }
         PoolError::Sv2ProtocolError(_) => {
             send_status(sender, e, error_handling::ErrorBranch::Break).await
-        }, 
+        }
         PoolError::TokioChannelRecv(_) => {
             send_status(sender, e, error_handling::ErrorBranch::Continue).await
-        },
+        }
         PoolError::TokioBroadcastChannelRecv(_) => {
             send_status(sender, e, error_handling::ErrorBranch::Continue).await
         }
diff --git a/roles/pool/src/lib/template_receiver/mod.rs b/roles/pool/src/lib/template_receiver/mod.rs
@@ -62,7 +62,7 @@ impl TemplateRx {
             None => Initiator::without_pk(),
         }?;
         let (mut receiver, mut sender, _, _) =
-            Connection::new(stream, HandshakeRole::Initiator(initiator),10)
+            Connection::new(stream, HandshakeRole::Initiator(initiator), 10)
                 .await
                 .unwrap();
 
@@ -164,7 +164,10 @@ impl TemplateRx {
         Ok(())
     }
 
-    async fn on_new_solution(self_: Arc<Mutex<Self>>,mut rx: tokio::sync::mpsc::Receiver<SubmitSolution<'static>>) {
+    async fn on_new_solution(
+        self_: Arc<Mutex<Self>>,
+        mut rx: tokio::sync::mpsc::Receiver<SubmitSolution<'static>>,
+    ) {
         let status_tx = self_.safe_lock(|s| s.status_tx.clone()).unwrap();
         while let Some(solution) = rx.recv().await {
             info!("Sending Solution to TP: {:?}", &solution);
diff --git a/roles/pool/src/lib/template_receiver/setup_connection.rs b/roles/pool/src/lib/template_receiver/setup_connection.rs
@@ -47,7 +47,8 @@ impl SetupConnectionHandler {
         let sv2_frame = sv2_frame.into();
         sender.send(sv2_frame)?;
 
-        let mut incoming: StdFrame = receiver.subscribe()
+        let mut incoming: StdFrame = receiver
+            .subscribe()
             .recv()
             .await?
             .try_into()
diff --git a/roles/pool/src/main.rs b/roles/pool/src/main.rs
@@ -74,9 +74,7 @@ fn init_tracing() {
     let console_layer = console_subscriber::spawn();
     tracing_subscriber::registry()
         .with(console_layer)
-        .with(
-            tracing_subscriber::fmt::layer()
-        )
+        .with(tracing_subscriber::fmt::layer())
         .init();
 }
 

Original file line number	Diff line number	Diff line change
`@@ -53,7 +53,7 @@ impl SetupConnectionHandler {`
`53`	`53`	`}`
`54`	`54`	`Ok(EitherFrame::Shutdown) => {`
`55`	`55`	`debug!("Shutdown message received");`
`56`		`- return Err(PoolError::RolesLogic(Error::Shutdown))`
	`56`	`+ return Err(PoolError::RolesLogic(Error::Shutdown));`
`57`	`57`	`}`
`58`	`58`	`Err(e) => {`
`59`	`59`	`error!("Error receiving message: {:?}", e);`