Switch to PollSender

common/client-core/Cargo.toml

@@ -29,6 +29,7 @@ sha2 = { workspace = true }
 si-scale = { workspace = true }
 thiserror = { workspace = true }
 url = { workspace = true, features = ["serde"] }
+tokio-util = { workspace = true }
 time = { workspace = true }
 tokio = { workspace = true, features = ["sync", "macros"] }
 tracing = { workspace = true }

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

@@ -67,6 +67,8 @@ use std::path::Path;
 use std::sync::Arc;
 use time::OffsetDateTime;
 use tokio::sync::mpsc::Sender;
+use tracing::*;
+use tokio_util::sync::PollSender;
 use url::Url;
 
 #[cfg(target_arch = "wasm32")]
@@ -116,7 +118,11 @@ impl ClientInput {
         &self,
         message: InputMessage,
     ) -> Result<(), tokio::sync::mpsc::error::SendError<InputMessage>> {
-        self.input_sender.send(message).await
+        if let Some(channel) = self.input_sender.get_ref() {
+            channel.send(message).await
+        } else {
+            Err(tokio::sync::mpsc::error::SendError(message))
+        }
     }
 }
 
@@ -1174,7 +1180,7 @@ where
             client_input: ClientInputStatus::AwaitingProducer {
                 client_input: ClientInput {
                     connection_command_sender: client_connection_tx,
-                    input_sender,
+                    input_sender: PollSender::new(input_sender),
                     client_request_sender,
                 },
             },

common/client-core/src/client/inbound_messages.rs

@@ -7,7 +7,7 @@ use nym_sphinx::forwarding::packet::MixPacket;
 use nym_sphinx::params::PacketType;
 use nym_task::connections::TransmissionLane;
 
-pub type InputMessageSender = tokio::sync::mpsc::Sender<InputMessage>;
+pub type InputMessageSender = tokio_util::sync::PollSender<InputMessage>;
 pub type InputMessageReceiver = tokio::sync::mpsc::Receiver<InputMessage>;
 
 #[derive(Debug)]

common/socks5-client-core/Cargo.toml

@@ -23,6 +23,7 @@ serde = { workspace = true, features = ["derive"] } # for config serialization/d
 tap = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread", "net", "signal"] }
+tokio-util = { workspace = true }
 url = { workspace = true }
 
 nym-bandwidth-controller = { workspace = true }

common/socks5-client-core/src/socks/client.rs

@@ -7,6 +7,7 @@ use super::{SocksVersion, RESERVED, SOCKS4_VERSION, SOCKS5_VERSION};
 use crate::config;
 use futures::channel::mpsc;
 use futures::task::{Context, Poll};
+use futures::SinkExt;
 use log::*;
 use nym_client_core::client::inbound_messages::{InputMessage, InputMessageSender};
 use nym_service_providers_common::interface::{ProviderInterfaceVersion, RequestVersion};

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

@@ -7,7 +7,7 @@ use log::{debug, error};
 use nym_socks5_requests::{ConnectionId, SocketData};
 use std::io;
 
