common/task: extract out spawn_with_report_error (#1837)

Cargo.lock

@@ -5700,6 +5700,8 @@ dependencies = [
  "log",
  "thiserror",
  "tokio",
+ "wasm-bindgen",
+ "wasm-bindgen-futures",
 ]
 
 [[package]]

clients/socks5/src/client/mod.rs

@@ -80,7 +80,7 @@ impl NymClient {
         client_input: ClientInput,
         client_output: ClientOutput,
         self_address: Recipient,
-        mut shutdown: ShutdownListener,
+        shutdown: ShutdownListener,
     ) {
         info!("Starting socks5 listener...");
         let auth_methods = vec![AuthenticationMethods::NoAuth as u8];
@@ -103,25 +103,18 @@ impl NymClient {
             shared_lane_queue_lengths,
             shutdown.clone(),
         );
-        tokio::spawn(async move {
-            // Ideally we should have a fully fledged task manager to check for errors in all
-            // tasks.
-            // However, pragmatically, we start out by at least reporting errors for some of the
-            // tasks that interact with the outside world and can fail in normal operation, such as
-            // network issues.
-            // TODO: replace this by a generic solution, such as a task manager that stores all
-            // JoinHandles of all spawned tasks.
-            if let Err(res) = sphinx_socks
-                .serve(
-                    input_sender,
-                    received_buffer_request_sender,
-                    connection_command_sender,
-                )
-                .await
-            {
-                shutdown.send_we_stopped(Box::new(res));
-            }
-        });
+        task::spawn_with_report_error(
+            async move {
+                sphinx_socks
+                    .serve(
+                        input_sender,
+                        received_buffer_request_sender,
+                        connection_command_sender,
+                    )
+                    .await
+            },
+            shutdown,
+        );
     }
 
     /// blocking version of `start` method. Will run forever (or until SIGINT is sent)

common/task/Cargo.toml

@@ -15,5 +15,11 @@ tokio = { version = "1.21.2", features = ["macros", "sync"] }
 version = "1.21.2"
 features = ["signal", "time"]
 
+[target."cfg(target_arch = \"wasm32\")".dependencies.wasm-bindgen-futures]
+version = "0.4"
+
+[target."cfg(target_arch = \"wasm32\")".dependencies.wasm-bindgen]
+version = "0.2.83"
+
 [dev-dependencies]
 tokio = { version = "1.21.2", features = ["rt-multi-thread", "net", "signal", "test-util", "macros"] }

common/task/src/lib.rs

@@ -4,7 +4,10 @@
 pub mod shutdown;
 #[cfg(not(target_arch = "wasm32"))]
 pub mod signal;
+pub mod spawn;
 
 pub use shutdown::{ShutdownListener, ShutdownNotifier};
 #[cfg(not(target_arch = "wasm32"))]
 pub use signal::{wait_for_signal, wait_for_signal_and_error};
+
+pub use spawn::spawn_with_report_error;

common/task/src/spawn.rs

@@ -0,0 +1,33 @@
+use crate::ShutdownListener;
+use std::future::Future;
+
+#[cfg(target_arch = "wasm32")]
+pub(crate) fn spawn<F>(future: F)
+where
+    F: Future<Output = ()> + 'static,
+{
+    wasm_bindgen_futures::spawn_local(future);
+}
+
+#[cfg(not(target_arch = "wasm32"))]
+pub(crate) fn spawn<F>(future: F)
+where
+    F: Future + Send + 'static,
+    F::Output: Send + 'static,
+{
+    tokio::spawn(future);
+}
+
+pub fn spawn_with_report_error<F, T, E>(future: F, mut shutdown: ShutdownListener)
+where
+    F: Future<Output = Result<T, E>> + Send + 'static,
+    T: 'static,
+    E: std::error::Error + Send + 'static,
+{
+    let future_that_sends = async move {
+        if let Err(err) = future.await {
+            shutdown.send_we_stopped(Box::new(err));
+        }
+    };
+    spawn(future_that_sends);
+}

nym-connect/Cargo.lock

@@ -5464,6 +5464,8 @@ dependencies = [
  "log",
  "thiserror",
  "tokio",
+ "wasm-bindgen",
+ "wasm-bindgen-futures",
 ]
 
 [[package]]