diff --git a/Cargo.lock b/Cargo.lock index 4ff0f9a693..bb62518ca4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4474,6 +4474,7 @@ dependencies = [ name = "nym-sdk" version = "0.1.0" dependencies = [ + "async-trait", "bip39", "dotenvy", "futures", diff --git a/documentation/docs/src/sdk/rust.md b/documentation/docs/src/sdk/rust.md index 68004b7576..bf52bb441a 100644 --- a/documentation/docs/src/sdk/rust.md +++ b/documentation/docs/src/sdk/rust.md @@ -54,7 +54,7 @@ If you're integrating mixnet functionality into an existing app and want to inte ``` ### Anonymous replies with SURBs -Both functions used to send messages through the mixnet (`send_str` and `send_bytes`) send a pre-determined number of SURBs along with their messages by default. +Both functions used to send messages through the mixnet (`send_message` and `send_plain_message`) send a pre-determined number of SURBs along with their messages by default. The number of SURBs is set [here](https://github.com/nymtech/nym/blob/release/{{platform_release_version}}/sdk/rust/nym-sdk/src/mixnet/client.rs#L34): diff --git a/sdk/rust/nym-sdk/Cargo.toml b/sdk/rust/nym-sdk/Cargo.toml index c034e733a1..e2b5d86d7e 100644 --- a/sdk/rust/nym-sdk/Cargo.toml +++ b/sdk/rust/nym-sdk/Cargo.toml @@ -6,6 +6,7 @@ edition = "2021" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] +async-trait = { workspace = true } bip39 = { workspace = true } nym-client-core = { path = "../../../common/client-core", features = ["fs-surb-storage"]} nym-crypto = { path = "../../../common/crypto" } diff --git a/sdk/rust/nym-sdk/examples/bandwidth.rs b/sdk/rust/nym-sdk/examples/bandwidth.rs index fbe100f037..b7f452fa92 100644 --- a/sdk/rust/nym-sdk/examples/bandwidth.rs +++ b/sdk/rust/nym-sdk/examples/bandwidth.rs @@ -1,4 +1,5 @@ use nym_sdk::mixnet; +use nym_sdk::mixnet::MixnetMessageSender; #[tokio::main] async fn main() { @@ -29,7 +30,10 @@ async fn main() { let our_address = client.nym_address(); // Send a message throughout the mixnet to ourselves - client.send_str(*our_address, "hello there").await; + client + .send_plain_message(*our_address, "hello there") + .await + .unwrap(); println!("Waiting for message"); if let Some(received) = client.wait_for_messages().await { diff --git a/sdk/rust/nym-sdk/examples/builder.rs b/sdk/rust/nym-sdk/examples/builder.rs index af6beec5f8..6ab50bc92c 100644 --- a/sdk/rust/nym-sdk/examples/builder.rs +++ b/sdk/rust/nym-sdk/examples/builder.rs @@ -1,4 +1,5 @@ use nym_sdk::mixnet; +use nym_sdk::mixnet::MixnetMessageSender; #[tokio::main] async fn main() { @@ -19,7 +20,10 @@ async fn main() { println!("Our client nym address is: {our_address}"); // Send a message through the mixnet to ourselves - client.send_str(*our_address, "hello there").await; + client + .send_plain_message(*our_address, "hello there") + .await + .unwrap(); println!("Waiting for message"); if let Some(received) = client.wait_for_messages().await { diff --git a/sdk/rust/nym-sdk/examples/builder_with_storage.rs b/sdk/rust/nym-sdk/examples/builder_with_storage.rs index ca7a732e5b..900b2d323f 100644 --- a/sdk/rust/nym-sdk/examples/builder_with_storage.rs +++ b/sdk/rust/nym-sdk/examples/builder_with_storage.rs @@ -1,4 +1,5 @@ use nym_sdk::mixnet; +use nym_sdk::mixnet::MixnetMessageSender; use std::path::PathBuf; #[tokio::main] @@ -26,7 +27,10 @@ async fn main() { println!("Our client nym address is: {our_address}"); // Send a message throught the mixnet to ourselves - client.send_str(*our_address, "hello there").await; + client + .send_plain_message(*our_address, "hello there") + .await + .unwrap(); println!("Waiting for message"); if let Some(received) = client.wait_for_messages().await { diff --git a/sdk/rust/nym-sdk/examples/custom_topology_provider.rs b/sdk/rust/nym-sdk/examples/custom_topology_provider.rs index 2ddf9d8ea0..7232f5ac58 100644 --- a/sdk/rust/nym-sdk/examples/custom_topology_provider.rs +++ b/sdk/rust/nym-sdk/examples/custom_topology_provider.rs @@ -2,6 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 use nym_sdk::mixnet; +use nym_sdk::mixnet::MixnetMessageSender; use nym_topology::provider_trait::{async_trait, TopologyProvider}; use nym_topology::{nym_topology_from_detailed, NymTopology}; use url::Url; @@ -69,7 +70,10 @@ async fn main() { println!("Our client nym address is: {our_address}"); // Send a message through the mixnet to ourselves - client.send_str(*our_address, "hello there").await; + client + .send_plain_message(*our_address, "hello there") + .await + .unwrap(); println!("Waiting for message (ctrl-c to exit)"); client diff --git a/sdk/rust/nym-sdk/examples/manually_handle_storage.rs b/sdk/rust/nym-sdk/examples/manually_handle_storage.rs index 3c5ef4f450..6130af5a25 100644 --- a/sdk/rust/nym-sdk/examples/manually_handle_storage.rs +++ b/sdk/rust/nym-sdk/examples/manually_handle_storage.rs @@ -3,6 +3,7 @@ use nym_client_core::client::base_client::storage::gateway_details::{ }; use nym_sdk::mixnet::{ self, EmptyReplyStorage, EphemeralCredentialStorage, KeyManager, KeyStore, MixnetClientStorage, + MixnetMessageSender, }; use nym_topology::provider_trait::async_trait; @@ -26,7 +27,10 @@ async fn main() { println!("Our client nym address is: {our_address}"); // Send important info up the pipe to a buddy - client.send_str(*our_address, "hello there").await; + client + .send_plain_message(*our_address, "hello there") + .await + .unwrap(); println!("Waiting for message"); if let Some(received) = client.wait_for_messages().await { diff --git a/sdk/rust/nym-sdk/examples/manually_overwrite_topology.rs b/sdk/rust/nym-sdk/examples/manually_overwrite_topology.rs index 3ef7e2257f..bad1c9651b 100644 --- a/sdk/rust/nym-sdk/examples/manually_overwrite_topology.rs +++ b/sdk/rust/nym-sdk/examples/manually_overwrite_topology.rs @@ -2,6 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 use nym_sdk::mixnet; +use nym_sdk::mixnet::MixnetMessageSender; use nym_topology::mix::Layer; use nym_topology::{mix, NymTopology}; use std::collections::BTreeMap; @@ -81,7 +82,10 @@ async fn main() { println!("Our client nym address is: {our_address}"); // Send a message through the mixnet to ourselves - client.send_str(*our_address, "hello there").await; + client + .send_plain_message(*our_address, "hello there") + .await + .unwrap(); println!("Waiting for message (ctrl-c to exit)"); client diff --git a/sdk/rust/nym-sdk/examples/parallel_sending_and_receiving.rs b/sdk/rust/nym-sdk/examples/parallel_sending_and_receiving.rs new file mode 100644 index 0000000000..e3690ce9e0 --- /dev/null +++ b/sdk/rust/nym-sdk/examples/parallel_sending_and_receiving.rs @@ -0,0 +1,55 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use futures::StreamExt; +use nym_sdk::mixnet; +use nym_sdk::mixnet::MixnetMessageSender; +use std::sync::Arc; +use tokio::sync::Notify; + +#[tokio::main] +async fn main() { + nym_bin_common::logging::setup_logging(); + + // Passing no config makes the client fire up an ephemeral session and figure stuff out on its own + let mut client = mixnet::MixnetClient::connect_new().await.unwrap(); + + // Be able to get our client address + let our_address = *client.nym_address(); + println!("Our client nym address is: {our_address}"); + + let receive_notify = Arc::new(Notify::new()); + let receive_notify2 = receive_notify.clone(); + let send_notify = Arc::new(Notify::new()); + let send_notify2 = send_notify.clone(); + + let sender = client.split_sender(); + + // receiving task + tokio::spawn(async move { + if let Some(received) = client.next().await { + for r in received { + println!("Received: {}", String::from_utf8_lossy(&r.message)); + } + } + + client.disconnect().await; + // notify that we're done and leave the task! + receive_notify.notify_waiters(); + }); + + // sending task + tokio::spawn(async move { + sender + .send_plain_message(our_address, "hello from a different task!") + .await + .unwrap(); + + send_notify.notify_waiters(); + }); + + // wait for both tasks to be done + println!("waiting for shutdown"); + send_notify2.notified().await; + receive_notify2.notified().await; +} diff --git a/sdk/rust/nym-sdk/examples/simple.rs b/sdk/rust/nym-sdk/examples/simple.rs index f2325e6048..14720ba24c 100644 --- a/sdk/rust/nym-sdk/examples/simple.rs +++ b/sdk/rust/nym-sdk/examples/simple.rs @@ -1,4 +1,5 @@ use nym_sdk::mixnet; +use nym_sdk::mixnet::MixnetMessageSender; #[tokio::main] async fn main() { @@ -12,7 +13,10 @@ async fn main() { println!("Our client nym address is: {our_address}"); // Send a message throught the mixnet to ourselves - client.send_str(*our_address, "hello there").await; + client + .send_plain_message(*our_address, "hello there") + .await + .unwrap(); println!("Waiting for message (ctrl-c to exit)"); client diff --git a/sdk/rust/nym-sdk/examples/surb-reply.rs b/sdk/rust/nym-sdk/examples/surb-reply.rs index 38c7e8257a..abe955b2b9 100644 --- a/sdk/rust/nym-sdk/examples/surb-reply.rs +++ b/sdk/rust/nym-sdk/examples/surb-reply.rs @@ -1,5 +1,6 @@ use nym_sdk::mixnet::{ - AnonymousSenderTag, MixnetClientBuilder, ReconstructedMessage, StoragePaths, + AnonymousSenderTag, MixnetClientBuilder, MixnetMessageSender, ReconstructedMessage, + StoragePaths, }; use std::path::PathBuf; @@ -28,7 +29,10 @@ async fn main() { println!("\nOur client nym address is: {our_address}"); // Send a message through the mixnet to ourselves using our nym address - client.send_str(*our_address, "hello there").await; + client + .send_plain_message(*our_address, "hello there") + .await + .unwrap(); // we're going to parse the sender_tag (AnonymousSenderTag) from the incoming message and use it to 'reply' to ourselves instead of our Nym address. // we know there will be a sender_tag since the sdk sends SURBs along with messages by default. @@ -57,7 +61,10 @@ async fn main() { // reply to self with it: note we use `send_str_reply` instead of `send_str` println!("Replying with using SURBs"); - client.send_str_reply(return_recipient, "hi an0n!").await; + client + .send_reply(return_recipient, "hi an0n!") + .await + .unwrap(); println!("Waiting for message (once you see it, ctrl-c to exit)\n"); client diff --git a/sdk/rust/nym-sdk/src/error.rs b/sdk/rust/nym-sdk/src/error.rs index 84283bc9a2..c451775d6c 100644 --- a/sdk/rust/nym-sdk/src/error.rs +++ b/sdk/rust/nym-sdk/src/error.rs @@ -90,6 +90,9 @@ pub enum Error { #[error("loaded shared gateway key without providing information about what gateway it corresponds to")] GatewayWithUnknownEndpoint, + + #[error("failed to send the provided message")] + MessageSendingFailure, } pub type Result = std::result::Result; diff --git a/sdk/rust/nym-sdk/src/mixnet.rs b/sdk/rust/nym-sdk/src/mixnet.rs index c1b85aaff3..d082756cd6 100644 --- a/sdk/rust/nym-sdk/src/mixnet.rs +++ b/sdk/rust/nym-sdk/src/mixnet.rs @@ -36,6 +36,7 @@ mod connection_state; mod native_client; mod paths; mod socks5_client; +mod traits; pub use client::{DisconnectedMixnetClient, IncludedSurbs, MixnetClientBuilder}; pub use config::{Config, KeyMode}; @@ -73,3 +74,4 @@ pub use nym_sphinx::{ pub use nym_topology::{provider_trait::TopologyProvider, NymTopology}; pub use paths::StoragePaths; pub use socks5_client::Socks5MixnetClient; +pub use traits::MixnetMessageSender; diff --git a/sdk/rust/nym-sdk/src/mixnet/native_client.rs b/sdk/rust/nym-sdk/src/mixnet/native_client.rs index 774b5c9868..e03d585ebb 100644 --- a/sdk/rust/nym-sdk/src/mixnet/native_client.rs +++ b/sdk/rust/nym-sdk/src/mixnet/native_client.rs @@ -1,5 +1,7 @@ -use crate::mixnet::client::{IncludedSurbs, MixnetClientBuilder}; -use crate::Result; +use crate::mixnet::client::MixnetClientBuilder; +use crate::mixnet::traits::MixnetMessageSender; +use crate::{Error, Result}; +use async_trait::async_trait; use futures::{Stream, StreamExt}; use nym_client_core::client::{ base_client::{ClientInput, ClientOutput, ClientState}, @@ -7,10 +9,9 @@ use nym_client_core::client::{ received_buffer::ReconstructedMessagesReceiver, }; use nym_sphinx::addressing::clients::Recipient; -use nym_sphinx::anonymous_replies::requests::AnonymousSenderTag; use nym_sphinx::{params::PacketType, receiver::ReconstructedMessage}; use nym_task::{ - connections::{ConnectionCommandSender, LaneQueueLengths, TransmissionLane}, + connections::{ConnectionCommandSender, LaneQueueLengths}, TaskManager, }; use nym_topology::NymTopology; @@ -74,9 +75,10 @@ impl MixnetClient { /// Get a shallow clone of [`MixnetClientSender`]. Useful if you want split the send and /// receive logic in different locations. - pub fn sender(&self) -> MixnetClientSender { + pub fn split_sender(&self) -> MixnetClientSender { MixnetClientSender { client_input: self.client_input.clone(), + packet_type: self.packet_type, } } @@ -112,132 +114,6 @@ impl MixnetClient { self.client_state.topology_accessor.release_manual_control() } - /// Sends stringy data to the supplied Nym address - /// - /// # Example - /// - /// ```no_run - /// use nym_sdk::mixnet; - /// - /// #[tokio::main] - /// async fn main() { - /// let address = "foobar"; - /// let recipient = mixnet::Recipient::try_from_base58_string(address).unwrap(); - /// let mut client = mixnet::MixnetClient::connect_new().await.unwrap(); - /// client.send_str(recipient, "hi").await; - /// } - /// ``` - pub async fn send_str(&self, address: Recipient, message: &str) { - self.send_bytes(address, message, IncludedSurbs::default()) - .await; - } - - /// Sends bytes to the supplied Nym address. There is the option to specify the number of - /// reply-SURBs to include. - /// - /// # Example - /// - /// ```no_run - /// use nym_sdk::mixnet; - /// - /// #[tokio::main] - /// async fn main() { - /// let address = "foobar"; - /// let recipient = mixnet::Recipient::try_from_base58_string(address).unwrap(); - /// let mut client = mixnet::MixnetClient::connect_new().await.unwrap(); - /// let surbs = mixnet::IncludedSurbs::default(); - /// client.send_bytes(recipient, "hi".to_owned().into_bytes(), surbs).await; - /// } - /// ``` - pub async fn send_bytes>( - &self, - address: Recipient, - message: M, - surbs: IncludedSurbs, - ) { - let lane = TransmissionLane::General; - let input_msg = match surbs { - IncludedSurbs::Amount(surbs) => InputMessage::new_anonymous( - address, - message.as_ref().to_vec(), - surbs, - lane, - self.packet_type, - ), - IncludedSurbs::ExposeSelfAddress => InputMessage::new_regular( - address, - message.as_ref().to_vec(), - lane, - self.packet_type, - ), - }; - self.send(input_msg).await - } - - /// Sends stringy reply data to the supplied anonymous recipient. - /// - /// # Example - /// - /// ```no_run - /// use nym_sdk::mixnet; - /// - /// #[tokio::main] - /// async fn main() { - /// let mut client = mixnet::MixnetClient::connect_new().await.unwrap(); - /// // note: the tag is something you would have received from a remote client sending you surbs! - /// let tag = mixnet::AnonymousSenderTag::try_from_base58_string("foobar").unwrap(); - /// client.send_str_reply(tag, "hi").await; - /// } - /// ``` - pub async fn send_str_reply(&self, recipient_tag: AnonymousSenderTag, message: &str) { - self.send_reply(recipient_tag, message).await; - } - - /// Sends binary reply data to the supplied anonymous recipient. - /// - /// # Example - /// - /// ```no_run - /// use nym_sdk::mixnet; - /// - /// #[tokio::main] - /// async fn main() { - /// let mut client = mixnet::MixnetClient::connect_new().await.unwrap(); - /// // note: the tag is something you would have received from a remote client sending you surbs! - /// let tag = mixnet::AnonymousSenderTag::try_from_base58_string("foobar").unwrap(); - /// client.send_reply(tag, b"hi").await; - /// } - /// ``` - pub async fn send_reply>(&self, recipient_tag: AnonymousSenderTag, message: M) { - let lane = TransmissionLane::General; - let input_msg = InputMessage::new_reply( - recipient_tag, - message.as_ref().to_vec(), - lane, - self.packet_type, - ); - self.send(input_msg).await - } - - /// Sends a [`InputMessage`] to the mixnet. This is the most low-level sending function, for - /// full customization. - async fn send(&self, message: InputMessage) { - if self.client_input.send(message).await.is_err() { - log::error!("Failed to send message"); - } - } - - /// Sends a [`InputMessage`] to the mixnet. This is the most low-level sending function, for - /// full customization. - /// - /// Waits until the message is actually sent, or close to being sent, until returning. - /// - /// NOTE: this not yet implemented. - #[allow(unused)] - async fn send_wait(&self, _message: InputMessage) { - todo!(); - } - /// Wait for messages from the mixnet pub async fn wait_for_messages(&mut self) -> Option> { self.reconstructed_receiver.next().await @@ -265,14 +141,7 @@ impl MixnetClient { pub struct MixnetClientSender { client_input: ClientInput, -} - -impl MixnetClientSender { - pub async fn send_input_message(&mut self, message: InputMessage) { - if self.client_input.send(message).await.is_err() { - log::error!("Failed to send message"); - } - } + packet_type: Option, } impl Stream for MixnetClient { @@ -282,3 +151,31 @@ impl Stream for MixnetClient { Pin::new(&mut self.reconstructed_receiver).poll_next(cx) } } + +#[async_trait] +impl MixnetMessageSender for MixnetClient { + fn packet_type(&self) -> Option { + self.packet_type + } + + async fn send(&self, message: InputMessage) -> Result<()> { + self.client_input + .send(message) + .await + .map_err(|_| Error::MessageSendingFailure) + } +} + +#[async_trait] +impl MixnetMessageSender for MixnetClientSender { + fn packet_type(&self) -> Option { + self.packet_type + } + + async fn send(&self, message: InputMessage) -> Result<()> { + self.client_input + .send(message) + .await + .map_err(|_| Error::MessageSendingFailure) + } +} diff --git a/sdk/rust/nym-sdk/src/mixnet/traits.rs b/sdk/rust/nym-sdk/src/mixnet/traits.rs new file mode 100644 index 0000000000..8eb78109f7 --- /dev/null +++ b/sdk/rust/nym-sdk/src/mixnet/traits.rs @@ -0,0 +1,119 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::mixnet::{AnonymousSenderTag, IncludedSurbs, Recipient}; +use crate::Result; +use async_trait::async_trait; +use nym_client_core::client::inbound_messages::InputMessage; +use nym_sphinx::params::PacketType; +use nym_task::connections::TransmissionLane; + +// defined to guarantee common interface regardless of whether you're using the full client +// or just the sending handler +#[async_trait] +pub trait MixnetMessageSender { + fn packet_type(&self) -> Option { + None + } + + /// Sends a [`InputMessage`] to the mixnet. This is the most low-level sending function, for + /// full customization. + async fn send(&self, message: InputMessage) -> Result<()>; + + /// Sends data to the supplied Nym address with the default surb behaviour. + /// + /// # Example + /// + /// ```no_run + /// use nym_sdk::mixnet::{self, MixnetMessageSender}; + /// + /// #[tokio::main] + /// async fn main() { + /// let address = "foobar"; + /// let recipient = mixnet::Recipient::try_from_base58_string(address).unwrap(); + /// let mut client = mixnet::MixnetClient::connect_new().await.unwrap(); + /// client.send_plain_message(recipient, "hi").await.unwrap(); + /// } + /// ``` + async fn send_plain_message(&self, address: Recipient, message: M) -> Result<()> + where + M: AsRef<[u8]> + Send, + { + self.send_message(address, message, IncludedSurbs::default()) + .await + } + + /// Sends bytes to the supplied Nym address. There is the option to specify the number of + /// reply-SURBs to include. + /// + /// # Example + /// + /// ```no_run + /// use nym_sdk::mixnet::{self, MixnetMessageSender}; + /// + /// #[tokio::main] + /// async fn main() { + /// let address = "foobar"; + /// let recipient = mixnet::Recipient::try_from_base58_string(address).unwrap(); + /// let mut client = mixnet::MixnetClient::connect_new().await.unwrap(); + /// let surbs = mixnet::IncludedSurbs::default(); + /// client.send_message(recipient, "hi".to_owned().into_bytes(), surbs).await.unwrap(); + /// } + /// ``` + async fn send_message( + &self, + address: Recipient, + message: M, + surbs: IncludedSurbs, + ) -> Result<()> + where + M: AsRef<[u8]> + Send, + { + let lane = TransmissionLane::General; + let input_msg = match surbs { + IncludedSurbs::Amount(surbs) => InputMessage::new_anonymous( + address, + message.as_ref().to_vec(), + surbs, + lane, + self.packet_type(), + ), + IncludedSurbs::ExposeSelfAddress => InputMessage::new_regular( + address, + message.as_ref().to_vec(), + lane, + self.packet_type(), + ), + }; + self.send(input_msg).await + } + + /// Sends reply data to the supplied anonymous recipient. + /// + /// # Example + /// + /// ```no_run + /// use nym_sdk::mixnet::{self, MixnetMessageSender}; + /// + /// #[tokio::main] + /// async fn main() { + /// let mut client = mixnet::MixnetClient::connect_new().await.unwrap(); + /// // note: the tag is something you would have received from a remote client sending you surbs! + /// let tag = mixnet::AnonymousSenderTag::try_from_base58_string("foobar").unwrap(); + /// client.send_reply(tag, b"hi").await.unwrap(); + /// } + /// ``` + async fn send_reply(&self, recipient_tag: AnonymousSenderTag, message: M) -> Result<()> + where + M: AsRef<[u8]> + Send, + { + let lane = TransmissionLane::General; + let input_msg = InputMessage::new_reply( + recipient_tag, + message.as_ref().to_vec(), + lane, + self.packet_type(), + ); + self.send(input_msg).await + } +} diff --git a/service-providers/common/examples/control_requests.rs b/service-providers/common/examples/control_requests.rs index 17af10448c..14886390f8 100644 --- a/service-providers/common/examples/control_requests.rs +++ b/service-providers/common/examples/control_requests.rs @@ -3,7 +3,9 @@ // use nym_client::client::config::{BaseClientConfig, Config, GatewayEndpointConfig}; // use nym_client::client::{DirectClient, KeyManager, Recipient, ReconstructedMessage, SocketClient}; -use nym_sdk::mixnet::{IncludedSurbs, MixnetClient, Recipient, ReconstructedMessage}; +use nym_sdk::mixnet::{ + IncludedSurbs, MixnetClient, MixnetMessageSender, Recipient, ReconstructedMessage, +}; use nym_service_providers_common::interface::{ ControlRequest, ControlResponse, ProviderInterfaceVersion, Request, Response, ResponseContent, }; @@ -51,34 +53,34 @@ async fn main() -> anyhow::Result<()> { // TODO: currently we HAVE TO use surbs unfortunately println!("Sending 'Health' request..."); client - .send_bytes( + .send_message( provider, full_request_health.into_bytes(), IncludedSurbs::new(10), ) - .await; + .await?; let response = wait_for_control_response(&mut client).await; println!("response to 'Health' request: {response:#?}"); println!("Sending 'BinaryInfo' request..."); client - .send_bytes( + .send_message( provider, full_request_binary_info.into_bytes(), IncludedSurbs::none(), ) - .await; + .await?; let response = wait_for_control_response(&mut client).await; println!("response to 'BinaryInfo' request: {response:#?}"); println!("Sending 'SupportedRequestVersions' request..."); client - .send_bytes( + .send_message( provider, full_request_versions.into_bytes(), IncludedSurbs::none(), ) - .await; + .await?; let response = wait_for_control_response(&mut client).await; println!("response to 'SupportedRequestVersions' request: {response:#?}"); diff --git a/service-providers/network-requester/examples/query.rs b/service-providers/network-requester/examples/query.rs index 5b89e6924b..87f410ee75 100644 --- a/service-providers/network-requester/examples/query.rs +++ b/service-providers/network-requester/examples/query.rs @@ -1,7 +1,9 @@ // Copyright 2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use nym_sdk::mixnet::{IncludedSurbs, MixnetClient, Recipient, ReconstructedMessage}; +use nym_sdk::mixnet::{ + IncludedSurbs, MixnetClient, MixnetMessageSender, Recipient, ReconstructedMessage, +}; use nym_service_providers_common::interface::{ ProviderInterfaceVersion, Request, Response, ResponseContent, }; @@ -47,23 +49,23 @@ async fn main() -> anyhow::Result<()> { ); println!("Sending 'OpenProxy' query..."); client - .send_bytes( + .send_message( provider, open_proxy_request.into_bytes(), IncludedSurbs::new(10), ) - .await; + .await?; let response = wait_for_response(&mut client).await; println!("response to 'OpenProxy' query: {response:#?}"); println!("Sending 'Description' query..."); client - .send_bytes( + .send_message( provider, description_request.into_bytes(), IncludedSurbs::none(), ) - .await; + .await?; let response = wait_for_response(&mut client).await; println!("response to 'Description' query: {response:#?}"); diff --git a/service-providers/network-requester/src/core.rs b/service-providers/network-requester/src/core.rs index 92ab7af231..cd9b57f2dd 100644 --- a/service-providers/network-requester/src/core.rs +++ b/service-providers/network-requester/src/core.rs @@ -16,6 +16,7 @@ use log::warn; use nym_bin_common::bin_info_owned; use nym_client_core::config::disk_persistence::CommonClientPaths; use nym_network_defaults::NymNetworkDetails; +use nym_sdk::mixnet::MixnetMessageSender; use nym_service_providers_common::interface::{ BinaryInformation, ProviderInterfaceVersion, Request, RequestVersion, }; @@ -223,7 +224,7 @@ impl NRServiceProviderBuilder { }; let stats_collector_clone = stats_collector.clone(); - let mixnet_client_sender = mixnet_client.sender(); + let mixnet_client_sender = mixnet_client.split_sender(); let self_address = *mixnet_client.nym_address(); // start the listener for mix messages @@ -297,7 +298,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( - mut mixnet_client_sender: nym_sdk::mixnet::MixnetClientSender, + mixnet_client_sender: nym_sdk::mixnet::MixnetClientSender, mut mix_input_reader: MixProxyReader, stats_collector: Option, ) { @@ -321,7 +322,7 @@ impl NRServiceProvider { } let response_message = msg.into_input_message(); - mixnet_client_sender.send_input_message(response_message).await; + mixnet_client_sender.send(response_message).await.unwrap(); } else { log::error!("Exiting: channel closed!"); break; diff --git a/tools/nym-nr-query/src/main.rs b/tools/nym-nr-query/src/main.rs index 97d376f6d5..2de1c08468 100644 --- a/tools/nym-nr-query/src/main.rs +++ b/tools/nym-nr-query/src/main.rs @@ -2,7 +2,7 @@ use std::fmt; use clap::{Parser, ValueEnum}; use nym_bin_common::output_format::OutputFormat; -use nym_sdk::mixnet::{self, IncludedSurbs}; +use nym_sdk::mixnet::{self, IncludedSurbs, MixnetMessageSender}; use nym_service_providers_common::interface::{ ControlRequest, ControlResponse, ProviderInterfaceVersion, Request, Response, ResponseContent, }; @@ -154,34 +154,37 @@ impl QueryClient { async fn query_bin_info(&mut self) -> ControlResponse { self.client - .send_bytes( + .send_message( self.provider, new_bin_info_request().into_bytes(), IncludedSurbs::new(10), ) - .await; + .await + .unwrap(); wait_for_control_response(&mut self.client).await } async fn query_supported_versions(&mut self) -> ControlResponse { self.client - .send_bytes( + .send_message( self.provider, new_supported_request_versions_request().into_bytes(), IncludedSurbs::new(10), ) - .await; + .await + .unwrap(); wait_for_control_response(&mut self.client).await } async fn query_open_proxy(&mut self) -> QueryResponse { self.client - .send_bytes( + .send_message( self.provider, new_open_proxy_request().into_bytes(), IncludedSurbs::new(10), ) - .await; + .await + .unwrap(); let response = wait_for_socks5_response(&mut self.client).await; response .content @@ -193,12 +196,13 @@ impl QueryClient { async fn ping(&mut self) -> PingResponse { let now = std::time::Instant::now(); self.client - .send_bytes( + .send_message( self.provider, new_ping_request().into_bytes(), IncludedSurbs::new(5), ) - .await; + .await + .unwrap(); let response = wait_for_control_response(&mut self.client).await; assert!(matches!(response, ControlResponse::Health)); let elapsed = now.elapsed();