From 203ef1adccb174cb405658128261db99e9ece5b9 Mon Sep 17 00:00:00 2001 From: Alex Emmet <111742636+Alex-Emmet@users.noreply.github.com> Date: Wed, 15 Jul 2026 01:53:48 +0200 Subject: [PATCH] format --- client/src/lib.rs | 12 ++++-------- client/tests/ping.rs | 9 +++++++-- common/src/lib.rs | 7 +++---- host/src/lib.rs | 17 ++++++----------- transport/src/connection.rs | 18 ++++++++---------- transport/src/pipe.rs | 11 ++++------- transport/tests/integration.rs | 26 +++++++++----------------- wasm/src/client.rs | 14 ++++++++------ wasm/src/pipe.rs | 6 ++---- wasm/src/transport.rs | 11 ++++------- 10 files changed, 55 insertions(+), 76 deletions(-) diff --git a/client/src/lib.rs b/client/src/lib.rs index b9ea6ac..73fd4df 100644 --- a/client/src/lib.rs +++ b/client/src/lib.rs @@ -2,9 +2,9 @@ use mtp_codec::{CommunicationValue, DataType, DataValue, PROTOCOL_VERSION, Versi use mtp_common::CommunicationError; #[cfg(feature = "pipes")] pub use mtp_common::PipeError; +use rand::Rng; use std::collections::HashMap; use std::sync::Arc; -use rand::Rng; use tokio::sync::{Mutex, mpsc}; use tokio::time::{Duration, Instant}; @@ -147,10 +147,7 @@ async fn run_dispatcher( Ok(mtp_transport::TransportEvent::Message(msg)) => { if msg.get_type() == pipe_req_type { let pipe_id = msg.get_id(); - let description = msg - .get_str(DataType::Description) - .unwrap_or("") - .to_string(); + let description = msg.get_str(DataType::Description).unwrap_or("").to_string(); let req = PipeRequest { pipe_id, description, @@ -510,9 +507,8 @@ fn connection_from_parts( let (app_tx, app_rx) = mpsc::channel::>( config.policy.receiver_queue_capacity, ); - let (pipe_req_tx, pipe_req_rx) = mpsc::channel::( - config.policy.receiver_queue_capacity, - ); + let (pipe_req_tx, pipe_req_rx) = + mpsc::channel::(config.policy.receiver_queue_capacity); let dispatcher = Arc::new(PipeDispatcher { pending_creations: Mutex::new(HashMap::new()), diff --git a/client/tests/ping.rs b/client/tests/ping.rs index b071864..e497db9 100644 --- a/client/tests/ping.rs +++ b/client/tests/ping.rs @@ -16,8 +16,13 @@ fn generate_self_signed_cert() -> (Vec, Vec) { async fn start_host(send_pongs: bool) -> (MTPHost, Vec) { let (cert_pem, key_pem) = generate_self_signed_cert(); let host = MTPHost::new( - HostConfig::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0, cert_pem.clone(), key_pem) - .with_pongs(send_pongs), + HostConfig::new( + IpAddr::V4(Ipv4Addr::LOCALHOST), + 0, + cert_pem.clone(), + key_pem, + ) + .with_pongs(send_pongs), ) .await .unwrap(); diff --git a/common/src/lib.rs b/common/src/lib.rs index 773bfca..75739f1 100644 --- a/common/src/lib.rs +++ b/common/src/lib.rs @@ -369,10 +369,9 @@ mod pipe_error_tests { #[test] fn test_pipe_error_from_connection_error() { - let pe: PipeError = CommunicationError::ConnectionError( - wtransport::error::ConnectionError::TimedOut, - ) - .into(); + let pe: PipeError = + CommunicationError::ConnectionError(wtransport::error::ConnectionError::TimedOut) + .into(); assert_eq!(pe, PipeError::ConnectionClosed); } diff --git a/host/src/lib.rs b/host/src/lib.rs index b251cc6..3f51b3d 100644 --- a/host/src/lib.rs +++ b/host/src/lib.rs @@ -5,13 +5,13 @@ use mtp_codec::{ use mtp_common::CommunicationError; #[cfg(feature = "pipes")] pub use mtp_common::PipeError; +#[cfg(feature = "pipes")] +use std::collections::HashMap; use std::net::IpAddr; #[cfg(feature = "crypto")] use std::pin::Pin; -use std::{error::Error, fmt}; -#[cfg(feature = "pipes")] -use std::collections::HashMap; use std::sync::Arc; +use std::{error::Error, fmt}; use tokio::sync::{Mutex, mpsc}; #[cfg(feature = "crypto")] use tokio::time::Duration; @@ -171,10 +171,7 @@ async fn run_dispatcher( Ok(mtp_transport::TransportEvent::Message(msg)) => { if msg.get_type() == pipe_req_type { let pipe_id = msg.get_id(); - let description = msg - .get_str(DataType::Description) - .unwrap_or("") - .to_string(); + let description = msg.get_str(DataType::Description).unwrap_or("").to_string(); let req = PipeRequest { pipe_id, description, @@ -519,8 +516,7 @@ impl MTPHost { receiver.respond_to_pings(sender.clone()); } - let (app_tx, app_rx) = - mpsc::channel(self.config.policy.receiver_queue_capacity); + let (app_tx, app_rx) = mpsc::channel(self.config.policy.receiver_queue_capacity); let (pipe_req_tx, pipe_req_rx) = mpsc::channel(self.config.policy.receiver_queue_capacity); @@ -597,8 +593,7 @@ impl MTPHost { receiver.respond_to_pings(sender.clone()); } - let (app_tx, app_rx) = - mpsc::channel(self.config.policy.receiver_queue_capacity); + let (app_tx, app_rx) = mpsc::channel(self.config.policy.receiver_queue_capacity); let (pipe_req_tx, pipe_req_rx) = mpsc::channel(self.config.policy.receiver_queue_capacity); diff --git a/transport/src/connection.rs b/transport/src/connection.rs index 387b13e..00ddad5 100644 --- a/transport/src/connection.rs +++ b/transport/src/connection.rs @@ -4,10 +4,10 @@ use crate::pipe::PipeReader; use mtp_codec::CommunicationValue; use mtp_common::CommunicationError; use std::sync::Arc; -use tokio::sync::{mpsc, Mutex, Notify, RwLock, Semaphore}; +use tokio::sync::{Mutex, Notify, RwLock, Semaphore, mpsc}; use tokio::time::{Duration, sleep, timeout}; -use wtransport::Connection; use tracing::{debug, info, instrument, trace}; +use wtransport::Connection; #[cfg(feature = "pipes")] #[derive(Debug)] @@ -119,10 +119,7 @@ impl Policy { self } - pub fn with_max_concurrent_stream_tasks( - mut self, - max_concurrent_stream_tasks: usize, - ) -> Self { + pub fn with_max_concurrent_stream_tasks(mut self, max_concurrent_stream_tasks: usize) -> Self { self.max_concurrent_stream_tasks = max_concurrent_stream_tasks; self } @@ -619,9 +616,7 @@ impl Receiver { policy.receiver_queue_capacity, ); #[cfg(feature = "pipes")] - let (pipe_tx, pipe_rx) = mpsc::channel::( - policy.receiver_queue_capacity, - ); + let (pipe_tx, pipe_rx) = mpsc::channel::(policy.receiver_queue_capacity); #[cfg(not(feature = "pipes"))] let (tx, rx) = mpsc::channel::>( policy.receiver_queue_capacity, @@ -659,7 +654,10 @@ impl Receiver { let cap_full = tx.capacity() == 0; if cap_full { - trace!(target = "mtp.transport", "accept loop paused: receiver queue full"); + trace!( + target = "mtp.transport", + "accept loop paused: receiver queue full" + ); tokio::select! { _ = close_rx.changed() => { if close_rx.borrow().is_some() { diff --git a/transport/src/pipe.rs b/transport/src/pipe.rs index 33157e7..75fbb37 100644 --- a/transport/src/pipe.rs +++ b/transport/src/pipe.rs @@ -9,13 +9,10 @@ pub struct PipeWriter { impl PipeWriter { pub async fn finish(mut self) -> Result<(), mtp_common::CommunicationError> { - self.stream - .finish() - .await - .map_err(|e| { - log::warn!("[PipeWriter] finish failed: {e}"); - mtp_common::CommunicationError::StreamWriteError(e) - }) + self.stream.finish().await.map_err(|e| { + log::warn!("[PipeWriter] finish failed: {e}"); + mtp_common::CommunicationError::StreamWriteError(e) + }) } pub fn abort(&mut self) -> Result<(), wtransport::error::ClosedStream> { diff --git a/transport/tests/integration.rs b/transport/tests/integration.rs index 276c975..326bcea 100644 --- a/transport/tests/integration.rs +++ b/transport/tests/integration.rs @@ -228,9 +228,7 @@ async fn test_receiver_backpressure_with_small_queue() { let mut h = start_test_host(cert_pem.clone(), key_pem).await; let url = format!("https://127.0.0.1:{}", h.local_addr().port()); let policy = Policy::default().with_receiver_queue_capacity(1); - let (client_tx, client_rx) = connect(&url, Some(cert_pem), policy.clone()) - .await - .unwrap(); + let (client_tx, client_rx) = connect(&url, Some(cert_pem), policy.clone()).await.unwrap(); let (_host_tx, host_rx) = h.next().await.unwrap(); let tm = TypeMap::latest(); @@ -242,13 +240,10 @@ async fn test_receiver_backpressure_with_small_queue() { } for i in 0..8u128 { - let received = tokio::time::timeout( - std::time::Duration::from_secs(5), - host_rx.receive(), - ) - .await - .unwrap() - .unwrap(); + let received = tokio::time::timeout(std::time::Duration::from_secs(5), host_rx.receive()) + .await + .unwrap() + .unwrap(); assert_numbered_message(&received, CommunicationType::Ping, i, &tm); } @@ -332,13 +327,10 @@ async fn test_semaphore_saturation_with_concurrent_streams() { } for i in 0..6u128 { - let received = tokio::time::timeout( - std::time::Duration::from_secs(5), - host_rx.receive(), - ) - .await - .unwrap() - .unwrap(); + let received = tokio::time::timeout(std::time::Duration::from_secs(5), host_rx.receive()) + .await + .unwrap() + .unwrap(); assert_numbered_message(&received, CommunicationType::Ping, i, &tm); } diff --git a/wasm/src/client.rs b/wasm/src/client.rs index 6da9156..f186960 100644 --- a/wasm/src/client.rs +++ b/wasm/src/client.rs @@ -58,7 +58,10 @@ impl WasmPipeHandle { match accepted { Ok(true) => { - let writer = self.transport.open_pipe(self.pipe_id, &self.description).await?; + let writer = self + .transport + .open_pipe(self.pipe_id, &self.description) + .await?; Ok(JsValue::from(writer)) } Ok(false) => Ok(JsValue::NULL), @@ -783,12 +786,11 @@ impl WasmClient { let pipe_id = random_pipe_id()?; let (tx, rx) = oneshot::channel(); - self.pending_pipe_creations - .borrow_mut() - .insert(pipe_id, tx); + self.pending_pipe_creations.borrow_mut().insert(pipe_id, tx); - let request = - CommunicationValue::new(CommunicationType::PipeRequest).with_id(pipe_id).add_typed_default( + let request = CommunicationValue::new(CommunicationType::PipeRequest) + .with_id(pipe_id) + .add_typed_default( DataType::Description, DataValue::Str(description.to_string()), ); diff --git a/wasm/src/pipe.rs b/wasm/src/pipe.rs index 4cf2bfc..b87c258 100644 --- a/wasm/src/pipe.rs +++ b/wasm/src/pipe.rs @@ -1,5 +1,5 @@ -use wasm_bindgen::prelude::*; use wasm_bindgen::JsCast; +use wasm_bindgen::prelude::*; use wasm_bindgen_futures::JsFuture; use crate::error::js_error; @@ -56,9 +56,7 @@ impl PipeWriter { let close_promise = close_fn .call0(&self.writer) .map_err(|e| js_error(&format!("close failed: {:?}", e)))?; - if let Err(e) = - JsFuture::from(close_promise.unchecked_into::()).await - { + if let Err(e) = JsFuture::from(close_promise.unchecked_into::()).await { crate::transport::log_stream_error_code(&e, "pipe writer close"); } release_writer_lock(&self.writer); diff --git a/wasm/src/transport.rs b/wasm/src/transport.rs index ab30cc5..b0ff763 100644 --- a/wasm/src/transport.rs +++ b/wasm/src/transport.rs @@ -467,8 +467,8 @@ impl WasmTransport { F: FnMut(JsValue), G: FnMut(crate::pipe::PipeReader), { - let pipe_request_type = mtp_codec::CommunicationType::PipeRequest - .to_id(&mtp_codec::TypeMap::latest()); + let pipe_request_type = + mtp_codec::CommunicationType::PipeRequest.to_id(&mtp_codec::TypeMap::latest()); loop { match self.next_frame().await { @@ -509,8 +509,7 @@ impl WasmTransport { } Err(e) => { let message = e.as_string().unwrap_or_else(|| format!("{:?}", e)); - let _ = - on_error.call1(&JsValue::NULL, &JsValue::from_str(&message)); + let _ = on_error.call1(&JsValue::NULL, &JsValue::from_str(&message)); } } } @@ -574,9 +573,7 @@ impl WasmTransport { let write_promise = write_fn .call1(&writer_val, &chunk) .map_err(|e| js_error(&format!("write failed: {:?}", e)))?; - if let Err(e) = - JsFuture::from(write_promise.unchecked_into::()).await - { + if let Err(e) = JsFuture::from(write_promise.unchecked_into::()).await { log_stream_error_code(&e, "open_pipe write"); release_writer_lock(&writer_val); return Err(e);