feat: add StreamTrait::now() and overhaul StreamInstant API (#1139)

- Adds now() to StreamTrait so callers can query the stream clock outside
  the audio callback on the same time base as callback/capture/playback
  timestamps.

- StreamInstant is reworked to mirror std::time::Instant with u64 storage
  (all stream clocks are monotonic and non-negative) and similar functions
  parameters, and return types.

Closes #472

Author: roderickvd

CHANGELOG.md

@@ -14,6 +14,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - `StreamConfig` now implements `Copy`.
 - `StreamTrait::buffer_size()` to query the stream's current buffer size in frames per callback.
 - `device_by_id` is now dispatched to each backend's implementation, allowing to override it.
+- `StreamTrait::now()` to query the current instant on the stream's clock.
+- `StreamInstant` API changed and extended to mirror `std::time::Instant`/`Duration`. See
+  [UPGRADING.md](UPGRADING.md) for migration details.
 - **ALSA**: `device_by_id` now accepts PCM shorthand names such as `hw:0,0` and `plughw:foo`.
 - **PipeWire**: New host for Linux and some BSDs using the PipeWire API.
 - **PulseAudio**: New host for Linux and some BSDs using the PulseAudio API.
@@ -48,6 +51,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **WebAudio**: Timestamps now include base and output latency.
 - **WebAudio**: Initial buffer scheduling offset now scales with buffer duration.
 
+### Removed
+
+- Replaced `StreamInstant::add()` and `sub()` by `checked_add()`/`+` and `checked_sub()`/`-`.
+
 ### Fixed
 
 - Reintroduce `audio_thread_priority` feature.

Cargo.toml

@@ -85,12 +85,14 @@ clap = { version = ">=4.0, <=4.5.57", features = ["derive"] }
 # When updating this, also update the "windows-version" matrix in the CI workflow.
 [target.'cfg(target_os = "windows")'.dependencies]
 windows = { version = ">=0.59, <=0.62", features = [
+    "Win32_Media",
     "Win32_Media_Audio",
     "Win32_Foundation",
     "Win32_Devices_Properties",
     "Win32_Media_KernelStreaming",
     "Win32_System_Com_StructuredStorage",
     "Win32_System_Threading",
+    "Win32_System_Performance",
     "Win32_Security",
     "Win32_System_SystemServices",
     "Win32_System_Variant",
@@ -183,6 +185,7 @@ ndk = { version = "0.9", default-features = false, features = [
 ] }
 ndk-context = "0.1"
 jni = "0.21"
+libc = "0.2"
 num-derive = "0.4"
 num-traits = "0.2"
 

UPGRADING.md

@@ -8,6 +8,12 @@ This guide covers breaking changes requiring code updates. See [CHANGELOG.md](CH
 - [ ] Optionally handle the new `DeviceBusy` variant for retryable device errors
 - [ ] Change `build_*_stream` call sites to pass `StreamConfig` by value (drop the `&`)
 - [ ] For custom hosts, change `DeviceTrait` implementations to accept `StreamConfig` by value.
+- [ ] Remove `instant.duration_since(e)` unwraps; it now returns `Duration` (saturating).
+- [ ] Change `instant.add(d)` to `instant.checked_add(d)` (or use `instant + d`).
+- [ ] Change `instant.sub(d)` to `instant.checked_sub(d)` (or use `instant - d`).
+- [ ] Update `StreamInstant::new(secs, nanos)` call sites: `secs` is now `u64`.
+- [ ] Update `StreamInstant::from_nanos(nanos)` call sites: `nanos` is now `u64`.
+- [ ] Update `duration_since` call sites to pass by value (drop the `&`).
 
 ## 1. Error enums are now `#[non_exhaustive]`
 
@@ -61,6 +67,73 @@ let stream = device.build_output_stream(config, data_fn, err_fn, None)?;
 
 If you implement `DeviceTrait` on your own type (via the `custom` feature), update your `build_input_stream_raw` and `build_output_stream_raw` signatures from `config: &StreamConfig` to `config: StreamConfig`. Any `config.clone()` calls before `move` closures can also be removed.
 
+## 4. `StreamInstant` API overhaul
+
+The `StreamInstant` API has been aligned with `std::time::Instant` and `std::time::Duration`.
+
+### `duration_since` now returns `Duration` (saturating)
+
+**What changed:** `duration_since` now returns `Duration` directly, saturating to `Duration::ZERO`
+when the argument is later than `self`, instead of returning `Option<Duration>`.
+
+```rust
+// Before (v0.17): returned Option<Duration>, argument by reference
+if let Some(d) = callback.duration_since(&start) {
+    println!("elapsed: {d:?}");
+}
+
+// After (v0.18): returns Duration (saturating), argument by value
+let d = callback.duration_since(start);
+println!("elapsed: {d:?}");
+
+// For the previous Option-returning behaviour, use checked_duration_since:
+if let Some(d) = callback.checked_duration_since(start) {
+    println!("elapsed: {d:?}");
+}
+```
+
+**Why:** Mirrors the saturating behavior of `std::time::Instant::saturating_duration_since` in the Rust standard library.
+
+### `add` / `sub` renamed to `checked_add` / `checked_sub`; operator impls added
+
+**What changed:** The `add` and `sub` methods (which returned `Option`) are replaced by
+`checked_add` / `checked_sub` with the same semantics. `+`, `-`, `+=`, and `-=` operator impls
+are also added.
+
+```rust
+// Before (v0.17)
+let future = instant.add(Duration::from_millis(10)).expect("overflow");
+let past   = instant.sub(Duration::from_millis(10)).expect("underflow");
+
+// After (v0.18): explicit checked form (same semantics):
+let future = instant.checked_add(Duration::from_millis(10)).expect("overflow");
+let past   = instant.checked_sub(Duration::from_millis(10)).expect("underflow");
+
+// Or use the operator (panics on overflow, like std::time::Instant):
+let future = instant + Duration::from_millis(10);
+let past   = instant - Duration::from_millis(10);
+
+// Subtract two instants to get a Duration (saturates to zero):
+let elapsed: Duration = later - earlier;
+```
+
+**Why:** Aligns the API with `std::time::Instant`, making `StreamInstant` more idiomatic.
+
+### `new` and `from_nanos` take unsigned integers
+
+**What changed:** The `secs` parameter of `StreamInstant::new` and the `nanos` parameter of
+`StreamInstant::from_nanos` are now `u64` instead of `i64`.
+
+```rust
+// Before (v0.17): negative seconds were accepted
+StreamInstant::new(-1_i64, 0);
+
+// After (v0.18): all stream clocks are non-negative
+StreamInstant::new(0_u64, 0);
+```
+
+**Why:** All audio host clocks are positive and monotonic; they are never negative.
+
 ---
 
 # Upgrading from v0.16 to v0.17

examples/custom.rs

@@ -2,6 +2,7 @@ use std::sync::{
     atomic::{AtomicBool, Ordering},
     Arc,
 };
+use std::time::Instant;
 
 use cpal::{
     traits::{DeviceTrait, HostTrait, StreamTrait},
@@ -19,7 +20,10 @@ struct MyDevice;
 // Only Send+Sync is needed
 struct MyStream {
     controls: Arc<StreamControls>,
-    // option is needed since joining a thread takes ownership,
+    // The instant the audio thread was started; shared with now() so that
+    // callback timestamps and now() are on the same time base.
+    start: Instant,
+    // Option is needed since joining a thread takes ownership,
     // and we want to do that on drop (gives us &mut self, not self)
     handle: Option<std::thread::JoinHandle<()>>,
 }
@@ -138,9 +142,9 @@ impl DeviceTrait for MyDevice {
             pause: AtomicBool::new(true), // streams are expected to start out paused by default
         });
 
+        let start = Instant::now();
         let thread_controls = controls.clone();
         let handle = std::thread::spawn(move || {
-            let start = std::time::Instant::now();
             let mut buffer = [0.0_f32; 4096];
             while !thread_controls.exit.load(Ordering::Relaxed) {
                 std::thread::sleep(std::time::Duration::from_secs_f32(
@@ -161,7 +165,7 @@ impl DeviceTrait for MyDevice {
                     )
                 };
 
-                let duration = std::time::Instant::now().duration_since(start);
+                let duration = Instant::now().duration_since(start);
                 let secs = duration.as_nanos() / 1_000_000_000;
                 let subsec_nanos = duration.as_nanos() - secs * 1_000_000_000;
                 let stream_instant = cpal::StreamInstant::new(secs as _, subsec_nanos as _);
@@ -178,6 +182,7 @@ impl DeviceTrait for MyDevice {
 
         Ok(MyStream {
             controls,
+            start,
             handle: Some(handle),
         })
     }
@@ -193,6 +198,11 @@ impl StreamTrait for MyStream {
         self.controls.pause.store(true, Ordering::Relaxed);
         Ok(())
     }
+
+    fn now(&self) -> cpal::StreamInstant {
+        let elapsed = self.start.elapsed();
+        cpal::StreamInstant::new(elapsed.as_secs(), elapsed.subsec_nanos())
+    }
 }
 
 // streams are expected to stop when dropped
@@ -315,9 +325,7 @@ pub fn make_stream(
         config,
         move |output: &mut [f32], _: &cpal::OutputCallbackInfo| {
             // for 0-1s play sine, 1-2s play square, 2-3s play saw, 3-4s play triangle_wave
-            let time_since_start = std::time::Instant::now()
-                .duration_since(time_at_start)
-                .as_secs_f32();
+            let time_since_start = Instant::now().duration_since(time_at_start).as_secs_f32();
             if time_since_start < 1.0 {
                 oscillator.set_waveform(Waveform::Sine);
             } else if time_since_start < 2.0 {

src/host/aaudio/convert.rs

@@ -1,5 +1,4 @@
-use std::convert::TryInto;
-use std::time::Duration;
+//! Time-conversion helpers for the AAudio backend.
 
 extern crate ndk;
 
@@ -8,21 +7,26 @@ use crate::{
     StreamInstant,
 };
 
-pub fn to_stream_instant(duration: Duration) -> StreamInstant {
-    StreamInstant::new(
-        duration.as_secs().try_into().unwrap(),
-        duration.subsec_nanos(),
-    )
+/// Returns a [`StreamInstant`] for the current moment.
+pub fn now_stream_instant() -> StreamInstant {
+    let mut ts = libc::timespec {
+        tv_sec: 0,
+        tv_nsec: 0,
+    };
+    let res = unsafe { libc::clock_gettime(libc::CLOCK_MONOTONIC, &mut ts) };
+    assert_eq!(res, 0, "clock_gettime(CLOCK_MONOTONIC) failed");
+    StreamInstant::new(ts.tv_sec as u64, ts.tv_nsec as u32)
 }
 
+/// Returns the [`StreamInstant`] of the most recent audio frame transferred by `stream`.
 pub fn stream_instant(stream: &ndk::audio::AudioStream) -> StreamInstant {
     let ts = stream
         .timestamp(ndk::audio::Clockid::Monotonic)
         .unwrap_or(ndk::audio::Timestamp {
             frame_position: 0,
             time_nanoseconds: 0,
         });
-    to_stream_instant(Duration::from_nanos(ts.time_nanoseconds as u64))
+    StreamInstant::from_nanos(ts.time_nanoseconds as u64)
 }
 
 impl From<ndk::audio::AudioError> for StreamError {

src/host/aaudio/mod.rs

@@ -6,12 +6,12 @@ use std::cmp;
 use std::convert::TryInto;
 use std::sync::atomic::{AtomicI32, Ordering};
 use std::sync::{Arc, Mutex};
-use std::time::{Duration, Instant};
+use std::time::Duration;
 use std::vec::IntoIter as VecIntoIter;
 
 extern crate ndk;
 
-use convert::{stream_instant, to_stream_instant};
+use convert::{now_stream_instant, stream_instant};
 use java_interface::{AudioDeviceInfo, AudioManager};
 
 use crate::traits::{DeviceTrait, HostTrait, StreamTrait};
@@ -316,13 +316,12 @@ where
     E: FnMut(StreamError) + Send + 'static,
 {
     let builder = configure_for_device(builder, device, config);
-    let created = Instant::now();
     let channel_count = config.channels as i32;
     let stream = builder
         .data_callback(Box::new(move |stream, data, num_frames| {
             let cb_info = InputCallbackInfo {
                 timestamp: InputStreamTimestamp {
-                    callback: to_stream_instant(created.elapsed()),
+                    callback: now_stream_instant(),
                     capture: stream_instant(stream),
                 },
             };
@@ -366,7 +365,6 @@ where
     E: FnMut(StreamError) + Send + 'static,
 {
     let builder = configure_for_device(builder, device, config);
-    let created = Instant::now();
     let channel_count = config.channels as i32;
     let tune_dynamically = config.buffer_size == BufferSize::Default;
 
@@ -378,7 +376,7 @@ where
             // Deliver audio data to user callback
             let cb_info = OutputCallbackInfo {
                 timestamp: OutputStreamTimestamp {
-                    callback: to_stream_instant(created.elapsed()),
+                    callback: now_stream_instant(),
                     playback: stream_instant(stream),
                 },
             };
@@ -719,6 +717,10 @@ impl StreamTrait for Stream {
         }
     }
 
+    fn now(&self) -> crate::StreamInstant {
+        now_stream_instant()
+    }
+
     fn buffer_size(&self) -> Option<crate::FrameCount> {
         let stream = self.inner.lock().ok()?;
 

src/host/alsa/mod.rs

@@ -27,8 +27,8 @@ use crate::{
     DefaultStreamConfigError, DeviceDescription, DeviceDescriptionBuilder, DeviceDirection,
     DeviceId, DeviceIdError, DeviceNameError, DevicesError, FrameCount, InputCallbackInfo,
     OutputCallbackInfo, PauseStreamError, PlayStreamError, SampleFormat, SampleRate, StreamConfig,
-    StreamError, SupportedBufferSize, SupportedStreamConfig, SupportedStreamConfigRange,
-    SupportedStreamConfigsError,
+    StreamError, StreamInstant, SupportedBufferSize, SupportedStreamConfig,
+    SupportedStreamConfigRange, SupportedStreamConfigsError,
 };
 
 mod enumerate;
@@ -1065,11 +1065,8 @@ fn process_input(
     }?;
     let delay_duration = frames_to_duration(delay_frames, stream.conf.sample_rate);
     let capture = callback
-        .sub(delay_duration)
-        .ok_or_else(|| BackendSpecificError {
-            description: "`capture` is earlier than representation supported by `StreamInstant`"
-                .to_string(),
-        })?;
+        .checked_sub(delay_duration)
+        .unwrap_or(StreamInstant::ZERO);
     let timestamp = crate::InputStreamTimestamp { callback, capture };
     let info = crate::InputCallbackInfo { timestamp };
     data_callback(&data, &info);
@@ -1098,12 +1095,7 @@ fn process_output(
             stream_timestamp_fallback(stream.creation_instant)
         }?;
         let delay_duration = frames_to_duration(delay_frames, stream.conf.sample_rate);
-        let playback = callback
-            .add(delay_duration)
-            .ok_or_else(|| BackendSpecificError {
-                description: "`playback` occurs beyond representation supported by `StreamInstant`"
-                    .to_string(),
-            })?;
+        let playback = callback + delay_duration;
         let timestamp = crate::OutputStreamTimestamp { callback, playback };
         let info = crate::OutputCallbackInfo { timestamp };
         data_callback(&mut data, &info);
@@ -1135,7 +1127,7 @@ fn process_output(
 #[inline]
 fn stream_timestamp_hardware(
     status: &alsa::pcm::Status,
-) -> Result<crate::StreamInstant, BackendSpecificError> {
+) -> Result<StreamInstant, BackendSpecificError> {
     let trigger_ts = status.get_trigger_htstamp();
     // trigger_htstamp records when the PCM stream started.
     // On the first few callbacks, it might not have been set yet,
@@ -1156,7 +1148,7 @@ fn stream_timestamp_hardware(
         );
         return Err(BackendSpecificError { description });
     }
-    Ok(crate::StreamInstant::from_nanos(nanos))
+    Ok(StreamInstant::from_nanos(nanos as u64))
 }
 
 // Use elapsed duration since stream creation as fallback when hardware timestamps are unavailable.
@@ -1165,12 +1157,13 @@ fn stream_timestamp_hardware(
 #[inline]
 fn stream_timestamp_fallback(
     creation: std::time::Instant,
-) -> Result<crate::StreamInstant, BackendSpecificError> {
+) -> Result<StreamInstant, BackendSpecificError> {
     let now = std::time::Instant::now();
     let duration = now.duration_since(creation);
-    crate::StreamInstant::from_nanos_i128(duration.as_nanos() as i128).ok_or(BackendSpecificError {
-        description: "stream duration has exceeded `StreamInstant` representation".to_string(),
-    })
+    Ok(StreamInstant::new(
+        duration.as_secs(),
+        duration.subsec_nanos(),
+    ))
 }
 
 // Adapted from `timestamp2ns` here:
@@ -1284,6 +1277,18 @@ impl StreamTrait for Stream {
         self.inner.channel.pause(true).ok();
         Ok(())
     }
+    fn now(&self) -> StreamInstant {
+        if self.inner.use_hw_timestamps {
+            if let Ok(status) = self.inner.channel.status() {
+                if let Ok(instant) = stream_timestamp_hardware(&status) {
+                    return instant;
+                }
+            }
+        }
+        stream_timestamp_fallback(self.inner.creation_instant)
+            .expect("stream duration exceeded `StreamInstant` range")
+    }
+
     fn buffer_size(&self) -> Option<FrameCount> {
         Some(self.inner.period_frames as FrameCount)
     }