Merge branch 'dev' into gst1.0-2020

.travis.yml

@@ -1,6 +1,6 @@
 language: rust
 rust:
-  - 1.33.0
+  - 1.40.0
   - stable
   - beta
   - nightly

COMPILING.md

@@ -13,7 +13,7 @@ curl https://sh.rustup.rs -sSf | sh
 
 Follow any prompts it gives you to install Rust. Once that’s done, Rust's standard tools should be setup and ready to use.
 
-*Note: The current minimum required Rust version is 1.33.0*
+*Note: The current minimum required Rust version at the time of writing is 1.40.0, you can find the current minimum version specified in the `.travis.yml` file.*
 
 #### Additional Rust tools - `rustfmt`
 To ensure a consistent codebase, we utilise [`rustfmt`](https://github.com/rust-lang/rustfmt), which is installed by default with `rustup` these days, else it can be installed manually with:

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "librespot"
-version = "0.1.1"
+version = "0.1.2"
 authors = ["Librespot Org"]
 license = "MIT"
 description = "An open source client library for Spotify, with support for Spotify Connect"
@@ -22,22 +22,22 @@ doc = false
 
 [dependencies.librespot-audio]
 path = "audio"
-version = "0.1.1"
+version = "0.1.2"
 [dependencies.librespot-connect]
 path = "connect"
-version = "0.1.1"
+version = "0.1.2"
 [dependencies.librespot-core]
 path = "core"
-version = "0.1.1"
+version = "0.1.2"
 [dependencies.librespot-metadata]
 path = "metadata"
-version = "0.1.1"
+version = "0.1.2"
 [dependencies.librespot-playback]
 path = "playback"
-version = "0.1.1"
+version = "0.1.2"
 [dependencies.librespot-protocol]
 path = "protocol"
-version = "0.1.1"
+version = "0.1.2"
 
 [dependencies]
 base64 = "0.10"
@@ -47,7 +47,7 @@ getopts = "0.2"
 hyper = "0.11"
 log = "0.4"
 num-bigint = "0.2"
-protobuf = "2.8.1"
+protobuf = "~2.14.0"
 rand = "0.7"
 rpassword = "3.0"
 tokio-core = "0.1"
@@ -77,9 +77,9 @@ default = ["librespot-playback/rodio-backend"]
 [package.metadata.deb]
 maintainer = "librespot-org"
 copyright = "2018 Paul Liétar"
-license_file = ["LICENSE", "4"]
+license-file = ["LICENSE", "4"]
 depends = "$auto"
-extended_description = """\
+extended-description = """\
 librespot is an open source client library for Spotify. It enables applications \
 to use Spotify's service, without using the official but closed-source \
 libspotify. Additionally, it will provide extra features which are not \

audio/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "librespot-audio"
-version = "0.1.1"
+version = "0.1.2"
 authors = ["Paul Lietar <paul@lietar.net>"]
 description="The audio fetching and processing logic for librespot"
 license="MIT"
@@ -8,7 +8,7 @@ edition = "2018"
 
 [dependencies.librespot-core]
 path = "../core"
-version = "0.1.1"
+version = "0.1.2"
 
 [dependencies]
 bit-set = "0.5"
@@ -23,7 +23,7 @@ tempfile = "3.1"
 aes-ctr = "0.3"
 
 librespot-tremor = { version = "0.1.0", optional = true }
-vorbis = { version ="0.1.0", optional = true }
+vorbis = { version ="0.0.14", optional = true }
 
 [features]
 with-tremor = ["librespot-tremor"]

connect/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "librespot-connect"
-version = "0.1.1"
+version = "0.1.2"
 authors = ["Paul Lietar <paul@lietar.net>"]
 description="The discovery and Spotify Connect logic for librespot"
 license="MIT"
@@ -8,13 +8,13 @@ edition = "2018"
 
 [dependencies.librespot-core]
 path = "../core"
-version = "0.1.1"
+version = "0.1.2"
 [dependencies.librespot-playback]
 path = "../playback"
-version = "0.1.1"
+version = "0.1.2"
 [dependencies.librespot-protocol]
 path = "../protocol"
-version = "0.1.1"
+version = "0.1.2"
 
 [dependencies]
 base64 = "0.10"
@@ -22,7 +22,7 @@ futures = "0.1"
 hyper = "0.11"
 log = "0.4"
 num-bigint = "0.2"
-protobuf = "2.8.1"
+protobuf = "~2.14.0"
 rand = "0.7"
 serde = "1.0"
 serde_derive = "1.0"

connect/src/spirc.rs

@@ -14,10 +14,12 @@ use crate::playback::mixer::Mixer;
 use crate::playback::player::{Player, PlayerEvent, PlayerEventChannel};
 use crate::protocol;
 use crate::protocol::spirc::{DeviceState, Frame, MessageType, PlayStatus, State, TrackRef};
+
 use librespot_core::config::ConnectConfig;
 use librespot_core::mercury::MercuryError;
 use librespot_core::session::Session;
 use librespot_core::spotify_id::{SpotifyAudioType, SpotifyId, SpotifyIdError};
+use librespot_core::util::url_encode;
 use librespot_core::util::SeqGenerator;
 use librespot_core::version;
 use librespot_core::volume::Volume;
@@ -249,7 +251,8 @@ impl Spirc {
         let ident = session.device_id().to_owned();
 
         // Uri updated in response to issue #288
-        let uri = format!("hm://remote/user/{}/", session.username());
+        debug!("canonical_username: {}", url_encode(&session.username()));
+        let uri = format!("hm://remote/user/{}/", url_encode(&session.username()));
 
         let subscription = session.mercury().subscribe(&uri as &str);
         let subscription = subscription
@@ -454,8 +457,8 @@ impl SpircTask {
             Ok(dur) => dur,
             Err(err) => err.duration(),
         };
-        ((dur.as_secs() as i64 + self.session.time_delta()) * 1000
-            + (dur.subsec_nanos() / 1000_000) as i64)
+        (dur.as_secs() as i64 + self.session.time_delta()) * 1000
+            + (dur.subsec_nanos() / 1000_000) as i64
     }
 
     fn ensure_mixer_started(&mut self) {
@@ -621,24 +624,8 @@ impl SpircTask {
                             self.play_status = SpircPlayStatus::Stopped;
                         }
                     },
-                    PlayerEvent::TimeToPreloadNextTrack { .. } => match self.play_status {
-                        SpircPlayStatus::Paused {
-                            ref mut preloading_of_next_track_triggered,
-                            ..
-                        }
-                        | SpircPlayStatus::Playing {
-                            ref mut preloading_of_next_track_triggered,
-                            ..
-                        } => {
-                            *preloading_of_next_track_triggered = true;
-                            if let Some(track_id) = self.preview_next_track() {
-                                self.player.preload(track_id);
-                            }
-                        }
-                        SpircPlayStatus::LoadingPause { .. }
-                        | SpircPlayStatus::LoadingPlay { .. }
-                        | SpircPlayStatus::Stopped => (),
-                    },
+                    PlayerEvent::TimeToPreloadNextTrack { .. } => self.handle_preload_next_track(),
+                    PlayerEvent::Unavailable { track_id, .. } => self.handle_unavailable(track_id),
                     _ => (),
                 }
             }
