diff --git a/cpp/moq/include/moq/moq.hpp b/cpp/moq/include/moq/moq.hpp index 3ce636c350d..26e2103406b 100644 --- a/cpp/moq/include/moq/moq.hpp +++ b/cpp/moq/include/moq/moq.hpp @@ -60,6 +60,7 @@ using Client = MoqClient; using FlateSnapshotProducer = MoqFlateSnapshotProducer; using FlateStreamProducer = MoqFlateStreamProducer; using GroupConsumer = MoqGroupConsumer; +using GroupDemand = MoqGroupDemand; using GroupProducer = MoqGroupProducer; using GroupRequest = MoqGroupRequest; using JsonSnapshotConsumer = MoqJsonSnapshotConsumer; diff --git a/dart/moq/lib/src/aliases.dart b/dart/moq/lib/src/aliases.dart index 18ad02441bd..ea9d88c7061 100644 --- a/dart/moq/lib/src/aliases.dart +++ b/dart/moq/lib/src/aliases.dart @@ -92,6 +92,9 @@ typedef TrackConsumer = MoqTrackConsumer; /// A request to produce one uncached group for a fetch consumer. typedef GroupRequest = MoqGroupRequest; +/// A watch-only handle to whether any caller still wants a requested group; holding it keeps nothing open. +typedef GroupDemand = MoqGroupDemand; + /// The write side of a single group: append frames to it. typedef GroupProducer = MoqGroupProducer; diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index 9995ab97b8c..dd89cb80cb3 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/dart/moq_ffi/lib/src/moq.dart @@ -5460,6 +5460,108 @@ class FfiConverterMoqTrackConsumer { } } +abstract class MoqGroupDemandInterface { + bool isUsed(); + int sequence(); + Future unused(); + Future used(); +} + +final _MoqGroupDemandFinalizer = Finalizer>((ptr) { + rustCall((status) => uniffi_moq_ffi_fn_free_moqgroupdemand(ptr, status)); +}); + +class MoqGroupDemand implements MoqGroupDemandInterface { + late final Pointer _ptr; + MoqGroupDemand._(this._ptr) { + _MoqGroupDemandFinalizer.attach(this, _ptr, detach: this); + } + factory MoqGroupDemand.lift(Pointer ptr) { + return MoqGroupDemand._(ptr); + } + Pointer uniffiClonePointer() { + return rustCall( + (status) => uniffi_moq_ffi_fn_clone_moqgroupdemand(_ptr, status), + ); + } + + void dispose() { + _MoqGroupDemandFinalizer.detach(this); + rustCall((status) => uniffi_moq_ffi_fn_free_moqgroupdemand(_ptr, status)); + } + + bool isUsed() { + return rustCallWithLifter( + (status) => uniffi_moq_ffi_fn_method_moqgroupdemand_is_used( + uniffiClonePointer(), + status, + ), + FfiConverterBool.lift, + null, + ); + } + + int sequence() { + return rustCallWithLifter( + (status) => uniffi_moq_ffi_fn_method_moqgroupdemand_sequence( + uniffiClonePointer(), + status, + ), + FfiConverterUInt64.lift, + null, + ); + } + + Future unused() { + return uniffiRustCallAsync( + () => + uniffi_moq_ffi_fn_method_moqgroupdemand_unused(uniffiClonePointer()), + ffi_moq_ffi_rust_future_poll_void, + ffi_moq_ffi_rust_future_complete_void, + ffi_moq_ffi_rust_future_free_void, + (_) {}, + moqExceptionErrorHandler, + ); + } + + Future used() { + return uniffiRustCallAsync( + () => uniffi_moq_ffi_fn_method_moqgroupdemand_used(uniffiClonePointer()), + ffi_moq_ffi_rust_future_poll_void, + ffi_moq_ffi_rust_future_complete_void, + ffi_moq_ffi_rust_future_free_void, + (_) {}, + moqExceptionErrorHandler, + ); + } +} + +class FfiConverterMoqGroupDemand { + static MoqGroupDemand lift(Pointer ptr) { + return MoqGroupDemand.lift(ptr); + } + + static Pointer lower(MoqGroupDemand value) { + return value.uniffiClonePointer(); + } + + static int allocationSize(MoqGroupDemand value) { + return 8; + } + + static LiftRetVal read(Uint8List buf) { + final handle = buf.buffer.asByteData(buf.offsetInBytes).getInt64(0); + final pointer = Pointer.fromAddress(handle); + return LiftRetVal(MoqGroupDemand.lift(pointer), 8); + } + + static int write(MoqGroupDemand value, Uint8List buf) { + final handle = lower(value); + buf.buffer.asByteData(buf.offsetInBytes).setInt64(0, handle.address); + return 8; + } +} + abstract class MoqTrackDemandInterface { bool isUsed(); String name(); @@ -7920,6 +8022,7 @@ class FfiConverterMoqGroupProducer { abstract class MoqGroupRequestInterface { void abort({required int errorCode}); MoqGroupProducer accept(); + MoqGroupDemand demand(); int priority(); int sequence(); } @@ -7968,6 +8071,17 @@ class MoqGroupRequest implements MoqGroupRequestInterface { ); } + MoqGroupDemand demand() { + return rustCallWithLifter( + (status) => uniffi_moq_ffi_fn_method_moqgrouprequest_demand( + uniffiClonePointer(), + status, + ), + FfiConverterMoqGroupDemand.lift, + moqExceptionErrorHandler, + ); + } + int priority() { return rustCallWithLifter( (status) => uniffi_moq_ffi_fn_method_moqgrouprequest_priority( @@ -10416,6 +10530,48 @@ external void uniffi_moq_ffi_fn_method_moqtrackconsumer_update( Pointer uniffiStatus, ); +@Native Function(Pointer, Pointer)>( + assetId: _uniffiAssetId, +) +external Pointer uniffi_moq_ffi_fn_clone_moqgroupdemand( + Pointer handle, + Pointer uniffiStatus, +); + +@Native, Pointer)>( + assetId: _uniffiAssetId, +) +external void uniffi_moq_ffi_fn_free_moqgroupdemand( + Pointer handle, + Pointer uniffiStatus, +); + +@Native, Pointer)>( + assetId: _uniffiAssetId, +) +external int uniffi_moq_ffi_fn_method_moqgroupdemand_is_used( + Pointer ptr, + Pointer uniffiStatus, +); + +@Native, Pointer)>( + assetId: _uniffiAssetId, +) +external int uniffi_moq_ffi_fn_method_moqgroupdemand_sequence( + Pointer ptr, + Pointer uniffiStatus, +); + +@Native Function(Pointer)>(assetId: _uniffiAssetId) +external Pointer uniffi_moq_ffi_fn_method_moqgroupdemand_unused( + Pointer ptr, +); + +@Native Function(Pointer)>(assetId: _uniffiAssetId) +external Pointer uniffi_moq_ffi_fn_method_moqgroupdemand_used( + Pointer ptr, +); + @Native Function(Pointer, Pointer)>( assetId: _uniffiAssetId, ) @@ -11624,6 +11780,14 @@ external Pointer uniffi_moq_ffi_fn_method_moqgrouprequest_accept( Pointer uniffiStatus, ); +@Native Function(Pointer, Pointer)>( + assetId: _uniffiAssetId, +) +external Pointer uniffi_moq_ffi_fn_method_moqgrouprequest_demand( + Pointer ptr, + Pointer uniffiStatus, +); + @Native, Pointer)>( assetId: _uniffiAssetId, ) @@ -12454,6 +12618,18 @@ external int uniffi_moq_ffi_checksum_method_moqtrackconsumer_recv_group(); @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_method_moqtrackconsumer_update(); +@Native(assetId: _uniffiAssetId) +external int uniffi_moq_ffi_checksum_method_moqgroupdemand_is_used(); + +@Native(assetId: _uniffiAssetId) +external int uniffi_moq_ffi_checksum_method_moqgroupdemand_sequence(); + +@Native(assetId: _uniffiAssetId) +external int uniffi_moq_ffi_checksum_method_moqgroupdemand_unused(); + +@Native(assetId: _uniffiAssetId) +external int uniffi_moq_ffi_checksum_method_moqgroupdemand_used(); + @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_method_moqtrackdemand_is_used(); @@ -12696,6 +12872,9 @@ external int uniffi_moq_ffi_checksum_method_moqgrouprequest_abort(); @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_method_moqgrouprequest_accept(); +@Native(assetId: _uniffiAssetId) +external int uniffi_moq_ffi_checksum_method_moqgrouprequest_demand(); + @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_method_moqgrouprequest_priority(); @@ -12947,6 +13126,18 @@ void _checkApiChecksums() { if (uniffi_moq_ffi_checksum_method_moqtrackconsumer_update() != 24851) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } + if (uniffi_moq_ffi_checksum_method_moqgroupdemand_is_used() != 55523) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } + if (uniffi_moq_ffi_checksum_method_moqgroupdemand_sequence() != 34558) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } + if (uniffi_moq_ffi_checksum_method_moqgroupdemand_unused() != 1305) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } + if (uniffi_moq_ffi_checksum_method_moqgroupdemand_used() != 16710) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } if (uniffi_moq_ffi_checksum_method_moqtrackdemand_is_used() != 62559) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } @@ -13204,6 +13395,9 @@ void _checkApiChecksums() { if (uniffi_moq_ffi_checksum_method_moqgrouprequest_accept() != 48242) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } + if (uniffi_moq_ffi_checksum_method_moqgrouprequest_demand() != 31288) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } if (uniffi_moq_ffi_checksum_method_moqgrouprequest_priority() != 1745) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } diff --git a/go/wrapper/moq_test.go b/go/wrapper/moq_test.go index 135caaf441e..5c91342ea6e 100644 --- a/go/wrapper/moq_test.go +++ b/go/wrapper/moq_test.go @@ -451,6 +451,13 @@ func TestFetchGroupAndServeDynamicMiss(t *testing.T) { if request.Sequence() != 7 || request.Priority() != 11 { t.Fatalf("unexpected request: sequence=%d priority=%d", request.Sequence(), request.Priority()) } + demand, err := request.Demand() + if err != nil { + t.Fatal(err) + } + if demand.Sequence() != 7 || !demand.IsUsed() { + t.Fatalf("unexpected demand: sequence=%d used=%v", demand.Sequence(), demand.IsUsed()) + } produced, err := request.Accept() if err != nil { t.Fatal(err) diff --git a/go/wrapper/publish.go b/go/wrapper/publish.go index 541fb9dff4e..e7e043a3fa0 100644 --- a/go/wrapper/publish.go +++ b/go/wrapper/publish.go @@ -373,6 +373,15 @@ func (r *GroupRequest) Priority() uint8 { return r.inner.Priority() } +// Demand returns a watch-only handle to whether any caller still wants the group. +func (r *GroupRequest) Demand() (*GroupDemand, error) { + inner, err := r.inner.Demand() + if err != nil { + return nil, err + } + return &GroupDemand{inner: inner}, nil +} + // Accept accepts the request and returns a producer for the group. func (r *GroupRequest) Accept() (*GroupProducer, error) { inner, err := r.inner.Accept() @@ -387,6 +396,37 @@ func (r *GroupRequest) Abort(errorCode uint16) error { return r.inner.Abort(errorCode) } +// GroupDemand watches the callers waiting on a requested group. +// +// It is weak: holding it does not keep the request alive. The last caller to +// leave withdraws the request, so once unused, demand never returns: drop the +// request. Waits fail once the request is answered: ErrClosed if it was +// dropped, otherwise the error the accept or reject left for the waiting +// fetches. +type GroupDemand struct { + inner *ffi.MoqGroupDemand +} + +// Sequence is the sequence of the group this watches. +func (d *GroupDemand) Sequence() uint64 { + return d.inner.Sequence() +} + +// IsUsed reports whether the group has at least one waiting caller right now. +func (d *GroupDemand) IsUsed() bool { + return d.inner.IsUsed() +} + +// Used blocks until the group has at least one waiting caller. +func (d *GroupDemand) Used(ctx context.Context) error { + return d.inner.Used(ctx) +} + +// Unused blocks until the group has no waiting callers. +func (d *GroupDemand) Unused(ctx context.Context) error { + return d.inner.Unused(ctx) +} + // AudioProducer pushes raw PCM and lets libopus encode it on the way out. type AudioProducer struct { inner *ffi.MoqAudioProducer diff --git a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt index 9e2849810ad..f02d7055d34 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt @@ -81,6 +81,8 @@ typealias TrackDemand = uniffi.moq.MoqTrackDemand typealias TrackConsumer = uniffi.moq.MoqTrackConsumer /** A request to produce one uncached group for a fetch consumer. */ typealias GroupRequest = uniffi.moq.MoqGroupRequest +/** A watch-only handle to whether any caller still wants a requested group; holding it keeps nothing open. */ +typealias GroupDemand = uniffi.moq.MoqGroupDemand /** The write side of a single group: append frames to it. */ typealias GroupProducer = uniffi.moq.MoqGroupProducer /** The read side of a single group: yields timestamped raw frames. */ diff --git a/py/moq-rs/docs/index.md b/py/moq-rs/docs/index.md index 89170a1178c..1368e929129 100644 --- a/py/moq-rs/docs/index.md +++ b/py/moq-rs/docs/index.md @@ -60,6 +60,7 @@ asyncio.run(main()) TrackRequest GroupProducer GroupRequest + GroupDemand AudioProducer VideoProducer ``` diff --git a/py/moq-rs/moq/__init__.py b/py/moq-rs/moq/__init__.py index 4b2519599b5..400eff93182 100644 --- a/py/moq-rs/moq/__init__.py +++ b/py/moq-rs/moq/__init__.py @@ -27,6 +27,7 @@ AudioProducer, BroadcastDynamic, BroadcastProducer, + GroupDemand, GroupProducer, GroupRequest, TrackDemand, @@ -106,6 +107,7 @@ "ProtocolKind", "FetchGroupOptions", "GroupConsumer", + "GroupDemand", "GroupRequest", "GroupProducer", "OriginConsumer", diff --git a/py/moq-rs/moq/publish.py b/py/moq-rs/moq/publish.py index ddf245f6986..84606b1071b 100644 --- a/py/moq-rs/moq/publish.py +++ b/py/moq-rs/moq/publish.py @@ -9,6 +9,7 @@ MoqAudioProducer, MoqBroadcastDynamic, MoqBroadcastProducer, + MoqGroupDemand, MoqGroupProducer, MoqGroupRequest, MoqTrackDemand, @@ -194,6 +195,37 @@ def abort(self, error_code: int) -> None: self._inner.abort(error_code) +class GroupDemand: + """A watch-only handle to the callers waiting on a requested group. + + Returned by ``GroupRequest.demand()``. Weak: holding it does not keep the + request alive. The last caller to leave withdraws the request, so once + unused, demand never returns: drop the request. Waits raise once the + request is answered: ``moq.Error.Closed`` if it was dropped, otherwise the + error the accept or reject left for the waiting fetches. + """ + + def __init__(self, inner: MoqGroupDemand) -> None: + self._inner = inner + + @property + def sequence(self) -> int: + """The sequence of the group this watches.""" + return self._inner.sequence() + + def is_used(self) -> bool: + """Whether the group has at least one waiting caller right now.""" + return self._inner.is_used() + + async def used(self) -> None: + """Wait until the group has at least one waiting caller.""" + await self._inner.used() + + async def unused(self) -> None: + """Wait until the group has no waiting callers.""" + await self._inner.unused() + + class GroupRequest: """A request to produce one uncached group for a fetch consumer.""" @@ -210,6 +242,10 @@ def priority(self) -> int: """The consumer's delivery priority for this fetch.""" return self._inner.priority() + def demand(self) -> GroupDemand: + """A watch-only handle to whether any caller still wants this group.""" + return GroupDemand(self._inner.demand()) + def accept(self) -> GroupProducer: """Accept the request and return a producer for the group.""" return GroupProducer(self._inner.accept()) diff --git a/py/moq-rs/tests/test_local.py b/py/moq-rs/tests/test_local.py index f69645b8611..4d040281bbc 100644 --- a/py/moq-rs/tests/test_local.py +++ b/py/moq-rs/tests/test_local.py @@ -429,6 +429,9 @@ async def test_fetch_group_and_serve_dynamic_miss(): request = await asyncio.wait_for(dynamic.requested_group(), timeout=5.0) assert request.sequence == 7 assert request.priority == 11 + demand = request.demand() + assert demand.sequence == 7 + assert demand.is_used() produced = request.accept() produced.write_frame(b"archive", timedelta(microseconds=140_000)) diff --git a/quest/m1/ffi-shape/README.md b/quest/m1/ffi-shape/README.md index 676c671413d..e47d299c229 100644 --- a/quest/m1/ffi-shape/README.md +++ b/quest/m1/ffi-shape/README.md @@ -75,6 +75,5 @@ runs `just test interop --all`. - [Codecs](/quest/m1/ffi-shape/codec.md) - audio and video producers, codec-only encoders, and decoders move under their own namespaces, named as in Rust; binding releases wait for it - [Named error fields](/quest/m1/ffi-shape/error-fields.md) - `MoqError` variants name their fields, so no binding exposes a positional `v1` - [Bindings](/quest/m0/broadcast-epoch/bindings.md) - the wrappers expose epochs and rename `session.epoch()`, on the reshaped wrappers -- [Group request demand](/quest/m1/ffi-shape/group-request-demand.md) - `MoqGroupRequest::demand()` in moq-ffi and every wrapper - [Track request demand](/quest/m1/ffi-shape/track-request-demand.md) - `MoqTrackRequest::demand()` in moq-ffi and every wrapper, built on group request demand and coordinated with Bindings - [Layers guide](/quest/m1/ffi-shape/layers-guide.md) - a `doc/lib` page maps each Rust layer to every binding's module, once Codecs adds `audio` and `video` diff --git a/quest/m1/ffi-shape/group-request-demand.md b/quest/m1/ffi-shape/group-request-demand.md deleted file mode 100644 index 0660f452341..00000000000 --- a/quest/m1/ffi-shape/group-request-demand.md +++ /dev/null @@ -1,14 +0,0 @@ -# [S] A group request reports its demand - -## Goal - -`MoqGroupRequest` gains `demand()` in moq-ffi and every wrapper (Python, Go, -Swift, Kotlin, Dart, C++), matching Rust's `group::Request::demand`, so a -group server can see when nobody still wants the group and stop producing it. - -## Plan - -Decided in #4868; the bullet was lost when #4946 removed `net.md`, and it -moved out of the FFI shape README into its own quest when the line landed -(2026-10-09). Mirror `MoqTrackProducer`'s existing `demand()`, which returns a -`MoqTrackDemand`. Additive in every binding. Wire: none. diff --git a/quest/m1/ffi-shape/track-request-demand.md b/quest/m1/ffi-shape/track-request-demand.md index 673b9d1256d..bfac9267e9a 100644 --- a/quest/m1/ffi-shape/track-request-demand.md +++ b/quest/m1/ffi-shape/track-request-demand.md @@ -29,10 +29,6 @@ Additive in every binding. It edits the same wrappers as that PR to avoid conflicts, without blocking on it. Run `just test interop --all`. Wire: none. -## Required - -- [Group request demand](/quest/m1/ffi-shape/group-request-demand.md) - establishes the binding shape this follows; the 2026-10-10 audit makes the existing start-after rule explicit - ## Related - [Bindings](/quest/m0/broadcast-epoch/bindings.md) - edits the same wrappers; land after it diff --git a/rs/moq-ffi/src/demand.rs b/rs/moq-ffi/src/demand.rs index f0aaa18dc9a..72223a44532 100644 --- a/rs/moq-ffi/src/demand.rs +++ b/rs/moq-ffi/src/demand.rs @@ -1,4 +1,4 @@ -//! Watching whether anyone is subscribed to a published track. +//! Watching whether anyone still wants a published track or a requested group. use std::sync::Arc; @@ -45,10 +45,52 @@ impl MoqTrackDemand { } } -/// Report a track released without an abort as [`MoqError::Closed`]. +/// A watch-only handle to the callers waiting on a requested group. /// -/// That is how a finished track ends, which a demand watcher expects, not the internal failure -/// `Dropped` means to a consumer. An aborted track keeps its reason. +/// Weak: holding it does not keep the request alive. The last caller to leave withdraws the +/// request, and a later fetch of the group queues a fresh one, so once unused, demand never +/// returns: drop the request. Waits fail once the request is answered: with `Closed` if it was +/// dropped, otherwise with the error the accept or reject left for the waiting fetches. +#[derive(uniffi::Object)] +pub struct MoqGroupDemand { + inner: moq_net::group::Demand, +} + +impl MoqGroupDemand { + pub(crate) fn new(inner: moq_net::group::Demand) -> Arc { + Arc::new(Self { inner }) + } +} + +#[uniffi::export] +impl MoqGroupDemand { + /// The sequence of the group this watches. + pub fn sequence(&self) -> u64 { + self.inner.sequence() + } + + /// Whether the group has at least one waiting caller right now, without waiting. + pub fn is_used(&self) -> bool { + self.inner.is_used() + } + + /// Wait until the group has at least one waiting caller. + pub async fn used(&self) -> Result<(), MoqError> { + let demand = self.inner.clone(); + crate::ffi::detached(async move { gone(demand.used().await) }).await + } + + /// Wait until the group has no waiting callers. + pub async fn unused(&self) -> Result<(), MoqError> { + let demand = self.inner.clone(); + crate::ffi::detached(async move { gone(demand.unused().await) }).await + } +} + +/// Report a track or group released without an abort as [`MoqError::Closed`]. +/// +/// That is how a finished track or request ends, which a demand watcher expects, not the internal +/// failure `Dropped` means to a consumer. An aborted track or rejected request keeps its reason. fn gone(result: Result<(), moq_net::Error>) -> Result<(), MoqError> { match result { Err(moq_net::Error::Dropped) => Err(MoqError::Closed), diff --git a/rs/moq-ffi/src/producer.rs b/rs/moq-ffi/src/producer.rs index 7c47b7ed500..8cd0d7c76fb 100644 --- a/rs/moq-ffi/src/producer.rs +++ b/rs/moq-ffi/src/producer.rs @@ -3,7 +3,7 @@ use std::sync::Arc; use moq_mux::catalog::hang::Extra; use crate::consumer::{MoqBroadcastConsumer, MoqGroupConsumer, MoqSubscription, MoqTrackConsumer}; -use crate::demand::MoqTrackDemand; +use crate::demand::{MoqGroupDemand, MoqTrackDemand}; use crate::error::MoqError; use crate::ffi::Task; use crate::media::MoqFrame; @@ -341,6 +341,12 @@ impl MoqGroupRequest { self.priority } + /// A handle that watches whether any caller still wants this group. + pub fn demand(&self) -> Result, MoqError> { + let guard = self.inner.lock().unwrap(); + Ok(MoqGroupDemand::new(guard.as_ref().ok_or(MoqError::Closed)?.demand())) + } + /// Accept the request and return a producer for filling the fetched group. pub fn accept(&self) -> Result, MoqError> { let _guard = crate::ffi::enter(); diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index 2a52a33942c..84b10be010d 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -1365,6 +1365,62 @@ async fn dynamic_track_serves_fetch_miss_and_priority() { assert_eq!(frame.timestamp_us, Some(100_000)); } +#[tokio::test] +async fn group_request_demand_ends_when_the_fetcher_leaves() { + let broadcast = MoqBroadcastProducer::new().unwrap(); + let track = broadcast.publish_track("events".into(), None).unwrap(); + let dynamic = track.dynamic().unwrap(); + let consumer = broadcast.consume().unwrap(); + + let fetch = tokio::spawn(async move { consumer.fetch_group("events".into(), 5, None).await }); + let request = tokio::time::timeout(TIMEOUT, dynamic.requested_group()) + .await + .expect("timed out waiting for group request") + .unwrap(); + let demand = request.demand().unwrap(); + assert_eq!(demand.sequence(), 5); + assert!(demand.is_used()); + + fetch.abort(); + tokio::time::timeout(TIMEOUT, demand.unused()) + .await + .expect("timed out waiting for the abandoned fetch to become unused") + .unwrap(); + assert!(!demand.is_used()); + + request.abort(404).unwrap(); + assert!(matches!(request.demand(), Err(MoqError::Closed))); +} + +/// An accepted request's demand ends with the `NotFound` the accept leaves for joined fetches +/// it can't cover, not `Closed`. +#[tokio::test] +async fn group_request_demand_fails_after_accept() { + let broadcast = MoqBroadcastProducer::new().unwrap(); + let track = broadcast.publish_track("events".into(), None).unwrap(); + let dynamic = track.dynamic().unwrap(); + let consumer = broadcast.consume().unwrap(); + + let fetch = tokio::spawn(async move { consumer.fetch_group("events".into(), 5, None).await }); + let request = tokio::time::timeout(TIMEOUT, dynamic.requested_group()) + .await + .expect("timed out waiting for group request") + .unwrap(); + let demand = request.demand().unwrap(); + let group = request.accept().unwrap(); + group.finish().unwrap(); + + let used = tokio::time::timeout(TIMEOUT, demand.used()) + .await + .expect("timed out waiting for an accepted demand to end"); + assert!(matches!(used, Err(MoqError::NotFound)), "got {used:?}"); + tokio::time::timeout(TIMEOUT, fetch) + .await + .expect("timed out waiting for fetch") + .expect("fetch task panicked") + .unwrap(); +} + #[tokio::test] async fn dynamic_track_rejects_fetch_miss() { let broadcast = MoqBroadcastProducer::new().unwrap(); diff --git a/swift/Sources/Moq/Dynamic.swift b/swift/Sources/Moq/Dynamic.swift index 28a3f4284a8..9ab9445c60f 100644 --- a/swift/Sources/Moq/Dynamic.swift +++ b/swift/Sources/Moq/Dynamic.swift @@ -32,6 +32,40 @@ public final class TrackRequest: Sendable { } } +/// A watch-only handle to the callers waiting on a requested group. +/// +/// Returned by `GroupRequest.demand()`. Weak: holding it does not keep the request alive. The last +/// caller to leave withdraws the request, so once unused, demand never returns: drop the request. +/// Waits throw once the request is answered: `MoqError.Closed` if it was dropped, otherwise the +/// error the accept or reject left for the waiting fetches. +public final class GroupDemand: Sendable { + let ffi: MoqGroupDemand + + init(_ ffi: MoqGroupDemand) { + self.ffi = ffi + } + + /// The sequence of the group this watches. + public var sequence: UInt64 { + ffi.sequence() + } + + /// Whether the group has at least one waiting caller right now. + public var isUsed: Bool { + ffi.isUsed() + } + + /// Suspend until the group has at least one waiting caller. + public func used() async throws { + try await ffi.used() + } + + /// Suspend until the group has no waiting callers. + public func unused() async throws { + try await ffi.unused() + } +} + /// A request to produce one uncached group for a fetch consumer. public final class GroupRequest: Sendable { let ffi: MoqGroupRequest @@ -50,6 +84,11 @@ public final class GroupRequest: Sendable { ffi.priority() } + /// A watch-only handle to whether any caller still wants this group. + public func demand() throws -> GroupDemand { + GroupDemand(try ffi.demand()) + } + /// Accept the request and return a producer for the group. public func accept() throws -> GroupProducer { GroupProducer(try ffi.accept())