client: fix error handling

src/bin/mgen-client.rs

@@ -8,8 +8,8 @@ use serde::Deserialize;
 use std::result::Result;
 use std::sync::Arc;
 use tokio::io::{split, AsyncWriteExt, ReadHalf, WriteHalf};
-use tokio::join;
 use tokio::net::TcpStream;
+use tokio::spawn;
 use tokio::sync::mpsc;
 use tokio::time::Duration;
 use tokio_rustls::{client::TlsStream, TlsConnector};
@@ -157,7 +157,7 @@ async fn reader(
                 if msg.body.has_attachment() {
                     let url = web_url(&target, msg.body.total_size(), &user);
                     let client = client.clone();
-                    tokio::spawn(async move {
+                    spawn(async move {
                         let mut res = client.get(url.clone()).await;
                         while res.is_err() {
                             log!("Error fetching: {}", res.unwrap_err());
@@ -428,38 +428,43 @@ async fn spawn_threads(config: FullConfig) -> Result<(), MessengerError> {
         .with_custom_certificate_verifier(Arc::new(NoCertificateVerification {}))
         .with_no_client_auth();
 
-    let reader = reader(
+    spawn(reader(
         web_server_params.clone(),
         retry,
         tls_config.clone(),
         reader_to_state,
         read_socket_updater_out,
         errs_in.clone(),
-    );
-    let writer = writer(
+    ));
+    spawn(writer(
         writer_from_state,
         writer_to_uploader,
         write_socket_updater_out,
         errs_in,
-    );
-    let uploader = uploader(
+    ));
+    spawn(uploader(
         web_server_params,
         retry,
         tls_config.clone(),
         uploader_from_writer,
-    );
-    let updater = socket_updater(
-        message_server_params,
-        retry,
-        tls_config,
-        errs_out,
-        read_socket_updater_in,
-        write_socket_updater_in,
-    );
-    let manager = manage_conversation(config, state_from_reader, state_to_writer);
-
-    join!(reader, writer, uploader, updater, manager)?;
-    Ok(())
+    ));
+    spawn(manage_conversation(
+        config,
+        state_from_reader,
+        state_to_writer,
+    ));
+
+    Err(MessengerError::Fatal(
+        socket_updater(
+            message_server_params,
+            retry,
+            tls_config,
+            errs_out,
+            read_socket_updater_in,
+            write_socket_updater_in,
+        )
+        .await,
+    ))
 }
 
 struct FullConfig {