@@ -777,6 +764,7 @@ impl SpircTask {
                 } = self.play_status
                 {
                     if preloading_of_next_track_triggered {
+                        // Get the next track_id in the playlist
                         if let Some(track_id) = self.preview_next_track() {
                             self.player.preload(track_id);
                         }
@@ -790,7 +778,11 @@ impl SpircTask {
             }
 
             MessageType::kMessageTypeNotify => {
-                if self.device.get_is_active() && frame.get_device_state().get_is_active() {
+                if self.device.get_is_active()
+                    && frame.get_device_state().get_is_active()
+                    && self.device.get_became_active_at()
+                        <= frame.get_device_state().get_became_active_at()
+                {
                     self.device.set_is_active(false);
                     self.state.set_status(PlayStatus::kPlayStatusStop);
                     self.player.stop();
@@ -904,6 +896,50 @@ impl SpircTask {
             .and_then(|(track_id, _)| Some(track_id))
     }
 
+    fn handle_preload_next_track(&mut self) {
+        // Requests the player thread to preload the next track
+        match self.play_status {
+            SpircPlayStatus::Paused {
+                ref mut preloading_of_next_track_triggered,
+                ..
+            }
+            | SpircPlayStatus::Playing {
+                ref mut preloading_of_next_track_triggered,
+                ..
+            } => {
+                *preloading_of_next_track_triggered = true;
+                if let Some(track_id) = self.preview_next_track() {
+                    self.player.preload(track_id);
+                }
+            }
+            SpircPlayStatus::LoadingPause { .. }
+            | SpircPlayStatus::LoadingPlay { .. }
+            | SpircPlayStatus::Stopped => (),
+        }
+    }
+
+    // Mark unavailable tracks so we can skip them later
+    fn handle_unavailable(&mut self, track_id: SpotifyId) {
+        let unavailables = self.get_track_index_for_spotify_id(&track_id, 0);
+        for &index in unavailables.iter() {
+            debug_assert_eq!(self.state.get_track()[index].get_gid(), track_id.to_raw());
+            let mut unplayable_track_ref = TrackRef::new();
+            unplayable_track_ref.set_gid(self.state.get_track()[index].get_gid().to_vec());
+            // Misuse context field to flag the track
+            unplayable_track_ref.set_context(String::from("NonPlayable"));
+            std::mem::swap(
+                &mut self.state.mut_track()[index],
+                &mut unplayable_track_ref,
+            );
+            debug!(
+                "Marked <{:?}> at {:?} as NonPlayable",
+                self.state.get_track()[index],
+                index,
+            );
+        }
+        self.handle_preload_next_track();
+    }
+
     fn handle_next(&mut self) {
         let mut new_index = self.consume_queued_track() as u32;
         let mut continue_playing = true;
@@ -919,7 +955,7 @@ impl SpircTask {
         if (context_uri.starts_with("spotify:station:")
             || context_uri.starts_with("spotify:dailymix:")
             // spotify:user:xxx:collection
-            || context_uri.starts_with(&format!("spotify:user:{}:collection",self.session.username())))
+            || context_uri.starts_with(&format!("spotify:user:{}:collection",url_encode(&self.session.username()))))
             && ((self.state.get_track().len() as u32) - new_index) < CONTEXT_FETCH_THRESHOLD
         {
             self.context_fut = self.resolve_station(&context_uri);
@@ -1137,10 +1173,32 @@ impl SpircTask {
         })
     }
 
+    // Helper to find corresponding index(s) for track_id
+    fn get_track_index_for_spotify_id(
+        &self,
+        track_id: &SpotifyId,
+        start_index: usize,
+    ) -> Vec<usize> {
+        let index: Vec<usize> = self.state.get_track()[start_index..]
+            .iter()
+            .enumerate()
+            .filter(|&(_, track_ref)| track_ref.get_gid() == track_id.to_raw())
+            .map(|(idx, _)| start_index + idx)
+            .collect();
+        // Sanity check
+        debug_assert!(!index.is_empty());
+        index
+    }
+
+    // Broken out here so we can refactor this later when we move to SpotifyObjectID or similar
+    fn track_ref_is_unavailable(&self, track_ref: &TrackRef) -> bool {
+        track_ref.get_context() == "NonPlayable"
+    }
+
     fn get_track_id_to_play_from_playlist(&self, index: u32) -> Option<(SpotifyId, u32)> {
-        let tracks_len = self.state.get_track().len() as u32;
+        let tracks_len = self.state.get_track().len();
 
-        let mut new_playlist_index = index;
+        let mut new_playlist_index = index as usize;
 
         if new_playlist_index >= tracks_len {
             new_playlist_index = 0;
@@ -1152,14 +1210,15 @@ impl SpircTask {
         // tracks in each frame either have a gid or uri (that may or may not be a valid track)
         // E.g - context based frames sometimes contain tracks with <spotify:meta:page:>
 
-        let mut track_ref = self.state.get_track()[new_playlist_index as usize].clone();
+        let mut track_ref = self.state.get_track()[new_playlist_index].clone();
         let mut track_id = self.get_spotify_id_for_track(&track_ref);
-        while track_id.is_err() || track_id.unwrap().audio_type == SpotifyAudioType::NonPlayable {
+        while self.track_ref_is_unavailable(&track_ref)
+            || track_id.is_err()
+            || track_id.unwrap().audio_type == SpotifyAudioType::NonPlayable
+        {
             warn!(
                 "Skipping track <{:?}> at position [{}] of {}",
-                track_ref.get_uri(),
-                new_playlist_index,
-                tracks_len
+                track_ref, new_playlist_index, tracks_len
             );
 
             new_playlist_index += 1;
@@ -1171,12 +1230,12 @@ impl SpircTask {
                 warn!("No playable track found in state: {:?}", self.state);
                 return None;
             }
-            track_ref = self.state.get_track()[index as usize].clone();
+            track_ref = self.state.get_track()[new_playlist_index].clone();
             track_id = self.get_spotify_id_for_track(&track_ref);
         }
 
         match track_id {
-            Ok(track_id) => Some((track_id, new_playlist_index)),
+            Ok(track_id) => Some((track_id, new_playlist_index as u32)),
             Err(_) => None,
         }
     }

core/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "librespot-core"
-version = "0.1.1"
+version = "0.1.2"
 authors = ["Paul Lietar <paul@lietar.net>"]
 build = "build.rs"
 description="The core functionality provided by librespot"
@@ -9,7 +9,7 @@ edition = "2018"
 
 [dependencies.librespot-protocol]
 path = "../protocol"
-version = "0.1.1"
+version = "0.1.2"
 
 [dependencies]
 base64 = "0.10"
@@ -25,7 +25,7 @@ log = "0.4"
 num-bigint = "0.2"
 num-integer = "0.1"
 num-traits = "0.2"
-protobuf = "2.8.1"
+protobuf = "~2.14.0"
 rand = "0.7"
 serde = "1.0"
 serde_derive = "1.0"

core/src/mercury/mod.rs

@@ -1,4 +1,5 @@
 use crate::protocol;
+use crate::util::url_encode;
 use byteorder::{BigEndian, ByteOrder};
 use bytes::Bytes;
 use futures::sync::{mpsc, oneshot};
@@ -192,7 +193,7 @@ impl MercuryManager {
         let header: protocol::mercury::Header = protobuf::parse_from_bytes(&header_data).unwrap();
 
         let response = MercuryResponse {
-            uri: header.get_uri().to_owned(),
+            uri: url_encode(header.get_uri()).to_owned(),
             status_code: header.get_status_code(),
             payload: pending.parts,
         };

core/src/util/mod.rs

@@ -12,6 +12,21 @@ pub fn rand_vec<G: Rng>(rng: &mut G, size: usize) -> Vec<u8> {
         .collect()
 }
 
+pub fn url_encode(inp: &str) -> String {
+    let mut encoded = String::new();
+
+    for c in inp.as_bytes().iter() {
+        match *c as char {
+            'A'..='Z' | 'a'..='z' | '0'..='9' | '-' | '_' | '.' | '~' | ':' | '/' => {
+                encoded.push(*c as char)
+            }
+            c => encoded.push_str(format!("%{:02X}", c as u32).as_str()),
+        };
+    }
+
+    encoded
+}
+
 pub fn powm(base: &BigUint, exp: &BigUint, modulus: &BigUint) -> BigUint {
     let mut base = base.clone();
     let mut exp = exp.clone();

metadata/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "librespot-metadata"
-version = "0.1.1"
+version = "0.1.2"
 authors = ["Paul Lietar <paul@lietar.net>"]
 description="The metadata logic for librespot"
 license="MIT"
@@ -10,12 +10,12 @@ edition = "2018"
 byteorder = "1.3"
 futures = "0.1"
 linear-map = "1.2"
-protobuf = "2.8.1"
+protobuf = "~2.14.0"
 log = "0.4"
 
 [dependencies.librespot-core]
 path = "../core"
-version = "0.1.1"
+version = "0.1.2"
 [dependencies.librespot-protocol]
 path = "../protocol"
-version = "0.1.1"
+version = "0.1.2"

playback/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "librespot-playback"
-version = "0.1.1"
+version = "0.1.2"
 authors = ["Sasha Hilton <sashahilton00@gmail.com>"]
 description="The audio playback logic for librespot"
 license="MIT"
@@ -8,13 +8,13 @@ edition = "2018"
 
 [dependencies.librespot-audio]
 path = "../audio"
-version = "0.1.1"
+version = "0.1.2"
 [dependencies.librespot-core]
 path = "../core"
-version = "0.1.1"
+version = "0.1.2"
 [dependencies.librespot-metadata]
 path = "../metadata"
-version = "0.1.1"
+version = "0.1.2"
 
 [dependencies]
 futures = "0.1"

playback/src/audio_backend/alsa.rs

@@ -1,12 +1,20 @@
 use super::{Open, Sink};
 use alsa::device_name::HintIter;
-use alsa::pcm::{Access, Format, HwParams, PCM};
+use alsa::pcm::{Access, Format, Frames, HwParams, PCM};
 use alsa::{Direction, Error, ValueOr};
+use std::cmp::min;
 use std::ffi::CString;
 use std::io;
 use std::process::exit;
 
-pub struct AlsaSink(Option<PCM>, String);
+const PREFERED_PERIOD_SIZE: Frames = 5512; // Period of roughly 125ms
+const BUFFERED_PERIODS: Frames = 4;
+
+pub struct AlsaSink {
+    pcm: Option<PCM>,
+    device: String,
+    buffer: Vec<i16>,
+}
 
 fn list_outputs() {
     for t in &["pcm", "ctl", "hwdep"] {
@@ -25,8 +33,9 @@ fn list_outputs() {
     }
 }
 
-fn open_device(dev_name: &str) -> Result<(PCM), Box<Error>> {
+fn open_device(dev_name: &str) -> Result<(PCM, Frames), Box<Error>> {
     let pcm = PCM::new(dev_name, Direction::Playback, false)?;
+    let mut period_size = PREFERED_PERIOD_SIZE;
     // http://www.linuxjournal.com/article/6735?page=0,1#N0x19ab2890.0x19ba78d8
     // latency = period_size * periods / (rate * bytes_per_frame)
     // For 16 Bit stereo data, one frame has a length of four bytes.
@@ -41,7 +50,8 @@ fn open_device(dev_name: &str) -> Result<(PCM), Box<Error>> {
         hwp.set_format(Format::s16())?;
         hwp.set_rate(44100, ValueOr::Nearest)?;
         hwp.set_channels(2)?;
-        hwp.set_buffer_size_near(22050)?; // ~ 0.5s latency
+        period_size = hwp.set_period_size_near(period_size, ValueOr::Greater)?;
+        hwp.set_buffer_size_near(period_size * BUFFERED_PERIODS)?;
         pcm.hw_params(&hwp)?;
 
         let swp = pcm.sw_params_current()?;
@@ -49,7 +59,7 @@ fn open_device(dev_name: &str) -> Result<(PCM), Box<Error>> {
         pcm.sw_params(&swp)?;
     }
 
-    Ok(pcm)
+    Ok((pcm, period_size))
 }
 
 impl Open for AlsaSink {
@@ -67,16 +77,24 @@ impl Open for AlsaSink {
         }
         .to_string();
 
-        AlsaSink(None, name)
+        AlsaSink {
+            pcm: None,
+            device: name,
+            buffer: vec![],
+        }
     }
 }
 
 impl Sink for AlsaSink {
     fn start(&mut self) -> io::Result<()> {
-        if self.0.is_none() {
-            let pcm = open_device(&self.1);
+        if self.pcm.is_none() {
+            let pcm = open_device(&self.device);
             match pcm {
-                Ok(p) => self.0 = Some(p),
+                Ok((p, period_size)) => {
+                    self.pcm = Some(p);
+                    // Create a buffer for all samples for a full period
+                    self.buffer = Vec::with_capacity((period_size * 2) as usize);
+                }
                 Err(e) => {
                     error!("Alsa error PCM open {}", e);
                     return Err(io::Error::new(
@@ -92,20 +110,39 @@ impl Sink for AlsaSink {
 
     fn stop(&mut self) -> io::Result<()> {
         {
-            let pcm = self.0.as_ref().unwrap();
+            let pcm = self.pcm.as_mut().unwrap();
+            // Write any leftover data in the period buffer
+            // before draining the actual buffer
+            let io = pcm.io_i16().unwrap();
+            match io.writei(&self.buffer[..]) {
+                Ok(_) => (),
+                Err(err) => pcm.try_recover(err, false).unwrap(),
+            }
             pcm.drain().unwrap();
         }
-        self.0 = None;
+        self.pcm = None;
         Ok(())
     }
 
     fn write(&mut self, data: &[i16]) -> io::Result<()> {
-        let pcm = self.0.as_mut().unwrap();
-        let io = pcm.io_i16().unwrap();
-
-        match io.writei(&data) {
-            Ok(_) => (),
-            Err(err) => pcm.try_recover(err, false).unwrap(),
+        let mut processed_data = 0;
+        while processed_data < data.len() {
+            let data_to_buffer = min(
+                self.buffer.capacity() - self.buffer.len(),
+                data.len() - processed_data,
+            );
+            self.buffer
+                .extend_from_slice(&data[processed_data..processed_data + data_to_buffer]);
+            processed_data += data_to_buffer;
+            if self.buffer.len() == self.buffer.capacity() {
+                let pcm = self.pcm.as_mut().unwrap();
+                let io = pcm.io_i16().unwrap();
+                match io.writei(&self.buffer) {
+                    Ok(_) => (),
+                    Err(err) => pcm.try_recover(err, false).unwrap(),
+                }
+                self.buffer.clear();
+            }
         }
 
         Ok(())

playback/src/player.rs

@@ -33,6 +33,15 @@ pub struct Player {
     play_request_id_generator: SeqGenerator<u64>,
 }
 
+#[derive(PartialEq, Debug, Clone, Copy)]
+pub enum SinkStatus {
+    Running,
+    Closed,
+    TemporarilyClosed,
+}
+
+pub type SinkEventCallback = Box<dyn Fn(SinkStatus) + Send>;
+
 struct PlayerInternal {
     session: Session,
     config: PlayerConfig,
@@ -41,7 +50,8 @@ struct PlayerInternal {
     state: PlayerState,
     preload: PlayerPreload,
     sink: Box<dyn Sink>,
-    sink_running: bool,
+    sink_status: SinkStatus,
+    sink_event_callback: Option<SinkEventCallback>,
     audio_filter: Option<Box<dyn AudioFilter + Send>>,
     event_senders: Vec<futures::sync::mpsc::UnboundedSender<PlayerEvent>>,
 }
@@ -61,6 +71,7 @@ enum PlayerCommand {
     Stop,
     Seek(u32),
     AddEventSender(futures::sync::mpsc::UnboundedSender<PlayerEvent>),
+    SetSinkEventCallback(Option<SinkEventCallback>),
     EmitVolumeSetEvent(u16),
 }
 
@@ -123,6 +134,11 @@ pub enum PlayerEvent {
         play_request_id: u64,
         track_id: SpotifyId,
     },
+    // The player was unable to load the requested track.
+    Unavailable {
+        play_request_id: u64,
+        track_id: SpotifyId,
+    },
     // The mixer volume was set to a new level.
     VolumeSet {
         volume: u16,
@@ -136,6 +152,9 @@ impl PlayerEvent {
             Loading {
                 play_request_id, ..
             }
+            | Unavailable {
+                play_request_id, ..
+            }
             | Started {
                 play_request_id, ..
             }
@@ -232,7 +251,8 @@ impl Player {
                 state: PlayerState::Stopped,
                 preload: PlayerPreload::None,
                 sink: sink_builder(),
-                sink_running: false,
+                sink_status: SinkStatus::Closed,
+                sink_event_callback: None,
                 audio_filter: audio_filter,
                 event_senders: [event_sender].to_vec(),
             };
@@ -308,6 +328,10 @@ impl Player {
         Box::new(result)
     }
 
+    pub fn set_sink_event_callback(&self, callback: Option<SinkEventCallback>) {
+        self.command(PlayerCommand::SetSinkEventCallback(callback));
+    }
+
     pub fn emit_volume_set_event(&self, volume: u16) {
         self.command(PlayerCommand::EmitVolumeSetEvent(volume));
     }
@@ -398,6 +422,7 @@ impl PlayerState {
         }
     }
 
+    #[allow(dead_code)]
     fn is_stopped(&self) -> bool {
         use self::PlayerState::*;
         match *self {
@@ -406,6 +431,14 @@ impl PlayerState {
         }
     }
 
+    fn is_loading(&self) -> bool {
+        use self::PlayerState::*;
+        match *self {
+            Loading { .. } => true,
+            _ => false,
+        }
+    }
+
     fn decoder(&mut self) -> Option<&mut Decoder> {
         use self::PlayerState::*;
         match *self {
@@ -748,8 +781,12 @@ impl Future for PlayerInternal {
                     }
                     Ok(Async::NotReady) => (),
                     Err(_) => {
-                        self.handle_player_stop();
-                        assert!(self.state.is_stopped());
+                        warn!("Unable to load <{:?}>\nSkipping to next track", track_id);
+                        assert!(self.state.is_loading());
+                        self.send_event(PlayerEvent::EndOfTrack {
+                            track_id,
+                            play_request_id,
+                        })
                     }
                 }
             }
@@ -769,7 +806,21 @@ impl Future for PlayerInternal {
                     }
                     Ok(Async::NotReady) => (),
                     Err(_) => {
+                        debug!("Unable to preload {:?}", track_id);
                         self.preload = PlayerPreload::None;
+                        // Let Spirc know that the track was unavailable.
+                        if let PlayerState::Playing {
+                            play_request_id, ..
+                        }
+                        | PlayerState::Paused {
+                            play_request_id, ..
+                        } = self.state
+                        {
+                            self.send_event(PlayerEvent::Unavailable {
+                                track_id,
+                                play_request_id,
+                            });
+                        }
                     }
                 }
             }
@@ -882,20 +933,41 @@ impl PlayerInternal {
     }
 
     fn ensure_sink_running(&mut self) {
-        if !self.sink_running {
+        if self.sink_status != SinkStatus::Running {
             trace!("== Starting sink ==");
+            if let Some(callback) = &mut self.sink_event_callback {
+                callback(SinkStatus::Running);
+            }
             match self.sink.start() {
-                Ok(()) => self.sink_running = true,
+                Ok(()) => self.sink_status = SinkStatus::Running,
                 Err(err) => error!("Could not start audio: {}", err),
             }
         }
     }
 
-    fn ensure_sink_stopped(&mut self) {
-        if self.sink_running {
-            trace!("== Stopping sink ==");
-            self.sink.stop().unwrap();
-            self.sink_running = false;
+    fn ensure_sink_stopped(&mut self, temporarily: bool) {
+        match self.sink_status {
+            SinkStatus::Running => {
+                trace!("== Stopping sink ==");
+                self.sink.stop().unwrap();
+                self.sink_status = if temporarily {
+                    SinkStatus::TemporarilyClosed
+                } else {
+                    SinkStatus::Closed
+                };
+                if let Some(callback) = &mut self.sink_event_callback {
+                    callback(self.sink_status);
+                }
+            }
+            SinkStatus::TemporarilyClosed => {
+                if !temporarily {
+                    self.sink_status = SinkStatus::Closed;
+                    if let Some(callback) = &mut self.sink_event_callback {
+                        callback(SinkStatus::Closed);
+                    }
+                }
+            }
+            SinkStatus::Closed => (),
         }
     }
 
@@ -921,7 +993,7 @@ impl PlayerInternal {
                 play_request_id,
                 ..
             } => {
-                self.ensure_sink_stopped();
+                self.ensure_sink_stopped(false);
                 self.send_event(PlayerEvent::Stopped {
                     track_id,
                     play_request_id,
@@ -968,7 +1040,7 @@ impl PlayerInternal {
         {
             self.state.playing_to_paused();
 
-            self.ensure_sink_stopped();
+            self.ensure_sink_stopped(false);
             let position_ms = Self::position_pcm_to_ms(stream_position_pcm);
             self.send_event(PlayerEvent::Paused {
                 track_id,
@@ -997,7 +1069,7 @@ impl PlayerInternal {
 
                     if let Err(err) = self.sink.write(&packet.data()) {
                         error!("Could not write audio: {}", err);
-                        self.ensure_sink_stopped();
+                        self.ensure_sink_stopped(false);
                     }
                 }
             }
@@ -1055,7 +1127,7 @@ impl PlayerInternal {
                 suggested_to_preload_next_track: false,
             };
         } else {
-            self.ensure_sink_stopped();
+            self.ensure_sink_stopped(false);
 
             self.state = PlayerState::Paused {
                 track_id: track_id,
@@ -1086,7 +1158,7 @@ impl PlayerInternal {
         position_ms: u32,
     ) {
         if !self.config.gapless {
-            self.ensure_sink_stopped();
+            self.ensure_sink_stopped(play);
         }
         // emit the correct player event
         match self.state {
@@ -1254,7 +1326,7 @@ impl PlayerInternal {
 
         // We need to load the track - either from scratch or by completing a preload.
         // In any case we go into a Loading state to load the track.
-        self.ensure_sink_stopped();
+        self.ensure_sink_stopped(play);
 
         self.send_event(PlayerEvent::Loading {
             track_id,
@@ -1302,7 +1374,6 @@ impl PlayerInternal {
     fn handle_command_preload(&mut self, track_id: SpotifyId) {
         debug!("Preloading track");
         let mut preload_track = true;
-
         // check whether the track is already loaded somewhere or being loaded.
         if let PlayerPreload::Loading {
             track_id: currently_loading,
@@ -1436,6 +1507,8 @@ impl PlayerInternal {
 
             PlayerCommand::AddEventSender(sender) => self.event_senders.push(sender),
 
+            PlayerCommand::SetSinkEventCallback(callback) => self.sink_event_callback = callback,
+
             PlayerCommand::EmitVolumeSetEvent(volume) => {
                 self.send_event(PlayerEvent::VolumeSet { volume })
             }
@@ -1540,6 +1613,9 @@ impl ::std::fmt::Debug for PlayerCommand {
             PlayerCommand::Stop => f.debug_tuple("Stop").finish(),
             PlayerCommand::Seek(position) => f.debug_tuple("Seek").field(&position).finish(),
             PlayerCommand::AddEventSender(_) => f.debug_tuple("AddEventSender").finish(),
+            PlayerCommand::SetSinkEventCallback(_) => {
+                f.debug_tuple("SetSinkEventCallback").finish()
+            }
             PlayerCommand::EmitVolumeSetEvent(volume) => {
                 f.debug_tuple("VolumeSet").field(&volume).finish()
             }
@@ -1547,6 +1623,51 @@ impl ::std::fmt::Debug for PlayerCommand {
     }
 }
 
+impl ::std::fmt::Debug for PlayerState {
+    fn fmt(&self, f: &mut ::std::fmt::Formatter<'_>) -> ::std::fmt::Result {
+        use PlayerState::*;
+        match *self {
+            Stopped => f.debug_struct("Stopped").finish(),
+            Loading {
+                track_id,
+                play_request_id,
+                ..
+            } => f
+                .debug_struct("Loading")
+                .field("track_id", &track_id)
+                .field("play_request_id", &play_request_id)
+                .finish(),
+            Paused {
+                track_id,
+                play_request_id,
+                ..
+            } => f
+                .debug_struct("Paused")
+                .field("track_id", &track_id)
+                .field("play_request_id", &play_request_id)
+                .finish(),
+            Playing {
+                track_id,
+                play_request_id,
+                ..
+            } => f
+                .debug_struct("Playing")
+                .field("track_id", &track_id)
+                .field("play_request_id", &play_request_id)
+                .finish(),
+            EndOfTrack {
+                track_id,
+                play_request_id,
+                ..
+            } => f
+                .debug_struct("EndOfTrack")
+                .field("track_id", &track_id)
+                .field("play_request_id", &play_request_id)
+                .finish(),
+            Invalid => f.debug_struct("Invalid").finish(),
+        }
+    }
+}
 struct Subfile<T: Read + Seek> {
     stream: T,
     offset: u64,

protocol/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "librespot-protocol"
-version = "0.1.1"
+version = "0.1.2"
 authors = ["Paul Liétar <paul@lietar.net>"]
 build = "build.rs"
 description="The protobuf logic for communicating with Spotify servers"
@@ -8,8 +8,8 @@ license="MIT"
 edition = "2018"
 
 [dependencies]
-protobuf = "~2.8.1"
+protobuf = "~2.14.0"
 
 [build-dependencies]
-protobuf-codegen-pure = "~2.8.1"
-protobuf-codegen = "~2.8.1"
+protobuf-codegen-pure = "~2.14.0"
+protobuf-codegen = "~2.14.0"

protocol/build.rs

@@ -29,20 +29,21 @@ fn main() {
         }
 
         // Build the paths to relevant files.
-        let src = &format!("proto/{}.proto", name);
-        let dest = &format!("src/{}.rs", name);
-
+        let src_fname = &format!("proto/{}.proto", name);
+        let dest_fname = &format!("src/{}.rs", name);
+        let src = Path::new(src_fname);
+        let dest = Path::new(dest_fname);
         // Get the contents of the existing generated file.
         let mut existing = "".to_string();
-        if Path::new(dest).exists() {
+        if dest.exists() {
             // Removing CRLF line endings if present.
             existing = read_to_string(dest).unwrap().replace("\r\n", "\n");
         }
 
-        println!("Regenerating {} from {}", dest, src);
+        println!("Regenerating {} from {}", dest.display(), src.display());
 
         // Parse the proto files as the protobuf-codegen-pure crate does.
-        let p = parse_and_typecheck(&["proto"], &[src]).expect("protoc");
+        let p = parse_and_typecheck(&[&Path::new("proto")], &[src]).expect("protoc");
         // But generate them with the protobuf-codegen crate directly.
         // Then we can keep the result in-memory.
         let result = protobuf_codegen::gen(&p.file_descriptors, &p.relative_paths, &customizations);

protocol/src/playlist4issues.rs

@@ -1,7 +1,7 @@
-// This file is generated by rust-protobuf 2.8.1. Do not edit
+// This file is generated by rust-protobuf 2.14.0. Do not edit
 // @generated
 
-// https://github.com/Manishearth/rust-clippy/issues/702
+// https://github.com/rust-lang/rust-clippy/issues/702
 #![allow(unknown_lints)]
 #![allow(clippy::all)]
 
@@ -24,9 +24,9 @@ use protobuf::ProtobufEnum as ProtobufEnum_imported_for_functions;
 
 /// Generated files are compatible only with the same version
 /// of protobuf runtime.
-const _PROTOBUF_VERSION_CHECK: () = ::protobuf::VERSION_2_8_1;
+// const _PROTOBUF_VERSION_CHECK: () = ::protobuf::VERSION_2_14_0;
 
-#[derive(PartialEq,Clone,Default)]
+#[derive(PartialEq, Clone, Default)]
 pub struct ClientIssue {
     // message fields
     level: ::std::option::Option<ClientIssue_Level>,
@@ -50,7 +50,6 @@ impl ClientIssue {
 
     // optional .ClientIssue.Level level = 1;
 
-
     pub fn get_level(&self) -> ClientIssue_Level {
         self.level.unwrap_or(ClientIssue_Level::LEVEL_UNKNOWN)
     }
@@ -69,7 +68,6 @@ impl ClientIssue {
 
     // optional .ClientIssue.Code code = 2;
 
-
     pub fn get_code(&self) -> ClientIssue_Code {
         self.code.unwrap_or(ClientIssue_Code::CODE_UNKNOWN)
     }
@@ -88,7 +86,6 @@ impl ClientIssue {
 
     // optional int32 repeatCount = 3;
 
-
     pub fn get_repeatCount(&self) -> i32 {
         self.repeatCount.unwrap_or(0)
     }
@@ -111,26 +108,44 @@ impl ::protobuf::Message for ClientIssue {
         true
     }
 
-    fn merge_from(&mut self, is: &mut ::protobuf::CodedInputStream<'_>) -> ::protobuf::ProtobufResult<()> {
+    fn merge_from(
+        &mut self,
+        is: &mut ::protobuf::CodedInputStream<'_>,
+    ) -> ::protobuf::ProtobufResult<()> {
         while !is.eof()? {
             let (field_number, wire_type) = is.read_tag_unpack()?;
             match field_number {
-                1 => {
-                    ::protobuf::rt::read_proto2_enum_with_unknown_fields_into(wire_type, is, &mut self.level, 1, &mut self.unknown_fields)?
-                },
-                2 => {
-                    ::protobuf::rt::read_proto2_enum_with_unknown_fields_into(wire_type, is, &mut self.code, 2, &mut self.unknown_fields)?
-                },
+                1 => ::protobuf::rt::read_proto2_enum_with_unknown_fields_into(
+                    wire_type,
+                    is,
+                    &mut self.level,
+                    1,
+                    &mut self.unknown_fields,
+                )?,
+                2 => ::protobuf::rt::read_proto2_enum_with_unknown_fields_into(
+                    wire_type,
+                    is,
+                    &mut self.code,
+                    2,
+                    &mut self.unknown_fields,
+                )?,
                 3 => {
                     if wire_type != ::protobuf::wire_format::WireTypeVarint {
-                        return ::std::result::Result::Err(::protobuf::rt::unexpected_wire_type(wire_type));
+                        return ::std::result::Result::Err(::protobuf::rt::unexpected_wire_type(
+                            wire_type,
+                        ));
                     }
                     let tmp = is.read_int32()?;
                     self.repeatCount = ::std::option::Option::Some(tmp);
-                },
+                }
                 _ => {
-                    ::protobuf::rt::read_unknown_or_skip_group(field_number, wire_type, is, self.mut_unknown_fields())?;
-                },
+                    ::protobuf::rt::read_unknown_or_skip_group(
+                        field_number,
+                        wire_type,
+                        is,
+                        self.mut_unknown_fields(),
+                    )?;
+                }
             };
         }
         ::std::result::Result::Ok(())
@@ -154,7 +169,10 @@ impl ::protobuf::Message for ClientIssue {
         my_size
     }
 
-    fn write_to_with_cached_sizes(&self, os: &mut ::protobuf::CodedOutputStream<'_>) -> ::protobuf::ProtobufResult<()> {
+    fn write_to_with_cached_sizes(
+        &self,
+        os: &mut ::protobuf::CodedOutputStream<'_>,
+    ) -> ::protobuf::ProtobufResult<()> {
         if let Some(v) = self.level {
             os.write_enum(1, v.value())?;
         }
@@ -199,45 +217,47 @@ impl ::protobuf::Message for ClientIssue {
     }
 
     fn descriptor_static() -> &'static ::protobuf::reflect::MessageDescriptor {
-        static mut descriptor: ::protobuf::lazy::Lazy<::protobuf::reflect::MessageDescriptor> = ::protobuf::lazy::Lazy {
-            lock: ::protobuf::lazy::ONCE_INIT,
-            ptr: 0 as *const ::protobuf::reflect::MessageDescriptor,
-        };
+        static mut descriptor: ::protobuf::lazy::Lazy<::protobuf::reflect::MessageDescriptor> =
+            ::protobuf::lazy::Lazy::INIT;
         unsafe {
             descriptor.get(|| {
                 let mut fields = ::std::vec::Vec::new();
-                fields.push(::protobuf::reflect::accessor::make_option_accessor::<_, ::protobuf::types::ProtobufTypeEnum<ClientIssue_Level>>(
+                fields.push(::protobuf::reflect::accessor::make_option_accessor::<
+                    _,
+                    ::protobuf::types::ProtobufTypeEnum<ClientIssue_Level>,
+                >(
                     "level",
-                    |m: &ClientIssue| { &m.level },
-                    |m: &mut ClientIssue| { &mut m.level },
+                    |m: &ClientIssue| &m.level,
+                    |m: &mut ClientIssue| &mut m.level,
                 ));
-                fields.push(::protobuf::reflect::accessor::make_option_accessor::<_, ::protobuf::types::ProtobufTypeEnum<ClientIssue_Code>>(
+                fields.push(::protobuf::reflect::accessor::make_option_accessor::<
+                    _,
+                    ::protobuf::types::ProtobufTypeEnum<ClientIssue_Code>,
+                >(
                     "code",
-                    |m: &ClientIssue| { &m.code },
-                    |m: &mut ClientIssue| { &mut m.code },
+                    |m: &ClientIssue| &m.code,
+                    |m: &mut ClientIssue| &mut m.code,
                 ));
-                fields.push(::protobuf::reflect::accessor::make_option_accessor::<_, ::protobuf::types::ProtobufTypeInt32>(
+                fields.push(::protobuf::reflect::accessor::make_option_accessor::<
+                    _,
+                    ::protobuf::types::ProtobufTypeInt32,
+                >(
                     "repeatCount",
-                    |m: &ClientIssue| { &m.repeatCount },
-                    |m: &mut ClientIssue| { &mut m.repeatCount },
+                    |m: &ClientIssue| &m.repeatCount,
+                    |m: &mut ClientIssue| &mut m.repeatCount,
                 ));
-                ::protobuf::reflect::MessageDescriptor::new::<ClientIssue>(
+                ::protobuf::reflect::MessageDescriptor::new_pb_name::<ClientIssue>(
                     "ClientIssue",
                     fields,
-                    file_descriptor_proto()
+                    file_descriptor_proto(),
                 )
             })
         }
     }
 
     fn default_instance() -> &'static ClientIssue {
-        static mut instance: ::protobuf::lazy::Lazy<ClientIssue> = ::protobuf::lazy::Lazy {
-            lock: ::protobuf::lazy::ONCE_INIT,
-            ptr: 0 as *const ClientIssue,
-        };
-        unsafe {
-            instance.get(ClientIssue::new)
-        }
+        static mut instance: ::protobuf::lazy::Lazy<ClientIssue> = ::protobuf::lazy::Lazy::INIT;
+        unsafe { instance.get(ClientIssue::new) }
     }
 }
 
@@ -257,12 +277,12 @@ impl ::std::fmt::Debug for ClientIssue {
 }
 
 impl ::protobuf::reflect::ProtobufValue for ClientIssue {
-    fn as_ref(&self) -> ::protobuf::reflect::ProtobufValueRef {
-        ::protobuf::reflect::ProtobufValueRef::Message(self)
+    fn as_ref(&self) -> ::protobuf::reflect::ReflectValueRef {
+        ::protobuf::reflect::ReflectValueRef::Message(self)
     }
 }
 
-#[derive(Clone,PartialEq,Eq,Debug,Hash)]
+#[derive(Clone, PartialEq, Eq, Debug, Hash)]
 pub enum ClientIssue_Level {
     LEVEL_UNKNOWN = 0,
     LEVEL_DEBUG = 1,
@@ -285,7 +305,7 @@ impl ::protobuf::ProtobufEnum for ClientIssue_Level {
             3 => ::std::option::Option::Some(ClientIssue_Level::LEVEL_NOTICE),
             4 => ::std::option::Option::Some(ClientIssue_Level::LEVEL_WARNING),
             5 => ::std::option::Option::Some(ClientIssue_Level::LEVEL_ERROR),
-            _ => ::std::option::Option::None
+            _ => ::std::option::Option::None,
         }
     }
 
@@ -302,20 +322,20 @@ impl ::protobuf::ProtobufEnum for ClientIssue_Level {
     }
 
     fn enum_descriptor_static() -> &'static ::protobuf::reflect::EnumDescriptor {
-        static mut descriptor: ::protobuf::lazy::Lazy<::protobuf::reflect::EnumDescriptor> = ::protobuf::lazy::Lazy {
-            lock: ::protobuf::lazy::ONCE_INIT,
-            ptr: 0 as *const ::protobuf::reflect::EnumDescriptor,
-        };
+        static mut descriptor: ::protobuf::lazy::Lazy<::protobuf::reflect::EnumDescriptor> =
+            ::protobuf::lazy::Lazy::INIT;
         unsafe {
             descriptor.get(|| {
-                ::protobuf::reflect::EnumDescriptor::new("ClientIssue_Level", file_descriptor_proto())
+                ::protobuf::reflect::EnumDescriptor::new_pb_name::<ClientIssue_Level>(
+                    "ClientIssue.Level",
+                    file_descriptor_proto(),
+                )
             })
         }
     }
 }
 
-impl ::std::marker::Copy for ClientIssue_Level {
-}
+impl ::std::marker::Copy for ClientIssue_Level {}
 
 impl ::std::default::Default for ClientIssue_Level {
     fn default() -> Self {
@@ -324,12 +344,12 @@ impl ::std::default::Default for ClientIssue_Level {
 }
 
 impl ::protobuf::reflect::ProtobufValue for ClientIssue_Level {
-    fn as_ref(&self) -> ::protobuf::reflect::ProtobufValueRef {
-        ::protobuf::reflect::ProtobufValueRef::Enum(self.descriptor())
+    fn as_ref(&self) -> ::protobuf::reflect::ReflectValueRef {
+        ::protobuf::reflect::ReflectValueRef::Enum(self.descriptor())
     }
 }
 
-#[derive(Clone,PartialEq,Eq,Debug,Hash)]
+#[derive(Clone, PartialEq, Eq, Debug, Hash)]
 pub enum ClientIssue_Code {
     CODE_UNKNOWN = 0,
     CODE_INDEX_OUT_OF_BOUNDS = 1,
@@ -352,7 +372,7 @@ impl ::protobuf::ProtobufEnum for ClientIssue_Code {
             3 => ::std::option::Option::Some(ClientIssue_Code::CODE_CACHED_CHANGE),
             4 => ::std::option::Option::Some(ClientIssue_Code::CODE_OFFLINE_CHANGE),
             5 => ::std::option::Option::Some(ClientIssue_Code::CODE_CONCURRENT_CHANGE),
-            _ => ::std::option::Option::None
+            _ => ::std::option::Option::None,
         }
     }
 
@@ -369,20 +389,20 @@ impl ::protobuf::ProtobufEnum for ClientIssue_Code {
     }
 
     fn enum_descriptor_static() -> &'static ::protobuf::reflect::EnumDescriptor {
-        static mut descriptor: ::protobuf::lazy::Lazy<::protobuf::reflect::EnumDescriptor> = ::protobuf::lazy::Lazy {
-            lock: ::protobuf::lazy::ONCE_INIT,
-            ptr: 0 as *const ::protobuf::reflect::EnumDescriptor,
-        };
+        static mut descriptor: ::protobuf::lazy::Lazy<::protobuf::reflect::EnumDescriptor> =
+            ::protobuf::lazy::Lazy::INIT;
         unsafe {
             descriptor.get(|| {
-                ::protobuf::reflect::EnumDescriptor::new("ClientIssue_Code", file_descriptor_proto())
+                ::protobuf::reflect::EnumDescriptor::new_pb_name::<ClientIssue_Code>(
+                    "ClientIssue.Code",
+                    file_descriptor_proto(),
+                )
             })
         }
     }
 }
 
-impl ::std::marker::Copy for ClientIssue_Code {
-}
+impl ::std::marker::Copy for ClientIssue_Code {}
 
 impl ::std::default::Default for ClientIssue_Code {
     fn default() -> Self {
@@ -391,12 +411,12 @@ impl ::std::default::Default for ClientIssue_Code {
 }
 
 impl ::protobuf::reflect::ProtobufValue for ClientIssue_Code {
-    fn as_ref(&self) -> ::protobuf::reflect::ProtobufValueRef {
-        ::protobuf::reflect::ProtobufValueRef::Enum(self.descriptor())
+    fn as_ref(&self) -> ::protobuf::reflect::ReflectValueRef {
+        ::protobuf::reflect::ReflectValueRef::Enum(self.descriptor())
     }
 }
 
-#[derive(PartialEq,Clone,Default)]
+#[derive(PartialEq, Clone, Default)]
 pub struct ClientResolveAction {
     // message fields
     code: ::std::option::Option<ClientResolveAction_Code>,
@@ -419,7 +439,6 @@ impl ClientResolveAction {
 
     // optional .ClientResolveAction.Code code = 1;
 
-
     pub fn get_code(&self) -> ClientResolveAction_Code {
         self.code.unwrap_or(ClientResolveAction_Code::CODE_UNKNOWN)
     }
@@ -438,9 +457,9 @@ impl ClientResolveAction {
 
     // optional .ClientResolveAction.Initiator initiator = 2;
 
-
     pub fn get_initiator(&self) -> ClientResolveAction_Initiator {
-        self.initiator.unwrap_or(ClientResolveAction_Initiator::INITIATOR_UNKNOWN)
+        self.initiator
+            .unwrap_or(ClientResolveAction_Initiator::INITIATOR_UNKNOWN)
     }
     pub fn clear_initiator(&mut self) {
         self.initiator = ::std::option::Option::None;
@@ -461,19 +480,35 @@ impl ::protobuf::Message for ClientResolveAction {
         true
     }
 
-    fn merge_from(&mut self, is: &mut ::protobuf::CodedInputStream<'_>) -> ::protobuf::ProtobufResult<()> {
+    fn merge_from(
+        &mut self,
+        is: &mut ::protobuf::CodedInputStream<'_>,
+    ) -> ::protobuf::ProtobufResult<()> {
         while !is.eof()? {
             let (field_number, wire_type) = is.read_tag_unpack()?;
             match field_number {
-                1 => {
-                    ::protobuf::rt::read_proto2_enum_with_unknown_fields_into(wire_type, is, &mut self.code, 1, &mut self.unknown_fields)?
-                },
-                2 => {
-                    ::protobuf::rt::read_proto2_enum_with_unknown_fields_into(wire_type, is, &mut self.initiator, 2, &mut self.unknown_fields)?
-                },
+                1 => ::protobuf::rt::read_proto2_enum_with_unknown_fields_into(
+                    wire_type,
+                    is,
+                    &mut self.code,
+                    1,
+                    &mut self.unknown_fields,
+                )?,
+                2 => ::protobuf::rt::read_proto2_enum_with_unknown_fields_into(
+                    wire_type,
+                    is,
+                    &mut self.initiator,
+                    2,
+                    &mut self.unknown_fields,
+                )?,
                 _ => {
-                    ::protobuf::rt::read_unknown_or_skip_group(field_number, wire_type, is, self.mut_unknown_fields())?;
-                },
+                    ::protobuf::rt::read_unknown_or_skip_group(
+                        field_number,
+                        wire_type,
+                        is,
+                        self.mut_unknown_fields(),
+                    )?;
+                }
             };
         }
         ::std::result::Result::Ok(())
@@ -494,7 +529,10 @@ impl ::protobuf::Message for ClientResolveAction {
         my_size
     }
 
-    fn write_to_with_cached_sizes(&self, os: &mut ::protobuf::CodedOutputStream<'_>) -> ::protobuf::ProtobufResult<()> {
+    fn write_to_with_cached_sizes(
+        &self,
+        os: &mut ::protobuf::CodedOutputStream<'_>,
+    ) -> ::protobuf::ProtobufResult<()> {
         if let Some(v) = self.code {
             os.write_enum(1, v.value())?;
         }
@@ -536,40 +574,40 @@ impl ::protobuf::Message for ClientResolveAction {
     }
 
     fn descriptor_static() -> &'static ::protobuf::reflect::MessageDescriptor {
-        static mut descriptor: ::protobuf::lazy::Lazy<::protobuf::reflect::MessageDescriptor> = ::protobuf::lazy::Lazy {
-            lock: ::protobuf::lazy::ONCE_INIT,
-            ptr: 0 as *const ::protobuf::reflect::MessageDescriptor,
-        };
+        static mut descriptor: ::protobuf::lazy::Lazy<::protobuf::reflect::MessageDescriptor> =
+            ::protobuf::lazy::Lazy::INIT;
         unsafe {
             descriptor.get(|| {
                 let mut fields = ::std::vec::Vec::new();
-                fields.push(::protobuf::reflect::accessor::make_option_accessor::<_, ::protobuf::types::ProtobufTypeEnum<ClientResolveAction_Code>>(
+                fields.push(::protobuf::reflect::accessor::make_option_accessor::<
+                    _,
+                    ::protobuf::types::ProtobufTypeEnum<ClientResolveAction_Code>,
+                >(
                     "code",
-                    |m: &ClientResolveAction| { &m.code },
-                    |m: &mut ClientResolveAction| { &mut m.code },
+                    |m: &ClientResolveAction| &m.code,
+                    |m: &mut ClientResolveAction| &mut m.code,
                 ));
-                fields.push(::protobuf::reflect::accessor::make_option_accessor::<_, ::protobuf::types::ProtobufTypeEnum<ClientResolveAction_Initiator>>(
+                fields.push(::protobuf::reflect::accessor::make_option_accessor::<
+                    _,
+                    ::protobuf::types::ProtobufTypeEnum<ClientResolveAction_Initiator>,
+                >(
                     "initiator",
-                    |m: &ClientResolveAction| { &m.initiator },
-                    |m: &mut ClientResolveAction| { &mut m.initiator },
+                    |m: &ClientResolveAction| &m.initiator,
+                    |m: &mut ClientResolveAction| &mut m.initiator,
                 ));
-                ::protobuf::reflect::MessageDescriptor::new::<ClientResolveAction>(
+                ::protobuf::reflect::MessageDescriptor::new_pb_name::<ClientResolveAction>(
                     "ClientResolveAction",
                     fields,
-                    file_descriptor_proto()
+                    file_descriptor_proto(),
                 )
             })
         }
     }
 
     fn default_instance() -> &'static ClientResolveAction {
-        static mut instance: ::protobuf::lazy::Lazy<ClientResolveAction> = ::protobuf::lazy::Lazy {
-            lock: ::protobuf::lazy::ONCE_INIT,
-            ptr: 0 as *const ClientResolveAction,
-        };
-        unsafe {
-            instance.get(ClientResolveAction::new)
-        }
+        static mut instance: ::protobuf::lazy::Lazy<ClientResolveAction> =
+            ::protobuf::lazy::Lazy::INIT;
+        unsafe { instance.get(ClientResolveAction::new) }
     }
 }
 
@@ -588,12 +626,12 @@ impl ::std::fmt::Debug for ClientResolveAction {
 }
 
 impl ::protobuf::reflect::ProtobufValue for ClientResolveAction {
-    fn as_ref(&self) -> ::protobuf::reflect::ProtobufValueRef {
-        ::protobuf::reflect::ProtobufValueRef::Message(self)
+    fn as_ref(&self) -> ::protobuf::reflect::ReflectValueRef {
+        ::protobuf::reflect::ReflectValueRef::Message(self)
     }
 }
 
-#[derive(Clone,PartialEq,Eq,Debug,Hash)]
+#[derive(Clone, PartialEq, Eq, Debug, Hash)]
 pub enum ClientResolveAction_Code {
     CODE_UNKNOWN = 0,
     CODE_NO_ACTION = 1,
@@ -618,7 +656,7 @@ impl ::protobuf::ProtobufEnum for ClientResolveAction_Code {
             4 => ::std::option::Option::Some(ClientResolveAction_Code::CODE_DISCARD_LOCAL_CHANGES),
             5 => ::std::option::Option::Some(ClientResolveAction_Code::CODE_SEND_DUMP),
             6 => ::std::option::Option::Some(ClientResolveAction_Code::CODE_DISPLAY_ERROR_MESSAGE),
-            _ => ::std::option::Option::None
+            _ => ::std::option::Option::None,
         }
     }
 
@@ -636,20 +674,20 @@ impl ::protobuf::ProtobufEnum for ClientResolveAction_Code {
     }
 
     fn enum_descriptor_static() -> &'static ::protobuf::reflect::EnumDescriptor {
-        static mut descriptor: ::protobuf::lazy::Lazy<::protobuf::reflect::EnumDescriptor> = ::protobuf::lazy::Lazy {
-            lock: ::protobuf::lazy::ONCE_INIT,
-            ptr: 0 as *const ::protobuf::reflect::EnumDescriptor,
-        };
+        static mut descriptor: ::protobuf::lazy::Lazy<::protobuf::reflect::EnumDescriptor> =
+            ::protobuf::lazy::Lazy::INIT;
         unsafe {
             descriptor.get(|| {
-                ::protobuf::reflect::EnumDescriptor::new("ClientResolveAction_Code", file_descriptor_proto())
+                ::protobuf::reflect::EnumDescriptor::new_pb_name::<ClientResolveAction_Code>(
+                    "ClientResolveAction.Code",
+                    file_descriptor_proto(),
+                )
             })
         }
     }
 }
 
-impl ::std::marker::Copy for ClientResolveAction_Code {
-}
+impl ::std::marker::Copy for ClientResolveAction_Code {}
 
 impl ::std::default::Default for ClientResolveAction_Code {
     fn default() -> Self {
@@ -658,12 +696,12 @@ impl ::std::default::Default for ClientResolveAction_Code {
 }
 
 impl ::protobuf::reflect::ProtobufValue for ClientResolveAction_Code {
-    fn as_ref(&self) -> ::protobuf::reflect::ProtobufValueRef {
-        ::protobuf::reflect::ProtobufValueRef::Enum(self.descriptor())
+    fn as_ref(&self) -> ::protobuf::reflect::ReflectValueRef {
+        ::protobuf::reflect::ReflectValueRef::Enum(self.descriptor())
     }
 }
 
-#[derive(Clone,PartialEq,Eq,Debug,Hash)]
+#[derive(Clone, PartialEq, Eq, Debug, Hash)]
 pub enum ClientResolveAction_Initiator {
     INITIATOR_UNKNOWN = 0,
     INITIATOR_SERVER = 1,
@@ -680,7 +718,7 @@ impl ::protobuf::ProtobufEnum for ClientResolveAction_Initiator {
             0 => ::std::option::Option::Some(ClientResolveAction_Initiator::INITIATOR_UNKNOWN),
             1 => ::std::option::Option::Some(ClientResolveAction_Initiator::INITIATOR_SERVER),
             2 => ::std::option::Option::Some(ClientResolveAction_Initiator::INITIATOR_CLIENT),
-            _ => ::std::option::Option::None
+            _ => ::std::option::Option::None,
         }
     }
 
@@ -694,20 +732,20 @@ impl ::protobuf::ProtobufEnum for ClientResolveAction_Initiator {
     }
 
     fn enum_descriptor_static() -> &'static ::protobuf::reflect::EnumDescriptor {
-        static mut descriptor: ::protobuf::lazy::Lazy<::protobuf::reflect::EnumDescriptor> = ::protobuf::lazy::Lazy {
-            lock: ::protobuf::lazy::ONCE_INIT,
-            ptr: 0 as *const ::protobuf::reflect::EnumDescriptor,
-        };
+        static mut descriptor: ::protobuf::lazy::Lazy<::protobuf::reflect::EnumDescriptor> =
+            ::protobuf::lazy::Lazy::INIT;
         unsafe {
             descriptor.get(|| {
-                ::protobuf::reflect::EnumDescriptor::new("ClientResolveAction_Initiator", file_descriptor_proto())
+                ::protobuf::reflect::EnumDescriptor::new_pb_name::<ClientResolveAction_Initiator>(
+                    "ClientResolveAction.Initiator",
+                    file_descriptor_proto(),
+                )
             })
         }
     }
 }
 
-impl ::std::marker::Copy for ClientResolveAction_Initiator {
-}
+impl ::std::marker::Copy for ClientResolveAction_Initiator {}
 
 impl ::std::default::Default for ClientResolveAction_Initiator {
     fn default() -> Self {
@@ -716,8 +754,8 @@ impl ::std::default::Default for ClientResolveAction_Initiator {
 }
 
 impl ::protobuf::reflect::ProtobufValue for ClientResolveAction_Initiator {
-    fn as_ref(&self) -> ::protobuf::reflect::ProtobufValueRef {
-        ::protobuf::reflect::ProtobufValueRef::Enum(self.descriptor())
+    fn as_ref(&self) -> ::protobuf::reflect::ReflectValueRef {
+        ::protobuf::reflect::ReflectValueRef::Enum(self.descriptor())
     }
 }
 
@@ -743,19 +781,14 @@ static file_descriptor_proto_data: &'static [u8] = b"\
     \x10\x01\x12\x14\n\x10INITIATOR_CLIENT\x10\x02\x1a\0:\0B\0b\x06proto2\
 ";
 
-static mut file_descriptor_proto_lazy: ::protobuf::lazy::Lazy<::protobuf::descriptor::FileDescriptorProto> = ::protobuf::lazy::Lazy {
-    lock: ::protobuf::lazy::ONCE_INIT,
-    ptr: 0 as *const ::protobuf::descriptor::FileDescriptorProto,
-};
+static mut file_descriptor_proto_lazy: ::protobuf::lazy::Lazy<
+    ::protobuf::descriptor::FileDescriptorProto,
+> = ::protobuf::lazy::Lazy::INIT;
 
 fn parse_descriptor_proto() -> ::protobuf::descriptor::FileDescriptorProto {
     ::protobuf::parse_from_bytes(file_descriptor_proto_data).unwrap()
 }
 
 pub fn file_descriptor_proto() -> &'static ::protobuf::descriptor::FileDescriptorProto {
-    unsafe {
-        file_descriptor_proto_lazy.get(|| {
-            parse_descriptor_proto()
-        })
-    }
+    unsafe { file_descriptor_proto_lazy.get(|| parse_descriptor_proto()) }
 }

protocol/src/pubsub.rs

@@ -1,7 +1,7 @@
-// This file is generated by rust-protobuf 2.8.1. Do not edit
+// This file is generated by rust-protobuf 2.14.0. Do not edit
 // @generated
 
-// https://github.com/Manishearth/rust-clippy/issues/702
+// https://github.com/rust-lang/rust-clippy/issues/702
 #![allow(unknown_lints)]
 #![allow(clippy::all)]
 
@@ -24,9 +24,9 @@ use protobuf::ProtobufEnum as ProtobufEnum_imported_for_functions;
 
 /// Generated files are compatible only with the same version
 /// of protobuf runtime.
-const _PROTOBUF_VERSION_CHECK: () = ::protobuf::VERSION_2_8_1;
+// const _PROTOBUF_VERSION_CHECK: () = ::protobuf::VERSION_2_14_0;
 
-#[derive(PartialEq,Clone,Default)]
+#[derive(PartialEq, Clone, Default)]
 pub struct Subscription {
     // message fields
     uri: ::protobuf::SingularField<::std::string::String>,
@@ -50,7 +50,6 @@ impl Subscription {
 
     // optional string uri = 1;
 
-
     pub fn get_uri(&self) -> &str {
         match self.uri.as_ref() {
             Some(v) => &v,
@@ -81,12 +80,13 @@ impl Subscription {
 
     // Take field
     pub fn take_uri(&mut self) -> ::std::string::String {
-        self.uri.take().unwrap_or_else(|| ::std::string::String::new())
+        self.uri
+            .take()
+            .unwrap_or_else(|| ::std::string::String::new())
     }
 
     // optional int32 expiry = 2;
 
-
     pub fn get_expiry(&self) -> i32 {
         self.expiry.unwrap_or(0)
     }
@@ -105,7 +105,6 @@ impl Subscription {
 
     // optional int32 status_code = 3;
 
-
     pub fn get_status_code(&self) -> i32 {
         self.status_code.unwrap_or(0)
     }
@@ -128,30 +127,42 @@ impl ::protobuf::Message for Subscription {
         true
     }
 
-    fn merge_from(&mut self, is: &mut ::protobuf::CodedInputStream<'_>) -> ::protobuf::ProtobufResult<()> {
+    fn merge_from(
+        &mut self,
+        is: &mut ::protobuf::CodedInputStream<'_>,
+    ) -> ::protobuf::ProtobufResult<()> {
         while !is.eof()? {
             let (field_number, wire_type) = is.read_tag_unpack()?;
             match field_number {
                 1 => {
                     ::protobuf::rt::read_singular_string_into(wire_type, is, &mut self.uri)?;
-                },
+                }
                 2 => {
                     if wire_type != ::protobuf::wire_format::WireTypeVarint {
-                        return ::std::result::Result::Err(::protobuf::rt::unexpected_wire_type(wire_type));
+                        return ::std::result::Result::Err(::protobuf::rt::unexpected_wire_type(
+                            wire_type,
+                        ));
                     }
                     let tmp = is.read_int32()?;
                     self.expiry = ::std::option::Option::Some(tmp);
-                },
+                }
                 3 => {
                     if wire_type != ::protobuf::wire_format::WireTypeVarint {
-                        return ::std::result::Result::Err(::protobuf::rt::unexpected_wire_type(wire_type));
+                        return ::std::result::Result::Err(::protobuf::rt::unexpected_wire_type(
+                            wire_type,
+                        ));
                     }
                     let tmp = is.read_int32()?;
                     self.status_code = ::std::option::Option::Some(tmp);
-                },
+                }
                 _ => {
-                    ::protobuf::rt::read_unknown_or_skip_group(field_number, wire_type, is, self.mut_unknown_fields())?;
-                },
+                    ::protobuf::rt::read_unknown_or_skip_group(
+                        field_number,
+                        wire_type,
+                        is,
+                        self.mut_unknown_fields(),
+                    )?;
+                }
             };
         }
         ::std::result::Result::Ok(())
@@ -175,7 +186,10 @@ impl ::protobuf::Message for Subscription {
         my_size
     }
 
-    fn write_to_with_cached_sizes(&self, os: &mut ::protobuf::CodedOutputStream<'_>) -> ::protobuf::ProtobufResult<()> {
+    fn write_to_with_cached_sizes(
+        &self,
+        os: &mut ::protobuf::CodedOutputStream<'_>,
+    ) -> ::protobuf::ProtobufResult<()> {
         if let Some(ref v) = self.uri.as_ref() {
             os.write_string(1, &v)?;
         }
@@ -220,45 +234,49 @@ impl ::protobuf::Message for Subscription {
     }
 
     fn descriptor_static() -> &'static ::protobuf::reflect::MessageDescriptor {
-        static mut descriptor: ::protobuf::lazy::Lazy<::protobuf::reflect::MessageDescriptor> = ::protobuf::lazy::Lazy {
-            lock: ::protobuf::lazy::ONCE_INIT,
-            ptr: 0 as *const ::protobuf::reflect::MessageDescriptor,
-        };
+        static mut descriptor: ::protobuf::lazy::Lazy<::protobuf::reflect::MessageDescriptor> =
+            ::protobuf::lazy::Lazy::INIT;
         unsafe {
             descriptor.get(|| {
                 let mut fields = ::std::vec::Vec::new();
-                fields.push(::protobuf::reflect::accessor::make_singular_field_accessor::<_, ::protobuf::types::ProtobufTypeString>(
-                    "uri",
-                    |m: &Subscription| { &m.uri },
-                    |m: &mut Subscription| { &mut m.uri },
-                ));
-                fields.push(::protobuf::reflect::accessor::make_option_accessor::<_, ::protobuf::types::ProtobufTypeInt32>(
+                fields.push(
+                    ::protobuf::reflect::accessor::make_singular_field_accessor::<
+                        _,
+                        ::protobuf::types::ProtobufTypeString,
+                    >(
+                        "uri",
+                        |m: &Subscription| &m.uri,
+                        |m: &mut Subscription| &mut m.uri,
+                    ),
+                );
+                fields.push(::protobuf::reflect::accessor::make_option_accessor::<
+                    _,
+                    ::protobuf::types::ProtobufTypeInt32,
+                >(
                     "expiry",
-                    |m: &Subscription| { &m.expiry },
-                    |m: &mut Subscription| { &mut m.expiry },
+                    |m: &Subscription| &m.expiry,
+                    |m: &mut Subscription| &mut m.expiry,
                 ));
-                fields.push(::protobuf::reflect::accessor::make_option_accessor::<_, ::protobuf::types::ProtobufTypeInt32>(
+                fields.push(::protobuf::reflect::accessor::make_option_accessor::<
+                    _,
+                    ::protobuf::types::ProtobufTypeInt32,
+                >(
                     "status_code",
-                    |m: &Subscription| { &m.status_code },
-                    |m: &mut Subscription| { &mut m.status_code },
+                    |m: &Subscription| &m.status_code,
+                    |m: &mut Subscription| &mut m.status_code,
                 ));
-                ::protobuf::reflect::MessageDescriptor::new::<Subscription>(
+                ::protobuf::reflect::MessageDescriptor::new_pb_name::<Subscription>(
                     "Subscription",
                     fields,
-                    file_descriptor_proto()
+                    file_descriptor_proto(),
                 )
             })
         }
     }
 
     fn default_instance() -> &'static Subscription {
-        static mut instance: ::protobuf::lazy::Lazy<Subscription> = ::protobuf::lazy::Lazy {
-            lock: ::protobuf::lazy::ONCE_INIT,
-            ptr: 0 as *const Subscription,
-        };
-        unsafe {
-            instance.get(Subscription::new)
-        }
+        static mut instance: ::protobuf::lazy::Lazy<Subscription> = ::protobuf::lazy::Lazy::INIT;
+        unsafe { instance.get(Subscription::new) }
     }
 }
 
@@ -278,8 +296,8 @@ impl ::std::fmt::Debug for Subscription {
 }
 
 impl ::protobuf::reflect::ProtobufValue for Subscription {
-    fn as_ref(&self) -> ::protobuf::reflect::ProtobufValueRef {
-        ::protobuf::reflect::ProtobufValueRef::Message(self)
+    fn as_ref(&self) -> ::protobuf::reflect::ReflectValueRef {
+        ::protobuf::reflect::ReflectValueRef::Message(self)
     }
 }
 
@@ -289,19 +307,14 @@ static file_descriptor_proto_data: &'static [u8] = b"\
     us_code\x18\x03\x20\x01(\x05B\0:\0B\0b\x06proto2\
 ";
 
-static mut file_descriptor_proto_lazy: ::protobuf::lazy::Lazy<::protobuf::descriptor::FileDescriptorProto> = ::protobuf::lazy::Lazy {
-    lock: ::protobuf::lazy::ONCE_INIT,
-    ptr: 0 as *const ::protobuf::descriptor::FileDescriptorProto,
-};
+static mut file_descriptor_proto_lazy: ::protobuf::lazy::Lazy<
+    ::protobuf::descriptor::FileDescriptorProto,
+> = ::protobuf::lazy::Lazy::INIT;
 
 fn parse_descriptor_proto() -> ::protobuf::descriptor::FileDescriptorProto {
     ::protobuf::parse_from_bytes(file_descriptor_proto_data).unwrap()
 }
 
 pub fn file_descriptor_proto() -> &'static ::protobuf::descriptor::FileDescriptorProto {
-    unsafe {
-        file_descriptor_proto_lazy.get(|| {
-            parse_descriptor_proto()
-        })
-    }
+    unsafe { file_descriptor_proto_lazy.get(|| parse_descriptor_proto()) }
 }

publish.sh

@@ -47,7 +47,7 @@ function publishCrates {
     crate_path="$WORKINGDIR/$CRATE"
     crate_path=${crate_path//\/\///}
     cd $crate_path
-
+    # Also need to update Cargo.lock in root directory
     crate_name=`echo $( awk -v FS="name = " 'NF>1{print $2; exit}' Cargo.toml )`
     echo "Publishing $crate_name to crates.io"
     if [ "$CRATE" == "protocol" ]
@@ -58,6 +58,7 @@ function publishCrates {
       cargo publish
     fi
     echo "Successfully published $crate_name to crates.io"
+    # Should sleep here for 30 seconds to allow Crates.io time to push updated package to edge servers.
   done
 }
 

src/main.rs

@@ -27,7 +27,7 @@ use librespot::playback::mixer::{self, Mixer, MixerConfig};
 use librespot::playback::player::{Player, PlayerEvent};
 
 mod player_event_handler;
-use crate::player_event_handler::run_program_on_events;
+use crate::player_event_handler::{emit_sink_event, run_program_on_events};
 
 fn device_id(name: &str) -> String {
     hex::encode(Sha1::digest(name.as_bytes()))
@@ -87,6 +87,7 @@ struct Setup {
     enable_discovery: bool,
     zeroconf_port: u16,
     player_event_program: Option<String>,
+    emit_sink_events: bool,
 }
 
 fn setup(args: &[String]) -> Setup {
@@ -111,6 +112,7 @@ fn setup(args: &[String]) -> Setup {
             "Run PROGRAM when playback is about to begin.",
             "PROGRAM",
         )
+        .optflag("", "emit-sink-events", "Run program set by --onevent before sink is opened and after it is closed.")
         .optflag("v", "verbose", "Enable verbose output")
         .optopt("u", "username", "Username to sign in with", "USERNAME")
         .optopt("p", "password", "Password", "PASSWORD")
@@ -359,6 +361,7 @@ fn setup(args: &[String]) -> Setup {
         mixer: mixer,
         mixer_config: mixer_config,
         player_event_program: matches.opt_str("onevent"),
+        emit_sink_events: matches.opt_present("emit-sink-events"),
     }
 }
 
@@ -386,6 +389,7 @@ struct Main {
 
     player_event_channel: Option<UnboundedReceiver<PlayerEvent>>,
     player_event_program: Option<String>,
+    emit_sink_events: bool,
 }
 
 impl Main {
@@ -412,6 +416,7 @@ impl Main {
 
             player_event_channel: None,
             player_event_program: setup.player_event_program,
+            emit_sink_events: setup.emit_sink_events,
         };
 
         if setup.enable_discovery {
@@ -481,6 +486,15 @@ impl Future for Main {
                             (backend)(device)
                         });
 
+                    if self.emit_sink_events {
+                        if let Some(player_event_program) = &self.player_event_program {
+                            let player_event_program = player_event_program.clone();
+                            player.set_sink_event_callback(Some(Box::new(move |sink_status| {
+                                emit_sink_event(sink_status, &player_event_program)
+                            })));
+                        }
+                    }
+
                     let (spirc, spirc_task) = Spirc::new(connect_config, session, player, mixer);
                     self.spirc = Some(spirc);
                     self.spirc_task = Some(spirc_task);

src/player_event_handler.rs

@@ -5,6 +5,9 @@ use std::io;
 use std::process::Command;
 use tokio_process::{Child, CommandExt};
 
+use futures::Future;
+use librespot::playback::player::SinkStatus;
+
 fn run_program(program: &str, env_vars: HashMap<&str, String>) -> io::Result<Child> {
     let mut v: Vec<&str> = program.split_whitespace().collect();
     info!("Running {:?} with environment variables {:?}", v, env_vars);
@@ -63,3 +66,16 @@ pub fn run_program_on_events(event: PlayerEvent, onevent: &str) -> Option<io::Re
     }
     Some(run_program(onevent, env_vars))
 }
+
+pub fn emit_sink_event(sink_status: SinkStatus, onevent: &str) {
+    let mut env_vars = HashMap::new();
+    env_vars.insert("PLAYER_EVENT", "sink".to_string());
+    let sink_status = match sink_status {
+        SinkStatus::Running => "running",
+        SinkStatus::TemporarilyClosed => "temporarily_closed",
+        SinkStatus::Closed => "closed",
+    };
+    env_vars.insert("SINK_STATUS", sink_status.to_string());
+
+    let _ = run_program(onevent, env_vars).and_then(|child| child.wait());
+}