diff --git a/c/moq/MIGRATE.md b/c/moq/MIGRATE.md index e2a9d73e5cd..a7fe99980a3 100644 --- a/c/moq/MIGRATE.md +++ b/c/moq/MIGRATE.md @@ -10,10 +10,9 @@ The hand-written crate never published as moq-c: a libmoq 0.6 user applies these `moq_announce_event.has_captures` is removed, changing the C ABI. Read `captures_len` directly: zero means the filter has no wildcards. Rebuild consumers against the regenerated header and matching native library. -### A subscription without `max_delay_present` has no limit +### Subscriptions take a `moq_delay` ([#7697](https://github.com/moq-dev/moq/pull/7697)) -A `moq_subscription` with `max_delay_present` false sets no subscriber limit, leaving the publisher's max age alone to bound it, instead of moq-net's default of zero. -Start from `moq_subscription_defaults()`, or set `max_delay_present`, where you relied on a zeroed struct; a NULL pointer still means the defaults. +`moq_subscription.max_delay_us` and `max_delay_present` are `moq_delay delay`. A zeroed `moq_subscription` is moq-net's defaults, the same as `moq_subscription_defaults()` and a NULL pointer: set `delay.max_us` to the old `max_delay_us`, and `delay.max_unlimited` for no subscriber limit. `delay.min_unstated` leaves the target latency unstated. ### Paths with an empty segment are refused ([#6648](https://github.com/moq-dev/moq/pull/6648)) diff --git a/c/moq/README.md b/c/moq/README.md index 7b22c0f366c..0eb5c0f9254 100644 --- a/c/moq/README.md +++ b/c/moq/README.md @@ -20,4 +20,6 @@ The generator is the one the C++ package pins; see [cpp/moq](../../cpp/moq/READM The implementation is C++17 over the generated C++ bindings, compiled into `libmoq.a` (`moq.lib`) beside moq-ffi's staticlib, installed as `libmoq_c_ffi.a` (`moq_c_ffi.lib`) so moq-cpp's copy can share the prefix, so a C program needs no C++ compiler, but its link adds the C++ runtime, which the CMake package and `moq-c.pc` name. Both libraries use the release CRT (`/MD`) on MSVC in every configuration, so an MSVC consumer must too (`CMAKE_MSVC_RUNTIME_LIBRARY=MultiThreadedDLL`); its default Debug runtime (`/MDd`) fails to link with LNK2038. +Start every subscription from `moq_subscription_defaults()` and change its fields: it holds moq-net's defaults, a zero `delay` that starts at the newest group and skips a superseded one at once. The bare record `moq_subscription_default()` returns does not, since the generated records carry no duration defaults: its unset `delay.max` sets no subscriber limit and its unset `delay.min` leaves the target latency unstated. + This package and the C++ package, `moq-cpp`, each embed moq-ffi, so one program links one of them, never both, and they install to separate prefixes. diff --git a/c/moq/test/probe.c b/c/moq/test/probe.c index e21c6057064..5ab02854386 100644 --- a/c/moq/test/probe.c +++ b/c/moq/test/probe.c @@ -266,12 +266,13 @@ int main(void) moq_dispatcher dispatcher = {loop_dispatch, loop_shutdown, NULL}; moq_set_dispatcher(&dispatcher); - /* Only the defaults carry moq-net's zero max delay; a bare record sets no subscriber limit. */ + /* Only the defaults carry moq-net's zero delay; a bare record leaves both bounds unset. */ moq_subscription *defaults = moq_subscription_defaults(); - CHECK(defaults->max_delay.has_value && defaults->max_delay.value == 0); + CHECK(defaults->delay.min.has_value && defaults->delay.min.value == 0); + CHECK(defaults->delay.max.has_value && defaults->delay.max.value == 0); moq_subscription_free(defaults); moq_subscription bare = moq_subscription_default(); - CHECK(!bare.max_delay.has_value); + CHECK(!bare.delay.max.has_value); /* A synchronous error is a returned value with Rust's Display message. */ moq_server_config unbound_config = moq_server_config_default(); diff --git a/cpp/moq/MIGRATE.md b/cpp/moq/MIGRATE.md index d561d680531..88b440592dc 100644 --- a/cpp/moq/MIGRATE.md +++ b/cpp/moq/MIGRATE.md @@ -2,6 +2,11 @@ ## Unreleased +### Subscriptions take a `moq::Delay` ([#7697](https://github.com/moq-dev/moq/pull/7697)) + +`moq::Subscription::max_delay` is `moq::Delay delay`: set `delay.max` to the old `max_delay`, or start from `moq::subscription_defaults()`. +`moq::Delay::min` is a `std::optional`: `std::nullopt` leaves the target latency unstated, which a decoder plays as zero. A bare `moq::Subscription{}` leaves both bounds unset, so its `delay.max` still sets no subscriber limit. + ### Announcement captures are required ([#7484](https://github.com/moq-dev/moq/pull/7484)) `moq::Announce.captures` is `std::vector` instead of an optional vector. Remove optional checks and dereferences; an empty vector means the filter has no wildcards. Regenerate bindings and rebuild with the matching native library. diff --git a/cpp/moq/test/probe.cpp b/cpp/moq/test/probe.cpp index 23e4aaacff3..706a8984065 100644 --- a/cpp/moq/test/probe.cpp +++ b/cpp/moq/test/probe.cpp @@ -142,11 +142,13 @@ void timestamp_roundtrip() ok(broadcast->close(), "close timestamp broadcast"); } -// Only the defaults carry moq-net's zero max delay; a bare record sets no subscriber limit. +// Only the defaults carry moq-net's zero delay; a bare record leaves both bounds unset. void subscription_defaults() { - CHECK(moq::subscription_defaults().max_delay == std::chrono::nanoseconds::zero()); - CHECK(!moq::Subscription{}.max_delay.has_value()); + auto defaults = moq::subscription_defaults(); + CHECK(defaults.delay.min == std::chrono::nanoseconds::zero()); + CHECK(defaults.delay.max == std::chrono::nanoseconds::zero()); + CHECK(!moq::Subscription{}.delay.max.has_value()); } // Exercise the installed audio publisher and consumer rather than only generated signatures. diff --git a/dart/moq/MIGRATE.md b/dart/moq/MIGRATE.md index f6279a8e838..f4a5f95ab7e 100644 --- a/dart/moq/MIGRATE.md +++ b/dart/moq/MIGRATE.md @@ -4,6 +4,11 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Subscriptions take a `Delay` ([#7697](https://github.com/moq-dev/moq/pull/7697)) + +`Subscription.maxDelay` is `delay`, a `Delay`, and a bare `Subscription()` is moq-net's defaults again: pass `delay: Delay(max: maxDelay)`, and `Delay(max: null)` for no subscriber limit. +`media.Delay` moves to `package:moq/moq.dart`. `Delay` and `Subscription` are extension types over the generated records, so they pass wherever those are taken; `Delay` defaults `min` and `max` to zero, and an explicit null `min` leaves the target latency unstated. + ### `Transport.iroh` is removed ([#7345](https://github.com/moq-dev/moq/pull/7345)) Native sessions no longer arrive over iroh, so `Transport` loses its `iroh` value. Drop the case, and regenerate bindings and rebuild with the matching native library. @@ -14,8 +19,8 @@ Native sessions no longer arrive over iroh, so `Transport` loses its `iroh` valu ### Decoder outputs take a `Delay` ([#7311](https://github.com/moq-dev/moq/pull/7311)) -`audio.DecoderOutput` replaces `maxDelay` and the playout `delay: Duration?` with `delay: media.Delay?` and `estimate: bool = true`. -`video.DecoderOutput.maxDelay` is `delay: media.Delay?`. +`audio.DecoderOutput` replaces `maxDelay` and the playout `delay: Duration?` with `delay: Delay?` and `estimate: bool = true`. +`video.DecoderOutput.maxDelay` is `delay: Delay?`. `Delay(min: ..., max: ...)`: `max` (null for no limit) is the old `maxDelay`, and `min` is the playout delay, where a cold join starts. A fixed audio playout delay `d` is `delay: Delay(min: d, max: ...)` with `estimate: false`. @@ -46,7 +51,6 @@ Their handles `OriginDynamic`, `BroadcastDynamic`, and `TrackDynamic` are `Origi The records carry durations as `Duration` fields instead of `*Us` microsecond integers, so the duration extension getters are gone: read `backoff.initial`, `stats.rtt`, `grant.expires`, `info.maxAge`, `fill.maxDelay()`, and `consumer.delay()` directly. `audio.EncoderOutput.frameDuration` is nullable: pass null where you passed `frameDurationUs: 0` for the codec's own frame. -`Subscription.maxDelay` defaults to null, so a `Subscription()` that omits it gets no staleness limit instead of moq-net's 0: start from `defaultSubscription()`, or pass `maxDelay: Duration.zero`, for the old default. A negative `Duration` throws `ArgumentError` wherever it crosses into the native library. ### Connection state diff --git a/dart/moq/lib/media.dart b/dart/moq/lib/media.dart index 674d24c0a79..83feaeee975 100644 --- a/dart/moq/lib/media.dart +++ b/dart/moq/lib/media.dart @@ -75,9 +75,6 @@ typedef ContainerInit = MoqContainerInit; /// A width and height pair, in pixels. typedef Dimensions = MoqDimensions; -/// How far behind the live edge a decoder plays: `min` is the playout delay and where a cold join starts, and `max`, null for no limit, is how far a stalled group may fall behind before it is skipped. The subscription carries both; a `min` past `max` starts at `max`. -typedef Delay = MoqDelay; - /// Where an audio or video decoder starts on a track that already holds groups: `latest` for a live player. typedef DecodeStart = MoqDecodeStart; diff --git a/dart/moq/lib/src/aliases.dart b/dart/moq/lib/src/aliases.dart index a9c38dde51b..f97c5c56074 100644 --- a/dart/moq/lib/src/aliases.dart +++ b/dart/moq/lib/src/aliases.dart @@ -3,7 +3,10 @@ import 'package:moq_ffi/moq_ffi.dart'; // Re-export the UniFFI types without their `Moq` prefix, so consumers spell // them the way the Rust API does. These are type aliases, not wrappers: the // values are the exact same objects, so every generated method applies -// unchanged and the prefixed names stay valid. +// unchanged and the prefixed names stay valid. `Delay` and `Subscription` are +// extension types instead, which are the same objects too but give an omitted +// delay bound the zero moq-net defaults to, where the generated record leaves +// it null. // // Four types are deliberately not aliased, because the unprefixed name is // already taken where this package is used: @@ -134,14 +137,39 @@ typedef Frame = MoqFrame; /// An exact media time: `value` units at `scale` units per second. typedef Timestamp = MoqTimestamp; -/// Tunes how a track subscription is delivered: priority, latency budget, ranges, and group ordering. -typedef Subscription = MoqSubscription; - -/// moq-net's defaults, whose `maxDelay` of zero skips a superseded group at once. +/// How far behind the live edge a subscriber or decoder plays: where it starts, and how late it may fall. +extension type Delay._(MoqDelay _delay) implements MoqDelay { + /// `min` is the target latency and where a cold join starts, and `max` is how far a + /// non-latest group may fall behind before it is skipped. Both default to zero, moq-net's + /// default. An explicit null `min` leaves it unstated, which a decoder plays as zero, and an + /// explicit null `max` sets no subscriber limit. A `min` past `max` starts at `max`. + Delay({Duration? min = Duration.zero, Duration? max = Duration.zero}) + : this._(MoqDelay(min: min, max: max)); +} + +/// Tunes how a track subscription is delivered: priority, delay, ranges, and group ordering. +extension type Subscription._(MoqSubscription _subscription) + implements MoqSubscription { + /// An omitted `delay` is a zero [Delay], so a bare `Subscription()` is moq-net's defaults. + Subscription({ + int priority = 0, + MoqDelay? delay, + List? ranges, + MoqOrder? order, + }) : this._( + MoqSubscription( + priority: priority, + delay: delay ?? Delay(), + ranges: ranges, + order: order, + ), + ); +} + +/// moq-net's defaults, whose zero `delay` starts at the newest group and skips a superseded one at once. /// -/// A null subscription means the same. A `Subscription()` leaves `maxDelay` null, which -/// sets no subscriber limit, so pass `maxDelay: Duration.zero` when building one to match. -Subscription defaultSubscription() => moqSubscriptionDefaults(); +/// A null subscription and a bare `Subscription()` mean the same. +Subscription defaultSubscription() => Subscription._(moqSubscriptionDefaults()); /// A span of a track to deliver: from `start` up to `end`, or onward when `end` is null. typedef Range = MoqRange; diff --git a/dart/moq/test/moq_test.dart b/dart/moq/test/moq_test.dart index e2cafbde1e4..0098a57d29c 100644 --- a/dart/moq/test/moq_test.dart +++ b/dart/moq/test/moq_test.dart @@ -68,8 +68,13 @@ void main() { }); test('a default subscription skips superseded groups at once', () { - expect(defaultSubscription().maxDelay, Duration.zero); - expect(Subscription().maxDelay, isNull); + for (final subscription in [defaultSubscription(), Subscription()]) { + expect(subscription.delay.min, Duration.zero); + expect(subscription.delay.max, Duration.zero); + } + final unset = Delay(min: null, max: null); + expect(unset.min, isNull); + expect(unset.max, isNull); }); test('a negative duration in a raw record is refused', () { @@ -77,7 +82,9 @@ void main() { final track = broadcast.publishTrack(name: 'negative', info: null); expect( () => track.consume( - subscription: Subscription(maxDelay: const Duration(microseconds: -1)), + subscription: Subscription( + delay: Delay(max: const Duration(microseconds: -1)), + ), ), throwsArgumentError, ); @@ -105,7 +112,7 @@ void main() { final consumer = track.consume( subscription: Subscription( - maxDelay: const Duration(seconds: 10), + delay: Delay(max: const Duration(seconds: 10)), ranges: [Range(start: RelativeMoqAnchor(2))], order: Order.ascending, ), @@ -759,10 +766,7 @@ void main() { final fixedOutput = moq_audio.DecoderOutput( format: moq_audio.SampleFormat.f32, - delay: moq_media.Delay( - min: Duration(milliseconds: 80), - max: Duration.zero, - ), + delay: Delay(min: Duration(milliseconds: 80)), estimate: false, ); final fixed = await moq_audio.Consumer.subscribe( diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index afad90c9868..b704af4ce14 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/dart/moq_ffi/lib/src/moq.dart @@ -8,7 +8,9 @@ import "dart:ffi"; import "dart:io" show Platform, File, Directory; import "dart:isolate"; import "dart:typed_data"; + import "package:ffi/ffi.dart"; + import "uniffi_runtime.dart"; export "uniffi_runtime.dart"; @@ -707,6 +709,65 @@ class FfiConverterMoqRatePolicy { } } +class MoqDelay { + final Duration? min; + final Duration? max; + MoqDelay({this.min, this.max}); +} + +class FfiConverterMoqDelay { + static MoqDelay lift(RustBuffer buf) { + return FfiConverterMoqDelay.read(buf.asUint8List()).value; + } + + static LiftRetVal read(Uint8List buf) { + int new_offset = buf.offsetInBytes; + final min_lifted = FfiConverterOptionalDuration.read( + Uint8List.view(buf.buffer, new_offset), + ); + final min = min_lifted.value; + new_offset += min_lifted.bytesRead; + final max_lifted = FfiConverterOptionalDuration.read( + Uint8List.view(buf.buffer, new_offset), + ); + final max = max_lifted.value; + new_offset += max_lifted.bytesRead; + return LiftRetVal( + MoqDelay(min: min, max: max), + new_offset - buf.offsetInBytes, + ); + } + + static RustBuffer lower(MoqDelay value) { + final total_length = + FfiConverterOptionalDuration.allocationSize(value.min) + + FfiConverterOptionalDuration.allocationSize(value.max) + + 0; + final buf = Uint8List(total_length); + write(value, buf); + return toRustBuffer(buf); + } + + static int write(MoqDelay value, Uint8List buf) { + int new_offset = buf.offsetInBytes; + new_offset += FfiConverterOptionalDuration.write( + value.min, + Uint8List.view(buf.buffer, new_offset), + ); + new_offset += FfiConverterOptionalDuration.write( + value.max, + Uint8List.view(buf.buffer, new_offset), + ); + return new_offset - buf.offsetInBytes; + } + + static int allocationSize(MoqDelay value) { + return FfiConverterOptionalDuration.allocationSize(value.min) + + FfiConverterOptionalDuration.allocationSize(value.max) + + 0; + } +} + class MoqFetchGroupOptions { final int priority; final Duration? maxDelay; @@ -827,12 +888,12 @@ class FfiConverterMoqRange { class MoqSubscription { final int priority; - final Duration? maxDelay; + final MoqDelay delay; final List? ranges; final MoqOrder? order; MoqSubscription({ this.priority = 0, - this.maxDelay, + required this.delay, this.ranges = null, this.order = null, }); @@ -850,11 +911,11 @@ class FfiConverterMoqSubscription { ); final priority = priority_lifted.value; new_offset += priority_lifted.bytesRead; - final maxDelay_lifted = FfiConverterOptionalDuration.read( + final delay_lifted = FfiConverterMoqDelay.read( Uint8List.view(buf.buffer, new_offset), ); - final maxDelay = maxDelay_lifted.value; - new_offset += maxDelay_lifted.bytesRead; + final delay = delay_lifted.value; + new_offset += delay_lifted.bytesRead; final ranges_lifted = FfiConverterOptionalSequenceMoqRange.read( Uint8List.view(buf.buffer, new_offset), ); @@ -868,7 +929,7 @@ class FfiConverterMoqSubscription { return LiftRetVal( MoqSubscription( priority: priority, - maxDelay: maxDelay, + delay: delay, ranges: ranges, order: order, ), @@ -879,7 +940,7 @@ class FfiConverterMoqSubscription { static RustBuffer lower(MoqSubscription value) { final total_length = FfiConverterInt8.allocationSize(value.priority) + - FfiConverterOptionalDuration.allocationSize(value.maxDelay) + + FfiConverterMoqDelay.allocationSize(value.delay) + FfiConverterOptionalSequenceMoqRange.allocationSize(value.ranges) + FfiConverterOptionalMoqOrder.allocationSize(value.order) + 0; @@ -894,8 +955,8 @@ class FfiConverterMoqSubscription { value.priority, Uint8List.view(buf.buffer, new_offset), ); - new_offset += FfiConverterOptionalDuration.write( - value.maxDelay, + new_offset += FfiConverterMoqDelay.write( + value.delay, Uint8List.view(buf.buffer, new_offset), ); new_offset += FfiConverterOptionalSequenceMoqRange.write( @@ -911,7 +972,7 @@ class FfiConverterMoqSubscription { static int allocationSize(MoqSubscription value) { return FfiConverterInt8.allocationSize(value.priority) + - FfiConverterOptionalDuration.allocationSize(value.maxDelay) + + FfiConverterMoqDelay.allocationSize(value.delay) + FfiConverterOptionalSequenceMoqRange.allocationSize(value.ranges) + FfiConverterOptionalMoqOrder.allocationSize(value.order) + 0; @@ -1766,65 +1827,6 @@ class FfiConverterMoqDatagram { } } -class MoqDelay { - final Duration min; - final Duration? max; - MoqDelay({required this.min, this.max}); -} - -class FfiConverterMoqDelay { - static MoqDelay lift(RustBuffer buf) { - return FfiConverterMoqDelay.read(buf.asUint8List()).value; - } - - static LiftRetVal read(Uint8List buf) { - int new_offset = buf.offsetInBytes; - final min_lifted = FfiConverterDuration.read( - Uint8List.view(buf.buffer, new_offset), - ); - final min = min_lifted.value; - new_offset += min_lifted.bytesRead; - final max_lifted = FfiConverterOptionalDuration.read( - Uint8List.view(buf.buffer, new_offset), - ); - final max = max_lifted.value; - new_offset += max_lifted.bytesRead; - return LiftRetVal( - MoqDelay(min: min, max: max), - new_offset - buf.offsetInBytes, - ); - } - - static RustBuffer lower(MoqDelay value) { - final total_length = - FfiConverterDuration.allocationSize(value.min) + - FfiConverterOptionalDuration.allocationSize(value.max) + - 0; - final buf = Uint8List(total_length); - write(value, buf); - return toRustBuffer(buf); - } - - static int write(MoqDelay value, Uint8List buf) { - int new_offset = buf.offsetInBytes; - new_offset += FfiConverterDuration.write( - value.min, - Uint8List.view(buf.buffer, new_offset), - ); - new_offset += FfiConverterOptionalDuration.write( - value.max, - Uint8List.view(buf.buffer, new_offset), - ); - return new_offset - buf.offsetInBytes; - } - - static int allocationSize(MoqDelay value) { - return FfiConverterDuration.allocationSize(value.min) + - FfiConverterOptionalDuration.allocationSize(value.max) + - 0; - } -} - class MoqDimensions { final int width; final int height; @@ -21788,7 +21790,7 @@ void _checkApiVersion() { } void _checkApiChecksums() { - if (uniffi_moq_ffi_checksum_func_moq_subscription_defaults() != 37074) { + if (uniffi_moq_ffi_checksum_func_moq_subscription_defaults() != 63642) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_func_moq_log_level() != 24625) { @@ -21915,7 +21917,7 @@ void _checkApiChecksums() { 17412) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } - if (uniffi_moq_ffi_checksum_method_moqtrackconsumer_recv_group() != 12949) { + if (uniffi_moq_ffi_checksum_method_moqtrackconsumer_recv_group() != 40781) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqtrackconsumer_update() != 39083) { @@ -21938,7 +21940,7 @@ void _checkApiChecksums() { 60932) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } - if (uniffi_moq_ffi_checksum_method_moqjsonsnapshotconsumer_next() != 37038) { + if (uniffi_moq_ffi_checksum_method_moqjsonsnapshotconsumer_next() != 5806) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqjsonsnapshotproducer_demand() != @@ -22474,7 +22476,7 @@ void _checkApiChecksums() { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_constructor_moqmediagroupconsumer_new() != - 58198) { + 43404) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_constructor_moqmediatimebase_new() != 21606) { diff --git a/doc/lib/c/index.md b/doc/lib/c/index.md index 33133509ef8..07d3786ef8a 100644 --- a/doc/lib/c/index.md +++ b/doc/lib/c/index.md @@ -102,7 +102,7 @@ inside. The header is the reference: each function carries a doc comment. - **Tokens.** Set `moq_client_config.tokens` instead of putting the credential in the URL. The first is the connection credential, sent in SETUP and, while an offered version can't carry it there (lite before 07), as the URL's `?jwt=`; each other token widens the grant over AUTH. A `?jwt=` already in the URL stays the connection credential, which is how to reach a relay older than this release. To replace expiring tokens, close the session and connect again. - **Threading.** Call any function from any thread. Callbacks run on moq-c's runtime thread; the callback and its `user_data` must be safe to use there. Synchronize access to data shared with other threads, and dispatch UI work to your UI thread. Raw publish calls block until the codec takes the frame, which paces a publisher. - **Zeroed config is the default.** A zeroed `moq_client_config` means the default for every field, so new fields don't break callers. A field whose default isn't zero has a `has_*` flag: `backoff_timeout_us = 0` means "retry forever" only with `has_backoff_timeout = true`. -- **Subscription defaults.** `moq_subscription_defaults()` returns moq-net's defaults, whose zero `max_delay_us` skips a superseded group at once; a NULL `moq_subscription` pointer means the same. Change its fields rather than zeroing one: a zeroed `moq_subscription` leaves `max_delay_present` false, which sets no subscriber limit. +- **Subscription defaults.** A zeroed `moq_subscription`, a NULL pointer, and `moq_subscription_defaults()` all mean moq-net's defaults, whose zero `delay` starts at the newest group and skips a superseded one at once. Set `delay.max_unlimited` for no subscriber limit, and `delay.min_unstated` to leave the target latency unstated. - **Compressed tracks.** `moq_track_info.compress` on a raw track sends each group DEFLATE-compressed to opt-in `moq-lite-07-wip` peers; other peers, and every reader here, get the original frames. - **fMP4 import.** `MOQ_CONTAINER_FORMAT_FMP4` splits each fragment into one frame per sample. - **Audio needs cuts.** Video groups at its keyframes, but audio forms a group only where you call `moq_publish_media_cut`: after every frame, or at a segment cadence to align with video. Pass that cadence as `moq_audio_init.hint.group_duration_us`. diff --git a/doc/lib/cpp/index.md b/doc/lib/cpp/index.md index 4b139d69663..a2084da6629 100644 --- a/doc/lib/cpp/index.md +++ b/doc/lib/cpp/index.md @@ -219,7 +219,7 @@ Hold the owned `BandwidthConsumer` returned by `session->recv_bandwidth()` to re ## Subscription defaults -`moq::subscription_defaults()` returns moq-net's defaults, whose zero `max_delay` skips a superseded group at once; passing `std::nullopt` for a subscription means the same. Change its fields rather than starting from `moq::Subscription{}`, whose unset `max_delay` sets no subscriber limit. +Start every subscription from `moq::subscription_defaults()` and change its fields. It holds moq-net's defaults, a zero `delay` that starts at the newest group and skips a superseded one at once, and passing `std::nullopt` for a subscription means the same. A bare `moq::Subscription{}` does not: the generated records carry no duration defaults, so its unset `delay.max` sets no subscriber limit and its unset `delay.min` leaves the target latency unstated. ## Track demand diff --git a/doc/lib/dart/index.md b/doc/lib/dart/index.md index 106c767cb72..05c0fef7842 100644 --- a/doc/lib/dart/index.md +++ b/doc/lib/dart/index.md @@ -159,7 +159,7 @@ shows which module holds each. - **Tokens.** Pass `ConnectOptions(tokens: [...])` instead of putting the credential in the URL. The first is the connection credential, sent in SETUP and, while an offered version can't carry it there (lite before 07), as the URL's `?jwt=`; each other token widens the grant over AUTH. `moq.tokens().set_(tokens: [...])` replaces them before they expire: the next dial presents them, a live AUTH session at once, and after an authorization refusal the session redials only if they changed. `await tokens().answered()` returns once the relay accepts them and throws its refusal with the relay's code. A `?jwt=` already in the URL stays the connection credential, which is how to reach a relay older than this release. - **Errors.** Read diagnostic payloads from `reason` and structured protocol failures from `details`. - **Disabled renditions.** A rendition with `enabled: false` has no frames coming until its publisher re-enables it. Select an enabled rendition from the latest catalog before opening a decoder. Keep one already playing while every rendition of its kind is disabled: the convenience decoders accept a disabled configuration, and a re-enable, such as an unmute, then plays without a resubscribe. Among the video renditions you can decode, prefer the highest `preference`; `0` is the default and `-1` marks a compatibility transcode. -- **Subscription defaults.** `defaultSubscription()` returns moq-net's defaults, whose zero `maxDelay` skips a superseded group at once; passing no subscription means the same. A `Subscription()` leaves `maxDelay` null, which sets no subscriber limit, so pass `maxDelay: Duration.zero` when building one to match. +- **Subscription defaults.** A bare `Subscription()`, passing no subscription, and `defaultSubscription()` all mean moq-net's defaults, whose zero `delay` starts at the newest group and skips a superseded one at once. `Delay` defaults both bounds to zero: pass `max: null` for no subscriber limit, or `min: null` to leave the target latency unstated. Decoders take the same `Delay`. - **Track demand.** Obtain a dynamic track request’s `demand()` before accepting it to watch subscribers through the handoff to its producer. The watch does not keep the track alive and ends when the track closes or the request is rejected. A producer publishes everything you write, subscribed or not, so a late subscriber can fetch the recent past. Capturing only while a producer's `demand()` reports a subscriber is an opt-in that saves the encode but leaves nothing from before the first one. ## Runtime diff --git a/doc/lib/go/index.md b/doc/lib/go/index.md index 5ad9e4eb92e..6cf412f6bd8 100644 --- a/doc/lib/go/index.md +++ b/doc/lib/go/index.md @@ -141,7 +141,7 @@ shows which module holds each. - **Tokens.** Set `ClientConfig.Tokens` instead of putting the credential in the URL. The first is the connection credential, sent in SETUP and, while an offered version can't carry it there (lite before 07), as the URL's `?jwt=`; each other token widens the grant over AUTH. `Session().Tokens().Set(...)` replaces a client session's tokens before they expire: the next dial presents them, a live AUTH session at once, and after an authorization refusal the session redials only if they changed. `Answered(ctx)` returns nil once the relay accepts them, or its refusal with the relay's code. A `?jwt=` already in the URL stays the connection credential, which is how to reach a relay older than this release. - **Errors.** Read diagnostic payloads from `Reason` and structured protocol failures from `Details`. - **Disabled renditions.** A rendition with `enabled: false` has no frames coming until its publisher re-enables it. Select an enabled rendition from the latest catalog before opening a decoder. Keep one already playing while every rendition of its kind is disabled: the convenience decoders accept a disabled configuration, and a re-enable, such as an unmute, then plays without a resubscribe. Among the video renditions you can decode, prefer the highest `Preference`; `0` is the default and `-1` marks a compatibility transcode. -- **Subscription defaults.** `moq.DefaultSubscription()` returns moq-net's defaults, whose zero `MaxDelay` skips a superseded group at once; a nil `*Subscription` means the same. Change its fields rather than starting from `Subscription{}`, whose nil `MaxDelay` sets no subscriber limit. +- **Subscription defaults.** A zero `moq.Subscription{}`, a nil `*Subscription`, and `moq.DefaultSubscription()` all mean moq-net's defaults, whose zero `Delay` starts at the newest group and skips a superseded one at once. Set `Delay.MaxUnlimited` for no subscriber limit, and `Delay.MinUnstated` to leave the target latency unstated. Decoders take the same `moq.Delay`. - **Track demand.** Obtain a dynamic track request’s `Demand()` before accepting it to watch subscribers through the handoff to its producer. The watch does not keep the track alive and ends when the track closes or the request is rejected. A producer publishes everything you write, subscribed or not, so a late subscriber can fetch the recent past. Capturing only while a producer's `Demand()` reports a subscriber is an opt-in that saves the encode but leaves nothing from before the first one. ## Runtime diff --git a/doc/lib/kt/index.md b/doc/lib/kt/index.md index d7c33fb9fe5..db04b5ac70a 100644 --- a/doc/lib/kt/index.md +++ b/doc/lib/kt/index.md @@ -114,7 +114,7 @@ shows which module holds each. - **Tokens.** Pass `ClientConfig(tokens = listOf(...))` instead of putting the credential in the URL. The first is the connection credential, sent in SETUP and, while an offered version can't carry it there (lite before 07), as the URL's `?jwt=`; each other token widens the grant over AUTH. `moq.tokens().set(listOf(...))` replaces them before they expire: the next dial presents them, a live AUTH session at once, and after an authorization refusal the session redials only if they changed. Suspending `tokens().answered()` returns once the relay accepts them and throws its refusal with the relay's code. A `?jwt=` already in the URL stays the connection credential, which is how to reach a relay older than this release. - **Errors.** Read diagnostic payloads from `reason` and structured protocol failures from `details`. - **Disabled renditions.** A rendition with `enabled: false` has no frames coming until its publisher re-enables it. Select an enabled rendition from the latest catalog before opening a decoder. Keep one already playing while every rendition of its kind is disabled: the convenience decoders accept a disabled configuration, and a re-enable, such as an unmute, then plays without a resubscribe. Among the video renditions you can decode, prefer the highest `preference`; `0` is the default and `-1` marks a compatibility transcode. -- **Subscription defaults.** `Subscription.default()` returns moq-net's defaults, whose zero `maxDelay` skips a superseded group at once; passing no subscription means the same. `copy` it to change a field; building a `Subscription(...)` must name `maxDelay`, where null sets no subscriber limit. +- **Subscription defaults.** `Subscription.default()` returns moq-net's defaults, whose zero `delay` starts at the newest group and skips a superseded one at once; passing no subscription means the same. `copy` it to change a field; building a `Subscription(...)` must name its `Delay(min, max)`, where a null `max` sets no subscriber limit and a null `min` leaves the target latency unstated. Decoders take the same `Delay`. - **Track demand.** Obtain a dynamic track request’s `demand()` before accepting it to watch subscribers through the handoff to its producer. The watch does not keep the track alive and ends when the track closes or the request is rejected. A producer publishes everything you write, subscribed or not, so a late subscriber can fetch the recent past. Capturing only while a producer's `demand()` reports a subscriber is an opt-in that saves the encode but leaves nothing from before the first one. ## Runtime diff --git a/doc/lib/py/index.md b/doc/lib/py/index.md index 3f1a8d5ae3f..8eebb126e12 100644 --- a/doc/lib/py/index.md +++ b/doc/lib/py/index.md @@ -129,7 +129,7 @@ shows which module holds each. - **Tokens.** Pass `tokens=[...]` instead of putting the credential in the URL. The first is the connection credential, sent in SETUP and, while an offered version can't carry it there (lite before 07), as the URL's `?jwt=`; each other token widens the grant over AUTH. `session.tokens().set([...])` replaces a client session's tokens before they expire: the next dial presents them, a live AUTH session at once, and after an authorization refusal the session redials only if they changed. `await tokens.answered()` returns once the relay accepts them and raises its refusal with the relay's code. A `?jwt=` already in the URL stays the connection credential, which is how to reach a relay older than this release. - **Errors.** Read diagnostic payloads from `reason` and structured protocol failures from `details`. - **Disabled renditions.** A rendition with `enabled: false` has no frames coming until its publisher re-enables it. Select an enabled rendition from the latest catalog before opening a decoder. Keep one already playing while every rendition of its kind is disabled: the convenience decoders accept a disabled configuration, and a re-enable, such as an unmute, then plays without a resubscribe. Among the video renditions you can decode, prefer the highest `preference`; `0` is the default and `-1` marks a compatibility transcode. -- **Subscription defaults.** `moq.Subscription.default()` returns moq-net's defaults, whose zero `max_delay` skips a superseded group at once; passing no subscription means the same. A bare `moq.Subscription()` has the same defaults; an explicit `max_delay=None` sets no subscriber limit. +- **Subscription defaults.** A bare `moq.Subscription()`, passing no subscription, and `moq.Subscription.default()` all mean moq-net's defaults, whose zero `delay` starts at the newest group and skips a superseded one at once. Pass `delay=moq.Delay(max=None)` for no subscriber limit, or `min=None` to leave the target latency unstated. Decoders take the same `moq.Delay`. - **Track demand.** Obtain a dynamic track request’s `demand()` before accepting it to watch subscribers through the handoff to its producer. The watch does not keep the track alive and ends when the track closes or the request is rejected. A producer publishes everything you write, subscribed or not, so a late subscriber can fetch the recent past. Capturing only while a producer's `demand()` reports a subscriber is an opt-in that saves the encode but leaves nothing from before the first one. ## Runtime diff --git a/doc/lib/swift/index.md b/doc/lib/swift/index.md index ec0ae5b804a..32fd9af777a 100644 --- a/doc/lib/swift/index.md +++ b/doc/lib/swift/index.md @@ -114,7 +114,7 @@ shows which module holds each. - **Tokens.** Pass `Client(tokens: [...])` instead of putting the credential in the URL. The first is the connection credential, sent in SETUP and, while an offered version can't carry it there (lite before 07), as the URL's `?jwt=`; each other token widens the grant over AUTH. `try session.tokens()?.set([...])` replaces them before they expire: the next dial presents them, a live AUTH session at once, and after an authorization refusal the session redials only if they changed. `try await tokens.answered()` returns once the relay accepts them and throws its refusal with the relay's code. A `?jwt=` already in the URL stays the connection credential, which is how to reach a relay older than this release. - **Errors.** Diagnostic cases use the `reason:` argument label; match their associated value to inspect it. Protocol errors carry `details`. - **Disabled renditions.** A rendition with `enabled: false` has no frames coming until its publisher re-enables it. Select an enabled rendition from the latest catalog before opening a decoder. Keep one already playing while every rendition of its kind is disabled: the convenience decoders accept a disabled configuration, and a re-enable, such as an unmute, then plays without a resubscribe. Among the video renditions you can decode, prefer the highest `preference`; `0` is the default and `-1` marks a compatibility transcode. -- **Subscription defaults.** `Subscription.default` holds moq-net's defaults, whose `maxDelay` of 0 skips a superseded group at once; passing no subscription means the same. Copy it to change a field; building a `Subscription(...)` must name `maxDelay`, where `nil` sets no subscriber limit. +- **Subscription defaults.** `Subscription.default` holds moq-net's defaults, whose zero `delay` starts at the newest group and skips a superseded one at once; passing no subscription means the same. Copy it to change a field; building a `Subscription(...)` must name its `Delay(min:max:)`, where a `nil` `max` sets no subscriber limit and a `nil` `min` leaves the target latency unstated. Decoders take the same `Delay`. - **Track demand.** Obtain a dynamic track request’s `demand()` before accepting it to watch subscribers through the handoff to its producer. The watch does not keep the track alive and ends when the track closes or the request is rejected. A producer publishes everything you write, subscribed or not, so a late subscriber can fetch the recent past. Capturing only while a producer's `demand()` reports a subscriber is an opt-in that saves the encode but leaves nothing from before the first one. ## Runtime diff --git a/go/wrapper/MIGRATE.md b/go/wrapper/MIGRATE.md index 8f0b8385259..ab7c8e208af 100644 --- a/go/wrapper/MIGRATE.md +++ b/go/wrapper/MIGRATE.md @@ -4,6 +4,12 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Subscriptions take a `moq.Delay` ([#7697](https://github.com/moq-dev/moq/pull/7697)) + +`Subscription.MaxDelay` is `Delay moq.Delay`, and the zero `Subscription{}` is moq-net's defaults, the same as `DefaultSubscription()`: set `Delay.Max` to the old `*MaxDelay`, and `Delay.MaxUnlimited` where `MaxDelay` was nil. +`media.Delay` is `moq.Delay`, a struct of its own whose zero value states zero for both: `Min` and `Max` are `time.Duration`, `MinUnstated` leaves the target latency unstated, and `MaxUnlimited` sets no limit. +`audio.DecoderOutput.Delay` and `video.DecoderOutput.Delay` are a `moq.Delay` value instead of a `*media.Delay`; the zero value is what nil meant. + ### `media.VideoInit` is a struct `media.VideoInit` is its own struct rather than an alias of the generated record, so its new `Lower` field takes a `*media.TrackProducer`. A composite literal with `Format`, `Data`, `Label`, and `Hint` compiles unchanged; build one that way where you converted a generated `MoqVideoInit`. @@ -17,7 +23,7 @@ How to update your code for each breaking change, newest first. `audio.DecoderOutput` replaces `MaxDelay` and the playout `Delay *time.Duration` with `Delay *media.Delay` and `Estimate *bool` (nil estimates). `video.DecoderOutput.MaxDelay` is `Delay *media.Delay`. `media.Delay{Min, Max}`: `Max` (nil for no limit) is the old `MaxDelay`, and `Min` is the playout delay, where a cold join starts. -A fixed audio playout delay `d` is `Delay: &media.Delay{Min: d, Max: &max}` with `Estimate` set to false. +A fixed audio playout delay `d` is `Delay: moq.Delay{Min: d, Max: max}` with `Estimate` set to false. ### Compression moves to the track diff --git a/go/wrapper/audio/audio.go b/go/wrapper/audio/audio.go index 1ccf0c68989..fe491a1bdb1 100644 --- a/go/wrapper/audio/audio.go +++ b/go/wrapper/audio/audio.go @@ -52,9 +52,9 @@ type DecoderOutput struct { SampleRate *uint32 Channels *uint32 // Delay.Min is the jitter buffer to hold ahead of the playhead, and where a cold join starts; - // Delay.Max bounds how stale a group may get before it is skipped, and adds no buffering. nil - // is zero for both. - Delay *media.Delay + // Delay.Max bounds how stale a group may get before it is skipped, and adds no buffering. The + // zero Delay is zero for both. + Delay moq.Delay // Estimate measures the playout delay from arrival timing, never below Delay.Min; false holds // exactly Delay.Min. nil estimates. Estimate *bool @@ -64,7 +64,8 @@ type DecoderOutput struct { } func (o DecoderOutput) ffi() (ffi.MoqAudioDecoderOutput, error) { - if err := delay(o.Delay, "audio"); err != nil { + delay, err := bridge.Delay(o.Delay) + if err != nil { return ffi.MoqAudioDecoderOutput{}, err } // Go's zero value is the native null: the oldest group. @@ -73,18 +74,7 @@ func (o DecoderOutput) ffi() (ffi.MoqAudioDecoderOutput, error) { start = &o.Start } estimate := o.Estimate == nil || *o.Estimate - return ffi.MoqAudioDecoderOutput{Format: o.Format, SampleRate: o.SampleRate, Channels: o.Channels, Delay: o.Delay, Estimate: estimate, Start: start}, nil -} - -// delay refuses a negative bound, which the FFI cannot carry. -func delay(d *media.Delay, kind string) error { - if d == nil { - return nil - } - if err := bridge.NonNegative(&d.Min, kind+" min delay"); err != nil { - return err - } - return bridge.NonNegative(d.Max, kind+" max delay") + return ffi.MoqAudioDecoderOutput{Format: o.Format, SampleRate: o.SampleRate, Channels: o.Channels, Delay: &delay, Estimate: estimate, Start: start}, nil } // Frame is interleaved PCM with its presentation timestamp. diff --git a/go/wrapper/bridge.go b/go/wrapper/bridge.go index 3209b931ca7..964965f008e 100644 --- a/go/wrapper/bridge.go +++ b/go/wrapper/bridge.go @@ -11,6 +11,7 @@ func init() { bridge.BroadcastConsumer = func(v any) *ffi.MoqBroadcastConsumer { return v.(*BroadcastConsumer).inner } bridge.TrackRequest = func(v any) *ffi.MoqTrackRequest { return v.(*TrackRequest).inner } bridge.Subscription = func(v any) (*ffi.MoqSubscription, error) { return subscriptionFFI(v.(*Subscription)) } + bridge.Delay = func(v any) (ffi.MoqDelay, error) { return v.(Delay).ffi() } bridge.BroadcastProducer = func(v any) *ffi.MoqBroadcastProducer { return v.(*BroadcastProducer).inner } bridge.TrackProducer = func(v any) *ffi.MoqTrackProducer { return v.(*TrackProducer).inner } diff --git a/go/wrapper/internal/bridge/handles.go b/go/wrapper/internal/bridge/handles.go index 1ed1fbdfeed..34f3483b201 100644 --- a/go/wrapper/internal/bridge/handles.go +++ b/go/wrapper/internal/bridge/handles.go @@ -41,6 +41,9 @@ var TrackRequest func(any) *ffi.MoqTrackRequest // Subscription converts a *moq.Subscription to the generated record. var Subscription func(any) (*ffi.MoqSubscription, error) +// Delay converts a moq.Delay to the generated record. +var Delay func(any) (ffi.MoqDelay, error) + // NonNegative refuses a negative duration, which the generated converter // would panic on rather than return an error. func NonNegative(value *time.Duration, name string) error { diff --git a/go/wrapper/json/json.go b/go/wrapper/json/json.go index 138f045919e..79c9d3977b1 100644 --- a/go/wrapper/json/json.go +++ b/go/wrapper/json/json.go @@ -161,7 +161,7 @@ func NewSnapshotConsumer(track *moq.TrackConsumer) (*SnapshotConsumer, error) { } // Next returns the next state in order, or nil when the track ends. Groups older -// than the subscription's MaxDelay are skipped, so a playhead reader sets that +// than the subscription's Delay.Max are skipped, so a playhead reader sets that // budget to its distance behind live. func (c *SnapshotConsumer) Next(ctx context.Context) (*Timed, error) { value, err := bridge.Call(ctx, c.inner.Cancel, c.inner.Next) diff --git a/go/wrapper/media/media.go b/go/wrapper/media/media.go index bbee93deab5..7a9ba2096a3 100644 --- a/go/wrapper/media/media.go +++ b/go/wrapper/media/media.go @@ -23,10 +23,6 @@ type ( Audio = ffi.MoqAudio // Catalog is a broadcast's manifest: its video and audio renditions plus display metadata. Catalog = ffi.MoqCatalog - // Delay is how far behind the live edge a decoder plays: Min is the playout delay and where a - // cold join starts, and Max, nil for no limit, is how far a stalled group may fall behind before - // it is skipped. The subscription carries both; a Min past Max starts at Max. - Delay = ffi.MoqDelay // Dimensions is a width and height in pixels. Dimensions = ffi.MoqDimensions // DecodeStart is where an audio or video decoder starts on a track that already holds groups. @@ -299,7 +295,7 @@ func FetchGroup(ctx context.Context, broadcast *moq.BroadcastConsumer, config Gr // // Pair it with moq.TrackConsumer.RecvGroup to read media in arrival order: frame timestamps are // container media time, not the transport's. Which groups arrive, and how long a stalled one -// stays readable, is the subscription's MaxDelay. Takes over group, whose handle is closed +// stays readable, is the subscription's Delay.Max. Takes over group, whose handle is closed // afterward; errors if it has already read a frame. func NewGroupConsumer(group *moq.GroupConsumer, container Container) (*GroupConsumer, error) { inner, err := ffi.NewMoqMediaGroupConsumer(bridge.GroupConsumer(group), container) diff --git a/go/wrapper/moq_test.go b/go/wrapper/moq_test.go index 9a1e813e68b..73942d39b48 100644 --- a/go/wrapper/moq_test.go +++ b/go/wrapper/moq_test.go @@ -288,11 +288,10 @@ func TestAudioConsumerDelay(t *testing.T) { } fixedDelay := 80 * time.Millisecond - noMax := time.Duration(0) estimate := false fixed, err := moqaudio.Subscribe(ctx, bc, track, rendition, moqaudio.DecoderOutput{ Format: moq.AudioSampleFormatF32, - Delay: &moqmedia.Delay{Min: fixedDelay, Max: &noMax}, + Delay: moq.Delay{Min: fixedDelay}, Estimate: &estimate, }) if err != nil { @@ -893,11 +892,10 @@ func TestAscendingSubscriptionReadsCachedBacklogOldestFirst(t *testing.T) { } order := moq.OrderAscending - maxDelay := 10 * time.Second consumer, err := track.Consume(&moq.Subscription{ - MaxDelay: &maxDelay, - Ranges: []moq.Range{{Start: moq.AnchorRelative{Groups: 2}}}, - Order: &order, + Delay: moq.Delay{Max: 10 * time.Second}, + Ranges: []moq.Range{{Start: moq.AnchorRelative{Groups: 2}}}, + Order: &order, }) if err != nil { t.Fatal(err) diff --git a/go/wrapper/records.go b/go/wrapper/records.go index 5ce0c0d6305..286554ca169 100644 --- a/go/wrapper/records.go +++ b/go/wrapper/records.go @@ -49,31 +49,74 @@ type Frame = ffi.MoqFrame // (nil from an untimed track), and payload. type Datagram = ffi.MoqDatagram -// Subscription holds subscriber-side delivery preferences: priority, the max delay -// a non-latest group may fall behind before it is skipped (zero skips at once, nil -// sets no subscriber limit and leaves the publisher's max age alone to bound it, and -// it caps every range), the ranges of the track to deliver (nil for everything -// MaxDelay still allows, while an empty non-nil slice asks for nothing), and the order -// the publisher sends pending groups in (nil for newest first). +// Delay is how far behind the live edge a subscriber or decoder plays: where it starts, and +// how late it may fall. The zero Delay is moq-net's default, a stated zero for both, which +// starts at the newest group and skips a superseded group at once. +type Delay struct { + // Min is the target latency: the playout or jitter-buffer delay held behind the live edge, + // and where a cold join starts. A Min past Max starts at Max, since the publisher serves + // nothing older. moq-lite-07 sends it upstream. + Min time.Duration + // MinUnstated leaves Min unstated, ignoring it. A decoder plays it as zero, while the + // subscription keeps it unstated, so it never pulls the aggregate across subscribers to zero. + MinUnstated bool + // Max is how far a non-latest group may fall behind the live edge before it is skipped. Zero + // skips immediately; a larger value tolerates that much reordering. It caps every range, + // and bounds lateness only, never the start. + Max time.Duration + // MaxUnlimited ignores Max and sets no subscriber limit, leaving the publisher's max age + // alone to bound it. + MaxUnlimited bool +} + +func (d Delay) ffi() (ffi.MoqDelay, error) { + var out ffi.MoqDelay + if !d.MinUnstated { + out.Min = &d.Min + } + if !d.MaxUnlimited { + out.Max = &d.Max + } + if err := bridge.NonNegative(out.Min, "min delay"); err != nil { + return out, err + } + return out, bridge.NonNegative(out.Max, "max delay") +} + +func delayFromFFI(d ffi.MoqDelay) Delay { + out := Delay{MinUnstated: d.Min == nil, MaxUnlimited: d.Max == nil} + if d.Min != nil { + out.Min = *d.Min + } + if d.Max != nil { + out.Max = *d.Max + } + return out +} + +// Subscription holds subscriber-side delivery preferences: priority, the Delay (where a cold +// join starts and how late a non-latest group may fall before it is skipped), the ranges of +// the track to deliver (nil for everything Delay.Max still allows, while an empty non-nil +// slice asks for nothing), and the order the publisher sends pending groups in (nil for +// newest first). // -// Start from DefaultSubscription for moq-net's defaults; the zero Subscription sets no -// max delay. +// The zero Subscription is moq-net's defaults, the same as DefaultSubscription. type Subscription struct { Priority int8 - MaxDelay *time.Duration + Delay Delay Ranges []Range Order *Order } -// DefaultSubscription returns moq-net's defaults, whose zero MaxDelay skips a superseded -// group at once. A nil *Subscription means the same; change a field of this one instead. +// DefaultSubscription returns moq-net's defaults, whose zero Delay starts at the newest group +// and skips a superseded group at once. A nil *Subscription means the same. func DefaultSubscription() Subscription { d := ffi.MoqSubscriptionDefaults() var ranges []Range if d.Ranges != nil { ranges = *d.Ranges } - return Subscription{Priority: d.Priority, MaxDelay: d.MaxDelay, Ranges: ranges, Order: d.Order} + return Subscription{Priority: d.Priority, Delay: delayFromFFI(d.Delay), Ranges: ranges, Order: d.Order} } func (s Subscription) ffi() (ffi.MoqSubscription, error) { @@ -81,12 +124,13 @@ func (s Subscription) ffi() (ffi.MoqSubscription, error) { if s.Ranges != nil { ranges = &s.Ranges } + delay, err := s.Delay.ffi() return ffi.MoqSubscription{ Priority: s.Priority, - MaxDelay: s.MaxDelay, + Delay: delay, Ranges: ranges, Order: s.Order, - }, bridge.NonNegative(s.MaxDelay, "max delay") + }, err } func subscriptionFFI(s *Subscription) (*ffi.MoqSubscription, error) { diff --git a/go/wrapper/records_internal_test.go b/go/wrapper/records_internal_test.go index 725d10d15d1..1ef5b175089 100644 --- a/go/wrapper/records_internal_test.go +++ b/go/wrapper/records_internal_test.go @@ -2,6 +2,7 @@ package moq import ( "math" + "reflect" "testing" "time" ) @@ -59,21 +60,27 @@ func TestTimestampToDurationZeroScale(t *testing.T) { TimestampToDuration(Timestamp{}) } -// DefaultSubscription carries moq-net's zero max delay, while the zero Subscription -// leaves it nil, which sets no subscriber limit. -func TestDefaultSubscriptionMaxDelay(t *testing.T) { - got, err := DefaultSubscription().ffi() +// The zero Subscription is moq-net's defaults, a stated zero Delay, and the flags reach the +// unstated minimum and the unlimited max. +func TestSubscriptionDelay(t *testing.T) { + if got := DefaultSubscription(); !reflect.DeepEqual(got, Subscription{}) { + t.Errorf("DefaultSubscription() = %+v, want the zero Subscription", got) + } + got, err := Subscription{}.ffi() if err != nil { t.Fatal(err) } - if got.MaxDelay == nil || *got.MaxDelay != 0 { - t.Errorf("DefaultSubscription MaxDelay = %v, want 0", got.MaxDelay) + if got.Delay.Min == nil || *got.Delay.Min != 0 || got.Delay.Max == nil || *got.Delay.Max != 0 { + t.Errorf("zero Subscription Delay = %+v, want a stated zero for both", got.Delay) } - got, err = Subscription{}.ffi() + got, err = Subscription{Delay: Delay{Min: time.Second, MinUnstated: true, Max: time.Second, MaxUnlimited: true}}.ffi() if err != nil { t.Fatal(err) } - if got.MaxDelay != nil { - t.Errorf("zero Subscription MaxDelay = %v, want nil", *got.MaxDelay) + if got.Delay.Min != nil || got.Delay.Max != nil { + t.Errorf("flagged Delay = %+v, want neither stated", got.Delay) + } + if _, err := (Subscription{Delay: Delay{Max: -time.Second}}).ffi(); err == nil { + t.Error("a negative max delay converted") } } diff --git a/go/wrapper/video/video.go b/go/wrapper/video/video.go index fd078739104..8111e4e0242 100644 --- a/go/wrapper/video/video.go +++ b/go/wrapper/video/video.go @@ -47,8 +47,8 @@ type Packet = ffi.MoqVideoPacket type DecoderOutput struct { Resize *media.Dimensions // Delay.Min is the playout delay, where a cold join starts; Delay.Max is how far a stalled - // group may fall behind before it is skipped. nil is zero for both. - Delay *media.Delay + // group may fall behind before it is skipped. The zero Delay is zero for both. + Delay moq.Delay Surface bool // Start is where to begin on a track that already holds groups; zero is media.DecodeStartOldest. // A live player sets media.DecodeStartLatest so a reopened decoder skips the cached backlog. @@ -61,14 +61,8 @@ func (o DecoderOutput) ffi() (ffi.MoqVideoDecoderOutput, error) { if o.Start != 0 { start = &o.Start } - out := ffi.MoqVideoDecoderOutput{Resize: o.Resize, Delay: o.Delay, Surface: o.Surface, Start: start} - if o.Delay == nil { - return out, nil - } - if err := bridge.NonNegative(&o.Delay.Min, "video min delay"); err != nil { - return out, err - } - return out, bridge.NonNegative(o.Delay.Max, "video max delay") + delay, err := bridge.Delay(o.Delay) + return ffi.MoqVideoDecoderOutput{Resize: o.Resize, Delay: &delay, Surface: o.Surface, Start: start}, err } // Producer encodes raw pictures and publishes them as a track in a broadcast's catalog. diff --git a/kt/moq/MIGRATE.md b/kt/moq/MIGRATE.md index ecb3c831ba8..98257b6dad4 100644 --- a/kt/moq/MIGRATE.md +++ b/kt/moq/MIGRATE.md @@ -4,14 +4,19 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Subscriptions take a `Delay` ([#7697](https://github.com/moq-dev/moq/pull/7697)) + +`Subscription.maxDelay` is `delay: Delay`: pass `delay = Delay(min = Duration.ZERO, max = maxDelay)`, where a null `max` sets no subscriber limit, or `copy` `Subscription.default()`. +`dev.moq.media.Delay` moves to `dev.moq.Delay`, and its `min` is nullable: null leaves the target latency unstated, which a decoder plays as zero. + ### Announcement captures are required ([#7484](https://github.com/moq-dev/moq/pull/7484)) `Announce.captures` is `List` instead of `List?`. Remove null checks and pass `captures = emptyList()` when constructing an announcement for a filter without wildcards. Rebuild with the matching generated bindings and native library. ### Decoder outputs take a `Delay` ([#7311](https://github.com/moq-dev/moq/pull/7311)) -`audio.DecoderOutput` replaces `maxDelay` and the playout `delay: Duration?` with `delay: media.Delay?` and `estimate: Boolean = true`. -`video.DecoderOutput.maxDelay` is `delay: media.Delay?`. +`audio.DecoderOutput` replaces `maxDelay` and the playout `delay: Duration?` with `delay: Delay?` and `estimate: Boolean = true`. +`video.DecoderOutput.maxDelay` is `delay: Delay?`. `Delay(min, max)`: `max` (null for no limit) is the old `maxDelay`, and `min` is the playout delay, where a cold join starts. A fixed audio playout delay `d` is `delay = Delay(min = d, max = ...)` with `estimate = false`. @@ -43,7 +48,6 @@ Their handles `OriginDynamic`, `BroadcastDynamic`, and `TrackDynamic` are `Origi The records carry durations as `java.time.Duration` fields instead of `*Us` microsecond integers, so the duration extension properties are gone: read `backoff.initial`, `stats.rtt`, `grant.expires`, `info.maxAge`, `fill.maxDelay()`, and `consumer.delay()` directly. Convert with `toKotlinDuration()` and `toJavaDuration()` where you want `kotlin.time.Duration`. `AudioEncoderOutput.frameDuration` is nullable: pass null where you passed `frameDurationUs = 0u` for the codec's own frame. -`Subscription.maxDelay` has no default: start from `Subscription.default()`, or pass `java.time.Duration.ZERO`, for the old default. Null sets no subscriber limit. `TrackConsumer.info()` no longer throws. ### Connection state diff --git a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt index fe561d01165..2789356ded4 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt @@ -181,16 +181,21 @@ typealias AnnounceRoute = uniffi.moq.MoqAnnounceRoute */ typealias Route = uniffi.moq.MoqRoute +/** Tunes how a track subscription is delivered: priority, delay, ranges, and group ordering. */ +typealias Subscription = uniffi.moq.MoqSubscription + /** - * Tunes how a track subscription is delivered: priority, latency budget, ranges, and group - * ordering. + * How far behind the live edge a subscriber or decoder plays: `min` is the target latency and where + * a cold join starts, null to leave it unstated (a decoder plays it as zero), and `max`, null for + * no subscriber limit, is how far a non-latest group may fall behind before it is skipped. A `min` + * past `max` starts at `max`. */ -typealias Subscription = uniffi.moq.MoqSubscription +typealias Delay = uniffi.moq.MoqDelay /** - * moq-net's defaults, whose `maxDelay` of zero skips a superseded group at once. A null - * subscription means the same; `copy` this one to change a field. A `Subscription(...)` must name - * `maxDelay`, where null sets no subscriber limit. + * moq-net's defaults, whose zero `delay` starts at the newest group and skips a superseded one at + * once. A null subscription means the same; `copy` this one to change a field. A + * `Subscription(...)` must name `delay`. */ fun uniffi.moq.MoqSubscription.Companion.default(): Subscription = uniffi.moq.moqSubscriptionDefaults() diff --git a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/json/Tracks.kt b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/json/Tracks.kt index 36ef697f80d..7db3dbee300 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/json/Tracks.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/json/Tracks.kt @@ -96,7 +96,7 @@ inline fun WindowEventPush.decode(): T = /** * Stream of each JSON state from a snapshot track, in order. * - * Groups older than the subscription's `maxDelay` are skipped, so a playhead reader sets that + * Groups older than the subscription's `delay.max` are skipped, so a playhead reader sets that * budget to its distance behind live. Call [SnapshotConsumer.latest] to skip to the newest state. */ fun SnapshotConsumer.values(): Flow = pull({ cancel() }) { next() } diff --git a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/media/Media.kt b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/media/Media.kt index 37037d8f006..d1bbec3d374 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/media/Media.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/media/Media.kt @@ -88,13 +88,6 @@ typealias Audio = uniffi.moq.MoqAudio /** A width and height pair, in pixels. */ typealias Dimensions = uniffi.moq.MoqDimensions -/** - * How far behind the live edge a decoder plays: `min` is the playout delay and where a cold join - * starts, and `max`, null for no limit, is how far a stalled group may fall behind before it is - * skipped. The subscription carries both; a `min` past `max` starts at `max`. - */ -typealias Delay = uniffi.moq.MoqDelay - /** * Where an audio or video decoder starts on a track that already holds groups: `LATEST` for a live * player. diff --git a/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt b/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt index cad86be4475..83188906403 100644 --- a/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt +++ b/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt @@ -309,8 +309,8 @@ class SmokeTest { @Test fun `a default subscription skips superseded groups at once`() { - assertEquals(java.time.Duration.ZERO, Subscription.default().maxDelay) - assertNull(Subscription(maxDelay = null).maxDelay) + assertEquals(java.time.Duration.ZERO, Subscription.default().delay.min) + assertEquals(java.time.Duration.ZERO, Subscription.default().delay.max) } @Test @@ -327,7 +327,11 @@ class SmokeTest { val consumer = track.consume( Subscription( - maxDelay = java.time.Duration.ofSeconds(10), + delay = + Delay( + min = java.time.Duration.ZERO, + max = java.time.Duration.ofSeconds(10), + ), ranges = listOf(Range(start = AnchorRelative(groups = 2uL))), order = Order.ASCENDING, ) diff --git a/py/moq-rs/MIGRATE.md b/py/moq-rs/MIGRATE.md index af31a32d54a..aad98da3ea3 100644 --- a/py/moq-rs/MIGRATE.md +++ b/py/moq-rs/MIGRATE.md @@ -4,15 +4,20 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Subscriptions take a `Delay` ([#7697](https://github.com/moq-dev/moq/pull/7697)) + +`Subscription.max_delay` is `delay`, a `moq.Delay`: pass `delay=moq.Delay(max=...)` for the old `max_delay=...`, and `moq.Delay(max=None)` for no subscriber limit. +`media.Delay` moves to `moq.Delay`, a dataclass whose `min` and `max` default to zero; `min=None` leaves the target latency unstated. The decoder outputs' `delay` defaults to `Delay()` instead of `None`. + ### Announcement captures are required ([#7484](https://github.com/moq-dev/moq/pull/7484)) `Announce.captures` is `list[str]` instead of `list[str] | None`. Remove `None` checks and pass `captures=[]` when constructing an announcement for a filter without wildcards. Rebuild with the matching generated bindings and native library. ### Decoder outputs take a `Delay` ([#7311](https://github.com/moq-dev/moq/pull/7311)) -`audio.DecoderOutput` replaces `max_delay` and the playout `delay: timedelta | None` with `delay: media.Delay | None` and `estimate: bool = True`. -`video.DecoderOutput.max_delay` is `delay: media.Delay | None`. -`media.Delay(min=..., max=...)`: `max` (`None` for no limit) is the old `max_delay`, and `min` is the playout delay, where a cold join starts. +`audio.DecoderOutput` replaces `max_delay` and the playout `delay: timedelta | None` with `delay: Delay` and `estimate: bool = True`. +`video.DecoderOutput.max_delay` is `delay: Delay`. +`Delay(min=..., max=...)`: `max` (`None` for no limit) is the old `max_delay`, and `min` is the playout delay, where a cold join starts. A fixed audio playout delay `d` is `delay=Delay(min=d, max=...)` with `estimate=False`. ### Compression moves to the track diff --git a/py/moq-rs/moq/__init__.py b/py/moq-rs/moq/__init__.py index 1725cd51a03..6ab6dbb7630 100644 --- a/py/moq-rs/moq/__init__.py +++ b/py/moq-rs/moq/__init__.py @@ -67,6 +67,7 @@ ConnectionState, ConnectionStats, Datagram, + Delay, ErrorScope, FetchGroupOptions, Frame, @@ -120,6 +121,7 @@ "ConnectionPhase", "ConnectionState", "Datagram", + "Delay", "Error", "ErrorScope", "Frame", diff --git a/py/moq-rs/moq/_records.py b/py/moq-rs/moq/_records.py index 45a2bebfd46..2f51ecfd577 100644 --- a/py/moq-rs/moq/_records.py +++ b/py/moq-rs/moq/_records.py @@ -21,6 +21,7 @@ MoqCongestionControl, MoqConnectionStats, MoqDatagram, + MoqDelay, MoqFrame, MoqGrant, MoqOrder, @@ -201,28 +202,53 @@ def _ffi(self) -> MoqRange: return MoqRange(start=self.start._ffi(), end=None if self.end is None else self.end._ffi()) +@dataclass(frozen=True) +class Delay: + """How far behind the live edge a subscriber or decoder plays, mirroring moq-net's ``track::Delay``. + + ``min`` is the target latency: the playout or jitter-buffer delay held behind the + live edge, and where a cold join starts. ``None`` leaves it unstated, which a decoder + plays as zero while the subscription keeps it unstated, so it never pulls the + aggregate across subscribers to zero. A ``min`` past ``max`` starts at ``max``. + ``max`` is how far a non-latest group may fall behind before it is skipped; zero + skips at once, ``None`` sets no subscriber limit and leaves the publisher's max age + alone to bound it, and it caps every range. + + A bare ``Delay()`` is moq-net's default, zero for both. + """ + + min: timedelta | None = timedelta(0) + max: timedelta | None = timedelta(0) + + @classmethod + def _from_ffi(cls, delay: MoqDelay) -> Delay: + return cls(min=delay.min, max=delay.max) + + def _ffi(self) -> MoqDelay: + return MoqDelay(min=self.min, max=self.max) + + @dataclass(frozen=True) class Subscription: """Subscriber-side delivery preferences, mirroring moq-net's ``track::Subscription``. ``priority`` is in ``-128..=127``, higher first, and defaults to 0, the midpoint. - ``max_delay`` is how far a non-latest group may fall behind before it is skipped; - zero skips at once, ``None`` sets no subscriber limit and leaves the publisher's max - age alone to bound it, and it caps every range. ``ranges`` are the spans of the track - to deliver, ``None`` for everything ``max_delay`` still allows. ``order`` is which - pending group the publisher sends first, ``None`` for newest first. + ``delay`` is where a cold join starts and how far a non-latest group may fall behind + before it is skipped. ``ranges`` are the spans of the track to deliver, ``None`` for + everything ``delay.max`` still allows. ``order`` is which pending group the publisher + sends first, ``None`` for newest first. A bare ``Subscription()`` has moq-net's defaults, as :meth:`default` does. """ priority: int = 0 - max_delay: timedelta | None = timedelta(0) + delay: Delay = Delay() ranges: Sequence[Range] | None = None order: MoqOrder | None = None @classmethod def default(cls) -> Subscription: - """moq-net's defaults, whose zero ``max_delay`` skips a superseded group at once. + """moq-net's defaults, whose zero ``delay`` starts at the newest group and skips a superseded one at once. A ``None`` subscription means the same; ``dataclasses.replace`` this one to change a field. @@ -230,7 +256,7 @@ def default(cls) -> Subscription: defaults = moq_subscription_defaults() return cls( priority=defaults.priority, - max_delay=defaults.max_delay, + delay=Delay._from_ffi(defaults.delay), # The FFI defaults never restrict ranges, so there is no record to convert. ranges=None, order=defaults.order, @@ -239,7 +265,7 @@ def default(cls) -> Subscription: def _ffi(self) -> MoqSubscription: return MoqSubscription( priority=self.priority, - max_delay=self.max_delay, + delay=self.delay._ffi(), ranges=None if self.ranges is None else [r._ffi() for r in self.ranges], order=self.order, ) diff --git a/py/moq-rs/moq/audio.py b/py/moq-rs/moq/audio.py index 79eeddeb41c..e01c0f8f100 100644 --- a/py/moq-rs/moq/audio.py +++ b/py/moq-rs/moq/audio.py @@ -19,9 +19,9 @@ ) from ._read import End, Gap, _end, _gap -from ._records import Timestamp +from ._records import Delay, Timestamp from ._stream import _Stream -from .media import Audio, AudioInit, CatalogProducer, DecodeStart, Delay +from .media import Audio, AudioInit, CatalogProducer, DecodeStart from .publish import TrackDemand from .session import Allocator, Reservation from .subscribe import BroadcastConsumer @@ -86,7 +86,7 @@ class DecoderOutput: ``delay.min`` is the jitter buffer to hold ahead of the playhead, and where a cold join starts; with ``estimate`` (the default) it floors the delay measured from arrival timing instead. ``delay.max`` only bounds how stale a group may - get before it is skipped, and adds no buffering. ``None`` is zero for both. + get before it is skipped, and adds no buffering. Both default to zero. ``start`` picks where to begin on a track that already holds groups, ``None`` for the oldest: a live player passes ``DecodeStart.LATEST`` so a reopened decoder skips the cached backlog. @@ -95,7 +95,7 @@ class DecoderOutput: format: SampleFormat sample_rate: int | None = None channels: int | None = None - delay: Delay | None = None + delay: Delay = Delay() estimate: bool = True start: DecodeStart | None = None @@ -104,7 +104,7 @@ def _ffi(self) -> MoqAudioDecoderOutput: format=self.format, sample_rate=self.sample_rate, channels=self.channels, - delay=self.delay, + delay=self.delay._ffi(), estimate=self.estimate, start=self.start, ) diff --git a/py/moq-rs/moq/json.py b/py/moq-rs/moq/json.py index e8a029e3dff..0eed010c2cb 100644 --- a/py/moq-rs/moq/json.py +++ b/py/moq-rs/moq/json.py @@ -126,7 +126,7 @@ def finish(self) -> None: class SnapshotConsumer(_Stream[Timed]): """Async iterator over a JSON snapshot track, yielding each state in order (lossy). - Each item is a :class:`Timed`. Groups older than the subscription's ``max_delay`` are skipped, + Each item is a :class:`Timed`. Groups older than the subscription's ``delay.max`` are skipped, so a playhead reader sets that budget to its distance behind live; :meth:`latest` skips to the newest state instead. Usable as an async context manager that cancels on exit. """ diff --git a/py/moq-rs/moq/media.py b/py/moq-rs/moq/media.py index 106277a431b..737fdd1a53a 100644 --- a/py/moq-rs/moq/media.py +++ b/py/moq-rs/moq/media.py @@ -40,7 +40,6 @@ MoqRendition, ) from moq_ffi import MoqDecodeStart as DecodeStart -from moq_ffi import MoqDelay as Delay from moq_ffi import MoqDimensions as Dimensions from moq_ffi import MoqVideo as Video from moq_ffi import MoqVideoFormat as VideoFormat @@ -374,7 +373,7 @@ def __init__(self, group: _subscribe.GroupConsumer, container: Container) -> Non Frame timestamps are container media time, not the transport's. Which groups arrive, and how long a stalled one stays readable, is the - subscription's ``max_delay``. Takes over ``group``, whose handle is + subscription's ``delay.max``. Takes over ``group``, whose handle is closed afterward; raises ``Error.AlreadyCommitted`` if it has already read a frame. """ @@ -467,7 +466,6 @@ async def catalog(broadcast: BroadcastConsumer) -> Catalog: "Video", "Catalog", "Container", - "Delay", "Dimensions", "DecodeStart", "AudioFormat", diff --git a/py/moq-rs/moq/types.py b/py/moq-rs/moq/types.py index ae87adc4a34..07ea1e05a49 100644 --- a/py/moq-rs/moq/types.py +++ b/py/moq-rs/moq/types.py @@ -71,6 +71,7 @@ ClientTls, ConnectionStats, Datagram, + Delay, Frame, Grant, QuicConfig, @@ -94,6 +95,7 @@ "ConnectionPhase", "ConnectionState", "Datagram", + "Delay", "ErrorScope", "Frame", "FetchGroupOptions", diff --git a/py/moq-rs/moq/video.py b/py/moq-rs/moq/video.py index 2c7c8ad88cd..bdc62fe2eca 100644 --- a/py/moq-rs/moq/video.py +++ b/py/moq-rs/moq/video.py @@ -17,9 +17,9 @@ from moq_ffi import MoqVideoFrameRate as FrameRate from ._read import End, Gap, _end, _gap -from ._records import Timestamp, VideoDecodedFrame +from ._records import Delay, Timestamp, VideoDecodedFrame from ._stream import _Stream -from .media import CatalogProducer, DecodeStart, Delay, Dimensions, Video +from .media import CatalogProducer, DecodeStart, Dimensions, Video from .publish import TrackDemand from .session import Allocator, Reservation from .subscribe import BroadcastConsumer @@ -63,21 +63,21 @@ class DecoderOutput: """What a :class:`Consumer` delivers: an optional resize, a delay, and the decoder's surface (macOS only). ``delay.min`` is the playout delay, where a cold join starts; ``delay.max`` is how - far behind the live edge a stalled group may fall before it is skipped. ``None`` is - zero for both. ``start`` picks where to begin on a track that already holds groups, + far behind the live edge a stalled group may fall before it is skipped. Both default + to zero. ``start`` picks where to begin on a track that already holds groups, ``None`` for the oldest: a live player passes ``DecodeStart.LATEST`` so a reopened decoder skips the cached backlog. """ resize: Dimensions | None = None - delay: Delay | None = None + delay: Delay = Delay() surface: bool = False start: DecodeStart | None = None def _ffi(self) -> MoqVideoDecoderOutput: return MoqVideoDecoderOutput( resize=self.resize, - delay=self.delay, + delay=self.delay._ffi(), surface=self.surface, start=self.start, ) diff --git a/py/moq-rs/tests/test_local.py b/py/moq-rs/tests/test_local.py index 7803aca57ba..6ba6d0b4cd6 100644 --- a/py/moq-rs/tests/test_local.py +++ b/py/moq-rs/tests/test_local.py @@ -486,10 +486,12 @@ def test_track_info_keeps_native_default_priority(): def test_default_subscription_skips_superseded_groups_at_once(): - """A bare record carries moq-net's zero max delay too; only an explicit None sets no limit.""" - assert moq.Subscription.default()._ffi().max_delay == timedelta(0) - assert moq.Subscription()._ffi().max_delay == timedelta(0) - assert moq.Subscription(max_delay=None)._ffi().max_delay is None + """A bare record carries moq-net's zero delay too; only an explicit None leaves a bound unset.""" + assert moq.Subscription.default() == moq.Subscription() + delay = moq.Subscription()._ffi().delay + assert (delay.min, delay.max) == (timedelta(0), timedelta(0)) + delay = moq.Subscription(delay=moq.Delay(min=None, max=None))._ffi().delay + assert (delay.min, delay.max) == (None, None) def test_string_sequences_refuse_a_bare_string(): @@ -1009,7 +1011,7 @@ async def test_raw_multiple_frames(): async for announcement in routes(consumer.announced()): broadcast_consumer = await consumer.request(announcement.prefix) - subscription = moq.Subscription(max_delay=timedelta(seconds=1), ranges=FROM_GROUP_0) + subscription = moq.Subscription(delay=moq.Delay(max=timedelta(seconds=1)), ranges=FROM_GROUP_0) raw_consumer = await broadcast_consumer.subscribe_track("commands", subscription) messages = [ @@ -1035,7 +1037,7 @@ async def test_raw_producer_consume_direct(): """Consume a raw track directly from the producer, no origin/broadcast plumbing.""" broadcast = moq.BroadcastProducer() track = broadcast.publish_track("direct") - consumer = track.consume(moq.Subscription(max_delay=timedelta(seconds=1), ranges=FROM_GROUP_0)) + consumer = track.consume(moq.Subscription(delay=moq.Delay(max=timedelta(seconds=1)), ranges=FROM_GROUP_0)) track.write_frame(b"hello") track.write_frame(b"world") @@ -1061,7 +1063,7 @@ async def test_ascending_subscription_reads_a_cached_backlog_oldest_first(): consumer = track.consume( moq.Subscription( - max_delay=timedelta(seconds=10), + delay=moq.Delay(max=timedelta(seconds=10)), ranges=[moq.Range(start=moq.Anchor.RELATIVE(groups=2))], order=moq.Order.ASCENDING, ) @@ -1086,7 +1088,7 @@ async def test_time_anchor_starts_at_the_group_containing_it(): start = moq.Anchor.TIME(timestamp=moq.Timestamp(1_500, 1_000)) consumer = track.consume( moq.Subscription( - max_delay=timedelta(seconds=10), + delay=moq.Delay(max=timedelta(seconds=10)), ranges=[moq.Range(start=start)], order=moq.Order.ASCENDING, ) @@ -1139,7 +1141,7 @@ async def test_raw_group_sequence(): async for announcement in routes(consumer.announced()): broadcast_consumer = await consumer.request(announcement.prefix) raw_consumer = await broadcast_consumer.subscribe_track( - "seq", moq.Subscription(max_delay=timedelta(seconds=1), ranges=FROM_GROUP_0) + "seq", moq.Subscription(delay=moq.Delay(max=timedelta(seconds=1)), ranges=FROM_GROUP_0) ) sent_sequences = [] @@ -1171,7 +1173,7 @@ async def test_default_iteration_is_sequence_order(): broadcast = create_announced(origin, "track/ordering") raw = broadcast.publish_track("ordering") - subscription = moq.Subscription(max_delay=timedelta(seconds=1), ranges=FROM_GROUP_0) + subscription = moq.Subscription(delay=moq.Delay(max=timedelta(seconds=1)), ranges=FROM_GROUP_0) seq_consumer = raw.consume(subscription) arr_consumer = raw.consume(subscription) @@ -1320,7 +1322,7 @@ async def test_read_frame_one_per_group(): """read_frame() returns the first frame of each successive group.""" broadcast = moq.BroadcastProducer() track = broadcast.publish_track("status") - consumer = track.consume(moq.Subscription(max_delay=timedelta(seconds=1), ranges=FROM_GROUP_0)) + consumer = track.consume(moq.Subscription(delay=moq.Delay(max=timedelta(seconds=1)), ranges=FROM_GROUP_0)) track.write_frame(b"ready") track.write_frame(b"running") @@ -1364,7 +1366,7 @@ async def test_read_frame_skips_remaining_frames_in_group(): """read_frame() only returns the first frame of a multi-frame group.""" broadcast = moq.BroadcastProducer() track = broadcast.publish_track("mixed") - consumer = track.consume(moq.Subscription(max_delay=timedelta(seconds=1), ranges=FROM_GROUP_0)) + consumer = track.consume(moq.Subscription(delay=moq.Delay(max=timedelta(seconds=1)), ranges=FROM_GROUP_0)) group = track.append_group() group.write_frame(b"first") @@ -1541,7 +1543,7 @@ async def test_audio_consumer_reports_its_playout_delay(): def output(delay: timedelta | None) -> moq.audio.DecoderOutput: if delay is None: return moq.audio.DecoderOutput(format=moq.AudioSampleFormat.F32) - fixed = moq.media.Delay(min=delay, max=timedelta(0)) + fixed = moq.Delay(min=delay) return moq.audio.DecoderOutput(format=moq.AudioSampleFormat.F32, delay=fixed, estimate=False) fixed = await moq.audio.Consumer.subscribe(broadcast_consumer, "mic", rendition, output(timedelta(milliseconds=80))) diff --git a/py/moq-rs/tests/test_testing_clock.py b/py/moq-rs/tests/test_testing_clock.py index d7c7ca6b3c8..e06bb6e3392 100644 --- a/py/moq-rs/tests/test_testing_clock.py +++ b/py/moq-rs/tests/test_testing_clock.py @@ -27,10 +27,10 @@ async def main(): origin = moq.OriginProducer() track = origin.create_broadcast("clock").broadcast.publish_track("track") from_group_0 = [moq.Range(start=moq.Anchor.ABSOLUTE(group=0))] - consumer = track.consume(moq.Subscription(max_delay=timedelta(seconds=1), ranges=from_group_0)) + consumer = track.consume(moq.Subscription(delay=moq.Delay(max=timedelta(seconds=1)), ranges=from_group_0)) # Group 0 stalls with no frame, so the in-order read waits on it until - # group 1 has been out for max_delay: a native timer is pending. + # group 1 has been out for delay.max: a native timer is pending. _stalled = track.append_group() later = track.append_group() later.write_frame(b"later") diff --git a/quest/m1/README.md b/quest/m1/README.md index 16fa5523e38..0dc50f8e763 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -150,7 +150,6 @@ per-frame CPU growth since #5926. - [A job's exact route continues its claim's readers](/quest/m1/announce-takeover.md) - a job's route inherits its claim's instance, replacing the ANNOUNCE_TAKEOVER predecessor design - [Subscribe ranges](/quest/m1/subscribe-ranges/README.md) - a lite-07 SUBSCRIBE asks for past and live ranges with absolute or relative ends, in either order, and replaces FETCH; merged ranges never starve a subscriber; relays fill misses by range, including over moq-transport - [Drop set](/quest/m1/drop-set.md) - each cursor judges missing groups against its own budget, so a narrow reader drops promptly and a widening re-offers groups already reported dropped on lite-07 -- [Binding subscriptions take a Delay { min, max }](/quest/m1/subscription-delay.md) - the bindings take the Rust and JS subscription Delay, and an omitted max means 0 in every one - [Datagram tracks](/quest/m1/datagram-tracks/README.md) - a track is datagrams or streams; datagram tracks have their own request and sequence, no ranges, no cache, and map to object ID 0 on moq-transport - [A step back smaller than the delay keeps the TS export](/quest/m1/ts-small-step-back.md) - audio restarts instead of regressing; video stays within its declared depth - [moq-boy's status loop stops on an errored track](/quest/m1/moq-boy-status-loop.md) - only schema errors are caught diff --git a/quest/m1/datagram-tracks/README.md b/quest/m1/datagram-tracks/README.md index 0b4771d3e6f..f787a92fe2a 100644 --- a/quest/m1/datagram-tracks/README.md +++ b/quest/m1/datagram-tracks/README.md @@ -32,8 +32,7 @@ Decided 2026-11-01 (maintainer, in `/quest-plan`): of those rules carry a datagram exception. - **Own request.** ✅ A datagram track is subscribed through its own lite-07 request, not SUBSCRIBE, and that request carries only a priority. ✅ It is - live from now: no start, end, order, or `Delay` - ([Subscription delay](/quest/m1/subscription-delay.md) exempts it). An update changes only + live from now: no start, end, order, or `Delay`. An update changes only the priority. A datagram still waits behind higher-priority traffic. - **No group terminology.** ✅ A datagram carries a sequence number and a timestamp, with no group. The sequence is the datagram track's own, not diff --git a/quest/m1/max-age-defaults.md b/quest/m1/max-age-defaults.md index de7f827b18b..e58ab86f732 100644 --- a/quest/m1/max-age-defaults.md +++ b/quest/m1/max-age-defaults.md @@ -83,13 +83,9 @@ One shared 30 s constant serves raw and media; `hang::container::MAX_AGE` reuses - **Catalog, MSF catalog, and HLS record tracks** inherit the raw default and become 30 s. Their newest group bootstraps a join, so this is harmless; switching them to zero is not part of this quest. - **Explicit policies stay:** GStreamer's 5 s opaque data, the JS catalog's `Number.MAX_SAFE_INTEGER`, room snapshots' one day, and archive playback's explicit no-limit republish. - **JS:** split authoring options from decoded `Info`. At producer construction an omitted or `undefined` `maxAge` is 30,000 ms, `null` is unlimited (matching [moq-dev/moq#7403](https://github.com/moq-dev/moq/pull/7403)'s `null` max-delay), and a number is that policy. `infoDefaults` and decoded `Info` keep absence as no publisher limit. `@moq/hang`'s `trackInfo` and `@moq/publish` gain the same `null`. -- **UniFFI and bindings:** fix the conversions that would otherwise inherit the new Rust default for explicit unlimited: UniFFI's `if let Some` in `rs/moq-ffi/src/producer.rs` and C's `if max_age_present` in `rs/moq-c/src/api.rs`. Where omission and unlimited can't be told apart (a UniFFI record default, a Go nil, a zeroed C struct), omitted means 30 s and an explicit unlimited flag carries unlimited, following the pattern [Subscriptions take a Delay](/quest/m1/subscription-delay.md) sets for max-delay. Python, Swift, Kotlin, Go, Dart, and C++ wrappers expose the same three cases. +- **UniFFI and bindings:** fix the conversions that would otherwise inherit the new Rust default for explicit unlimited: UniFFI's `if let Some` in `rs/moq-ffi/src/producer.rs` and C's `if max_age_present` in `rs/moq-c/src/api.rs`. Where omission and unlimited can't be told apart (a UniFFI record default, a Go nil, a zeroed C struct), omitted means 30 s and an explicit unlimited flag carries unlimited, following the pattern the bindings' subscription `Delay` sets for max-delay: Go's `moq.Delay.MaxUnlimited` and moq-c's `moq_delay.max_unlimited`. Python, Swift, Kotlin, Go, Dart, and C++ wrappers expose the same three cases. - **Relay:** no change. Unset `duration` keeps no local age ceiling and its 30 s idle expiry; byte budgets and cache expiry are untouched. - **Migration:** raw publishers relying on unlimited history must request unlimited, or a longer window, at construction, before publishing a new track or broadcast identity. Readers asking for history still set a sufficient max-delay and range. The PR title carries `!`, its Impact bullet says how to migrate, migration notes go under `## Unreleased`, and the Rust, JS, binding, and user docs change together. - **Tests:** cover authored omission versus explicit unlimited in each binding, decoded absence staying unlimited, timed and untimed supersession at 30 s, a late join to an unchanged newest snapshot, a direct fetch of a superseded group inside and past the window, media segment fetch, and archive replay. Use mocked time and the existing harnesses, and run all-language interop for the API and FFI scope. Re-check the overlapping PRs listed above rather than duplicating their fixes. Public API: breaking in Rust, JS, and every binding (the authored default, and mux `with_max_age(None)`). Wire: none; only authored advertisements change. - -## Related - -- [Subscriptions take a Delay](/quest/m1/subscription-delay.md) - the max-delay binding defaults and explicit-unlimited flags this quest's bindings mirror diff --git a/quest/m1/subscription-delay.md b/quest/m1/subscription-delay.md deleted file mode 100644 index 22a37b9f341..00000000000 --- a/quest/m1/subscription-delay.md +++ /dev/null @@ -1,28 +0,0 @@ -# [M] Binding subscriptions take a Delay { min, max } - -## Goal - -Every binding's subscription takes the same `Delay { min, max }` that moq-net's `track::Subscription` and `@moq/net`'s `Track.Subscription` took in [moq-dev/moq#7597](https://github.com/moq-dev/moq/pull/7597), in place of a bare max delay. That covers FFI `MoqSubscription`, moq-c `moq_subscription`, and the Python, Swift, Kotlin, Go, Dart, and C++ wrappers. An omitted max means 0 in every binding, including Dart, Go, and C. - -## Plan - -Decided 2026-10-30 (maintainer, follow-up planning): m1, because it breaks the public API and doing it before the cut avoids a second break. -[moq-dev/moq#7597](https://github.com/moq-dev/moq/pull/7597) landed the Rust and JS half: `track::Delay` / `Track.Delay`, `Subscription.delay`, the lite-07 `Subscriber Min Age`, and its aggregation. The bindings still expose `max_delay` and map it onto `delay.max`, stating a zero `min`. - -- **Representation:** decided 2026-11-02 (maintainer): the shared `Delay` has a nullable minimum. FFI `MoqDelay.min` becomes an explicit `null` (`Option`) that defaults to a stated zero, and C and Go get explicit absence flags so a zero-initialized value stays a stated zero. Media consumers read an unstated minimum as zero locally, and subscription metadata keeps it. -- **FFI:** `MoqSubscription.max_delay` becomes `delay: MoqDelay`, the same record the decoders take. `moq_subscription_defaults()` returns moq-net's defaults. -- **Binding defaults:** this change also fixes the bindings where an omitted max is still unlimited. - - Dart: an optional nullable parameter. - - Go: a nil `MaxDelay` in a zero `Subscription{}`. - - C: `max_delay_present` is false in a zeroed struct. - - Each one's omitted or zeroed form becomes 0, and unlimited stays expressible. Dart defaults the parameter to `Duration.zero`, so an explicit `null` stays unlimited. Go's nil and C's unset `present` can't mean both, so each gains an explicit unlimited flag in the delay, as `moq_delay.max_unlimited` already does for decoders; the PR names it. -- **Not datagram subscriptions:** a datagram track's request carries only a priority and takes no `Delay` ([Datagram tracks](/quest/m1/datagram-tracks/README.md), maintainer 2026-11-01). -- Update `doc/lib/{py,swift,kt,go,dart,c,cpp}` and add MIGRATE notes under `## Unreleased` for each breaking binding change. - -Public API: breaking in every binding. Wire: none. - -## Related - -- [A max-delay decoder joins an earlier group](https://github.com/moq-dev/moq/pull/7311) - its PR, #7311, adds `Delay` -- [Ordered reads wait only on owed groups](https://github.com/moq-dev/moq/pull/7403) - the Rust rule; its PR, #7403, settles the max-delay defaults and touches the same fields diff --git a/rs/moq-c/MIGRATE.md b/rs/moq-c/MIGRATE.md index 8a782e11d76..e09b312af74 100644 --- a/rs/moq-c/MIGRATE.md +++ b/rs/moq-c/MIGRATE.md @@ -4,6 +4,11 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Subscriptions take a `moq_delay` ([#7697](https://github.com/moq-dev/moq/pull/7697)) + +`moq_subscription.max_delay_us` and `max_delay_present` are `moq_delay delay`. A zeroed `moq_subscription` is moq-net's defaults again, the same as `moq_subscription_defaults()` and a NULL pointer: set `delay.max_us` to the old `max_delay_us`, and `delay.max_unlimited` where `max_delay_present` was false. +`moq_delay` gains `min_unstated`, after `max_unlimited`, which leaves the target latency unstated; the struct keeps its layout, and a zeroed one still states zero. + ### JSON window edits take an optional timestamp `moq_publish_json_window_push` and `moq_publish_json_window_pop` take a final @@ -21,11 +26,6 @@ Recompile callers against the regenerated `moq.h`. `moq_video_decoder_output.max_delay_us` and `moq_audio_decoder_output.max_delay_us` are `moq_delay delay`, which changes both structs' layout. Set `delay.max_us` to the old `max_delay_us`, `delay.max_unlimited` for no limit, and `delay.min_us` to your playout delay, where a cold join starts. A zeroed `moq_delay` is zero for both. -### A subscription without `max_delay_present` has no limit - -A `moq_subscription` with `max_delay_present` false sets no subscriber limit, leaving the publisher's max age alone to bound it, instead of moq-net's default of zero. -Start from `moq_subscription_defaults()`, or set `max_delay_present`, where you relied on a zeroed struct; a NULL pointer still means the defaults. - ### Compression moves to the track `moq_json_snapshot_config.compression`, `moq_json_stream_config`, `moq_binary_config`, and the `moq_publish_binary_snapshot*` and `moq_publish_binary_stream*` functions are removed. diff --git a/rs/moq-c/c-tests/decode-output.c b/rs/moq-c/c-tests/decode-output.c index 42f998cc526..c8dbaf59741 100644 --- a/rs/moq-c/c-tests/decode-output.c +++ b/rs/moq-c/c-tests/decode-output.c @@ -34,6 +34,7 @@ _Static_assert(MOQ_VIDEO_PIXEL_FORMAT_RGBA == 1, "RGBA discriminant moved"); _Static_assert(offsetof(moq_delay, min_us) == 0, "min_us moved"); _Static_assert(offsetof(moq_delay, max_us) == 8, "max_us moved"); _Static_assert(offsetof(moq_delay, max_unlimited) == 16, "max_unlimited moved"); +_Static_assert(offsetof(moq_delay, min_unstated) == 17, "min_unstated moved"); _Static_assert(sizeof(moq_delay) == 24, "delay size changed"); _Static_assert(offsetof(moq_video_decoder_output, delay) == 0, "delay moved"); _Static_assert(offsetof(moq_video_decoder_output, format) == 24, "format moved"); diff --git a/rs/moq-c/src/api.rs b/rs/moq-c/src/api.rs index ab0f050c394..eefa5157a81 100644 --- a/rs/moq-c/src/api.rs +++ b/rs/moq-c/src/api.rs @@ -651,7 +651,7 @@ pub struct moq_track_info { pub priority: i8, /// Maximum age of a non-latest group before the publisher evicts it, in microseconds. - /// The publisher-side half of `moq_subscription.max_delay_us`. + /// The publisher-side half of `moq_subscription.delay.max_us`. pub max_age_us: u64, /// Whether `max_age_us` is set. When false, the publisher imposes no age limit. pub max_age_present: bool, @@ -708,35 +708,51 @@ pub enum moq_demand { MOQ_DEMAND_UNWANTED = 2, } -/// How far behind the live edge a media consumer plays. +/// How far behind the live edge a subscriber or media consumer plays: where it starts, and how +/// late it may fall. /// -/// A zeroed struct is the default, and so is a null pointer where one is taken: start at -/// the newest group and skip aggressively. +/// A zeroed struct is moq-net's default, and so is a null pointer where one is taken: start at +/// the newest group and skip a superseded one at once. #[repr(C)] #[derive(Clone, Copy, Default)] pub struct moq_delay { /// The target latency in microseconds: the playout or jitter-buffer delay held behind - /// the live edge, and where a cold join starts. The subscription carries it too, which - /// moq-lite-07 sends upstream. A `min_us` past the max starts at the max, since the - /// publisher serves nothing older. + /// the live edge, and where a cold join starts. moq-lite-07 sends it upstream. A + /// `min_us` past the max starts at the max, since the publisher serves nothing older. + /// Ignored when `min_unstated` is set. pub min_us: u64, - /// How far a group may fall behind the live edge before it is skipped, in - /// microseconds. It bounds lateness only, never the start. Ignored when - /// `max_unlimited` is set. + /// How far a non-latest group may fall behind the live edge before it is skipped, in + /// microseconds. Zero skips immediately. Enforced by the publisher's cache and by any + /// local buffering, and it caps every range. It bounds lateness only, never the start. + /// Ignored when `max_unlimited` is set. pub max_us: u64, - /// No max: wait on a stalled group as long as anything could. + /// No subscriber max: wait on a stalled group as long as the publisher's max age allows. pub max_unlimited: bool, + /// Leave the minimum unstated. A media consumer plays it as zero, while the subscription + /// keeps it unstated, so it never pulls the aggregate across subscribers to zero. + pub min_unstated: bool, } impl From for moq_net::track::Delay { fn from(delay: moq_delay) -> Self { Self { - min: Some(std::time::Duration::from_micros(delay.min_us)), + min: (!delay.min_unstated).then(|| std::time::Duration::from_micros(delay.min_us)), max: (!delay.max_unlimited).then(|| std::time::Duration::from_micros(delay.max_us)), } } } +impl From for moq_delay { + fn from(delay: moq_net::track::Delay) -> Self { + Self { + min_us: delay.min.map(micros).unwrap_or_default(), + max_us: delay.max.map(micros).unwrap_or_default(), + max_unlimited: delay.max.is_none(), + min_unstated: delay.min.is_none(), + } + } +} + /// A [moq_delay] from a pointer that may be null, for the defaults. /// /// # Safety @@ -748,24 +764,19 @@ pub(crate) unsafe fn parse_delay(delay: *const moq_delay) -> moq_net::track::Del /// Subscriber-side raw track delivery preferences. /// /// A null [moq_consume_track] or [moq_consume_track_update] `subscription` -/// pointer uses the moq-net defaults, which [moq_subscription_defaults] returns as a -/// struct to change. A zeroed struct differs only in `max_delay_present`: it sets no -/// subscriber limit. +/// pointer uses the moq-net defaults, and so does a zeroed struct, which +/// [moq_subscription_defaults] returns to change. #[repr(C)] pub struct moq_subscription { /// Delivery priority. Higher values preempt lower ones under contention; the default 0 is /// the midpoint. pub priority: i8, - /// How far a non-latest group may fall behind the live edge before it is skipped, in microseconds. - /// Zero skips immediately. Enforced by the publisher's cache and by any local buffering. - /// It caps every range. - pub max_delay_us: u64, - /// Whether `max_delay_us` is set. When false, there is no subscriber limit, leaving the - /// publisher's max age alone to bound it. - pub max_delay_present: bool, + /// How far behind the live edge this subscriber plays: where a cold join starts, and how + /// late a non-latest group may fall before it is skipped. + pub delay: moq_delay, - /// The spans of the track to deliver, or NULL for everything `max_delay_us` still + /// The spans of the track to deliver, or NULL for everything `delay.max_us` still /// considers fresh. A non-NULL pointer with `ranges_len` 0 asks for nothing. /// /// A request to the publisher, aggregated across every subscriber, and this @@ -852,9 +863,7 @@ pub(crate) unsafe fn parse_subscription( let mut out = moq_net::track::Subscription::default() .with_priority(subscription.priority) .with_order(order); - let max_delay = std::time::Duration::from_micros(subscription.max_delay_us); - out = - out.with_delay(moq_net::track::Delay::default().with_max(subscription.max_delay_present.then_some(max_delay))); + out = out.with_delay(subscription.delay.into()); if !subscription.ranges.is_null() { let ranges = unsafe { std::slice::from_raw_parts(subscription.ranges, subscription.ranges_len) }; let ranges = ranges @@ -1282,9 +1291,9 @@ pub extern "C" fn moq_qlog_supported() -> bool { moq_tokio::qlog_supported() } -/// A [moq_subscription] with moq-net's defaults, whose zero `max_delay_us` skips a -/// superseded group at once. A NULL subscription pointer means the same; start from -/// this struct to change a field. +/// A [moq_subscription] with moq-net's defaults, whose zero `delay` starts at the newest +/// group and skips a superseded one at once. A NULL subscription pointer or a zeroed struct +/// means the same; start from this struct to change a field. /// /// Returned by value because there is nothing to fail. #[unsafe(no_mangle)] @@ -1296,8 +1305,7 @@ pub extern "C" fn moq_subscription_defaults() -> moq_subscription { }; moq_subscription { priority: defaults.priority, - max_delay_us: defaults.delay.max.map(micros).unwrap_or_default(), - max_delay_present: defaults.delay.max.is_some(), + delay: defaults.delay.into(), ranges: std::ptr::null(), ranges_len: 0, order: order as u32, diff --git a/rs/moq-c/src/test.rs b/rs/moq-c/src/test.rs index 49864cd5ffa..68115583dc5 100644 --- a/rs/moq-c/src/test.rs +++ b/rs/moq-c/src/test.rs @@ -25,6 +25,7 @@ const TEN_SECONDS: moq_delay = moq_delay { min_us: 0, max_us: 10_000_000, max_unlimited: false, + min_unstated: false, }; /// Convert a positive `i32` return value to `u32`, panicking on error. @@ -1768,8 +1769,10 @@ fn publish_track_invalid_broadcast() { let subscription = moq_subscription { priority: 1, - max_delay_us: 0, - max_delay_present: false, + delay: moq_delay { + max_unlimited: true, + ..moq_delay::default() + }, ranges: std::ptr::null(), ranges_len: 0, order: 0, @@ -1992,8 +1995,10 @@ fn raw_track_publish_consume() { // finish draining if the second becomes visible while the callback runs. let subscription = moq_subscription { priority: 0, - max_delay_us: 1_000_000, - max_delay_present: true, + delay: moq_delay { + max_us: 1_000_000, + ..moq_delay::default() + }, ranges: std::ptr::null(), ranges_len: 0, order: 0, @@ -2226,8 +2231,10 @@ fn raw_track_subscription_options_and_update() { ]; let subscription = moq_subscription { priority: 5, - max_delay_us: 25_000, - max_delay_present: true, + delay: moq_delay { + max_us: 25_000, + ..moq_delay::default() + }, ranges: ranges.as_ptr(), ranges_len: ranges.len(), order: 0, @@ -2337,8 +2344,10 @@ fn raw_track_time_anchor() { }]; let subscription = moq_subscription { priority: 5, - max_delay_us: 1_000_000, - max_delay_present: true, + delay: moq_delay { + max_us: 1_000_000, + ..moq_delay::default() + }, ranges: ranges.as_ptr(), ranges_len: ranges.len(), order: 0, @@ -2435,8 +2444,10 @@ fn raw_track_update_refuses_time_on_untimed() { }]; let update = moq_subscription { priority: 3, - max_delay_us: 1_000_000, - max_delay_present: true, + delay: moq_delay { + max_us: 1_000_000, + ..moq_delay::default() + }, ranges: ranges.as_ptr(), ranges_len: ranges.len(), order: 0, @@ -2492,8 +2503,10 @@ fn raw_track_relative_update_counts_from_the_track_edge() { }; let subscribe = |ranges: &[moq_range]| moq_subscription { priority: 0, - max_delay_us: 60_000_000, - max_delay_present: true, + delay: moq_delay { + max_us: 60_000_000, + ..moq_delay::default() + }, ranges: ranges.as_ptr(), ranges_len: ranges.len(), order: 0, @@ -3249,8 +3262,10 @@ fn publish_finish_serves_finished_tracks_until_closed() { let frame_cb = Callback::new(); let subscription = moq_subscription { priority: 0, - max_delay_us: 0, - max_delay_present: false, + delay: moq_delay { + max_unlimited: true, + ..moq_delay::default() + }, ranges: std::ptr::null(), ranges_len: 0, order: 0, @@ -6508,16 +6523,27 @@ fn a_zeroed_config_is_the_defaults() { assert!(parsed.quic.resolve().keylog.is_none()); } -/// `moq_subscription_defaults` reads back as moq-net's defaults, while a zeroed struct -/// sets no subscriber limit. +/// `moq_subscription_defaults` and a zeroed struct both read back as moq-net's defaults, and +/// the flags reach the unstated minimum and the unlimited max. #[test] fn subscription_defaults_are_moq_net_defaults() { let defaults = unsafe { crate::api::parse_subscription(&moq_subscription_defaults()) }.unwrap(); assert_eq!(defaults, moq_net::track::Subscription::default()); - let zeroed: moq_subscription = unsafe { std::mem::zeroed() }; - let zeroed = unsafe { crate::api::parse_subscription(&zeroed) }.unwrap(); - assert_eq!(zeroed.delay.max, None); + let mut zeroed: moq_subscription = unsafe { std::mem::zeroed() }; + assert_eq!( + unsafe { crate::api::parse_subscription(&zeroed) }.unwrap(), + moq_net::track::Subscription::default() + ); + + zeroed.delay = moq_delay { + min_us: 5, + max_us: 5, + max_unlimited: true, + min_unstated: true, + }; + let flagged = unsafe { crate::api::parse_subscription(&zeroed) }.unwrap(); + assert_eq!(flagged.delay, moq_net::track::Delay { min: None, max: None }); } /// `moq_client_defaults` has to agree with what a zeroed struct actually dials, diff --git a/rs/moq-ffi/MIGRATE.md b/rs/moq-ffi/MIGRATE.md index 943c1c7e8f6..247823c3be5 100644 --- a/rs/moq-ffi/MIGRATE.md +++ b/rs/moq-ffi/MIGRATE.md @@ -5,6 +5,12 @@ These are the UniFFI names; each language wrapper has its own `MIGRATE.md` in it ## Unreleased +### Subscriptions take a `MoqDelay` ([#7697](https://github.com/moq-dev/moq/pull/7697)) + +`MoqSubscription.max_delay` is `delay: MoqDelay`, the record the decoder outputs take: set `delay.max` to the old `max_delay`. +`MoqDelay.min` is `Option`: null leaves the target latency unstated, which a decoder plays as zero, while `moq_subscription_defaults()` states zero. Wrap a decoder's `min: d` as `Some(d)`. +Kotlin and Swift name both fields, and the Python, Go, and Dart wrappers default an omitted one to zero. A bare generated C or C++ record leaves both unset, so start from `moq_subscription_defaults()` there. + ### `MoqTransport::Iroh` is removed ([#7345](https://github.com/moq-dev/moq/pull/7345)) moq-tokio no longer accepts iroh sessions, so `MoqTransport` loses its `Iroh` variant and the later variants renumber; Go loses `TransportIroh`. @@ -58,9 +64,6 @@ It defaults to empty in every generated binding; a Rust struct literal adds `tok Every duration is a UniFFI `Duration` instead of an integer microsecond field, and drops its `_us` suffix: each language receives its own duration type (`timedelta`, `TimeInterval`, `java.time.Duration`, `time.Duration`, Dart `Duration`, C++ `std::chrono` nanoseconds). The renamed fields are `MoqClientConfig.{timeout, race, resolution_delay}`, `MoqQuicConfig.{idle_timeout, keep_alive}`, `MoqWebSocketConfig.delay`, `MoqBackoff.{initial, max, timeout}`, `MoqConnectionStats.rtt`, `MoqGrant.expires`, `MoqTrackInfo.max_age`, `MoqSubscription.max_delay`, `MoqFetchGroupOptions.max_delay`, `MoqAudio.jitter`, `MoqAudioDecoderOutput.{max_delay, delay}`, and `MoqVideoDecoderOutput.max_delay`; the methods are `MoqFill::max_delay()` and `MoqAudioConsumer::delay()`. `MoqAudioEncoderOutput.frame_duration_us: u32` is `frame_duration: Option`: pass null where you passed 0 for the codec's own frame. -`MoqSubscription.max_delay` has no default, since UniFFI can't default a duration to 0: Kotlin and Swift must name it, `0` for moq-net's default or null for no subscriber limit. -Dart, Go, and C can still omit it (Dart's optional parameter, Go's nil `*time.Duration`, a zeroed C struct), which means no subscriber limit; name it there too. -Start from `moq_subscription_defaults()` where you relied on that default; a null subscription argument still means it. `MoqTrackConsumer::info()` no longer fails. A duration handed out saturates at `i64::MAX` nanoseconds (about 292 years), so C, C++, and Go hold every value. diff --git a/rs/moq-ffi/src/audio.rs b/rs/moq-ffi/src/audio.rs index 6570e536f31..adc29cff582 100644 --- a/rs/moq-ffi/src/audio.rs +++ b/rs/moq-ffi/src/audio.rs @@ -170,11 +170,11 @@ pub struct MoqAudioDecoderOutput { /// [`MoqAudioConsumer::delay`], and where a cold join starts; with /// [`estimate`](Self::estimate) it floors the measured delay instead. A `min` /// beyond the track's retention is refused. `max` is the congestion-control - /// knob [`MoqSubscription::max_delay`](crate::consumer::MoqSubscription::max_delay) - /// carries: a stalled group more than this far behind a newer one is skipped, - /// with no buffering added. `None` keeps the moq-mux default of zero for both. + /// knob of [`MoqSubscription::delay`](crate::consumer::MoqSubscription::delay): + /// a stalled group more than this far behind a newer one is skipped, with no + /// buffering added. `None` keeps the moq-mux default of zero for both. #[uniffi(default = None)] - pub delay: Option, + pub delay: Option, /// Measure the playout delay from arrival timing, never below `delay.min`. /// `false` holds exactly `delay.min`. #[uniffi(default = true)] diff --git a/rs/moq-ffi/src/consumer.rs b/rs/moq-ffi/src/consumer.rs index 1550d77d4dd..3d254df1e6b 100644 --- a/rs/moq-ffi/src/consumer.rs +++ b/rs/moq-ffi/src/consumer.rs @@ -16,27 +16,59 @@ fn raw_frame(frame: moq_net::frame::Frame) -> MoqFrame { } } +/// How far behind the live edge a subscriber plays, mirroring [`moq_net::track::Delay`]: where +/// it starts, and how late it may fall. +/// +/// [`moq_subscription_defaults`] carries moq-net's default, zero for both. Uniffi can't default +/// a duration, so Kotlin and Swift name both fields, the Python, Go, and Dart wrappers default +/// an omitted one to zero, and a bare C or C++ record leaves both unset. +#[derive(Clone, Copy, Debug, uniffi::Record)] +pub struct MoqDelay { + /// The target latency: the playout or jitter-buffer delay held behind the live edge, and + /// where a cold join starts, or null to leave it unstated. A media consumer plays an + /// unstated minimum as zero, while the subscription keeps it unstated, so it never pulls + /// the aggregate across subscribers to zero. moq-lite-07 sends it upstream. A `min` past + /// `max` starts at `max`, since the publisher serves nothing older. + pub min: Option, + /// How far a non-latest group may fall behind the live edge before it is skipped, or null + /// for no subscriber limit, leaving the publisher's max age alone to bound it. `0` skips + /// immediately; a larger value tolerates that much reordering. It bounds lateness only, + /// never the start. + /// + /// Enforced both by the publisher's cache (sent on the wire) and by any local buffering, + /// such as [`MoqMediaTrackConsumer`]'s jitter buffer. It caps every range. + pub max: Option, +} + +impl From for moq_net::track::Delay { + fn from(delay: MoqDelay) -> Self { + Self::default().with_min(delay.min).with_max(delay.max) + } +} + +impl From for MoqDelay { + fn from(delay: moq_net::track::Delay) -> Self { + Self { + min: delay.min, + max: delay.max, + } + } +} + /// Subscriber-side delivery preferences, mirroring [`moq_net::track::Subscription`]. /// -/// Start from [`moq_subscription_defaults`] for moq-net's defaults (priority 0, a `max_delay` -/// of 0, everything still fresh, newest first) and change the fields you care about. Kotlin and -/// Swift make a record built field by field name `max_delay`; Dart, Go, and C can still omit it, -/// which means no subscriber limit. +/// Start from [`moq_subscription_defaults`] for moq-net's defaults (priority 0, a zero +/// `delay`, everything still fresh, newest first) and change the fields you care about. #[derive(Clone, uniffi::Record)] pub struct MoqSubscription { /// Delivery priority; higher values preempt lower ones under bandwidth contention. /// The default 0 is the midpoint. #[uniffi(default = 0)] pub priority: i8, - /// Maximum age of a non-latest group before it is skipped. - /// `0` skips immediately; a larger value tolerates that much reordering; null sets no - /// subscriber limit and leaves the publisher's max age alone to bound it. Required, - /// since uniffi can't default a duration to `0`. - /// - /// Enforced both by the publisher's cache (sent on the wire) and by any local - /// buffering, such as [`MoqMediaTrackConsumer`]'s jitter buffer. It caps every range. - pub max_delay: Option, - /// The spans of the track to deliver, or null for everything `max_delay` still + /// How far behind the live edge this subscriber plays: where a cold join starts, and how + /// late a non-latest group may fall before it is skipped. + pub delay: MoqDelay, + /// The spans of the track to deliver, or null for everything `delay.max` still /// considers fresh. An empty list asks for nothing. /// /// A request to the publisher, aggregated across every subscriber, and this @@ -48,7 +80,8 @@ pub struct MoqSubscription { pub order: Option, } -/// A subscription with moq-net's defaults, which skips a superseded group at once. +/// A subscription with moq-net's defaults, which starts at the newest group and skips a +/// superseded group at once. /// /// A null subscription argument means the same; build from this one to change a field. #[uniffi::export] @@ -56,7 +89,7 @@ pub fn moq_subscription_defaults() -> MoqSubscription { let defaults = moq_net::track::Subscription::default(); MoqSubscription { priority: defaults.priority, - max_delay: defaults.delay.max, + delay: defaults.delay.into(), ranges: None, order: None, } @@ -174,7 +207,7 @@ impl TryFrom for moq_net::track::Subscription { fn try_from(s: MoqSubscription) -> Result { let mut subscription = moq_net::track::Subscription::default() .with_priority(s.priority) - .with_delay(moq_net::track::Delay::default().with_max(s.max_delay)); + .with_delay(s.delay.into()); if let Some(ranges) = s.ranges { subscription = subscription.with_ranges( ranges @@ -575,7 +608,7 @@ impl MoqTrackConsumer { /// /// Groups are returned as they arrive on the wire, which may be out of sequence /// order (e.g. if a later group lands before an earlier one on a separate stream). - /// One that drifts further behind the live edge than the subscription's `max_delay` + /// One that drifts further behind the live edge than the subscription's `delay.max` /// is skipped, or ends its pending read if already returned; raise it to keep stalled groups. /// /// The first call commits this track to arrival order: sequence-order reads diff --git a/rs/moq-ffi/src/json.rs b/rs/moq-ffi/src/json.rs index ade243ec600..9ad9378c11c 100644 --- a/rs/moq-ffi/src/json.rs +++ b/rs/moq-ffi/src/json.rs @@ -246,7 +246,7 @@ impl MoqJsonSnapshotConsumer { /// Get the next state in order. Returns `None` once the track ends. /// - /// Groups older than the subscription's `max_delay` are skipped, so a playhead + /// Groups older than the subscription's `delay.max` are skipped, so a playhead /// reader sets that budget to its distance behind live. pub async fn next(&self) -> Result, MoqError> { let value = self diff --git a/rs/moq-ffi/src/media.rs b/rs/moq-ffi/src/media.rs index 335f0017253..feba781c4ec 100644 --- a/rs/moq-ffi/src/media.rs +++ b/rs/moq-ffi/src/media.rs @@ -22,27 +22,6 @@ pub struct MoqDimensions { pub height: u32, } -/// How far behind the live edge a decoder plays, the counterpart of `moq_net::track::Delay`. -#[derive(Clone, Copy, Debug, uniffi::Record)] -pub struct MoqDelay { - /// The target latency: the playout or jitter-buffer delay held behind the live edge, and - /// where a cold join starts. The subscription carries it too, which moq-lite-07 sends - /// upstream. A `min` past `max` starts at `max`, since the publisher serves nothing older. - pub min: std::time::Duration, - /// How far a group may fall behind the live edge before it is skipped, or null for no limit. - /// It bounds lateness only, never the start. - pub max: Option, -} - -impl From for moq_net::track::Delay { - fn from(delay: MoqDelay) -> Self { - Self { - min: Some(delay.min), - max: delay.max, - } - } -} - /// Where an audio or video decoder starts on a track that already holds groups. #[derive(Clone, Copy, Debug, Default, PartialEq, Eq, uniffi::Enum)] pub enum MoqDecodeStart { @@ -1755,7 +1734,7 @@ impl MoqMediaGroupConsumer { /// Pair it with [`MoqTrackConsumer::recv_group`](crate::consumer::MoqTrackConsumer::recv_group) /// to read media in arrival order: frame timestamps are container media time, not the /// transport's. Which groups arrive, and how long a stalled one stays readable, is the - /// subscription's `max_delay`. Takes over `group`, whose handle is closed afterward; errors + /// subscription's `delay.max`. Takes over `group`, whose handle is closed afterward; errors /// if it has already read a frame. #[uniffi::constructor] pub fn new(group: &MoqGroupConsumer, container: MoqContainer) -> Result, MoqError> { diff --git a/rs/moq-ffi/src/producer.rs b/rs/moq-ffi/src/producer.rs index 0836bac0e0b..8787053c394 100644 --- a/rs/moq-ffi/src/producer.rs +++ b/rs/moq-ffi/src/producer.rs @@ -41,7 +41,7 @@ pub struct MoqTrackInfo { pub priority: i8, /// Maximum age of a non-latest group before the publisher evicts it. Null imposes no /// publisher age limit. This is the publisher-side half of - /// [`MoqSubscription::max_delay`](crate::consumer::MoqSubscription::max_delay). + /// [`MoqDelay::max`](crate::consumer::MoqDelay::max). #[uniffi(default = None)] pub max_age: Option, /// How many times its first frame a group's later bytes may grow before a joining diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index c055599ad6b..b76777ec62e 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -6,7 +6,7 @@ use crate::consumer::MoqBroadcastConsumer; use crate::consumer::MoqFetchGroupOptions; use crate::consumer::MoqGroupConsumer; use crate::consumer::MoqTrackConsumer; -use crate::consumer::{MoqAnchor, MoqOrder, MoqRange, MoqSubscription, moq_subscription_defaults}; +use crate::consumer::{MoqAnchor, MoqDelay, MoqOrder, MoqRange, MoqSubscription, moq_subscription_defaults}; use crate::error::{MoqError, MoqProtocolError, MoqProtocolKind}; use crate::json::{ MoqJsonSnapshotConfig, MoqJsonSnapshotConsumer, MoqJsonSnapshotProducer, MoqJsonStreamConsumer, @@ -728,8 +728,8 @@ async fn audio_subscribes_a_held_disabled_rendition() { "voice".into(), muted, MoqAudioDecoderOutput { - delay: Some(crate::media::MoqDelay { - min: Duration::ZERO, + delay: Some(MoqDelay { + min: Some(Duration::ZERO), max: Some(Duration::from_secs(30)), }), ..audio_output() @@ -795,8 +795,8 @@ async fn audio_raw_publish_from_many_threads() { MoqAudioDecoderOutput { // The track's whole retention: reading starts only after finish, and the budget // ages groups by wall clock, so a stalled host would otherwise skip the start. - delay: Some(crate::media::MoqDelay { - min: Duration::ZERO, + delay: Some(MoqDelay { + min: Some(Duration::ZERO), max: Some(Duration::from_secs(30)), }), ..audio_output() @@ -945,8 +945,8 @@ async fn raw_audio_decode_delay() { "microphone".into(), rendition.clone(), MoqAudioDecoderOutput { - delay: delay.map(|min| crate::media::MoqDelay { - min, + delay: delay.map(|min| MoqDelay { + min: Some(min), max: Some(Duration::ZERO), }), estimate: delay.is_none(), @@ -1044,8 +1044,8 @@ async fn raw_audio_decode_reports_encoder_ends() { snapshot.audio["microphone"].clone(), MoqAudioDecoderOutput { // Reading starts only after finish, so keep every group. - delay: Some(crate::media::MoqDelay { - min: Duration::ZERO, + delay: Some(MoqDelay { + min: Some(Duration::ZERO), max: Some(Duration::from_secs(30)), }), ..audio_output() @@ -1296,7 +1296,10 @@ async fn raw_track_update_does_not_wait_for_pending_read() { consumer .update(MoqSubscription { priority: 10, - max_delay: Some(Duration::from_millis(25)), + delay: MoqDelay { + min: Some(Duration::ZERO), + max: Some(Duration::from_millis(25)), + }, ranges: Some(vec![MoqRange { start: MoqAnchor::Absolute { group: 0 }, end: None, @@ -4282,8 +4285,8 @@ async fn video_decode_start() { MoqVideoDecoderOutput { start, // The track's whole retention, so a slow host does not skip the first group. - delay: Some(crate::media::MoqDelay { - min: Duration::ZERO, + delay: Some(MoqDelay { + min: Some(Duration::ZERO), max: Some(Duration::from_secs(30)), }), ..Default::default() @@ -5353,7 +5356,10 @@ async fn raw_track_update_during_pending_group_still_reads_datagram() { consumer .update(MoqSubscription { priority: 10, - max_delay: Some(Duration::from_millis(25)), + delay: MoqDelay { + min: Some(Duration::ZERO), + max: Some(Duration::from_millis(25)), + }, ranges: Some(vec![MoqRange { start: MoqAnchor::Absolute { group: 0 }, end: None, @@ -6600,6 +6606,7 @@ async fn server_new_validates_the_config() { fn subscription_defaults_are_moq_net_defaults() { let defaults = moq_net::track::Subscription::try_from(moq_subscription_defaults()).unwrap(); assert_eq!(defaults, moq_net::track::Subscription::default()); + assert_eq!(defaults.delay.min, Some(Duration::ZERO)); assert_eq!(defaults.delay.max, Some(Duration::ZERO)); } @@ -8496,7 +8503,10 @@ async fn ascending_subscription_reads_a_cached_backlog_oldest_first() { let subscription = MoqSubscription { priority: 0, - max_delay: Some(Duration::from_secs(10)), + delay: MoqDelay { + min: Some(Duration::ZERO), + max: Some(Duration::from_secs(10)), + }, ranges: Some(vec![MoqRange { start: MoqAnchor::Relative { groups: 2 }, end: None, @@ -8705,7 +8715,10 @@ async fn time_anchors_start_at_the_containing_group() { let since = |timestamp_us| MoqSubscription { priority: 0, - max_delay: Some(Duration::from_secs(10)), + delay: MoqDelay { + min: Some(Duration::ZERO), + max: Some(Duration::from_secs(10)), + }, ranges: Some(vec![MoqRange { start: MoqAnchor::Time { timestamp: us(timestamp_us), @@ -8922,7 +8935,10 @@ async fn raw_read_frame_moves_on_when_next_group_takes_its_group() { let track = broadcast.publish_track("events".into(), Some(timed_info())).unwrap(); let consumer = track .consume(Some(MoqSubscription { - max_delay: None, + delay: MoqDelay { + min: Some(Duration::ZERO), + max: None, + }, ..moq_subscription_defaults() })) .unwrap(); @@ -8991,7 +9007,10 @@ async fn raw_next_group_wakes_read_frame_after_unlocking() { let track = broadcast.publish_track("events".into(), Some(timed_info())).unwrap(); let consumer = track .consume(Some(MoqSubscription { - max_delay: None, + delay: MoqDelay { + min: Some(Duration::ZERO), + max: None, + }, ..moq_subscription_defaults() })) .unwrap(); @@ -9028,7 +9047,10 @@ async fn raw_track_concurrent_reads_take_every_group_once() { // No age limit, so a group nobody has read yet is never skipped for a newer one. let consumer = track .consume(Some(MoqSubscription { - max_delay: None, + delay: MoqDelay { + min: Some(Duration::ZERO), + max: None, + }, ..moq_subscription_defaults() })) .unwrap(); diff --git a/rs/moq-ffi/src/video.rs b/rs/moq-ffi/src/video.rs index e5e43380d16..708e9a97214 100644 --- a/rs/moq-ffi/src/video.rs +++ b/rs/moq-ffi/src/video.rs @@ -1050,7 +1050,7 @@ pub struct MoqVideoDecoderOutput { /// `None` keeps the moq-mux default of zero for both (start at the newest group, /// skip aggressively). #[uniffi(default = None)] - pub delay: Option, + pub delay: Option, /// Keep each frame in the surface its decoder produced, for /// [`MoqVideoDecodedFrame::surface`]. `false`, the default, downloads every /// frame to CPU memory as it is decoded, so @@ -1872,10 +1872,11 @@ mod rung_tests { .subscribe_track( "high".into(), Some(crate::consumer::MoqSubscription { - priority: 0, - max_delay: None, - ranges: None, - order: None, + delay: crate::consumer::MoqDelay { + min: Some(std::time::Duration::ZERO), + max: None, + }, + ..crate::consumer::moq_subscription_defaults() }), ) .await diff --git a/swift/MIGRATE.md b/swift/MIGRATE.md index 3ff5db0b9d9..cc7bd5e5c43 100644 --- a/swift/MIGRATE.md +++ b/swift/MIGRATE.md @@ -4,16 +4,21 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Subscriptions take a `Delay` ([#7697](https://github.com/moq-dev/moq/pull/7697)) + +`Subscription.maxDelay` is `delay: Delay`: pass `delay: Delay(min: 0, max: maxDelay)`, where a `nil` `max` sets no subscriber limit, or start from `Subscription.default`. +`Media.Delay` moves to the root `Delay`, and its `min` is optional: `nil` leaves the target latency unstated, which a decoder plays as zero. + ### Announcement captures are required ([#7484](https://github.com/moq-dev/moq/pull/7484)) `Announce.captures` is `[String]` instead of `[String]?`. Remove optional checks and pass `captures: []` when constructing an announcement for a filter without wildcards. Rebuild with the matching MoqFFI package and native library. -### Decoder outputs take a `Media.Delay` ([#7311](https://github.com/moq-dev/moq/pull/7311)) +### Decoder outputs take a `Delay` ([#7311](https://github.com/moq-dev/moq/pull/7311)) -`Audio.DecoderOutput` replaces `maxDelay` and the playout `delay: TimeInterval?` with `delay: Media.Delay?` and `estimate: Bool = true`. -`Video.DecoderOutput.maxDelay` is `delay: Media.Delay?`. -`Media.Delay(min:max:)`: `max` (nil for no limit) is the old `maxDelay`, and `min` is the playout delay, where a cold join starts. -A fixed audio playout delay `d` is `delay: Media.Delay(min: d, max: ...)` with `estimate: false`. +`Audio.DecoderOutput` replaces `maxDelay` and the playout `delay: TimeInterval?` with `delay: Delay?` and `estimate: Bool = true`. +`Video.DecoderOutput.maxDelay` is `delay: Delay?`. +`Delay(min:max:)`: `max` (nil for no limit) is the old `maxDelay`, and `min` is the playout delay, where a cold join starts. +A fixed audio playout delay `d` is `delay: Delay(min: d, max: ...)` with `estimate: false`. ### Compression moves to the track @@ -42,7 +47,6 @@ Their handles `OriginDynamic`, `BroadcastDynamic`, and `TrackDynamic` are `Origi Every duration is a `TimeInterval` in seconds instead of a `UInt64` `...Us` field, and drops the suffix: `Client(timeout:race:resolutionDelay:)`, `Backoff(initial:max:timeout:)`, `QuicConfig(idleTimeout:keepAlive:)`, `WebSocketConfig(delay:)`, `ConnectionStats.rtt`, `Grant.expires`, `TrackInfo.maxAge`, `Subscription.maxDelay`, `Fill.maxDelay`, `Audio.Consumer.delay`, and the decoder outputs' `maxDelay` and `delay`. Divide a microsecond value by 1,000,000. A negative interval traps. `AudioEncoderOutput.frameDuration` is optional: pass `nil` where you passed `frameDurationUs: 0` for the codec's own frame. -`Subscription.maxDelay` has no default: start from `Subscription.default`, or pass `0`, for the old default. `nil` sets no subscriber limit. `TrackConsumer.info()` no longer throws. ### Connection state diff --git a/swift/Sources/Moq/Aliases.swift b/swift/Sources/Moq/Aliases.swift index 4c6bca3e995..ad36ebfff33 100644 --- a/swift/Sources/Moq/Aliases.swift +++ b/swift/Sources/Moq/Aliases.swift @@ -65,9 +65,14 @@ public typealias Announce = MoqFFI.MoqAnnounce /// What an `AnnounceConsumer` yields: `.start`, `.update`, `.end`, or /// `.restart` (request the path again), each carrying an `Announce`. public typealias AnnounceEvent = MoqFFI.MoqAnnounceEvent -/// Per-subscription delivery preferences: priority, latency budget, the ranges -/// of the track to deliver, and group ordering. +/// Per-subscription delivery preferences: priority, delay, the ranges of the track to +/// deliver, and group ordering. public typealias Subscription = MoqFFI.MoqSubscription +/// How far behind the live edge a subscriber or decoder plays: `min` is the target latency and +/// where a cold join starts, `nil` to leave it unstated (a decoder plays it as zero), and `max`, +/// `nil` for no subscriber limit, is how far a non-latest group may fall behind before it is +/// skipped. A `min` past `max` starts at `max`. +public typealias Delay = MoqFFI.MoqDelay /// Options for fetching one complete group by sequence. public typealias FetchGroupOptions = MoqFFI.MoqFetchGroupOptions /// Publisher-side track properties: priority, group ordering, max age, @@ -133,8 +138,8 @@ extension Subscription { /// Which pending group the publisher sends first: `.ascending` or `.descending`. public typealias Order = MoqFFI.MoqOrder - /// moq-net's defaults, whose `maxDelay` of 0 skips a superseded group at once. A `nil` - /// subscription means the same; copy this one to change a field. A `Subscription(...)` - /// must name `maxDelay`, where `nil` sets no subscriber limit. + /// moq-net's defaults, whose zero `delay` starts at the newest group and skips a superseded + /// one at once. A `nil` subscription means the same; copy this one to change a field. A + /// `Subscription(...)` must name `delay`. public static var `default`: Subscription { moqSubscriptionDefaults() } } diff --git a/swift/Sources/Moq/AsyncStream.swift b/swift/Sources/Moq/AsyncStream.swift index bf1ab56c33e..9e6223ab71f 100644 --- a/swift/Sources/Moq/AsyncStream.swift +++ b/swift/Sources/Moq/AsyncStream.swift @@ -5,7 +5,7 @@ import MoqFFI /// /// Pull-per-demand: `next` is called only when the consumer asks for the next /// element, never ahead of it. This is what preserves the FFI jitter buffer's -/// fall-behind skipping (`maxDelay` GoP-skipping): draining eagerly would +/// fall-behind skipping (`delay.max` GoP-skipping): draining eagerly would /// move the backlog into an unbounded Swift buffer, so a slow consumer would /// grow memory and receive stale groups instead of skipping forward. /// diff --git a/swift/Sources/Moq/Json.swift b/swift/Sources/Moq/Json.swift index c4717bb1c89..431f1864696 100644 --- a/swift/Sources/Moq/Json.swift +++ b/swift/Sources/Moq/Json.swift @@ -78,7 +78,7 @@ extension Json { /// Read side of a JSON snapshot track (lossy latest-value). Iterating yields each state in /// order with its timestamp: `for try await state in json { ... }`. Groups older than the - /// subscription's `maxDelay` are skipped; `latest()` skips to the newest state instead. + /// subscription's `delay.max` are skipped; `latest()` skips to the newest state instead. public final class SnapshotConsumer: AsyncSequence, Sendable { /// The decoded snapshot state emitted by this sequence. public typealias Element = Timed diff --git a/swift/Sources/Moq/Media.swift b/swift/Sources/Moq/Media.swift index 80fdf822acf..f9748be482c 100644 --- a/swift/Sources/Moq/Media.swift +++ b/swift/Sources/Moq/Media.swift @@ -29,10 +29,6 @@ public enum Media { public typealias Audio = MoqFFI.MoqAudio /// A width and height in pixels. public typealias Dimensions = MoqFFI.MoqDimensions - /// How far behind the live edge a decoder plays: `min` is the playout delay and where a - /// cold join starts, and `max`, nil for no limit, is how far a stalled group may fall - /// behind before it is skipped. The subscription carries both; a `min` past `max` starts at `max`. - public typealias Delay = MoqFFI.MoqDelay /// Where an audio or video decoder starts on a track that already holds /// groups: `.latest` for a live player. public typealias DecodeStart = MoqFFI.MoqDecodeStart @@ -197,7 +193,7 @@ public enum Media { /// /// Pair it with ``Moq/TrackConsumer/recvGroup()`` to read media in arrival order: frame /// timestamps are container media time, not the transport's. Which groups arrive, and how - /// long a stalled one stays readable, is the subscription's `maxDelay`. Takes over + /// long a stalled one stays readable, is the subscription's `delay.max`. Takes over /// `group`, whose handle is closed afterward; throws if it has already read a frame. public convenience init(group: Moq.GroupConsumer, container: Container) throws { self.init(try MoqMediaGroupConsumer(group: group.ffi, container: container)) diff --git a/swift/Tests/MoqTests/SmokeTests.swift b/swift/Tests/MoqTests/SmokeTests.swift index bc090bcebb9..6d211b6d179 100644 --- a/swift/Tests/MoqTests/SmokeTests.swift +++ b/swift/Tests/MoqTests/SmokeTests.swift @@ -27,8 +27,8 @@ private func assertTimeOverflow( final class SmokeTests: XCTestCase { func testDefaultSubscriptionSkipsSupersededGroupsAtOnce() { - XCTAssertEqual(Subscription.default.maxDelay, 0) - XCTAssertNil(Subscription(maxDelay: nil).maxDelay) + XCTAssertEqual(Subscription.default.delay.min, 0) + XCTAssertEqual(Subscription.default.delay.max, 0) } func testContainerTimebaseBelongsToOneBroadcast() throws { @@ -71,7 +71,7 @@ final class SmokeTests: XCTestCase { } let consumer = try track.consume( subscription: Subscription( - maxDelay: 10, + delay: Delay(min: 0, max: 10), ranges: [Subscription.Range(start: .relative(groups: 2))], order: .ascending )) @@ -709,7 +709,7 @@ final class SmokeTests: XCTestCase { let fixed = try await Audio.Consumer.subscribe( broadcast: consumer, id: "mic", rendition: rendition, - output: AudioDecoderOutput(format: .f32, delay: Media.Delay(min: 0.08, max: 0), estimate: false) + output: AudioDecoderOutput(format: .f32, delay: Delay(min: 0.08, max: 0), estimate: false) ) let estimated = try await Audio.Consumer.subscribe( broadcast: consumer, id: "mic", rendition: rendition, diff --git a/test/interop/clients/cpp/main.cpp b/test/interop/clients/cpp/main.cpp index 37a3fda1adf..41e897c65e9 100644 --- a/test/interop/clients/cpp/main.cpp +++ b/test/interop/clients/cpp/main.cpp @@ -364,8 +364,8 @@ int subscribe( } const auto &[id, video] = *catalog->video.begin(); - moq::Subscription subscription; - subscription.max_delay = + auto subscription = moq::subscription_defaults(); + subscription.delay.max = finish ? std::chrono::duration_cast(std::chrono::duration(timeout)) : MAX_DELAY; auto media = until( moq::MediaTrackConsumer::subscribe(consumer, {moq::Rendition::kVideo{id, video}, subscription}), read_deadline, diff --git a/test/interop/clients/go/main.go b/test/interop/clients/go/main.go index 8ab4f683d5d..c433cf4e05e 100644 --- a/test/interop/clients/go/main.go +++ b/test/interop/clients/go/main.go @@ -50,9 +50,6 @@ const maxDelay = 5 * time.Second // finishLinger is how long a finished publisher stays up for the late readers. const finishLinger = 10 * time.Second -// durationPtr is d as the optional duration the wrapper's options take. -func durationPtr(d time.Duration) *time.Duration { return &d } - // Synthetic audio: a 48 kHz tone per channel at 0.5 amplitude, encoded as Opus. const ( audioTrack = "tone" @@ -340,7 +337,7 @@ func finishSubscribe(ctx context.Context, client *moq.Client, broadcast string, for id, video = range catalog.Video { break } - media, err := moqmedia.NewTrackConsumer(ctx, consumer, moqmedia.TrackConfig{Rendition: moqmedia.VideoRendition(id, video), Subscription: &moq.Subscription{MaxDelay: durationPtr(maxDelay)}}) + media, err := moqmedia.NewTrackConsumer(ctx, consumer, moqmedia.TrackConfig{Rendition: moqmedia.VideoRendition(id, video), Subscription: &moq.Subscription{Delay: moq.Delay{Max: maxDelay}}}) if err != nil { return err } @@ -405,7 +402,7 @@ func finishSubscribe(ctx context.Context, client *moq.Client, broadcast string, break } } - healthAnnounced, err := client.Follow(broadcast + ".health", nil) + healthAnnounced, err := client.Follow(broadcast+".health", nil) if err != nil { return err } @@ -465,7 +462,7 @@ func decodeAudio(ctx context.Context, consumer *moq.BroadcastConsumer, id string out += "." + codec decoder, err := moqaudio.Subscribe(ctx, consumer, id, rendition, moqaudio.DecoderOutput{ Format: moq.AudioSampleFormatF32, - Delay: &moqmedia.Delay{Max: durationPtr(maxDelay)}, + Delay: moq.Delay{Max: maxDelay}, }) if err != nil { return err @@ -564,7 +561,7 @@ func subscribe(ctx context.Context, url, broadcast string, opts dialOptions, tim break } - media, err := moqmedia.NewTrackConsumer(ctx, consumer, moqmedia.TrackConfig{Rendition: moqmedia.VideoRendition(id, video), Subscription: &moq.Subscription{MaxDelay: durationPtr(maxDelay)}}) + media, err := moqmedia.NewTrackConsumer(ctx, consumer, moqmedia.TrackConfig{Rendition: moqmedia.VideoRendition(id, video), Subscription: &moq.Subscription{Delay: moq.Delay{Max: maxDelay}}}) if err != nil { return err } diff --git a/test/interop/clients/python/interop.py b/test/interop/clients/python/interop.py index cf0a899a87f..59c320f9f32 100644 --- a/test/interop/clients/python/interop.py +++ b/test/interop/clients/python/interop.py @@ -196,7 +196,7 @@ async def _finish_subscribe(client: moq.Client, broadcast: str, frames: int) -> raise RuntimeError("catalog ended without video") id, video = next(iter(catalog.video.items())) async with await moq.media.TrackConsumer.subscribe( - consumer, id, video, subscription=moq.Subscription(max_delay=MAX_DELAY) + consumer, id, video, subscription=moq.Subscription(delay=moq.Delay(max=MAX_DELAY)) ) as media: count = 0 async for frame in media: @@ -260,9 +260,7 @@ async def _decode_audio( consumer, id, audio, - moq.audio.DecoderOutput( - format=moq.audio.SampleFormat.F32, delay=moq.media.Delay(min=timedelta(0), max=MAX_DELAY) - ), + moq.audio.DecoderOutput(format=moq.audio.SampleFormat.F32, delay=moq.Delay(max=MAX_DELAY)), ) frame_bytes = 4 * audio.channel_count decoded = 0 @@ -312,7 +310,7 @@ async def subscribe( id, video = next(iter(catalog.video.items())) media = await moq.media.TrackConsumer.subscribe( - consumer, id, video, subscription=moq.Subscription(max_delay=MAX_DELAY) + consumer, id, video, subscription=moq.Subscription(delay=moq.Delay(max=MAX_DELAY)) ) total = 0