Remove async-channel

protocols/v2/framing-sv2/src/framing.rs

@@ -31,7 +31,7 @@ type Slice = buffer_sv2::Slice;
 /// A wrapper used when generic reference to a frame is needed, but the kind of frame ([`Sv2Frame`]
 /// or [`HandShakeFrame`]) does not matter. Note that after the initial handshake is complete
 /// between two Sv2 roles, all further messages are framed with [`Sv2Frame`].
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub enum Frame<T, B> {
     HandShake(HandShakeFrame),
     Sv2(Sv2Frame<T, B>),
@@ -239,7 +239,7 @@ impl<T, B> TryFrom<Frame<T, B>> for Sv2Frame<T, B> {
 /// Contains only the serialized payload with a fixed length and is only used during Noise
 /// handshake process. Once the handshake is complete, regular Sv2 communication switches to
 /// [`Sv2Frame`] for ongoing communication.
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct HandShakeFrame {
     payload: Slice,
 }

roles/jd-client/Cargo.toml

@@ -17,7 +17,6 @@ path = "src/lib/mod.rs"
 
 [dependencies]
 stratum-common = { path = "../../common" }
-async-channel = "1.5.1"
 async-recursion = "0.3.2"
 binary_sv2 = { path = "../../protocols/v2/binary-sv2/binary-sv2" }
 buffer_sv2 = { path = "../../utils/buffer" }

roles/jd-client/src/lib/downstream.rs

@@ -3,7 +3,7 @@ use super::{
     status::{self, State},
     upstream_sv2::Upstream as UpstreamMiningNode,
 };
-use async_channel::{Receiver, SendError, Sender};
+
 use roles_logic_sv2::{
     channel_logic::channel_factory::{OnNewShare, PoolChannelFactory, Share},
     common_messages_sv2::{SetupConnection, SetupConnectionSuccess},
@@ -36,12 +36,12 @@ pub type EitherFrame = StandardEitherFrame<Message>;
 /// downstream do not make much sense.
 #[derive(Debug)]
 pub struct DownstreamMiningNode {
-    receiver: Receiver<EitherFrame>,
-    sender: Sender<EitherFrame>,
+    receiver: tokio::sync::broadcast::Sender<EitherFrame>,
+    sender: tokio::sync::broadcast::Sender<EitherFrame>,
     pub status: DownstreamMiningNodeStatus,
     #[allow(dead_code)]
     pub prev_job_id: Option<u32>,
-    solution_sender: Sender<SubmitSolution<'static>>,
+    solution_sender: tokio::sync::mpsc::Sender<SubmitSolution<'static>>,
     withhold: bool,
     task_collector: Arc<Mutex<Vec<AbortHandle>>>,
     tx_status: status::Sender,
@@ -154,10 +154,10 @@ use std::sync::Arc;
 impl DownstreamMiningNode {
     #[allow(clippy::too_many_arguments)]
     pub fn new(
-        receiver: Receiver<EitherFrame>,
-        sender: Sender<EitherFrame>,
+        receiver: tokio::sync::broadcast::Sender<EitherFrame>,
+        sender: tokio::sync::broadcast::Sender<EitherFrame>,
         upstream: Option<Arc<Mutex<UpstreamMiningNode>>>,
-        solution_sender: Sender<SubmitSolution<'static>>,
+        solution_sender: tokio::sync::mpsc::Sender<SubmitSolution<'static>>,
         withhold: bool,
         task_collector: Arc<Mutex<Vec<AbortHandle>>>,
         tx_status: status::Sender,
@@ -207,7 +207,7 @@ impl DownstreamMiningNode {
                 .unwrap();
             Self::set_channel_factory(self_mutex.clone());
 
-            while let Ok(message) = receiver.recv().await {
+            while let Ok(message) = receiver.subscribe().recv().await {
                 let incoming: StdFrame = message.try_into().unwrap();
                 Self::next(self_mutex, incoming).await;
             }
@@ -341,10 +341,11 @@ impl DownstreamMiningNode {
     pub async fn send(
         self_mutex: &Arc<Mutex<Self>>,
         sv2_frame: StdFrame,
-    ) -> Result<(), SendError<StdFrame>> {
+        // Error needs to be taken care
+    ) -> Result<(), ()> {
         let either_frame = sv2_frame.into();
         let sender = self_mutex.safe_lock(|self_| self_.sender.clone()).unwrap();
-        match sender.send(either_frame).await {
+        match sender.send(either_frame) {
             Ok(_) => Ok(()),
             Err(_) => {
                 todo!()
@@ -592,7 +593,7 @@ impl
                             coinbase_tx: coinbase.try_into()?,
                         };
                         // The below channel should never be full is ok to block
-                        solution_sender.send_blocking(solution).unwrap();
+                        solution_sender.blocking_send(solution).unwrap();
                         if !self.status.is_solo_miner() {
                             {
                                 let jd = self.jd.clone();
@@ -671,7 +672,7 @@ use tokio::{
 pub async fn listen_for_downstream_mining(
     address: SocketAddr,
     upstream: Option<Arc<Mutex<UpstreamMiningNode>>>,
-    solution_sender: Sender<SubmitSolution<'static>>,
+    solution_sender: tokio::sync::mpsc::Sender<SubmitSolution<'static>>,
     withhold: bool,
     authority_public_key: Secp256k1PublicKey,
     authority_secret_key: Secp256k1SecretKey,
@@ -707,7 +708,14 @@ pub async fn listen_for_downstream_mining(
             jd,
         );
 
-        let mut incoming: StdFrame = node.receiver.recv().await.unwrap().try_into().unwrap();
+        let mut incoming: StdFrame = node
+            .receiver
+            .subscribe()
+            .recv()
+            .await
+            .unwrap()
+            .try_into()
+            .unwrap();
         let message_type = incoming.get_header().unwrap().msg_type();
         let payload = incoming.payload();
         let routing_logic = roles_logic_sv2::routing_logic::CommonRoutingLogic::None;