diff --git a/host/src/connection.rs b/host/src/connection.rs index ad7c330..453588b 100644 --- a/host/src/connection.rs +++ b/host/src/connection.rs @@ -3,6 +3,7 @@ use mtp_codec::{CommunicationType, DataType, DataValue}; use mtp_codec::{CommunicationValue, Version, registry::VersionedCodec}; use mtp_common::CommunicationError; use std::net::SocketAddr; +#[cfg(feature = "pipes")] use std::sync::Arc; #[cfg(feature = "pipes")] use tokio::sync::{Mutex, mpsc}; @@ -11,6 +12,7 @@ use tokio::sync::{Mutex, mpsc}; use crate::error::random_client_id; #[cfg(feature = "pipes")] use crate::pipe::{PipeDispatcher, PipeReceiver, PipeRequest, PipeSender, run_dispatcher}; +#[cfg(feature = "pipes")] use mtp_transport::Policy; mod connection_capability { @@ -161,6 +163,9 @@ where } /// Construct an MTP connection with an explicit policy for pipe dispatch. + // The shared transport constructor keeps its argument order aligned with + // `from_transport_parts_with_remote_addr`; policy is required only here. + #[allow(clippy::too_many_arguments)] pub fn from_transport_parts_with_policy( version: Version, codec: VersionedCodec, @@ -255,35 +260,6 @@ impl MTPConnection { client_public_key: None, } } - - pub fn from_transport_parts_with_policy( - version: Version, - codec: VersionedCodec, - sender: S, - receiver: R, - path: String, - description: Option, - remote_addr: Option, - _policy: Arc, - ) -> Self { - Self { - version, - codec, - sender, - receiver, - path, - remote_addr, - description, - _pipe_stream: std::marker::PhantomData, - _dispatcher_task: tokio::spawn(async {}), - #[cfg(feature = "crypto")] - auth_state: crate::error::AuthState::Unauthenticated, - #[cfg(feature = "crypto")] - client_id: random_client_id(), - #[cfg(feature = "crypto")] - client_public_key: None, - } - } } #[cfg(not(feature = "pipes"))] diff --git a/host/src/engine.rs b/host/src/engine.rs index ce94111..7d6464e 100644 --- a/host/src/engine.rs +++ b/host/src/engine.rs @@ -33,8 +33,7 @@ pub trait HandshakeSender: Send + Sync { pub trait HandshakeReceiver: Send + Sync { fn receive( &self, - ) -> impl std::future::Future> - + Send; + ) -> impl std::future::Future> + Send; } /// The result of a successful handshake, containing everything needed to @@ -136,7 +135,10 @@ impl HandshakeEngine { } }; - let negotiated = match self.registry.negotiate(std::slice::from_ref(&client_version)) { + let negotiated = match self + .registry + .negotiate(std::slice::from_ref(&client_version)) + { Some(v) => v, None => { send_rejection_generic( @@ -300,24 +302,23 @@ impl HandshakeEngine { _ => 0, }; - if cid > 0 { - if let Some(bundle) = + if cid > 0 + && let Some(bundle) = (self.config.get_existing_client)(cid, description.clone()).await - { - return self - .complete_auth_handshake( - sender, - receiver, - Flow::Login { id: cid, bundle }, - CommunicationType::IdentificationResponse, - &negotiated, - &codec, - description, - version_str, - client_version, - ) - .await; - } + { + return self + .complete_auth_handshake( + sender, + receiver, + Flow::Login { id: cid, bundle }, + CommunicationType::IdentificationResponse, + &negotiated, + &codec, + description, + version_str, + client_version, + ) + .await; } // Unknown or zero ID: fall back to guest @@ -634,8 +635,8 @@ impl HandshakeEngine { DataValue::UnsignedNumber(client_nonce), ) .add_typed_default(DataType::Signature, DataValue::Bytes(host_sig)); - response = response - .add_typed_default(DataType::Version, DataValue::Str(negotiated.to_string())); + response = + response.add_typed_default(DataType::Version, DataValue::Str(negotiated.to_string())); if pq_enabled { response = response.add_typed_default(DataType::PqSignature, DataValue::Bytes(host_pq_sig)); @@ -722,19 +723,15 @@ fn extract_register_bundle( msg: &CommunicationValue, ) -> Result { match msg.get_data(DataType::PublicKeys) { - DataValue::Bytes(b) => mtp_crypto::PublicKeyBundle::from_bytes(b).map_err(|_| { - AcceptError::AuthenticationFailed("invalid public key bundle".into()) - }), + DataValue::Bytes(b) => mtp_crypto::PublicKeyBundle::from_bytes(b) + .map_err(|_| AcceptError::AuthenticationFailed("invalid public key bundle".into())), _ => Err(AcceptError::AuthenticationFailed( "missing public keys".into(), )), } } -async fn send_rejection_generic( - sender: &S, - reason: RejectionReason, -) { +async fn send_rejection_generic(sender: &S, reason: RejectionReason) { let response = match &reason { RejectionReason::BadVersion { supported_versions } => { CommunicationValue::new(CommunicationType::ErrorBadVersion) @@ -815,7 +812,9 @@ impl HandshakeSender for mtp_transport::G } } -impl HandshakeReceiver for mtp_transport::GenericReceiver { +impl HandshakeReceiver + for mtp_transport::GenericReceiver +{ fn receive( &self, ) -> impl std::future::Future> + Send diff --git a/host/src/handshake.rs b/host/src/handshake.rs index 328de89..32a23fb 100644 --- a/host/src/handshake.rs +++ b/host/src/handshake.rs @@ -1,7 +1,10 @@ -#[cfg(not(feature = "crypto"))] -use mtp_codec::{Version, registry::{Registry, VersionedCodec}}; #[cfg(feature = "crypto")] use mtp_codec::registry::Registry; +#[cfg(not(feature = "crypto"))] +use mtp_codec::{ + Version, + registry::{Registry, VersionedCodec}, +}; use mtp_transport::{Receiver, Sender}; use std::sync::Arc; use std::time::Instant; @@ -136,9 +139,7 @@ impl HandshakeContext { #[cfg(feature = "crypto")] { Ok(Some(self.connection_from_handshake_result( - sender, - receiver, - result, + sender, receiver, result, ))) } #[cfg(not(feature = "crypto"))] diff --git a/host/src/lib.rs b/host/src/lib.rs index 211b3a6..c071db2 100644 --- a/host/src/lib.rs +++ b/host/src/lib.rs @@ -11,7 +11,7 @@ pub use MTPHost as Host; pub use config::HostConfig; pub use config::Policy; pub use connection::{MTPConnection, MtpReceiverLike, MtpSenderLike}; -pub use engine::{HandshakeEngine, HandshakeResult, HandshakeReceiver, HandshakeSender}; +pub use engine::{HandshakeEngine, HandshakeReceiver, HandshakeResult, HandshakeSender}; pub use error::AcceptError; pub use handshake::MTPHost; pub use mtp_transport::Receiver; diff --git a/mtp-webserver/src/server.rs b/mtp-webserver/src/server.rs index b8f9dfd..4edc521 100644 --- a/mtp-webserver/src/server.rs +++ b/mtp-webserver/src/server.rs @@ -352,9 +352,11 @@ fn build_endpoint( transport.keep_alive_interval(Some(keep_alive)); } if let Some(idle_timeout) = config.policy.max_idle_timeout { - transport.max_idle_timeout(Some(idle_timeout.try_into().map_err( - |error| CommunicationError::Other(format!("{error}")), - )?)); + transport.max_idle_timeout(Some( + idle_timeout + .try_into() + .map_err(|error| CommunicationError::Other(format!("{error}")))?, + )); } Arc::new(transport) }); diff --git a/mtp-webserver/src/transport.rs b/mtp-webserver/src/transport.rs index 1b2eb42..f96aa88 100644 --- a/mtp-webserver/src/transport.rs +++ b/mtp-webserver/src/transport.rs @@ -267,17 +267,13 @@ pub(crate) async fn accept_web_connection( send_pongs: bool, policy: Policy, host_config: Arc, - auth_semaphore: Arc, + _auth_semaphore: Arc, ) -> Result { #[cfg(feature = "crypto")] { - let permit = auth_semaphore - .clone() - .acquire_owned() - .await - .map_err(|_| { - AcceptError::AuthenticationFailed("authentication service stopped".into()) - })?; + let permit = _auth_semaphore.clone().acquire_owned().await.map_err(|_| { + AcceptError::AuthenticationFailed("authentication service stopped".into()) + })?; let result = tokio::time::timeout( host_config.auth_timeout, accept_web_connection_inner(session, path, quinn, send_pongs, policy, host_config), @@ -327,8 +323,20 @@ async fn accept_web_connection_inner( if send_pongs { receiver.respond_to_pings(sender.clone()).await; } + #[cfg(feature = "pipes")] + let connection: WebMTPConnection = mtp_host::MTPConnection::from_transport_parts_with_policy( + negotiated, + codec, + sender, + receiver, + path, + description.clone(), + Some(remote_addr), + policy, + ); + #[cfg(not(feature = "pipes"))] let connection: WebMTPConnection = - mtp_host::MTPConnection::from_transport_parts_with_policy( + mtp_host::MTPConnection::from_transport_parts_with_remote_addr( negotiated, codec, sender, @@ -336,12 +344,11 @@ async fn accept_web_connection_inner( path, description.clone(), Some(remote_addr), - policy, ); #[cfg(not(feature = "crypto"))] { connection.receiver.set_max_message_size(max_message_size); - return Ok(connection); + Ok(connection) } #[cfg(feature = "crypto")] let mut connection = connection; @@ -364,13 +371,14 @@ async fn accept_web_connection_inner( // Unauthenticated: send accepted response with guest ID (or ID 0) if !is_allow_auth && !is_force_auth { - let response = mtp_codec::CommunicationValue::new(CommunicationType::IdentificationResponse) - .add_typed_default(DataType::Connected, DataValue::BoolTrue) - .add_typed_default( - DataType::Version, - DataValue::Str(connection.version.to_string()), - ) - .add_typed_default(DataType::Id, DataValue::UnsignedNumber(0)); + let response = + mtp_codec::CommunicationValue::new(CommunicationType::IdentificationResponse) + .add_typed_default(DataType::Connected, DataValue::BoolTrue) + .add_typed_default( + DataType::Version, + DataValue::Str(connection.version.to_string()), + ) + .add_typed_default(DataType::Id, DataValue::UnsignedNumber(0)); connection .sender .send(&response) @@ -405,16 +413,31 @@ async fn accept_web_connection_inner( if let Some(bundle) = (_host_config.get_existing_client)(id, description.clone()).await { - (id, Some(bundle), CommunicationType::IdentificationResponse, false) + ( + id, + Some(bundle), + CommunicationType::IdentificationResponse, + false, + ) } else { // Unknown client: fall back to guest let guest_id = H3TransportConnection::assign_guest_id(&_host_config).await?; - (guest_id, None, CommunicationType::IdentificationResponse, true) + ( + guest_id, + None, + CommunicationType::IdentificationResponse, + true, + ) } } else { // ID zero or missing: fall back to guest let guest_id = H3TransportConnection::assign_guest_id(&_host_config).await?; - (guest_id, None, CommunicationType::IdentificationResponse, true) + ( + guest_id, + None, + CommunicationType::IdentificationResponse, + true, + ) } } else if first_type_opt == reg_type { // Registration: always authenticate (both AllowAuth and ForceAuth) @@ -442,7 +465,12 @@ async fn accept_web_connection_inner( let bundle = (_host_config.get_existing_client)(id, description.clone()) .await .ok_or_else(|| AcceptError::AuthenticationFailed("unknown client id".into()))?; - (id, Some(bundle), CommunicationType::IdentificationResponse, false) + ( + id, + Some(bundle), + CommunicationType::IdentificationResponse, + false, + ) } else { return Err(AcceptError::AuthenticationFailed( "unexpected authentication message".into(), @@ -453,10 +481,14 @@ async fn accept_web_connection_inner( // Guest path: skip challenge/response, send accepted with guest ID if is_guest { let guest_id = client_id; - let response = mtp_codec::CommunicationValue::new(CommunicationType::IdentificationResponse) - .add_typed_default(DataType::Connected, DataValue::BoolTrue) - .add_typed_default(DataType::Version, DataValue::Str(connection.version.to_string())) - .add_typed_default(DataType::Id, DataValue::UnsignedNumber(guest_id as u128)); + let response = + mtp_codec::CommunicationValue::new(CommunicationType::IdentificationResponse) + .add_typed_default(DataType::Connected, DataValue::BoolTrue) + .add_typed_default( + DataType::Version, + DataValue::Str(connection.version.to_string()), + ) + .add_typed_default(DataType::Id, DataValue::UnsignedNumber(guest_id as u128)); connection .sender .send(&response) @@ -513,14 +545,11 @@ async fn accept_web_connection_inner( let host_config = _host_config.clone(); let pq_signer = host_pq_signer.clone(); async move { - let signer = - Ed25519Signer::new(&host_config.host_keyring.sig_cl_secret_key) - .map_err(|e| AcceptError::AuthenticationFailed(e.to_string()))?; + let signer = Ed25519Signer::new(&host_config.host_keyring.sig_cl_secret_key) + .map_err(|e| AcceptError::AuthenticationFailed(e.to_string()))?; if let Some(pq_signer) = pq_signer { mtp_crypto::sign_parallel::sign_dual_parallel_shared_pq( - signer, - pq_signer, - payload, + signer, pq_signer, payload, ) .await .map_err(|e| AcceptError::AuthenticationFailed(e.to_string())) @@ -535,21 +564,20 @@ async fn accept_web_connection_inner( let sign_challenge_started = Instant::now(); let (sig, pq_sig) = host_sign(auth::challenge_payload(client_id, server_challenge)).await?; tracing::debug!(elapsed = ?sign_challenge_started.elapsed(), "web authentication handshake: sign challenge"); - let mut challenge = - mtp_codec::CommunicationValue::new(CommunicationType::Challenge) - .add_typed_default( - DataType::ServerNonce, - DataValue::UnsignedNumber(server_challenge), - ) - .add_typed_default(DataType::Signature, DataValue::Bytes(sig)) - .add_typed_default( - DataType::RequirePq, - if _host_config.require_pq { - DataValue::BoolTrue - } else { - DataValue::BoolFalse - }, - ); + let mut challenge = mtp_codec::CommunicationValue::new(CommunicationType::Challenge) + .add_typed_default( + DataType::ServerNonce, + DataValue::UnsignedNumber(server_challenge), + ) + .add_typed_default(DataType::Signature, DataValue::Bytes(sig)) + .add_typed_default( + DataType::RequirePq, + if _host_config.require_pq { + DataValue::BoolTrue + } else { + DataValue::BoolFalse + }, + ); if pq_enabled { challenge = challenge.add_typed_default(DataType::PqSignature, DataValue::Bytes(pq_sig)); @@ -579,8 +607,7 @@ async fn accept_web_connection_inner( } }; tracing::debug!(elapsed = ?receive_proof_started.elapsed(), "web authentication handshake: receive client proof"); - if Some(proof.get_type()) != CommunicationType::ChallengeResponse.try_to_id(&tm) - { + if Some(proof.get_type()) != CommunicationType::ChallengeResponse.try_to_id(&tm) { return Err(AcceptError::AuthenticationFailed( "missing challenge response".into(), )); @@ -605,11 +632,7 @@ async fn accept_web_connection_inner( DataValue::Bytes(bytes) => bytes.as_slice(), _ => &[], }; - let payload = if first.get_type() - == CommunicationType::Register - .try_to_id(&tm) - .unwrap() - { + let payload = if first.get_type() == CommunicationType::Register.try_to_id(&tm).unwrap() { auth::register_proof_payload( &version.to_string(), &client_bundle.as_bytes(), @@ -643,9 +666,13 @@ async fn accept_web_connection_inner( tracing::debug!(elapsed = ?verify_proof_started.elapsed(), "web authentication handshake: verify client proof"); if !proof_ok { - let rejection = mtp_codec::CommunicationValue::new(CommunicationType::IdentificationResponse) - .add_typed_default(DataType::Connected, DataValue::BoolFalse) - .add_typed_default(DataType::ErrorMessage, DataValue::Str("client proof signature invalid".into())); + let rejection = + mtp_codec::CommunicationValue::new(CommunicationType::IdentificationResponse) + .add_typed_default(DataType::Connected, DataValue::BoolFalse) + .add_typed_default( + DataType::ErrorMessage, + DataValue::Str("client proof signature invalid".into()), + ); let _ = connection.sender.send(&rejection).await; connection.sender.close(); return Err(AcceptError::AuthenticationFailed( @@ -698,5 +725,3 @@ async fn accept_web_connection_inner( Ok(connection) } } - - diff --git a/src/sdk/index.ts b/src/sdk/index.ts index 4144178..9123c18 100644 --- a/src/sdk/index.ts +++ b/src/sdk/index.ts @@ -857,6 +857,10 @@ export class MTPClient { return this.raw.client.state; } + get pingMs(): number | null { + return this.raw.client.ping_ms ?? null; + } + async #loadStoredCredentials() { if (this.#credentials || !this.#options.storage) { return; diff --git a/transport/src/generic_connection.rs b/transport/src/generic_connection.rs index 7dfb960..1855c90 100644 --- a/transport/src/generic_connection.rs +++ b/transport/src/generic_connection.rs @@ -198,7 +198,9 @@ impl GenericReceiver { let (pipe_tx, pipe_rx) = mpsc::channel(policy.receiver_queue_capacity); let ping_sender: Arc>>> = Arc::new(RwLock::new(None)); let max_message_size = Arc::new(AtomicU64::new( - policy.handshake_max_message_size.min(policy.max_message_size), + policy + .handshake_max_message_size + .min(policy.max_message_size), )); let task_ping_sender = ping_sender.clone(); let task_connection = connection.clone(); @@ -214,8 +216,8 @@ impl GenericReceiver { loop { // Backpressure: stop accepting new streams if the output queue is full. #[cfg(feature = "pipes")] - let cap_full = task_accept_task_tx.capacity() == 0 - || task_accept_task_pipe_tx.capacity() == 0; + let cap_full = + task_accept_task_tx.capacity() == 0 || task_accept_task_pipe_tx.capacity() == 0; #[cfg(not(feature = "pipes"))] let cap_full = task_accept_task_tx.capacity() == 0; @@ -269,10 +271,7 @@ impl GenericReceiver { { let close_error = CommunicationError::StreamError; let _ = tx.send(Err(close_error.clone())).await; - connection.close( - policy.application_close_code, - b"max frames exceeded", - ); + connection.close(policy.application_close_code, b"max frames exceeded"); break; } let mut len = [0; 4]; @@ -291,12 +290,12 @@ impl GenericReceiver { break; } Err(_) => { - tracing::warn!("MTP receive stream timed out while reading frame header"); - let _ = tx.send(Err(CommunicationError::StreamError)).await; - connection.close( - policy.application_close_code, - b"frame header timeout", + tracing::warn!( + "MTP receive stream timed out while reading frame header" ); + let _ = tx.send(Err(CommunicationError::StreamError)).await; + connection + .close(policy.application_close_code, b"frame header timeout"); break; } } @@ -308,10 +307,7 @@ impl GenericReceiver { if len as u64 > frame_limit { tracing::warn!(len, "MTP receive stream frame is too large"); let _ = tx.send(Err(CommunicationError::MessageTooLarge)).await; - connection.close( - policy.application_close_code, - b"frame too large", - ); + connection.close(policy.application_close_code, b"frame too large"); break; } let target_len = len as usize; @@ -322,10 +318,8 @@ impl GenericReceiver { "MTP receive stream could not reserve frame body" ); let _ = tx.send(Err(CommunicationError::MessageTooLarge)).await; - connection.close( - policy.application_close_code, - b"frame allocation failed", - ); + connection + .close(policy.application_close_code, b"frame allocation failed"); break; } while body.len() < target_len { @@ -345,10 +339,8 @@ impl GenericReceiver { "MTP receive stream failed while reading frame body" ); let _ = tx.send(Err(CommunicationError::StreamError)).await; - connection.close( - policy.application_close_code, - b"frame body read error", - ); + connection + .close(policy.application_close_code, b"frame body read error"); break; } body.extend_from_slice(&chunk[..chunk_len]); @@ -361,12 +353,10 @@ impl GenericReceiver { Ok(message) => message, Err(_) => { tracing::warn!("MTP receive stream contained an invalid frame"); - let _ = tx.send(Err(CommunicationError::ParseCommunicationValue)) + let _ = tx + .send(Err(CommunicationError::ParseCommunicationValue)) .await; - connection.close( - policy.application_close_code, - b"invalid frame", - ); + connection.close(policy.application_close_code, b"invalid frame"); break; } }; diff --git a/wasm/src/client.rs b/wasm/src/client.rs index 3c7aba3..91dc57a 100644 --- a/wasm/src/client.rs +++ b/wasm/src/client.rs @@ -44,9 +44,19 @@ fn route_incoming_frame( on_message: &js_sys::Function, subscriptions: &Rc>>, pending_requests: &Rc>>, + pending_pings: &Rc>>, + ping_ms: &Rc>>, ) { let message_type = frame_type(frame); + if message_type.as_deref() == Some("Pong") { + if let Some(sent_at) = frame_id(frame).and_then(|id| pending_pings.borrow_mut().remove(&id)) + { + ping_ms.set(Some(js_sys::Date::now() - sent_at)); + } + return; + } + if let Some(request_id) = frame_id(frame) { let pending = pending_requests.borrow_mut().remove(&request_id); if let Some(pending) = pending { @@ -149,6 +159,8 @@ pub struct WasmClient { next_subscription_id: Rc>, pending_requests: Rc>>, ping_timer: Rc>>, + pending_pings: Rc>>, + ping_ms: Rc>>, pending_pipe_creations: Rc>>>>, pending_pipes: Rc>>>, on_pipe_request: Rc>>, @@ -184,6 +196,8 @@ impl WasmClient { next_subscription_id: Rc::new(Cell::new(1)), pending_requests: Rc::new(RefCell::new(HashMap::new())), ping_timer: Rc::new(RefCell::new(None)), + pending_pings: Rc::new(RefCell::new(HashMap::new())), + ping_ms: Rc::new(Cell::new(None)), pending_pipe_creations: Rc::new(RefCell::new(HashMap::new())), pending_pipes: Rc::new(RefCell::new(HashMap::new())), on_pipe_request: Rc::new(RefCell::new(None)), @@ -200,6 +214,11 @@ impl WasmClient { self.state.get() as u8 } + #[wasm_bindgen(getter)] + pub fn ping_ms(&self) -> Option { + self.ping_ms.get() + } + #[wasm_bindgen] pub async fn connect(&self, config: &ConnectionConfig) -> Result<(), JsValue> { self.set_state(ConnectionState::Connecting); @@ -582,11 +601,17 @@ impl WasmClient { }; let interval_ms = interval_ms.max(1_000) as i32; let on_error = self.on_error.clone(); + let pending_pings = self.pending_pings.clone(); let closure = Closure::wrap(Box::new(move || { let transport = transport.clone(); let on_error = on_error.clone(); + let pending_pings = pending_pings.clone(); wasm_bindgen_futures::spawn_local(async move { - let timestamp = js_sys::Date::now() as u64; + let sent_at = js_sys::Date::now(); + pending_pings + .borrow_mut() + .retain(|_, pending_at| sent_at - *pending_at < interval_ms as f64 * 3.0); + let timestamp = sent_at as u64; let frame = CommunicationValue::new(CommunicationType::Ping) .add_typed_default( DataType::Description, @@ -596,12 +621,16 @@ impl WasmClient { DataType::Timestamp, DataValue::UnsignedNumber(timestamp as u128), ) - .with_sender(client_id) + .with_sender(client_id); + let ping_id = frame.get_id(); + let frame = frame .to_bytes() .map_err(|e| js_error(format!("encode ping failed: {}", e))); match frame { Ok(frame) => { + pending_pings.borrow_mut().insert(ping_id, sent_at); if let Err(error) = transport.send_frame(&frame).await { + pending_pings.borrow_mut().remove(&ping_id); let _ = on_error.call1(&JsValue::NULL, &error); } } @@ -629,6 +658,8 @@ impl WasmClient { #[wasm_bindgen] pub fn stop_protocol_pings(&self) { + self.pending_pings.borrow_mut().clear(); + self.ping_ms.set(None); let Some(timer) = self.ping_timer.borrow_mut().take() else { return; }; @@ -741,6 +772,10 @@ impl WasmClient { let pending_requests = self.pending_requests.clone(); let loop_pending_requests = pending_requests.clone(); let ping_timer = self.ping_timer.clone(); + let pending_pings = self.pending_pings.clone(); + let loop_pending_pings = pending_pings.clone(); + let ping_ms = self.ping_ms.clone(); + let loop_ping_ms = ping_ms.clone(); let pending_pipe_creations = self.pending_pipe_creations.clone(); let pending_pipes = self.pending_pipes.clone(); let on_pipe_request = self.on_pipe_request.clone(); @@ -808,6 +843,8 @@ impl WasmClient { &on_msg, &subscriptions, &loop_pending_requests, + &loop_pending_pings, + &loop_ping_ms, ); }, on_err.clone(), @@ -822,6 +859,8 @@ impl WasmClient { .await; state.set(ConnectionState::Disconnected); stop_ping_timer(&ping_timer); + pending_pings.borrow_mut().clear(); + ping_ms.set(None); reject_pending_requests(&pending_requests, "disconnected"); client_pipe::reject_pending_pipe_creations(&pending_pipe_creations, "disconnected"); }); diff --git a/wasm/types/mtp_wasm.d.ts b/wasm/types/mtp_wasm.d.ts index b2ce979..58c3cda 100644 --- a/wasm/types/mtp_wasm.d.ts +++ b/wasm/types/mtp_wasm.d.ts @@ -121,6 +121,7 @@ export class WasmClient implements DisposableWasmObject { subscribe(message_type: string, callback: MessageCallback): number; unsubscribe(id: number): boolean; static is_supported(): boolean; + readonly ping_ms: number | undefined; readonly state: ConnectionState; }