From edd9fef468f910b24387bbd772304c214639ebad Mon Sep 17 00:00:00 2001 From: Drazen Date: Wed, 12 Jun 2024 21:35:17 +0200 Subject: [PATCH] Use Sink always --- common/client-core/src/client/base_client/mod.rs | 14 ++++---------- common/socks5/proxy-helpers/src/ordered_sender.rs | 9 ++++----- .../socks5/proxy-helpers/src/proxy_runner/mod.rs | 2 +- .../examples/parallel_sending_and_receiving.rs | 2 +- sdk/rust/nym-sdk/src/mixnet/native_client.rs | 4 ++-- sdk/rust/nym-sdk/src/mixnet/traits.rs | 8 ++++---- service-providers/network-requester/src/core.rs | 4 ++-- 7 files changed, 18 insertions(+), 25 deletions(-) diff --git a/common/client-core/src/client/base_client/mod.rs b/common/client-core/src/client/base_client/mod.rs index c50e7ec7f8..66bbdc3191 100644 --- a/common/client-core/src/client/base_client/mod.rs +++ b/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> { - 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> { + self.input_sender.send(message).await } } diff --git a/common/socks5/proxy-helpers/src/ordered_sender.rs b/common/socks5/proxy-helpers/src/ordered_sender.rs index a94cfe97dc..02c465037b 100644 --- a/common/socks5/proxy-helpers/src/ordered_sender.rs +++ b/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!") } } diff --git a/common/socks5/proxy-helpers/src/proxy_runner/mod.rs b/common/socks5/proxy-helpers/src/proxy_runner/mod.rs index 730e4b8eb0..dcf6298520 100644 --- a/common/socks5/proxy-helpers/src/proxy_runner/mod.rs +++ b/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; diff --git a/sdk/rust/nym-sdk/examples/parallel_sending_and_receiving.rs b/sdk/rust/nym-sdk/examples/parallel_sending_and_receiving.rs index ca9b843717..643aec1fb2 100644 --- a/sdk/rust/nym-sdk/examples/parallel_sending_and_receiving.rs +++ b/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 { diff --git a/sdk/rust/nym-sdk/src/mixnet/native_client.rs b/sdk/rust/nym-sdk/src/mixnet/native_client.rs index f262fe0e87..2f9777cd2f 100644 --- a/sdk/rust/nym-sdk/src/mixnet/native_client.rs +++ b/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 diff --git a/sdk/rust/nym-sdk/src/mixnet/traits.rs b/sdk/rust/nym-sdk/src/mixnet/traits.rs index 8eb78109f7..a20a7cb996 100644 --- a/sdk/rust/nym-sdk/src/mixnet/traits.rs +++ b/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(&self, address: Recipient, message: M) -> Result<()> + async fn send_plain_message(&mut self, address: Recipient, message: M) -> Result<()> where M: AsRef<[u8]> + Send, { @@ -61,7 +61,7 @@ pub trait MixnetMessageSender { /// } /// ``` async fn send_message( - &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(&self, recipient_tag: AnonymousSenderTag, message: M) -> Result<()> + async fn send_reply(&mut self, recipient_tag: AnonymousSenderTag, message: M) -> Result<()> where M: AsRef<[u8]> + Send, { diff --git a/service-providers/network-requester/src/core.rs b/service-providers/network-requester/src/core.rs index ff05e08a49..f4a349efeb 100644 --- a/service-providers/network-requester/src/core.rs +++ b/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, stats_collector: Option, packet_type: PacketType,