Refactor notifier background thread spawning

src/cmd/rec.rs

@@ -6,6 +6,7 @@ use crate::config::Config;
 use crate::encoder::{AsciicastEncoder, Encoder, RawEncoder, TextEncoder};
 use crate::locale;
 use crate::logger;
+use crate::notifier;
 use crate::pty;
 use crate::recorder::Output;
 use crate::recorder::{self, KeyBindings};
@@ -43,6 +44,8 @@ impl Command for cli::Record {
             logger::info!("Press <ctrl+d> or type 'exit' to end");
         }
 
+        let notifier = notifier::threaded(notifier);
+
         {
             let mut tty = self.get_tty()?;
             let mut recorder = recorder::Recorder::new(output, record_input, keys, notifier);

src/cmd/stream.rs

@@ -4,6 +4,7 @@ use crate::cli;
 use crate::config::Config;
 use crate::locale;
 use crate::logger;
+use crate::notifier;
 use crate::pty;
 use crate::streamer::{self, KeyBindings};
 use crate::tty::{self, FixedSizeTty};
@@ -71,6 +72,8 @@ impl Command for cli::Stream {
             logger::info!("Press <ctrl+d> or type 'exit' to end");
         }
 
+        let notifier = notifier::threaded(notifier);
+
         {
             let mut tty = self.get_tty()?;
 

src/notifier.rs

@@ -1,4 +1,6 @@
 use anyhow::Result;
+use std::sync::mpsc;
+use std::thread;
 use std::{
     env,
     ffi::OsStr,
@@ -94,12 +96,46 @@ impl Notifier for NullNotifier {
 }
 
 fn exec<S: AsRef<OsStr>>(command: &mut Command, args: &[S]) -> Result<()> {
-    command
+    let status = command
         .stdin(Stdio::null())
         .stdout(Stdio::null())
         .stderr(Stdio::null())
         .args(args)
         .status()?;
 
-    Ok(())
+    if status.success() {
+        Ok(())
+    } else {
+        Err(anyhow::anyhow!(
+            "exit status: {}",
+            status.code().unwrap_or(0)
+        ))
+    }
+}
+
+#[derive(Clone)]
+pub struct ThreadedNotifier(mpsc::Sender<String>);
+
+pub fn threaded(mut notifier: Box<dyn Notifier>) -> ThreadedNotifier {
+    let (tx, rx) = mpsc::channel();
+
+    thread::spawn(move || {
+        for message in &rx {
+            if notifier.notify(message).is_err() {
+                break;
+            }
+        }
+
+        for _ in rx {}
+    });
+
+    ThreadedNotifier(tx)
+}
+
+impl Notifier for ThreadedNotifier {
+    fn notify(&mut self, message: String) -> Result<()> {
+        self.0.send(message)?;
+
+        Ok(())
+    }
 }

src/recorder.rs

@@ -9,11 +9,11 @@ use std::sync::mpsc;
 use std::thread;
 use std::time::{Duration, SystemTime};
 
-pub struct Recorder {
+pub struct Recorder<N> {
     output: Option<Box<dyn Output + Send>>,
     record_input: bool,
     keys: KeyBindings,
-    notifier: Option<Box<dyn Notifier>>,
+    notifier: N,
     sender: mpsc::Sender<Message>,
     receiver: Option<mpsc::Receiver<Message>>,
     handle: Option<util::JoinHandle>,
@@ -38,15 +38,14 @@ enum Message {
     Input(u64, Vec<u8>),
     Resize(u64, tty::TtySize),
     Marker(u64),
-    Notification(String),
 }
 
-impl Recorder {
+impl<N: Notifier> Recorder<N> {
     pub fn new(
         output: Box<dyn Output + Send>,
         record_input: bool,
         keys: KeyBindings,
-        notifier: Box<dyn Notifier>,
+        notifier: N,
     ) -> Self {
         let (sender, receiver) = mpsc::channel();
 
@@ -54,7 +53,7 @@ impl Recorder {
             output: Some(output),
             record_input,
             keys,
-            notifier: Some(notifier),
+            notifier,
             sender,
             receiver: Some(receiver),
             handle: None,
@@ -72,21 +71,18 @@ impl Recorder {
         }
     }
 
-    fn notify<S: ToString>(&self, text: S) {
-        let msg = Message::Notification(text.to_string());
-
-        self.sender
-            .send(msg)
+    fn notify<S: ToString>(&mut self, text: S) {
+        self.notifier
+            .notify(text.to_string())
             .expect("notification send should succeed");
     }
 }
 
-impl pty::Handler for Recorder {
+impl<N: Notifier> pty::Handler for Recorder<N> {
     fn start(&mut self, tty_size: tty::TtySize, theme: Option<tty::Theme>) {
         let mut output = self.output.take().unwrap();
         let _ = output.header(SystemTime::now(), tty_size, theme);
         let receiver = self.receiver.take().unwrap();
-        let mut notifier = self.notifier.take().unwrap();
 
         let handle = thread::spawn(move || {
             use Message::*;
@@ -122,10 +118,6 @@ impl pty::Handler for Recorder {
                     Marker(time) => {
                         let _ = output.event(Event::marker(time, String::new()));
                     }
-
-                    Notification(text) => {
-                        let _ = notifier.notify(text);
-                    }
                 }
             }
 

src/streamer/forwarder.rs

@@ -1,3 +1,5 @@
+use crate::notifier::Notifier;
+
 use super::alis;
 use super::session;
 use crate::api;
@@ -10,7 +12,6 @@ use std::borrow::Cow;
 use std::pin::Pin;
 use std::time::Duration;
 use tokio::net::TcpStream;
-use tokio::sync::mpsc;
 use tokio::time::{interval, sleep, timeout};
 use tokio_stream::wrappers::errors::BroadcastStreamRecvError;
 use tokio_stream::wrappers::IntervalStream;
@@ -25,10 +26,10 @@ const PING_TIMEOUT: u64 = 10;
 const SEND_TIMEOUT: u64 = 10;
 const MAX_RECONNECT_DELAY: u64 = 5000;
 
-pub async fn forward(
+pub async fn forward<N: Notifier>(
     url: url::Url,
-    clients_tx: mpsc::Sender<session::Client>,
-    notifier_tx: std::sync::mpsc::Sender<String>,
+    clients_tx: tokio::sync::mpsc::Sender<session::Client>,
+    mut notifier: N,
     shutdown_token: tokio_util::sync::CancellationToken,
 ) {
     info!("forwarding to {url}");
@@ -45,9 +46,9 @@ pub async fn forward(
             _ = sleep(Duration::from_secs(3)) => {
                 if reconnect_attempt > 0 {
                     if connection_count == 0 {
-                        let _ = notifier_tx.send("Connected to the server".to_string());
+                        let _ = notifier.notify("Connected to the server".to_string());
                     } else {
-                        let _ = notifier_tx.send("Reconnected to the server".to_string());
+                        let _ = notifier.notify("Reconnected to the server".to_string());
                     }
                 }
 
@@ -62,7 +63,7 @@ pub async fn forward(
             Ok(true) => break,
 
             Ok(false) => {
-                let _ = notifier_tx.send("Stream halted by the server".to_string());
+                let _ = notifier.notify("Stream halted by the server".to_string());
                 break;
             }
 
@@ -75,8 +76,8 @@ pub async fn forward(
                     // but doesn't properly perform the protocol negotiation.
                     // This applies to asciinema-server v20241103 and earlier.
 
-                    let _ = notifier_tx
-                        .send("The server version is too old, forwarding failed".to_string());
+                    let _ = notifier
+                        .notify("The server version is too old, forwarding failed".to_string());
 
                     break;
                 }
@@ -88,7 +89,7 @@ pub async fn forward(
                         // This happens when the server doesn't support our protocol (version).
                         // This applies to asciinema-server versions newer than v20241103.
 
-                        let _ = notifier_tx.send(
+                        let _ = notifier.notify(
                             "CLI not compatible with the server, forwarding failed".to_string(),
                         );
 
@@ -100,11 +101,11 @@ pub async fn forward(
 
                 if reconnect_attempt == 0 {
                     if connection_count == 0 {
-                        let _ = notifier_tx
-                            .send("Cannot connect to the server, retrying...".to_string());
+                        let _ = notifier
+                            .notify("Cannot connect to the server, retrying...".to_string());
                     } else {
-                        let _ = notifier_tx
-                            .send("Disconnected from the server, reconnecting...".to_string());
+                        let _ = notifier
+                            .notify("Disconnected from the server, reconnecting...".to_string());
                     }
                 }
             }
@@ -123,7 +124,7 @@ pub async fn forward(
 
 async fn connect_and_forward(
     url: &url::Url,
-    clients_tx: &mpsc::Sender<session::Client>,
+    clients_tx: &tokio::sync::mpsc::Sender<session::Client>,
 ) -> anyhow::Result<bool> {
     let uri: Uri = url.to_string().parse()?;
 
@@ -139,7 +140,7 @@ async fn connect_and_forward(
 }
 
 async fn event_stream(
-    clients_tx: &mpsc::Sender<session::Client>,
+    clients_tx: &tokio::sync::mpsc::Sender<session::Client>,
 ) -> anyhow::Result<impl Stream<Item = anyhow::Result<Message>>> {
     let stream = alis::stream(clients_tx)
         .await?

src/streamer/mod.rs

@@ -10,24 +10,20 @@ use crate::util;
 use std::net;
 use std::thread;
 use std::time::Duration;
-use tokio::sync::mpsc;
 use tracing::info;
 
-pub struct Streamer {
+pub struct Streamer<N> {
     record_input: bool,
     keys: KeyBindings,
-    notifier: Option<Box<dyn Notifier>>,
-    notifier_rx: Option<std::sync::mpsc::Receiver<String>>,
-    pty_rx: Option<mpsc::UnboundedReceiver<Event>>,
+    notifier: N,
+    pty_rx: Option<tokio::sync::mpsc::UnboundedReceiver<Event>>,
     paused: bool,
     prefix_mode: bool,
     listener: Option<net::TcpListener>,
     forward_url: Option<url::Url>,
     // XXX: field (drop) order below is crucial for correct shutdown
-    pty_tx: mpsc::UnboundedSender<Event>,
-    notifier_tx: std::sync::mpsc::Sender<String>,
+    pty_tx: tokio::sync::mpsc::UnboundedSender<Event>,
     event_loop_handle: Option<util::JoinHandle>,
-    notifier_handle: Option<util::JoinHandle>,
 }
 
 enum Event {
@@ -37,24 +33,20 @@ enum Event {
     Marker(u64),
 }
 
-impl Streamer {
+impl<N: Notifier> Streamer<N> {
     pub fn new(
         listener: Option<net::TcpListener>,
         forward_url: Option<url::Url>,
         record_input: bool,
         keys: KeyBindings,
-        notifier: Box<dyn Notifier>,
+        notifier: N,
     ) -> Self {
-        let (notifier_tx, notifier_rx) = std::sync::mpsc::channel();
-        let (pty_tx, pty_rx) = mpsc::unbounded_channel();
+        let (pty_tx, pty_rx) = tokio::sync::mpsc::unbounded_channel();
 
         Self {
             record_input,
             keys,
-            notifier: Some(notifier),
-            notifier_tx,
-            notifier_rx: Some(notifier_rx),
-            notifier_handle: None,
+            notifier,
             pty_tx,
             pty_rx: Some(pty_rx),
             event_loop_handle: None,
@@ -69,20 +61,20 @@ impl Streamer {
         time.as_micros() as u64
     }
 
-    fn notify<S: ToString>(&self, message: S) {
+    fn notify<S: ToString>(&mut self, message: S) {
         let message = message.to_string();
         info!(message);
 
-        self.notifier_tx
-            .send(message)
+        self.notifier
+            .notify(message)
             .expect("notification send should succeed");
     }
 }
 
-impl pty::Handler for Streamer {
+impl<N: Notifier + Clone + 'static> pty::Handler for Streamer<N> {
     fn start(&mut self, tty_size: tty::TtySize, theme: Option<tty::Theme>) {
         let pty_rx = self.pty_rx.take().unwrap();
-        let (clients_tx, mut clients_rx) = mpsc::channel(1);
+        let (clients_tx, mut clients_rx) = tokio::sync::mpsc::channel(1);
         let shutdown_token = tokio_util::sync::CancellationToken::new();
         let runtime = build_tokio_runtime();
 
@@ -98,7 +90,7 @@ impl pty::Handler for Streamer {
             runtime.spawn(forwarder::forward(
                 url,
                 clients_tx,
-                self.notifier_tx.clone(),
+                self.notifier.clone(),
                 shutdown_token.clone(),
             ))
         });
@@ -120,15 +112,6 @@ impl pty::Handler for Streamer {
                 let _ = clients_rx.recv().await;
             });
         }));
-
-        let mut notifier = self.notifier.take().unwrap();
-        let notifier_rx = self.notifier_rx.take().unwrap();
-
-        self.notifier_handle = wrap_thread_handle(thread::spawn(move || {
-            for message in notifier_rx {
-                let _ = notifier.notify(message);
-            }
-        }));
     }
 
     fn output(&mut self, time: Duration, data: &[u8]) -> bool {
@@ -188,8 +171,8 @@ impl pty::Handler for Streamer {
 }
 
 async fn event_loop(
-    mut events: mpsc::UnboundedReceiver<Event>,
-    clients: &mut mpsc::Receiver<session::Client>,
+    mut events: tokio::sync::mpsc::UnboundedReceiver<Event>,
+    clients: &mut tokio::sync::mpsc::Receiver<session::Client>,
     tty_size: tty::TtySize,
     theme: Option<tty::Theme>,
 ) {