diff --git a/gateway/src/node/internal_service_providers/authenticator/mixnet_listener.rs b/gateway/src/node/internal_service_providers/authenticator/mixnet_listener.rs index 3607ff7926..713620e35f 100644 --- a/gateway/src/node/internal_service_providers/authenticator/mixnet_listener.rs +++ b/gateway/src/node/internal_service_providers/authenticator/mixnet_listener.rs @@ -928,7 +928,7 @@ impl MixnetListener { // When an incoming mixnet message triggers a response that we send back. async fn handle_response( - &self, + &mut self, response: Vec, recipient: Option, sender_tag: Option, diff --git a/service-providers/ip-packet-router/src/clients/connected_client_handler.rs b/service-providers/ip-packet-router/src/clients/connected_client_handler.rs index ba9f9b3290..d047819d31 100644 --- a/service-providers/ip-packet-router/src/clients/connected_client_handler.rs +++ b/service-providers/ip-packet-router/src/clients/connected_client_handler.rs @@ -219,14 +219,14 @@ impl MixnetMessageSinkTranslator for ToIprDataResponse { #[cfg(test)] mod tests { - use std::sync::{Arc, Mutex}; - use async_trait::async_trait; use bytes::Bytes; use futures::SinkExt; use nym_sdk::mixnet::{AnonymousSenderTag, MixnetMessageSender, MixnetMessageSink}; + use std::sync::{Arc, Mutex}; use tokio::sync::Notify; use tokio_util::codec::FramedWrite; + use tokio_util::sync::PollSender; use super::*; @@ -264,12 +264,15 @@ mod tests { #[async_trait] impl MixnetMessageSender for MockMixnetClientSender { - async fn send(&self, message: InputMessage) -> std::result::Result<(), nym_sdk::Error> { + async fn send(&mut self, message: InputMessage) -> std::result::Result<(), nym_sdk::Error> { let mut sent_messages = self.sent_messages.lock().unwrap(); sent_messages.push(message); self.notify.notify_one(); Ok(()) } + fn sender(&mut self) -> &mut PollSender { + todo!() + } } #[tokio::test] diff --git a/service-providers/ip-packet-router/src/mixnet_listener.rs b/service-providers/ip-packet-router/src/mixnet_listener.rs index fc5b2b20d6..cfbf5fce21 100644 --- a/service-providers/ip-packet-router/src/mixnet_listener.rs +++ b/service-providers/ip-packet-router/src/mixnet_listener.rs @@ -607,7 +607,7 @@ impl MixnetListener { // When an incoming mixnet message triggers a response that we send back, such as during // connect handshake. - async fn handle_response(&self, response: VersionedResponse) -> Result<()> { + async fn handle_response(&mut self, response: VersionedResponse) -> Result<()> { let send_to = response.reply_to.clone(); let response_bytes = response.try_into_bytes()?; let input_message = @@ -622,7 +622,7 @@ impl MixnetListener { // A single incoming request can trigger multiple responses, such as when data requests contain // multiple IP packets. - async fn handle_responses(&self, responses: Vec) { + async fn handle_responses(&mut self, responses: Vec) { for response in responses { match response { Ok(Some(response)) => { diff --git a/service-providers/network-requester/Cargo.toml b/service-providers/network-requester/Cargo.toml index 60c4d192c4..c073d0a974 100644 --- a/service-providers/network-requester/Cargo.toml +++ b/service-providers/network-requester/Cargo.toml @@ -43,8 +43,6 @@ 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 } @@ -62,6 +60,35 @@ 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" } +>>>>>>> 76c3fb42d (Update service providers for &mut self send signature) [dev-dependencies] tempfile = { workspace = true } diff --git a/service-providers/network-requester/src/core.rs b/service-providers/network-requester/src/core.rs index 6ca9b3b588..9b6cf86655 100644 --- a/service-providers/network-requester/src/core.rs +++ b/service-providers/network-requester/src/core.rs @@ -7,6 +7,7 @@ use crate::reply::MixnetMessage; use crate::request_filter::RequestFilter; use crate::{reply, socks5}; use async_trait::async_trait; +use futures::SinkExt; use futures::channel::{mpsc, oneshot}; use futures::stream::StreamExt; use log::{debug, error, warn}; @@ -15,7 +16,7 @@ use nym_client_core::HardcodedTopologyProvider; use nym_client_core::client::mix_traffic::transceiver::GatewayTransceiver; use nym_client_core::config::disk_persistence::CommonClientPaths; use nym_network_defaults::NymNetworkDetails; -use nym_sdk::mixnet::{MixnetMessageSender, TopologyProvider}; +use nym_sdk::mixnet::TopologyProvider; use nym_service_providers_common::ServiceProvider; use nym_service_providers_common::interface::{ BinaryInformation, ProviderInterfaceVersion, Request, RequestVersion, @@ -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); @@ -240,6 +242,8 @@ impl NRServiceProviderBuilder { // going to be used by `mixnet_response_listener` let (mix_input_sender, mix_input_receiver) = tokio::sync::mpsc::channel::(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(), @@ -337,7 +341,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, packet_type: PacketType, ) { @@ -346,7 +350,7 @@ impl NRServiceProvider { socks5_msg = mix_input_reader.recv() => { if let Some(msg) = socks5_msg { let response_message = msg.into_input_message(packet_type); - mixnet_client_sender.send(response_message).await.unwrap(); + nym_sdk::mixnet::MixnetMessageSender::send(&mut mixnet_client_sender, response_message).await.unwrap(); } else { log::error!("Exiting: channel closed!"); break; @@ -364,7 +368,7 @@ impl NRServiceProvider { return_address: reply::MixnetAddress, biggest_packet_size: PacketSize, controller_sender: ControllerSender, - mix_input_sender: MixProxySender, + mut mix_input_sender: MixProxySender, lane_queue_lengths: LaneQueueLengths, shutdown: ShutdownTracker, ) { @@ -457,7 +461,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