Update service providers for &mut self send signature

2026-01-11 21:29:48 +00:00
parent 9224f01d49
commit c7b0eed15f
5 changed files with 47 additions and 13 deletions
@@ -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<u8>,
        recipient: Option<Recipient>,
        sender_tag: Option<AnonymousSenderTag>,
@@ -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<nym_sdk::mixnet::InputMessage> {
+            todo!()
+        }
    }

    #[tokio::test]
@@ -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<PacketHandleResult>) {
+    async fn handle_responses(&mut self, responses: Vec<PacketHandleResult>) {
        for response in responses {
            match response {
                Ok(Some(response)) => {
@@ -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 }
@@ -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::<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(),
@@ -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<MixnetMessage>,
        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<MixnetMessage>,
+        mut mix_input_sender: MixProxySender<MixnetMessage>,
        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