Use Sink always

common/client-core/src/client/base_client/mod.rs

@@ -69,7 +69,7 @@ use std::path::Path;
 use std::sync::Arc;
 use time::OffsetDateTime;
 use tokio::sync::mpsc::Sender;
-use tokio_util::sync::PollSender;
+use tokio_util::sync::{PollSendError, PollSender};
 use tracing::*;
 use url::Url;
 
@@ -116,15 +116,8 @@ pub struct ClientInput {
 }
 
 impl ClientInput {
-    pub async fn send(
-        &self,
-        message: InputMessage,
-    ) -> Result<(), tokio::sync::mpsc::error::SendError<InputMessage>> {
-        if let Some(channel) = self.input_sender.get_ref() {
-            channel.send(message).await
-        } else {
-            Err(tokio::sync::mpsc::error::SendError(message))
-        }
+    pub async fn send(&mut self, message: InputMessage) -> Result<(), PollSendError<InputMessage>> {
+        self.input_sender.send(message).await
     }
 }
 

common/socks5/proxy-helpers/src/ordered_sender.rs

@@ -3,6 +3,7 @@
 
 use crate::proxy_runner::MixProxySender;
 use bytes::Bytes;
+use futures::SinkExt;
 use log::{debug, error};
 use nym_socks5_requests::{ConnectionId, SocketData};
 use std::io;
@@ -55,11 +56,9 @@ where
         (self.mix_message_adapter)(data)
     }
 
-    async fn send_message(&self, message: S) {
-        if let Some(sender) = self.mixnet_sender.get_ref() {
-            if sender.send(message).await.is_err() {
-                panic!("BatchRealMessageReceiver has stopped receiving!")
-            }
+    async fn send_message(&mut self, message: S) {
+        if self.mixnet_sender.send(message).await.is_err() {
+            panic!("BatchRealMessageReceiver has stopped receiving!")
         }
     }
 

common/socks5/proxy-helpers/src/proxy_runner/mod.rs

@@ -5,11 +5,11 @@ use crate::connection_controller::ConnectionReceiver;
 use crate::ordered_sender::OrderedMessageSender;
 use nym_socks5_requests::{ConnectionId, SocketData};
 use nym_task::connections::LaneQueueLengths;
-use nym_task::ShutdownTracker;
-use tokio_util::sync::PollSender;
+<<<<<<< HEAD
 use std::fmt::Debug;
 use std::{sync::Arc, time::Duration};
 use tokio::{net::TcpStream, sync::Notify};
+use tokio_util::sync::PollSender;
 
 mod inbound;
 mod outbound;

sdk/rust/nym-sdk/examples/parallel_sending_and_receiving.rs

@@ -16,7 +16,7 @@ async fn main() {
     let our_address = *client.nym_address();
     println!("Our client nym address is: {our_address}");
 
-    let sender = client.split_sender();
+    let mut sender = client.split_sender();
 
     // receiving task
     let receiving_task_handle = tokio::spawn(async move {

sdk/rust/nym-sdk/src/mixnet/native_client.rs

@@ -453,7 +453,7 @@ impl MixnetMessageSender for MixnetClient {
         self.packet_type
     }
 
-    async fn send(&self, message: InputMessage) -> Result<()> {
+    async fn send(&mut self, message: InputMessage) -> Result<()> {
         if self.stream_mode.load(Ordering::SeqCst) {
             tracing::warn!("send() called after stream mode activated");
             return Err(Error::StreamModeActive);
@@ -471,7 +471,7 @@ impl MixnetMessageSender for MixnetClientSender {
         self.packet_type
     }
 
-    async fn send(&self, message: InputMessage) -> Result<()> {
+    async fn send(&mut self, message: InputMessage) -> Result<()> {
         if self.stream_mode.load(Ordering::SeqCst) {
             tracing::warn!("send() called after stream mode activated");
             return Err(Error::StreamModeActive);

sdk/rust/nym-sdk/src/mixnet/traits.rs

@@ -18,7 +18,7 @@ pub trait MixnetMessageSender {
 
     /// Sends a [`InputMessage`] to the mixnet. This is the most low-level sending function, for
     /// full customization.
-    async fn send(&self, message: InputMessage) -> Result<()>;
+    async fn send(&mut self, message: InputMessage) -> Result<()>;
 
     /// Sends data to the supplied Nym address with the default surb behaviour.
     ///
@@ -35,7 +35,7 @@ pub trait MixnetMessageSender {
     ///     client.send_plain_message(recipient, "hi").await.unwrap();
     /// }
     /// ```
-    async fn send_plain_message<M>(&self, address: Recipient, message: M) -> Result<()>
+    async fn send_plain_message<M>(&mut self, address: Recipient, message: M) -> Result<()>
     where
         M: AsRef<[u8]> + Send,
     {
@@ -61,7 +61,7 @@ pub trait MixnetMessageSender {
     /// }
     /// ```
     async fn send_message<M>(
-        &self,
+        &mut self,
         address: Recipient,
         message: M,
         surbs: IncludedSurbs,
@@ -103,7 +103,7 @@ pub trait MixnetMessageSender {
     ///     client.send_reply(tag, b"hi").await.unwrap();
     /// }
     /// ```
-    async fn send_reply<M>(&self, recipient_tag: AnonymousSenderTag, message: M) -> Result<()>
+    async fn send_reply<M>(&mut self, recipient_tag: AnonymousSenderTag, message: M) -> Result<()>
     where
         M: AsRef<[u8]> + Send,
     {

service-providers/network-requester/src/core.rs

@@ -353,7 +353,7 @@ impl NRServiceProvider {
     /// Listens for any messages from `mix_reader` that should be written back to the mix network
     /// via the `websocket_writer`.
     async fn mixnet_response_listener(
-        mixnet_client_sender: nym_sdk::mixnet::MixnetClientSender,
+        mut mixnet_client_sender: nym_sdk::mixnet::MixnetClientSender,
         mut mix_input_reader: MixProxyReader<MixnetMessage>,
         packet_type: PacketType,
     ) {