GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

gateway-client: disable shutdown signalling for wasm

Browse Source
This commit is contained in:
Jon Häggblad
2022-09-09 15:00:13 +02:00
parent 351adb7f7b
commit f3ed0bb11f
4 changed files with 67 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/client-libs/gateway-client/Cargo.toml
+3 -1
View File
@@ -27,7 +27,6 @@ gateway-requests = { path = "../../../gateway/gateway-requests" }
network-defaults = { path = "../../network-defaults" }
nymsphinx = { path = "../../nymsphinx" }
pemstore = { path = "../../pemstore" }
task = { path = "../../task" }
validator-client = { path = "../validator-client", optional = true }
[dependencies.tungstenite]
@@ -49,6 +48,9 @@ version = "0.14"
[target."cfg(not(target_arch = \"wasm32\"))".dependencies.credential-storage]
path = "../../credential-storage"
[target."cfg(not(target_arch = \"wasm32\"))".dependencies.task]
path = "../../task"
# wasm-only dependencies
[target."cfg(target_arch = \"wasm32\")".dependencies.wasm-bindgen]
version = "0.2"
common/client-libs/gateway-client/src/client.rs
+38 -12
View File
@@ -30,6 +30,7 @@ use rand::rngs::OsRng;
use std::convert::TryFrom;
use std::sync::Arc;
use std::time::Duration;
#[cfg(not(target_arch = "wasm32"))]
use task::ShutdownListener;
use tungstenite::protocol::Message;
@@ -67,6 +68,7 @@ pub struct GatewayClient {
/// Delay between each subsequent reconnection attempt.
reconnection_backoff: Duration,
#[cfg(not(target_arch = "wasm32"))]
/// Listen to shutdown messages.
shutdown: Option<ShutdownListener>,
}
@@ -84,7 +86,7 @@ impl GatewayClient {
ack_sender: AcknowledgementSender,
response_timeout_duration: Duration,
bandwidth_controller: Option<BandwidthController<PersistentStorage>>,
shutdown: Option<ShutdownListener>,
#[cfg(not(target_arch = "wasm32"))] shutdown: Option<ShutdownListener>,
) -> Self {
GatewayClient {
authenticated: false,
@@ -96,12 +98,18 @@ impl GatewayClient {
local_identity,
shared_key,
connection: SocketState::NotConnected,
packet_router: PacketRouter::new(ack_sender, mixnet_message_sender, shutdown.clone()),
packet_router: PacketRouter::new(
ack_sender,
mixnet_message_sender,
#[cfg(not(target_arch = "wasm32"))]
shutdown.clone(),
),
response_timeout_duration,
bandwidth_controller,
should_reconnect_on_failure: true,
reconnection_attempts: DEFAULT_RECONNECTION_ATTEMPTS,
reconnection_backoff: DEFAULT_RECONNECTION_BACKOFF,
#[cfg(not(target_arch = "wasm32"))]
shutdown,
}
}
@@ -129,7 +137,7 @@ impl GatewayClient {
gateway_owner: String,
local_identity: Arc<identity::KeyPair>,
response_timeout_duration: Duration,
shutdown: Option<ShutdownListener>,
#[cfg(not(target_arch = "wasm32"))] shutdown: Option<ShutdownListener>,
) -> Self {
use futures::channel::mpsc;
@@ -137,7 +145,12 @@ impl GatewayClient {
// perfectly fine here, because it's not meant to be used
let (ack_tx, _) = mpsc::unbounded();
let (mix_tx, _) = mpsc::unbounded();
let packet_router = PacketRouter::new(ack_tx, mix_tx, shutdown.clone());
let packet_router = PacketRouter::new(
ack_tx,
mix_tx,
#[cfg(not(target_arch = "wasm32"))]
shutdown.clone(),
);
GatewayClient {
authenticated: false,
@@ -155,6 +168,7 @@ impl GatewayClient {
should_reconnect_on_failure: false,
reconnection_attempts: DEFAULT_RECONNECTION_ATTEMPTS,
reconnection_backoff: DEFAULT_RECONNECTION_BACKOFF,
#[cfg(not(target_arch = "wasm32"))]
shutdown,
}
}
@@ -287,16 +301,27 @@ impl GatewayClient {
let mut fused_timeout = timeout.fuse();
let mut fused_stream = conn.fuse();
// Bit of an ugly workaround for selecting on an `Option`. The unwrap is lazy so we use
// this bool inside the `select` to guard against unwrapping in the `None` case.
let shutdown_is_some = self.shutdown.is_some();
let shutdown_recv_lazy = async { self.shutdown.clone().unwrap().recv().await };
tokio::pin!(shutdown_recv_lazy);
// Bit of an ugly workaround for selecting on an `Option` without having access to
// `tokio::select`
#[cfg(not(target_arch = "wasm32"))]
let shutdown = {
let m_shutdown = self.shutdown.clone();
async {
if let Some(mut s) = m_shutdown {
s.recv().await
}
}
.fuse()
};
#[cfg(not(target_arch = "wasm32"))]
tokio::pin!(shutdown);
#[cfg(target_arch = "wasm32")]
let mut shutdown = Box::pin(async {}.fuse());
loop {
tokio::select! {
biased;
_ = &mut shutdown_recv_lazy, if shutdown_is_some => {
futures::select! {
_ = shutdown => {
log::trace!("GatewayClient control response: Received shutdown");
log::debug!("GatewayClient control response: Exiting");
break Err(GatewayClientError::ConnectionClosedGatewayShutdown);
@@ -745,6 +770,7 @@ impl GatewayClient {
.as_ref()
.expect("no shared key present even though we're authenticated!"),
),
#[cfg(not(target_arch = "wasm32"))]
self.shutdown.clone(),
)
}
common/client-libs/gateway-client/src/packet_router.rs
+5 -1
View File
@@ -8,6 +8,7 @@ use futures::channel::mpsc;
use log::*;
use nymsphinx::addressing::nodes::MAX_NODE_ADDRESS_UNPADDED_LEN;
use nymsphinx::params::packet_sizes::PacketSize;
#[cfg(not(target_arch = "wasm32"))]
use task::ShutdownListener;
use crate::error::GatewayClientError;
@@ -22,6 +23,7 @@ pub type AcknowledgementReceiver = mpsc::UnboundedReceiver<Vec<Vec<u8>>>;
pub struct PacketRouter {
ack_sender: AcknowledgementSender,
mixnet_message_sender: MixnetMessageSender,
#[cfg(not(target_arch = "wasm32"))]
shutdown: Option<ShutdownListener>,
}
@@ -29,11 +31,12 @@ impl PacketRouter {
pub fn new(
ack_sender: AcknowledgementSender,
mixnet_message_sender: MixnetMessageSender,
shutdown: Option<ShutdownListener>,
#[cfg(not(target_arch = "wasm32"))] shutdown: Option<ShutdownListener>,
) -> Self {
PacketRouter {
ack_sender,
mixnet_message_sender,
#[cfg(not(target_arch = "wasm32"))]
shutdown,
}
}
@@ -72,6 +75,7 @@ impl PacketRouter {
if !received_messages.is_empty() {
trace!("routing 'real'");
if let Err(err) = self.mixnet_message_sender.unbounded_send(received_messages) {
#[cfg(not(target_arch = "wasm32"))]
if let Some(shutdown) = &mut self.shutdown {
if shutdown.is_shutdown_poll() {
// This should ideally not happen, but it's ok
common/client-libs/gateway-client/src/socket_state.rs
+21 -9
View File
@@ -11,6 +11,7 @@ use gateway_requests::registration::handshake::SharedKeys;
use gateway_requests::BinaryResponse;
use log::*;
use std::sync::Arc;
#[cfg(not(target_arch = "wasm32"))]
use task::ShutdownListener;
use tungstenite::Message;
@@ -90,7 +91,7 @@ impl PartiallyDelegated {
conn: WsConn,
packet_router: PacketRouter,
shared_key: Arc<SharedKeys>,
shutdown: Option<ShutdownListener>,
#[cfg(not(target_arch = "wasm32"))] shutdown: Option<ShutdownListener>,
) -> Self {
// when called for, it NEEDS TO yield back the stream so that we could merge it and
// read control request responses.
@@ -104,16 +105,27 @@ impl PartiallyDelegated {
let mut fused_stream = (&mut stream).fuse();
let mut packet_router = packet_router;
// Bit of an ugly workaround for selecting on an `Option`. The unwrap is lazy so we use
// this bool inside the `select` to guard against unwrapping in the `None` case.
let shutdown_is_some = shutdown.is_some();
let shutdown_recv_lazy = async { shutdown.unwrap().recv().await };
tokio::pin!(shutdown_recv_lazy);
// Bit of an ugly workaround for selecting on an `Option` without having access to
// `tokio::select`
#[cfg(not(target_arch = "wasm32"))]
let shutdown = {
let m_shutdown = shutdown.clone();
async {
if let Some(mut s) = m_shutdown {
s.recv().await
}
}
.fuse()
};
#[cfg(not(target_arch = "wasm32"))]
tokio::pin!(shutdown);
#[cfg(target_arch = "wasm32")]
let mut shutdown = Box::pin(async {}.fuse());
let ret_err = loop {
tokio::select! {
biased;
_ = &mut shutdown_recv_lazy, if shutdown_is_some => {
futures::select! {
_ = shutdown => {
log::trace!("GatewayClient listener: Received shutdown");
log::debug!("GatewayClient listener: Exiting");
return;