Better errors in PulseAudio backend (#801)

* More meaningful error messages
* Use F32 if a user requests F64 (F64 is not supported by PulseAudio)
* Move all code that can fail to `start` where errors can be returned to prevent panics
* Use drain in `stop`

Author: JasonLG1979

CHANGELOG.md

@@ -44,7 +44,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - [playback] `alsamixer`: make `--volume-ctrl {linear|log}` work as expected
 - [playback] `alsa`, `gstreamer`, `pulseaudio`: always output in native endianness
 - [playback] `alsa`: revert buffer size to ~500 ms
-- [playback] `alsa`, `pipe`: better error handling
+- [playback] `alsa`, `pipe`, `pulseaudio`: better error handling
 
 ## [0.2.0] - 2021-05-04
 

playback/Cargo.toml

@@ -53,7 +53,7 @@ rand_distr = "0.4"
 [features]
 alsa-backend = ["alsa", "thiserror"]
 portaudio-backend = ["portaudio-rs"]
-pulseaudio-backend = ["libpulse-binding", "libpulse-simple-binding"]
+pulseaudio-backend = ["libpulse-binding", "libpulse-simple-binding", "thiserror"]
 jackaudio-backend = ["jack"]
 rodio-backend = ["rodio", "cpal", "thiserror"]
 rodiojack-backend = ["rodio", "cpal/jack", "thiserror"]

playback/src/audio_backend/pulseaudio.rs

@@ -3,82 +3,117 @@ use crate::config::AudioFormat;
 use crate::convert::Converter;
 use crate::decoder::AudioPacket;
 use crate::{NUM_CHANNELS, SAMPLE_RATE};
-use libpulse_binding::{self as pulse, stream::Direction};
+use libpulse_binding::{self as pulse, error::PAErr, stream::Direction};
 use libpulse_simple_binding::Simple;
 use std::io;
+use thiserror::Error;
 
 const APP_NAME: &str = "librespot";
 const STREAM_NAME: &str = "Spotify endpoint";
 
+#[derive(Debug, Error)]
+enum PulseError {
+    #[error("Error starting PulseAudioSink, invalid PulseAudio sample spec")]
+    InvalidSampleSpec,
+    #[error("Error starting PulseAudioSink, could not connect to PulseAudio server, {0}")]
+    ConnectionRefused(PAErr),
+    #[error("Error stopping PulseAudioSink, failed to drain PulseAudio server buffer, {0}")]
+    DrainFailure(PAErr),
+    #[error("Error in PulseAudioSink, Not connected to PulseAudio server")]
+    NotConnected,
+    #[error("Error writing from PulseAudioSink to PulseAudio server, {0}")]
+    OnWrite(PAErr),
+}
+
 pub struct PulseAudioSink {
     s: Option<Simple>,
-    ss: pulse::sample::Spec,
     device: Option<String>,
     format: AudioFormat,
 }
 
 impl Open for PulseAudioSink {
     fn open(device: Option<String>, format: AudioFormat) -> Self {
-        info!("Using PulseAudio sink with format: {:?}", format);
+        let mut actual_format = format;
+
+        if actual_format == AudioFormat::F64 {
+            warn!("PulseAudio currently does not support F64 output");
+            actual_format = AudioFormat::F32;
+        }
+
+        info!("Using PulseAudioSink with format: {:?}", actual_format);
+
+        Self {
+            s: None,
+            device,
+            format: actual_format,
+        }
+    }
+}
+
+impl Sink for PulseAudioSink {
+    fn start(&mut self) -> io::Result<()> {
+        if self.s.is_some() {
+            return Ok(());
+        }
 
         // PulseAudio calls S24 and S24_3 different from the rest of the world
-        let pulse_format = match format {
+        let pulse_format = match self.format {
             AudioFormat::F32 => pulse::sample::Format::FLOAT32NE,
             AudioFormat::S32 => pulse::sample::Format::S32NE,
             AudioFormat::S24 => pulse::sample::Format::S24_32NE,
             AudioFormat::S24_3 => pulse::sample::Format::S24NE,
             AudioFormat::S16 => pulse::sample::Format::S16NE,
-            _ => {
-                unimplemented!("PulseAudio currently does not support {:?} output", format)
-            }
+            _ => unreachable!(),
         };
 
         let ss = pulse::sample::Spec {
             format: pulse_format,
             channels: NUM_CHANNELS,
             rate: SAMPLE_RATE,
         };
-        debug_assert!(ss.is_valid());
 
-        Self {
-            s: None,
-            ss,
-            device,
-            format,
-        }
-    }
-}
-
-impl Sink for PulseAudioSink {
-    fn start(&mut self) -> io::Result<()> {
-        if self.s.is_some() {
-            return Ok(());
+        if !ss.is_valid() {
+            return Err(io::Error::new(
+                io::ErrorKind::Other,
+                PulseError::InvalidSampleSpec,
+            ));
         }
 
-        let device = self.device.as_deref();
         let result = Simple::new(
-            None,                // Use the default server.
-            APP_NAME,            // Our application's name.
-            Direction::Playback, // Direction.
-            device,              // Our device (sink) name.
-            STREAM_NAME,         // Description of our stream.
-            &self.ss,            // Our sample format.
-            None,                // Use default channel map.
-            None,                // Use default buffering attributes.
+            None,                   // Use the default server.
+            APP_NAME,               // Our application's name.
+            Direction::Playback,    // Direction.
+            self.device.as_deref(), // Our device (sink) name.
+            STREAM_NAME,            // Description of our stream.
+            &ss,                    // Our sample format.
+            None,                   // Use default channel map.
+            None,                   // Use default buffering attributes.
         );
+
         match result {
             Ok(s) => {
                 self.s = Some(s);
-                Ok(())
             }
-            Err(e) => Err(io::Error::new(
-                io::ErrorKind::ConnectionRefused,
-                e.to_string().unwrap(),
-            )),
+            Err(e) => {
+                return Err(io::Error::new(
+                    io::ErrorKind::ConnectionRefused,
+                    PulseError::ConnectionRefused(e),
+                ));
+            }
         }
+
+        Ok(())
     }
 
     fn stop(&mut self) -> io::Result<()> {
+        let s = self
+            .s
+            .as_mut()
+            .ok_or_else(|| io::Error::new(io::ErrorKind::NotConnected, PulseError::NotConnected))?;
+
+        s.drain()
+            .map_err(|e| io::Error::new(io::ErrorKind::Other, PulseError::DrainFailure(e)))?;
+
         self.s = None;
         Ok(())
     }
@@ -88,20 +123,15 @@ impl Sink for PulseAudioSink {
 
 impl SinkAsBytes for PulseAudioSink {
     fn write_bytes(&mut self, data: &[u8]) -> io::Result<()> {
-        if let Some(s) = &self.s {
-            match s.write(data) {
-                Ok(_) => Ok(()),
-                Err(e) => Err(io::Error::new(
-                    io::ErrorKind::BrokenPipe,
-                    e.to_string().unwrap(),
-                )),
-            }
-        } else {
-            Err(io::Error::new(
-                io::ErrorKind::NotConnected,
-                "Not connected to PulseAudio",
-            ))
-        }
+        let s = self
+            .s
+            .as_mut()
+            .ok_or_else(|| io::Error::new(io::ErrorKind::NotConnected, PulseError::NotConnected))?;
+
+        s.write(data)
+            .map_err(|e| io::Error::new(io::ErrorKind::Other, PulseError::OnWrite(e)))?;
+
+        Ok(())
     }
 }