From 7b2f8a4ed1b95b64bba0716c78fe52e788a7e436 Mon Sep 17 00:00:00 2001 From: durch Date: Fri, 14 Jun 2024 11:23:36 +0200 Subject: [PATCH] WASM changes --- Cargo.lock | 1 + common/client-core/src/client/inbound_messages.rs | 2 +- common/socks5/proxy-helpers/src/ordered_sender.rs | 4 ++-- wasm/client/Cargo.toml | 1 + wasm/client/src/client.rs | 5 +++-- wasm/client/src/helpers.rs | 6 +++++- wasm/mix-fetch/src/client.rs | 11 +++++++++-- 7 files changed, 22 insertions(+), 8 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index cba59cf03d..f0f2c68eaa 100644 --- a/Cargo.lock +++ b/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", diff --git a/common/client-core/src/client/inbound_messages.rs b/common/client-core/src/client/inbound_messages.rs index a859f7aa3b..9ea8f270fe 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 2377ddcb6d..c9ef8f3f1f 100644 --- a/wasm/client/Cargo.toml +++ b/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 } diff --git a/wasm/client/src/client.rs b/wasm/client/src/client.rs index 7d6df14b35..ebeb5e6cfb 100644 --- a/wasm/client/src/client.rs +++ b/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; #[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 @@ -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, diff --git a/wasm/client/src/helpers.rs b/wasm/client/src/helpers.rs index d568d76250..c96b634759 100644 --- a/wasm/client/src/helpers.rs +++ b/wasm/client/src/helpers.rs @@ -1,6 +1,7 @@ // Copyright 2022-2023 - Nym Technologies SA // 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) -> 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/src/client.rs b/wasm/mix-fetch/src/client.rs index 6fd5d642ac..51aba4d0bb 100644 --- a/wasm/mix-fetch/src/client.rs +++ b/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>, 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