Use Sink always

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

@@ -35,6 +35,7 @@ use crate::init::{
 };
 use crate::{config, spawn_future};
 use futures::channel::mpsc;
+use futures::SinkExt;
 use log::{debug, error, info, warn};
 use nym_bandwidth_controller::BandwidthController;
 use nym_client_core_gateways_storage::{GatewayDetails, GatewaysDetailsStore};
@@ -59,7 +60,7 @@ use std::fmt::Debug;
 use std::os::raw::c_int as RawFd;
 use std::path::Path;
 use std::sync::Arc;
-use tokio_util::sync::PollSender;
+use tokio_util::sync::{PollSendError, PollSender};
 use url::Url;
 
 #[cfg(all(
@@ -79,15 +80,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

@@ -6,10 +6,10 @@ use crate::ordered_sender::OrderedMessageSender;
 use nym_socks5_requests::{ConnectionId, SocketData};
 use nym_task::connections::LaneQueueLengths;
 use nym_task::TaskClient;
-use tokio_util::sync::PollSender;
 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

@@ -294,7 +294,7 @@ impl MixnetMessageSender for MixnetClient {
         self.packet_type
     }
 
-    async fn send(&self, message: InputMessage) -> Result<()> {
+    async fn send(&mut self, message: InputMessage) -> Result<()> {
         self.client_input
             .send(message)
             .await
@@ -308,7 +308,7 @@ impl MixnetMessageSender for MixnetClientSender {
         self.packet_type
     }
 
-    async fn send(&self, message: InputMessage) -> Result<()> {
+    async fn send(&mut self, message: InputMessage) -> Result<()> {
         self.client_input
             .send(message)
             .await

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

@@ -39,9 +39,9 @@ use nym_statistics_common::collector::StatisticsSender;
 use nym_task::connections::LaneQueueLengths;
 use nym_task::manager::TaskHandle;
 use nym_task::TaskClient;
-use tokio_util::sync::PollSender;
 use std::path::Path;
 use std::sync::atomic::{AtomicUsize, Ordering};
+use tokio_util::sync::PollSender;
 
 // Since it's an atomic, it's safe to be kept static and shared across threads
 static ACTIVE_PROXIES: AtomicUsize = AtomicUsize::new(0);
@@ -404,7 +404,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>,
         stats_collector: Option<ServiceStatisticsCollector>,
         packet_type: PacketType,