add watch_with_volume API

CHANGELOG.md

@@ -6,6 +6,12 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ## [Unreleased]
 
+### Added
+- There is now a `watch_with_volume` method on the API which allows API users to observe both changes to SoundBlaster settings and changes to the Windows volume settings at the same time without needing to run two threads.
+
+### Changed
+- The output of the `watch` command is now different due to using the `watch_with_volume` API.
+
 ## [3.0.0] - 2019-01-14
 
 This release unfortunately renames the `-f` command line parameter to `-i` to allow for a new `-f` to specify the file format.

src/com/event.rs

@@ -1,16 +1,16 @@
-use futures::{Async, Stream};
 use futures::executor::{self, Notify, NotifyHandle, Spawn};
+use futures::{Async, Stream};
 
-use std::ptr;
 use std::mem;
+use std::ptr;
 use std::sync::{Arc, Mutex};
 
 use winapi::um::combaseapi::{CoWaitForMultipleObjects, CWMO_DISPATCH_CALLS};
 use winapi::um::handleapi::CloseHandle;
 use winapi::um::synchapi::{CreateEventW, SetEvent};
 use winapi::um::winbase::INFINITE;
 
-use crate::hresult::{check};
+use crate::hresult::check;
 
 struct ComUnparkState {
     handles: Vec<usize>,
@@ -119,7 +119,10 @@ pub struct ComEventIterator<S> {
     inner: Spawn<S>,
 }
 
-impl<S, I, E> ComEventIterator<S> where S: Stream<Item = I, Error = E> {
+impl<S, I, E> ComEventIterator<S>
+where
+    S: Stream<Item = I, Error = E>,
+{
     pub fn new(stream: S) -> Self {
         let park = ComUnpark::new();
         let id = park.allocate_id();
@@ -131,7 +134,10 @@ impl<S, I, E> ComEventIterator<S> where S: Stream<Item = I, Error = E> {
     }
 }
 
-impl<S, I, E> Iterator for ComEventIterator<S> where S: Stream<Item = I, Error = E> {
+impl<S, I, E> Iterator for ComEventIterator<S>
+where
+    S: Stream<Item = I, Error = E>,
+{
     type Item = Result<I, E>;
 
     fn next(&mut self) -> Option<Self::Item> {

src/lib.rs

@@ -21,6 +21,9 @@ pub mod media;
 pub mod soundcore;
 mod winapiext;
 
+use futures::stream::Fuse;
+use futures::{Async, Poll, Stream};
+
 use indexmap::IndexMap;
 
 use std::collections::BTreeSet;
@@ -30,9 +33,11 @@ use std::fmt;
 
 use slog::Logger;
 
-use crate::media::{DeviceEnumerator, Endpoint};
+use crate::com::event::ComEventIterator;
+use crate::media::{DeviceEnumerator, Endpoint, VolumeEvents, VolumeNotification};
 use crate::soundcore::{
-    SoundCore, SoundCoreEventIterator, SoundCoreFeature, SoundCoreParamValue, SoundCoreParameter,
+    SoundCore, SoundCoreEvent, SoundCoreEventIterator, SoundCoreEvents, SoundCoreFeature,
+    SoundCoreParamValue, SoundCoreParameter,
 };
 
 pub use crate::hresult::Win32Error;
@@ -257,6 +262,82 @@ pub fn watch(
     Ok(core.events()?)
 }
 
+/// Either a SoundCoreEvent or a VolumeNotification.
+#[derive(Debug)]
+pub enum SoundCoreOrVolumeEvent {
+    /// A SoundCoreEvent.
+    SoundCore(SoundCoreEvent),
+    /// A VolumeNotification.
+    Volume(VolumeNotification),
+}
+
+struct SoundCoreAndVolumeEvents {
+    sound_core: Fuse<SoundCoreEvents>,
+    volume: Fuse<VolumeEvents>,
+}
+
+impl Stream for SoundCoreAndVolumeEvents {
+    type Item = SoundCoreOrVolumeEvent;
+    type Error = Win32Error;
+
+    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+        if let Async::Ready(Some(item)) = self.sound_core.poll()? {
+            return Ok(Async::Ready(Some(SoundCoreOrVolumeEvent::SoundCore(item))));
+        }
+        if let Async::Ready(Some(item)) = self.volume.poll().unwrap() {
+            return Ok(Async::Ready(Some(SoundCoreOrVolumeEvent::Volume(item))));
+        }
+        Ok(Async::NotReady)
+    }
+}
+
+/// Iterates over volume change events and also events produced through the
+/// SoundCore API.
+///
+/// This iterator will block until the next event is available.
+pub struct SoundCoreAndVolumeEventIterator {
+    inner: ComEventIterator<SoundCoreAndVolumeEvents>,
+}
+
+impl Iterator for SoundCoreAndVolumeEventIterator {
+    type Item = Result<SoundCoreOrVolumeEvent, Win32Error>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        self.inner.next()
+    }
+}
+
+/// Gets the sequence of events for a device.
+///
+/// If `device_id` is None, the system default output device will be used.
+///
+/// # Examples
+///
+/// ```
+/// for event in watch_with_volume(logger.clone(), None) {
+///     println!("{:?}", event);
+/// }
+/// ```
+pub fn watch_with_volume(
+    logger: &Logger,
+    device_id: Option<&OsStr>,
+) -> Result<SoundCoreAndVolumeEventIterator, Box<Error>> {
+    let endpoint = get_endpoint(logger.clone(), device_id)?;
+    let id = endpoint.id()?;
+    let clsid = endpoint.clsid()?;
+    let core = SoundCore::for_device(&clsid, &id, logger.clone())?;
+
+    let core_events = core.event_stream()?;
+    let volume_events = endpoint.event_stream()?;
+
+    Ok(SoundCoreAndVolumeEventIterator {
+        inner: ComEventIterator::new(SoundCoreAndVolumeEvents {
+            sound_core: core_events.fuse(),
+            volume: volume_events.fuse(),
+        }),
+    })
+}
+
 #[derive(Debug)]
 struct UnsupportedValueError {
     feature: String,

src/main.rs

@@ -708,7 +708,7 @@ fn set(logger: &Logger, matches: &ArgMatches) -> Result<(), Box<Error>> {
 }
 
 fn watch(logger: &Logger, matches: &ArgMatches) -> Result<(), Box<Error>> {
-    for event in sbz_switch::watch(logger, matches.value_of_os("device"))? {
+    for event in sbz_switch::watch_with_volume(logger, matches.value_of_os("device"))? {
         println!("{:?}", event);
     }
     Ok(())

src/media/event.rs

@@ -0,0 +1,121 @@
+use futures::task::AtomicTask;
+use futures::{Async, Poll, Stream};
+
+use std::clone::Clone;
+use std::fmt;
+use std::sync::{mpsc, Arc};
+
+use winapi::shared::guiddef::GUID;
+use winapi::shared::winerror::E_ABORT;
+use winapi::um::endpointvolume::{IAudioEndpointVolume, IAudioEndpointVolumeCallback};
+
+use super::AudioEndpointVolumeCallback;
+use crate::com::ComObject;
+use crate::hresult::{check, Win32Error};
+
+/// Describes a volume change event.
+///
+/// [Official documentation](https://docs.microsoft.com/en-us/windows/desktop/api/endpointvolume/ns-endpointvolume-audio_volume_notification_data)
+pub struct VolumeNotification {
+    /// The ID that was provided when changing the volume.
+    pub event_context: GUID,
+    /// Is the endpoint now muted?
+    pub is_muted: bool,
+    /// The new volume level of the endpoint.
+    pub volume: f32,
+}
+
+impl fmt::Debug for VolumeNotification {
+    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+        fmt.debug_struct("VolumeNotification")
+            .field(
+                "event_context",
+                &format_args!(
+                    "{{{:08X}-{:04X}-{:04X}-{:02X}{:02X}-{:02X}{:02X}{:02X}{:02X}{:02X}{:02X}}}",
+                    self.event_context.Data1,
+                    self.event_context.Data2,
+                    self.event_context.Data3,
+                    self.event_context.Data4[0],
+                    self.event_context.Data4[1],
+                    self.event_context.Data4[2],
+                    self.event_context.Data4[3],
+                    self.event_context.Data4[4],
+                    self.event_context.Data4[5],
+                    self.event_context.Data4[6],
+                    self.event_context.Data4[7]
+                ),
+            )
+            .field("is_muted", &self.is_muted)
+            .field("volume", &self.volume)
+            .finish()
+    }
+}
+
+pub(crate) struct VolumeEvents {
+    volume: ComObject<IAudioEndpointVolume>,
+    events: mpsc::Receiver<VolumeNotification>,
+    task: Arc<AtomicTask>,
+    callback: ComObject<IAudioEndpointVolumeCallback>,
+}
+
+impl VolumeEvents {
+    pub fn new(volume: ComObject<IAudioEndpointVolume>) -> Result<Self, Win32Error> {
+        let task = Arc::new(AtomicTask::new());
+        let (tx, rx) = mpsc::channel();
+
+        let tx_task = task.clone();
+        unsafe {
+            let callback = AudioEndpointVolumeCallback::wrap(move |e| {
+                match tx.send(VolumeNotification {
+                    event_context: e.guidEventContext,
+                    is_muted: e.bMuted != 0,
+                    volume: e.fMasterVolume,
+                }) {
+                    Ok(()) => {
+                        tx_task.notify();
+                        Ok(())
+                    }
+                    Err(_) => Err(Win32Error::new(E_ABORT)),
+                }
+            });
+
+            let result = check((*volume).RegisterControlChangeNotify(callback));
+            if let Err(error) = result {
+                (*callback).Release();
+                return Err(error);
+            }
+
+            Ok(Self {
+                volume,
+                events: rx,
+                task,
+                callback: ComObject::take(callback),
+            })
+        }
+    }
+}
+
+impl Stream for VolumeEvents {
+    type Item = VolumeNotification;
+    type Error = ();
+
+    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+        self.task.register();
+        match self.events.try_recv() {
+            Ok(e) => Ok(Async::Ready(Some(e))),
+            Err(mpsc::TryRecvError::Empty) => Ok(Async::NotReady),
+            Err(mpsc::TryRecvError::Disconnected) => Ok(Async::Ready(None)),
+        }
+    }
+}
+
+impl Drop for VolumeEvents {
+    fn drop(&mut self) {
+        unsafe {
+            check(
+                (*self.volume).UnregisterControlChangeNotify(&*self.callback as *const _ as *mut _),
+            )
+            .unwrap();
+        }
+    }
+}