WASM changes

Cargo.lock

@@ -5324,6 +5324,7 @@ dependencies = [
  "serde-wasm-bindgen 0.6.5",
  "serde_json",
  "thiserror 2.0.17",
+ "tokio",
  "tokio_with_wasm",
  "tsify",
  "wasm-bindgen",

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

@@ -32,6 +32,7 @@ once_cell = { workspace = true }
 thiserror = { workspace = true }
 tsify = { workspace = true, features = ["js"] }
 web-sys = { workspace = true }
+tokio = { workspace = true, default-features = false, features = ["sync"] }
 
 nym-bin-common = { workspace = true }
 nym-wasm-client-core = { workspace = true }

wasm/client/src/client.rs

@@ -35,6 +35,7 @@ use nym_wasm_utils::{check_promise_result, console_error, console_log};
 use serde::{Deserialize, Serialize};
 use std::sync::Arc;
 use tokio_with_wasm::sync::mpsc;
+use tokio::sync::RwLock;
 use tsify::Tsify;
 use wasm_bindgen::prelude::*;
 use wasm_bindgen_futures::future_to_promise;
@@ -54,7 +55,7 @@ pub type ClientRequestSender = mpsc::Sender<ClientRequest>;
 #[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
@@ -253,7 +254,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,
             _task_manager: started_client.shutdown_handle,

wasm/client/src/helpers.rs

@@ -1,6 +1,7 @@
 // Copyright 2022-2023 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
+use futures::SinkExt;
 use js_sys::Promise;
 use nym_wasm_client_core::client::base_client::{ClientInput, ClientState};
 use nym_wasm_client_core::client::inbound_messages::InputMessage;
@@ -10,6 +11,7 @@ use nym_wasm_client_core::NymTopology;
 use nym_wasm_utils::error::simple_js_error;
 use nym_wasm_utils::{check_promise_result, console_log};
 use std::sync::Arc;
+use tokio::sync::RwLock;
 use wasm_bindgen::JsValue;
 use wasm_bindgen_futures::future_to_promise;
 
@@ -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/src/client.rs

@@ -8,6 +8,7 @@ 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_bin_common::bin_info;
 use nym_socks5_requests::RemoteAddress;
@@ -24,7 +25,9 @@ use nym_wasm_client_core::{IdentityKey, QueryReqwestRpcNyxdClient, Recipient};
 use nym_wasm_utils::console_log;
 use nym_wasm_utils::error::PromisableResult;
 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;
 
@@ -36,7 +39,7 @@ pub struct MixFetchClient {
 
     self_address: Recipient,
 
-    client_input: ClientInput,
+    client_input: Arc<RwLock<ClientInput>>,
 
     requests: ActiveRequests,
 
@@ -185,7 +188,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,
             _shutdown_manager: Mutex::new(started_client.shutdown_handle),
         })
@@ -261,6 +264,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
@@ -288,6 +293,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