diff --git a/Cargo.lock b/Cargo.lock index d709c54890..f21e6b9cc9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4255,6 +4255,7 @@ dependencies = [ "serde-wasm-bindgen 0.6.5", "serde_json", "thiserror", + "tokio", "tsify", "wasm-bindgen", "wasm-bindgen-futures", diff --git a/common/client-core/src/client/inbound_messages.rs b/common/client-core/src/client/inbound_messages.rs index af251c2373..0c9032c0d0 100644 --- a/common/client-core/src/client/inbound_messages.rs +++ b/common/client-core/src/client/inbound_messages.rs @@ -1,4 +1,3 @@ -use hyper::body::Buf; // Copyright 2020-2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 use nym_sphinx::addressing::clients::Recipient; @@ -8,6 +7,7 @@ use nym_sphinx::params::PacketType; use nym_task::connections::TransmissionLane; use serde::{Deserialize, Serialize}; use tokio_util::{ + bytes::Buf, bytes::BytesMut, codec::{Decoder, Encoder}, }; diff --git a/common/socks5/proxy-helpers/src/ordered_sender.rs b/common/socks5/proxy-helpers/src/ordered_sender.rs index 02c465037b..de30293eb0 100644 --- a/common/socks5/proxy-helpers/src/ordered_sender.rs +++ b/common/socks5/proxy-helpers/src/ordered_sender.rs @@ -8,7 +8,7 @@ use log::{debug, error}; use nym_socks5_requests::{ConnectionId, SocketData}; use std::io; -pub(crate) struct OrderedMessageSender { +pub(crate) struct OrderedMessageSender { connection_id: ConnectionId, // addresses are provided for better logging local_destination_address: String, @@ -19,7 +19,7 @@ pub(crate) struct OrderedMessageSender { mix_message_adapter: F, } -impl OrderedMessageSender +impl OrderedMessageSender where F: Fn(SocketData) -> S, { diff --git a/wasm/client/Cargo.toml b/wasm/client/Cargo.toml index 48589a20c2..e8c4422211 100644 --- a/wasm/client/Cargo.toml +++ b/wasm/client/Cargo.toml @@ -1,6 +1,9 @@ [package] name = "nym-client-wasm" -authors = ["Dave Hrycyszyn ", "Jedrzej Stuczynski "] +authors = [ + "Dave Hrycyszyn ", + "Jedrzej Stuczynski ", +] version = "1.3.0-rc.0" edition = "2021" keywords = ["nym", "sphinx", "wasm", "webassembly", "privacy", "client"] @@ -22,15 +25,16 @@ serde_json = { workspace = true } serde-wasm-bindgen = { workspace = true } wasm-bindgen = { workspace = true } wasm-bindgen-futures = { workspace = true } -thiserror = { workspace = true } +thiserror = { workspace = true } tsify = { workspace = true, features = ["js"] } +tokio = { workspace = true, default-features = false, features = ["sync"] } nym-bin-common = { path = "../../common/bin-common" } wasm-client-core = { path = "../../common/wasm/client-core" } wasm-utils = { path = "../../common/wasm/utils" } nym-node-tester-utils = { path = "../../common/node-tester-utils", optional = true } -nym-node-tester-wasm = { path = "../node-tester", optional = true} +nym-node-tester-wasm = { path = "../node-tester", optional = true } [dev-dependencies] wasm-bindgen-test = { workspace = true } diff --git a/wasm/client/src/client.rs b/wasm/client/src/client.rs index ad54cb5e6f..eb4a2c727a 100644 --- a/wasm/client/src/client.rs +++ b/wasm/client/src/client.rs @@ -14,6 +14,7 @@ use crate::response_pusher::ResponsePusher; use js_sys::Promise; use serde::{Deserialize, Serialize}; use std::sync::Arc; +use tokio::sync::RwLock; use tsify::Tsify; use wasm_bindgen::prelude::*; use wasm_bindgen_futures::future_to_promise; @@ -50,7 +51,7 @@ pub(crate) const NODE_TESTER_CLIENT_ID: &str = "_nym-node-tester-client"; #[wasm_bindgen] pub struct NymClient { self_address: String, - client_input: Arc, + client_input: Arc>, client_state: Arc, // keep track of the "old" topology for the purposes of node tester @@ -196,7 +197,7 @@ impl NymClientBuilder { Ok(NymClient { self_address, - client_input: Arc::new(client_input), + client_input: Arc::new(RwLock::new(client_input)), client_state: Arc::new(started_client.client_state), _full_topology: None, // this cannot failed as we haven't passed an external task manager diff --git a/wasm/client/src/helpers.rs b/wasm/client/src/helpers.rs index e8570d02b9..92951e49d4 100644 --- a/wasm/client/src/helpers.rs +++ b/wasm/client/src/helpers.rs @@ -1,8 +1,10 @@ // Copyright 2022-2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +use futures::SinkExt; use js_sys::Promise; use std::sync::Arc; +use tokio::sync::RwLock; use wasm_bindgen::JsValue; use wasm_bindgen_futures::future_to_promise; use wasm_client_core::client::base_client::{ClientInput, ClientState}; @@ -48,10 +50,11 @@ pub(crate) trait InputSender { fn send_messages(&self, messages: Vec) -> Promise; } -impl InputSender for Arc { +impl InputSender for Arc> { fn send_message(&self, message: InputMessage) -> Promise { let this = Arc::clone(self); future_to_promise(async move { + let mut this = this.write().await; match this.input_sender.send(message).await { Ok(_) => Ok(JsValue::null()), Err(_) => Err(simple_js_error( @@ -64,6 +67,7 @@ impl InputSender for Arc { fn send_messages(&self, messages: Vec) -> Promise { let this = Arc::clone(self); future_to_promise(async move { + let mut this = this.write().await; for message in messages { if this.input_sender.send(message).await.is_err() { return Err(simple_js_error( diff --git a/wasm/mix-fetch/Cargo.toml b/wasm/mix-fetch/Cargo.toml index fd187f04e1..f64d2001e1 100644 --- a/wasm/mix-fetch/Cargo.toml +++ b/wasm/mix-fetch/Cargo.toml @@ -24,7 +24,7 @@ tokio = { workspace = true, features = ["sync"] } url = { workspace = true } wasm-bindgen = { workspace = true } wasm-bindgen-futures = { workspace = true } -thiserror = { workspace = true } +thiserror = { workspace = true } tsify = { workspace = true, features = ["js"] } nym-bin-common = { path = "../../common/bin-common" } diff --git a/wasm/mix-fetch/src/client.rs b/wasm/mix-fetch/src/client.rs index 9e4bff5e32..55bc92580d 100644 --- a/wasm/mix-fetch/src/client.rs +++ b/wasm/mix-fetch/src/client.rs @@ -8,10 +8,13 @@ use crate::go_bridge::goWasmSetMixFetchRequestTimeout; use crate::request_writer::RequestWriter; use crate::socks_helpers::{socks5_connect_request, socks5_data_request}; use crate::{config, RequestId}; +use futures::SinkExt; use js_sys::Promise; use nym_socks5_requests::RemoteAddress; use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; use tokio::sync::Mutex; +use tokio::sync::RwLock; use wasm_bindgen::prelude::*; use wasm_bindgen_futures::future_to_promise; use wasm_client_core::client::base_client::{BaseClientBuilder, ClientInput, ClientOutput}; @@ -34,7 +37,7 @@ pub struct MixFetchClient { self_address: Recipient, - client_input: ClientInput, + client_input: Arc>, requests: ActiveRequests, @@ -131,7 +134,7 @@ impl MixFetchClientBuilder { invalidated: AtomicBool::new(false), mix_fetch_config: self.config.mix_fetch, self_address, - client_input, + client_input: Arc::new(RwLock::new(client_input)), requests: active_requests, // this cannot failed as we haven't passed an external task manager _task_manager: Mutex::new(started_client.task_handle.try_into_task_manager().unwrap()), @@ -208,6 +211,8 @@ impl MixFetchClient { // the expect here is fine as it implies an unrecoverable failure since one of the client core // tasks has terminated self.client_input + .write() + .await .input_sender .send(input) .await @@ -235,6 +240,8 @@ impl MixFetchClient { // the expect here is fine as it implies an unrecoverable failure since one of the client core // tasks has terminated self.client_input + .write() + .await .input_sender .send(input) .await