format

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
+                    let description = msg.get_str(DataType::Description).unwrap_or("").to_string();
-                        .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::<Result<CommunicationValue, CommunicationError>>(
             config.policy.receiver_queue_capacity,
         );
-        let (pipe_req_tx, pipe_req_rx) = mpsc::channel::<PipeRequest>(
+        let (pipe_req_tx, pipe_req_rx) =
-            config.policy.receiver_queue_capacity,
+            mpsc::channel::<PipeRequest>(config.policy.receiver_queue_capacity);
-        );
 
         let dispatcher = Arc::new(PipeDispatcher {
             pending_creations: Mutex::new(HashMap::new()),

client/tests/ping.rs

@@ -16,7 +16,12 @@ fn generate_self_signed_cert() -> (Vec<u8>, Vec<u8>) {
 async fn start_host(send_pongs: bool) -> (MTPHost, Vec<u8>) {
     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)
+        HostConfig::new(
+            IpAddr::V4(Ipv4Addr::LOCALHOST),
+            0,
+            cert_pem.clone(),
+            key_pem,
+        )
         .with_pongs(send_pongs),
     )
     .await

common/src/lib.rs

@@ -369,9 +369,8 @@ mod pipe_error_tests {
 
     #[test]
     fn test_pipe_error_from_connection_error() {
-        let pe: PipeError = CommunicationError::ConnectionError(
+        let pe: PipeError =
-            wtransport::error::ConnectionError::TimedOut,
+            CommunicationError::ConnectionError(wtransport::error::ConnectionError::TimedOut)
-        )
                 .into();
         assert_eq!(pe, PipeError::ConnectionClosed);
     }

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
+                    let description = msg.get_str(DataType::Description).unwrap_or("").to_string();
-                        .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) =
+            let (app_tx, app_rx) = mpsc::channel(self.config.policy.receiver_queue_capacity);
-                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) =
+            let (app_tx, app_rx) = mpsc::channel(self.config.policy.receiver_queue_capacity);
-                mpsc::channel(self.config.policy.receiver_queue_capacity);
             let (pipe_req_tx, pipe_req_rx) =
                 mpsc::channel(self.config.policy.receiver_queue_capacity);
 

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(
+    pub fn with_max_concurrent_stream_tasks(mut self, max_concurrent_stream_tasks: usize) -> Self {
-        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::<PipeReader>(
+        let (pipe_tx, pipe_rx) = mpsc::channel::<PipeReader>(policy.receiver_queue_capacity);
-            policy.receiver_queue_capacity,
-        );
         #[cfg(not(feature = "pipes"))]
         let (tx, rx) = mpsc::channel::<Result<CommunicationValue, CommunicationError>>(
             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() {

transport/src/pipe.rs

@@ -9,10 +9,7 @@ pub struct PipeWriter {
 
 impl PipeWriter {
     pub async fn finish(mut self) -> Result<(), mtp_common::CommunicationError> {
-        self.stream
+        self.stream.finish().await.map_err(|e| {
-            .finish()
-            .await
-            .map_err(|e| {
             log::warn!("[PipeWriter] finish failed: {e}");
             mtp_common::CommunicationError::StreamWriteError(e)
         })

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())
+    let (client_tx, client_rx) = connect(&url, Some(cert_pem), policy.clone()).await.unwrap();
-        .await
-        .unwrap();
     let (_host_tx, host_rx) = h.next().await.unwrap();
 
     let tm = TypeMap::latest();
@@ -242,10 +240,7 @@ async fn test_receiver_backpressure_with_small_queue() {
     }
 
     for i in 0..8u128 {
-        let received = tokio::time::timeout(
+        let received = tokio::time::timeout(std::time::Duration::from_secs(5), host_rx.receive())
-            std::time::Duration::from_secs(5),
-            host_rx.receive(),
-        )
             .await
             .unwrap()
             .unwrap();
@@ -332,10 +327,7 @@ async fn test_semaphore_saturation_with_concurrent_streams() {
     }
 
     for i in 0..6u128 {
-        let received = tokio::time::timeout(
+        let received = tokio::time::timeout(std::time::Duration::from_secs(5), host_rx.receive())
-            std::time::Duration::from_secs(5),
-            host_rx.receive(),
-        )
             .await
             .unwrap()
             .unwrap();

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
+        self.pending_pipe_creations.borrow_mut().insert(pipe_id, tx);
-            .borrow_mut()
-            .insert(pipe_id, tx);
 
-        let request =
+        let request = CommunicationValue::new(CommunicationType::PipeRequest)
-            CommunicationValue::new(CommunicationType::PipeRequest).with_id(pipe_id).add_typed_default(
+            .with_id(pipe_id)
+            .add_typed_default(
                 DataType::Description,
                 DataValue::Str(description.to_string()),
             );

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) =
+        if let Err(e) = JsFuture::from(close_promise.unchecked_into::<js_sys::Promise>()).await {
-            JsFuture::from(close_promise.unchecked_into::<js_sys::Promise>()).await
-        {
             crate::transport::log_stream_error_code(&e, "pipe writer close");
         }
         release_writer_lock(&self.writer);

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
+        let pipe_request_type =
-            .to_id(&mtp_codec::TypeMap::latest());
+            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 _ =
+                            let _ = on_error.call1(&JsValue::NULL, &JsValue::from_str(&message));
-                                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) =
+        if let Err(e) = JsFuture::from(write_promise.unchecked_into::<js_sys::Promise>()).await {
-            JsFuture::from(write_promise.unchecked_into::<js_sys::Promise>()).await
-        {
             log_stream_error_code(&e, "open_pipe write");
             release_writer_lock(&writer_val);
             return Err(e);