WASM changes

Cargo.lock

@@ -4255,6 +4255,7 @@ dependencies = [
  "serde-wasm-bindgen 0.6.5",
  "serde_json",
  "thiserror",
+ "tokio",
  "tsify",
  "wasm-bindgen",
  "wasm-bindgen-futures",

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

@@ -1,4 +1,3 @@
-use hyper::body::Buf;
 // Copyright 2020-2023 - Nym Technologies SA <contact@nymtech.net>
 // 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},
 };

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<F, S: Send> {
+pub(crate) struct OrderedMessageSender<F, S: Send + 'static> {
     connection_id: ConnectionId,
     // addresses are provided for better logging
     local_destination_address: String,
@@ -19,7 +19,7 @@ pub(crate) struct OrderedMessageSender<F, S: Send> {
     mix_message_adapter: F,
 }
 
-impl<F, S: Send> OrderedMessageSender<F, S>
+impl<F, S: Send + 'static> OrderedMessageSender<F, S>
 where
     F: Fn(SocketData) -> S,
 {

wasm/client/Cargo.toml

@@ -1,6 +1,9 @@
 [package]
 name = "nym-client-wasm"
-authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>", "Jedrzej Stuczynski <andrew@nymtech.net>"]
+authors = [
+    "Dave Hrycyszyn <futurechimp@users.noreply.github.com>",
+    "Jedrzej Stuczynski <andrew@nymtech.net>",
+]
 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 }

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<ClientInput>,
+    client_input: Arc<RwLock<ClientInput>>,
     client_state: Arc<ClientState>,
 
     // 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

wasm/client/src/helpers.rs

@@ -1,8 +1,10 @@
 // Copyright 2022-2023 - Nym Technologies SA <contact@nymtech.net>
 // 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<InputMessage>) -> Promise;
 }
 
-impl InputSender for Arc<ClientInput> {
+impl InputSender for Arc<RwLock<ClientInput>> {
     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<ClientInput> {
     fn send_messages(&self, messages: Vec<InputMessage>) -> 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(

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" }

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<RwLock<ClientInput>>,
 
     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