-pub(crate) struct OrderedMessageSender<F, S> {
+pub(crate) struct OrderedMessageSender<F, S: Send> {
     connection_id: ConnectionId,
     // addresses are provided for better logging
     local_destination_address: String,
@@ -18,7 +18,7 @@ pub(crate) struct OrderedMessageSender<F, S> {
     mix_message_adapter: F,
 }
 
-impl<F, S> OrderedMessageSender<F, S>
+impl<F, S: Send> OrderedMessageSender<F, S>
 where
     F: Fn(SocketData) -> S,
 {
@@ -56,8 +56,10 @@ where
     }
 
     async fn send_message(&self, message: S) {
-        if self.mixnet_sender.send(message).await.is_err() {
-            panic!("BatchRealMessageReceiver has stopped receiving!")
+        if let Some(sender) = self.mixnet_sender.get_ref() {
+            if sender.send(message).await.is_err() {
+                panic!("BatchRealMessageReceiver has stopped receiving!")
+            }
         }
     }
 

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

@@ -74,7 +74,7 @@ async fn wait_for_lane(
     }
 }
 
-pub(super) async fn run_inbound<F, S>(
+pub(super) async fn run_inbound<F, S: Send>(
     mut reader: OwnedReadHalf,
     mut message_sender: OrderedMessageSender<F, S>,
     connection_id: ConnectionId,

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

@@ -6,6 +6,7 @@ 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;
 use std::fmt::Debug;
 use std::{sync::Arc, time::Duration};
 use tokio::{net::TcpStream, sync::Notify};
@@ -35,7 +36,7 @@ impl From<(Vec<u8>, bool)> for ProxyMessage {
     }
 }
 
-pub type MixProxySender<S> = tokio::sync::mpsc::Sender<S>;
+pub type MixProxySender<S> = PollSender<S>;
 pub type MixProxyReader<S> = tokio::sync::mpsc::Receiver<S>;
 
 // TODO: when we finally get to implementing graceful shutdown,

service-providers/network-requester/Cargo.toml

@@ -4,11 +4,10 @@
 [package]
 name = "nym-network-requester"
 license = "GPL-3.0"
-version = "1.1.73"
+version = "1.1.68"
 authors.workspace = true
 edition.workspace = true
 rust-version = "1.85"
-publish = false
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
@@ -38,30 +37,39 @@ tap = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["net", "rt-multi-thread", "macros"] }
 tokio-tungstenite = { workspace = true }
+tokio-util = { workspace = true }
 url = { workspace = true }
 time = { workspace = true }
 zeroize = { workspace = true }
 
 # internal
-nym-async-file-watcher = { workspace = true }
-nym-bin-common = { workspace = true, features = ["output_format", "clap", "basic_tracing"] }
-nym-client-core = { workspace = true, features = ["cli", "fs-gateways-storage", "fs-surb-storage"] }
-nym-client-websocket-requests = { workspace = true }
-nym-config = { workspace = true }
-nym-credentials = { workspace = true }
-nym-credential-storage = { workspace = true }
-nym-crypto = { workspace = true }
-nym-network-defaults = { workspace = true }
-nym-ordered-buffer = { workspace = true }
-nym-sdk = { workspace = true }
-nym-service-providers-common = { workspace = true }
-nym-socks5-proxy-helpers = { workspace = true }
-nym-socks5-requests = { workspace = true }
-nym-sphinx = { workspace = true }
-nym-task = { workspace = true }
-nym-types = { workspace = true }
-nym-exit-policy = { workspace = true, features = ["client"] }
-nym-id = { workspace = true }
+
+
+nym-async-file-watcher = { path = "../../common/async-file-watcher" }
+nym-client-core = { path = "../../common/client-core", features = [
+    "cli",
+    "fs-gateways-storage",
+    "fs-surb-storage",
+] }
+nym-bin-common = { path = "../../common/bin-common", features = [
+    "output_format",
+] }
+nym-client-websocket-requests = { path = "../../clients/native/websocket-requests" }
+nym-config = { path = "../../common/config" }
+nym-credentials = { path = "../../common/credentials" }
+nym-credential-storage = { path = "../../common/credential-storage" }
+nym-crypto = { path = "../../common/crypto" }
+nym-network-defaults = { path = "../../common/network-defaults" }
+nym-ordered-buffer = { path = "../../common/socks5/ordered-buffer" }
+nym-sdk = { path = "../../sdk/rust/nym-sdk" }
+nym-service-providers-common = { path = "../common" }
+nym-socks5-proxy-helpers = { path = "../../common/socks5/proxy-helpers" }
+nym-socks5-requests = { path = "../../common/socks5/requests" }
+nym-sphinx = { path = "../../common/nymsphinx" }
+nym-task = { path = "../../common/task" }
+nym-types = { path = "../../common/types" }
+nym-exit-policy = { path = "../../common/exit-policy", features = ["client"] }
+nym-id = { path = "../../common/nym-id" }
 
 [dev-dependencies]
 tempfile = { workspace = true }

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

@@ -9,6 +9,7 @@ use crate::{reply, socks5};
 use async_trait::async_trait;
 use futures::channel::{mpsc, oneshot};
 use futures::stream::StreamExt;
+use futures::SinkExt;
 use log::{debug, error, warn};
 use nym_bin_common::bin_info_owned;
 use nym_client_core::HardcodedTopologyProvider;
@@ -37,6 +38,7 @@ use nym_task::ShutdownTracker;
 use nym_task::connections::LaneQueueLengths;
 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);
@@ -252,6 +254,8 @@ impl NRServiceProviderBuilder {
         // going to be used by `mixnet_response_listener`
         let (mix_input_sender, mix_input_receiver) = tokio::sync::mpsc::channel::<MixnetMessage>(1);
 
+        let mix_input_sender = PollSender::new(mix_input_sender);
+
         // Controller for managing all active connections.
         let (mut active_connections_controller, controller_sender) = Controller::new(
             mixnet_client.connection_command_sender(),
@@ -376,7 +380,7 @@ impl NRServiceProvider {
         return_address: reply::MixnetAddress,
         biggest_packet_size: PacketSize,
         controller_sender: ControllerSender,
-        mix_input_sender: MixProxySender<MixnetMessage>,
+        mut mix_input_sender: MixProxySender<MixnetMessage>,
         lane_queue_lengths: LaneQueueLengths,
         shutdown: ShutdownTracker,
     ) {
@@ -469,7 +473,7 @@ impl NRServiceProvider {
             .unwrap_or(traffic_config.primary_packet_size);
 
         let controller_sender_clone = self.controller_sender.clone();
-        let mix_input_sender_clone = self.mix_input_sender.clone();
+        let mut mix_input_sender_clone = self.mix_input_sender.clone();
         let lane_queue_lengths_clone = self.mixnet_client.shared_lane_queue_lengths();
 
         // we're just cloning the underlying pointer, nothing expensive is happening here