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`

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,48 +3,49 @@ 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;
 
-        // PulseAudio calls S24 and S24_3 different from the rest of the world
+        if actual_format == AudioFormat::F64 {
-        let pulse_format = match format {
+            warn!("PulseAudio currently does not support F64 output");
-            AudioFormat::F32 => pulse::sample::Format::FLOAT32NE,
+            actual_format = AudioFormat::F32;
-            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)
-            }
-        };
 
-        let ss = pulse::sample::Spec {
+        info!("Using PulseAudioSink with format: {:?}", actual_format);
-            format: pulse_format,
-            channels: NUM_CHANNELS,
-            rate: SAMPLE_RATE,
-        };
-        debug_assert!(ss.is_valid());
 
         Self {
             s: None,
-            ss,
             device,
-            format,
+            format: actual_format,
         }
     }
 }
@@ -55,30 +56,64 @@ impl Sink for PulseAudioSink {
             return Ok(());
         }
 
-        let device = self.device.as_deref();
+        // PulseAudio calls S24 and S24_3 different from the rest of the world
+        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,
+            _ => unreachable!(),
+        };
+
+        let ss = pulse::sample::Spec {
+            format: pulse_format,
+            channels: NUM_CHANNELS,
+            rate: SAMPLE_RATE,
+        };
+
+        if !ss.is_valid() {
+            return Err(io::Error::new(
+                io::ErrorKind::Other,
+                PulseError::InvalidSampleSpec,
+            ));
+        }
+
         let result = Simple::new(
-            None,                // Use the default server.
+            None,                   // Use the default server.
-            APP_NAME,            // Our application's name.
+            APP_NAME,               // Our application's name.
-            Direction::Playback, // Direction.
+            Direction::Playback,    // Direction.
-            device,              // Our device (sink) name.
+            self.device.as_deref(), // Our device (sink) name.
-            STREAM_NAME,         // Description of our stream.
+            STREAM_NAME,            // Description of our stream.
-            &self.ss,            // Our sample format.
+            &ss,                    // Our sample format.
-            None,                // Use default channel map.
+            None,                   // Use default channel map.
-            None,                // Use default buffering attributes.
+            None,                   // Use default buffering attributes.
         );
+
         match result {
             Ok(s) => {
                 self.s = Some(s);
-                Ok(())
             }
-            Err(e) => Err(io::Error::new(
+            Err(e) => {
-                io::ErrorKind::ConnectionRefused,
+                return Err(io::Error::new(
-                e.to_string().unwrap(),
+                    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 {
+        let s = self
-            match s.write(data) {
+            .s
-                Ok(_) => Ok(()),
+            .as_mut()
-                Err(e) => Err(io::Error::new(
+            .ok_or_else(|| io::Error::new(io::ErrorKind::NotConnected, PulseError::NotConnected))?;
-                    io::ErrorKind::BrokenPipe,
+
-                    e.to_string().unwrap(),
+        s.write(data)
-                )),
+            .map_err(|e| io::Error::new(io::ErrorKind::Other, PulseError::OnWrite(e)))?;
-            }
+
-        } else {
+        Ok(())
-            Err(io::Error::new(
-                io::ErrorKind::NotConnected,
-                "Not connected to PulseAudio",
-            ))
-        }
     }
 }