replace hook processes with tokio::process

Previously, handling too many librespot events in a row, the spawned hook might not have been finished. In this case, the events would stack up, since the event channel then would not be polled.

By using the tokio::process module, we gain support for waiting for processes asynchronously and even if an event would not be handled directly, it will be handled after the previous hook process has finished.

Author: eladyn

Cargo.toml

@@ -28,7 +28,7 @@ serde = { version = "1.0.115", features = ["derive"] }
 sha-1 = "0.9"
 structopt = "0.3.17"
 syslog = "4"
-tokio = {version = "1.6.1", features = ["signal", "rt-multi-thread"] }
+tokio = {version = "1.6.1", features = ["signal", "rt-multi-thread", "process"] }
 tokio-compat = { version = "0.1.6", features = ["rt-current-thread"] }
 tokio-compat-02 = "0.2.0"
 tokio-stream = "0.1.7"

src/main_loop.rs

@@ -88,7 +88,7 @@ pub(crate) struct MainLoopState {
     pub(crate) autoplay: bool,
     pub(crate) volume_ctrl: VolumeCtrl,
     pub(crate) initial_volume: Option<u16>,
-    pub(crate) running_event_program: Option<Child>,
+    pub(crate) running_event_program: Option<Pin<Box<Child>>>,
     pub(crate) shell: String,
     pub(crate) device_type: DeviceType,
     // Command line option should still be available without dbus_mpris feature
@@ -120,15 +120,20 @@ impl Future for MainLoopState {
             }
 
             if let Some(mut child) = self.running_event_program.take() {
-                match child.try_wait() {
-                    // Still running...
-                    Ok(None) => self.running_event_program = Some(child),
-                    // Exited with error...
-                    Err(e) => error!("{}", e),
-                    // Exited without error...
-                    Ok(Some(_)) => (),
+                // check if child has already exited
+                if let Poll::Ready(result) = child.as_mut().poll(cx) {
+                    match result {
+                        // Exited without error...
+                        Ok(_) => (),
+                        // Exited with error...
+                        Err(e) => error!("{}", e),
+                    }
+                } else {
+                    // drop the Box that holds a reference to our child
+                    self.running_event_program = Some(child);
                 }
             }
+
             if self.running_event_program.is_none() {
                 if let Some(ref mut player_event_channel) = self.spotifyd_state.player_event_channel
                 {
@@ -139,7 +144,15 @@ impl Future for MainLoopState {
                         }
                         if let Some(ref cmd) = self.spotifyd_state.player_event_program {
                             match spawn_program_on_event(&self.shell, cmd, event) {
-                                Ok(child) => self.running_event_program = Some(child),
+                                Ok(child) => {
+                                    self.running_event_program = Some({
+                                        let mut child = Box::pin(child);
+                                        // We poll the child once, so the waker is awoken once the process finishes.
+                                        // Polling on a Poll::Ready(...) is allowed in this case.
+                                        let _ = child.as_mut().poll(cx);
+                                        child
+                                    })
+                                }
                                 Err(e) => error!("{}", e),
                             }
                         }

src/process.rs

@@ -1,18 +1,22 @@
 use crate::error::Error;
+use futures::Future;
 use librespot_playback::player::PlayerEvent;
 use log::info;
 use std::{
     collections::HashMap,
-    io::{self, Read, Write},
-    process::{Command, ExitStatus, Stdio},
+    io::{self, Write},
+    pin::Pin,
+    process::{Output, Stdio},
+    task::{Context, Poll},
 };
+use tokio::process::Command;
 
 /// Blocks while provided command is run in a subprocess using the provided
 /// shell. If successful, returns the contents of the subprocess's `stdout` as a
 /// `String`.
 pub(crate) fn run_program(shell: &str, cmd: &str) -> Result<String, Error> {
     info!("Running {:?} using {:?}", cmd, shell);
-    let output = Command::new(shell)
+    let output = std::process::Command::new(shell)
         .arg("-c")
         .arg(cmd)
         .output()
@@ -150,96 +154,69 @@ pub(crate) fn spawn_program_on_event(
     spawn_program(shell, cmd, env)
 }
 
-/// Same as a `std::process::Child` except when this `Child` exits:
+/// Wraps a process into a Future that executes something after the process has
+/// exited:
 /// * successfully: It writes the contents of it's stdout to the stdout of the
 ///   main process.
 /// * unsuccesfully: It returns an error that includes the contents it's stderr
 ///   as well as information on the command that was run and the shell that
 ///   invoked it.
-#[derive(Debug)]
 pub(crate) struct Child {
     cmd: String,
-    inner: std::process::Child,
+    future: Pin<Box<dyn Future<Output = io::Result<Output>>>>,
     shell: String,
 }
 
 impl Child {
-    pub(crate) fn new(cmd: String, child: std::process::Child, shell: String) -> Self {
+    pub(crate) fn new(cmd: String, child: tokio::process::Child, shell: String) -> Self {
         Self {
             cmd,
-            inner: child,
+            future: Box::pin(child.wait_with_output()),
             shell,
         }
     }
+}
 
-    #[allow(unused)]
-    pub(crate) fn wait(&mut self) -> Result<(), Error> {
-        match self.inner.wait() {
-            Ok(status) => {
-                self.write_output(status)?;
-                Ok(())
-            }
-            Err(e) => Err(Error::subprocess_with_err(&self.shell, &self.cmd, e)),
-        }
-    }
-
-    #[allow(unused)]
-    pub(crate) fn try_wait(&mut self) -> Result<Option<()>, Error> {
-        match self.inner.try_wait() {
-            Ok(Some(status)) => {
-                self.write_output(status)?;
-                Ok(Some(()))
+impl Future for Child {
+    type Output = Result<(), Error>;
+
+    fn poll(mut self: Pin<&mut Child>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        if let Poll::Ready(result) = self.future.as_mut().poll(cx) {
+            let output = match result {
+                Ok(output) => output,
+                Err(err) => {
+                    return Poll::Ready(Err(Error::subprocess_with_err(
+                        &self.shell,
+                        &self.cmd,
+                        err,
+                    )));
+                }
+            };
+
+            if output.status.success() {
+                // If successful, write subprocess's stdout to main process's stdout...
+                let stdout = io::stdout();
+                let mut writer = stdout.lock();
+
+                writer
+                    .write_all(&output.stdout)
+                    .map_err(|e| Error::subprocess_with_err(&self.shell, &self.cmd, e))?;
+
+                writer
+                    .flush()
+                    .map_err(|e| Error::subprocess_with_err(&self.shell, &self.cmd, e))?;
+
+                Poll::Ready(Ok(()))
+            } else {
+                // If unsuccessful, return an error that includes the contents of stderr...
+                let stderr = String::from_utf8(output.stderr);
+                match stderr {
+                    Ok(stderr) => Err(Error::subprocess_with_str(&self.shell, &self.cmd, &stderr)),
+                    Err(_) => Err(Error::subprocess(&self.shell, &self.cmd)),
+                }?
             }
-            Ok(None) => Ok(None),
-            Err(e) => Err(Error::subprocess_with_err(&self.shell, &self.cmd, e)),
-        }
-    }
-
-    fn write_output(&mut self, status: ExitStatus) -> Result<(), Error> {
-        if status.success() {
-            // If successful, write subprocess's stdout to main process's stdout...
-            let mut stdout_of_child = self.inner.stdout.as_mut().unwrap();
-            let reader = &mut stdout_of_child;
-
-            let stdout_of_main = io::stdout();
-            let mut guard = stdout_of_main.lock();
-            let writer = &mut guard;
-
-            io::copy(reader, writer)
-                .map_err(|e| Error::subprocess_with_err(&self.shell, &self.cmd, e))?;
-
-            writer
-                .flush()
-                .map_err(|e| Error::subprocess_with_err(&self.shell, &self.cmd, e))?;
-
-            Ok(())
         } else {
-            // If unsuccessful, return an error that includes the contents of stderr...
-            let mut buf = String::new();
-            match self.inner.stderr.as_mut().unwrap().read_to_string(&mut buf) {
-                Ok(_nread) => Err(Error::subprocess_with_str(&self.shell, &self.cmd, &buf)),
-                Err(e) => Err(Error::subprocess_with_err(&self.shell, &self.cmd, e)),
-            }
+            Poll::Pending
         }
     }
 }
-
-impl std::ops::Deref for Child {
-    type Target = std::process::Child;
-
-    fn deref(&self) -> &Self::Target {
-        &self.inner
-    }
-}
-
-impl std::ops::DerefMut for Child {
-    fn deref_mut(&mut self) -> &mut Self::Target {
-        &mut self.inner
-    }
-}
-
-impl From<Child> for std::process::Child {
-    fn from(child: Child) -> Self {
-        child.inner
-    }
-}