From fc855b1ca941b83367bb5e93dd1f6811ffd4e182 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 15:32:21 -0700 Subject: [PATCH 1/6] chore: claim quest/m1/ffi-shape/group-request-demand From e6be96768bcab3a4e87e5e02d54157490263dac7 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 15:47:23 -0700 Subject: [PATCH 2/6] feat(ffi): a group request reports its demand MoqGroupRequest gains demand(), returning a new MoqGroupDemand that mirrors moq_net::group::Request::demand: a weak handle a group server can watch to stop producing once every fetch caller has left. Wrapped in Python, Go, and Swift, and aliased in Kotlin, Dart, and C++. Co-Authored-By: Claude Opus 5.5 --- cpp/moq/include/moq/moq.hpp | 1 + dart/moq/lib/src/aliases.dart | 3 + dart/moq_ffi/lib/src/moq.dart | 194 ++++++++++++++++++ go/wrapper/moq_test.go | 7 + go/wrapper/publish.go | 38 ++++ .../kotlin/dev/moq/Aliases.kt | 2 + py/moq-rs/docs/index.md | 1 + py/moq-rs/moq/__init__.py | 2 + py/moq-rs/moq/publish.py | 35 ++++ py/moq-rs/tests/test_local.py | 3 + quest/m1/ffi-shape/README.md | 1 - quest/m1/ffi-shape/group-request-demand.md | 14 -- rs/moq-ffi/src/demand.rs | 49 ++++- rs/moq-ffi/src/producer.rs | 8 +- rs/moq-ffi/src/test.rs | 27 +++ swift/Sources/Moq/Dynamic.swift | 38 ++++ 16 files changed, 403 insertions(+), 20 deletions(-) delete mode 100644 quest/m1/ffi-shape/group-request-demand.md 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..7f0f57f94a7 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,35 @@ 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 return ErrClosed once the request is released. +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..f655c55d58a 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,36 @@ 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 + ``moq.Error.Closed`` once the request is released. + """ + + 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 +241,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 e226be5520b..798c0c3f18f 100644 --- a/quest/m1/ffi-shape/README.md +++ b/quest/m1/ffi-shape/README.md @@ -75,5 +75,4 @@ 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 - [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 f8697014869..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 `MoqTrackRequest`'s existing `demand()`, which returns a -`MoqTrackDemand`. Additive in every binding. Wire: none. diff --git a/rs/moq-ffi/src/demand.rs b/rs/moq-ffi/src/demand.rs index f0aaa18dc9a..d26a496ef29 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,51 @@ 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 with `Closed` once the request is released. +#[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 3191e8ebae1..776aaccb13b 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -1365,6 +1365,33 @@ 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))); +} + #[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..31101e8764c 100644 --- a/swift/Sources/Moq/Dynamic.swift +++ b/swift/Sources/Moq/Dynamic.swift @@ -32,6 +32,39 @@ 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 `MoqError.Closed` once the request is released. +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 +83,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()) From 4dae919d73dd01a1787664c57232b9894d61f327 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 06:02:37 -0700 Subject: [PATCH 3/6] feat(moq-c): a group request reports its demand Adds moq_group_request_demand, a demand watcher over a group request's waiting fetches, sharing the track demand watcher via a small trait. Co-Authored-By: Claude Opus 5.5 --- rs/moq-c/README.md | 1 + rs/moq-c/src/api.rs | 31 ++++++++++++++++++++++- rs/moq-c/src/publish.rs | 44 ++++++++++++++++++++++++++++----- rs/moq-c/src/test.rs | 54 +++++++++++++++++++++++++++++++++++++++++ 4 files changed, 123 insertions(+), 7 deletions(-) diff --git a/rs/moq-c/README.md b/rs/moq-c/README.md index b4eabd8fff9..df90debee24 100644 --- a/rs/moq-c/README.md +++ b/rs/moq-c/README.md @@ -105,6 +105,7 @@ int32_t moq_track_request_free(uint32_t request); int32_t moq_group_request_sequence(uint32_t request, uint64_t *dst); int32_t moq_group_request_priority(uint32_t request, uint8_t *dst); int32_t moq_group_request_frame_start(uint32_t request, uint64_t *dst); +int32_t moq_group_request_demand(uint32_t request, moq_status_callback on_demand, void *user_data); int32_t moq_group_request_accept(uint32_t request); int32_t moq_group_request_abort(uint32_t request, uint16_t error_code); int32_t moq_group_request_free(uint32_t request); diff --git a/rs/moq-c/src/api.rs b/rs/moq-c/src/api.rs index 87fb0872011..35199ea5f64 100644 --- a/rs/moq-c/src/api.rs +++ b/rs/moq-c/src/api.rs @@ -2142,7 +2142,8 @@ pub unsafe extern "C" fn moq_publish_media_demand( } /// Stop a demand watcher from [moq_publish_track_demand], [moq_publish_media_demand], -/// [`crate::moq_encode_video_demand`], or [`crate::moq_encode_audio_demand`]. +/// [moq_group_request_demand], [`crate::moq_encode_video_demand`], or +/// [`crate::moq_encode_audio_demand`]. /// /// Returns immediately: zero on success, or a negative code if already closed. The /// watcher's `on_demand` callback still fires once more with a terminal `0`, and @@ -2890,6 +2891,34 @@ pub unsafe extern "C" fn moq_group_request_frame_start(request: u32, dst: *mut u }) } +/// Watch whether any fetch still waits on a group request. See [moq_publish_media_demand] +/// for the callback contract. +/// +/// The last fetch to leave withdraws the request and a later fetch of the group starts a +/// fresh one, so once `MOQ_DEMAND_UNUSED` fires, demand never returns: free the request. +/// The watcher does not keep the request alive. It ends with `0` once the request is +/// freed, or negative once it is accepted or rejected. +/// +/// Returns a non-zero watcher handle on success, or a negative code on failure. +/// +/// # Safety +/// - `on_demand` must be non-NULL. +/// - The caller must keep `user_data` valid until the terminal (`<= 0`) `on_demand` callback. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn moq_group_request_demand( + request: u32, + on_demand: ffi::moq_status_callback, + user_data: *mut c_void, +) -> i32 { + ffi::enter(move || { + let request = ffi::parse_id(request)?; + let on_demand = unsafe { ffi::OnStatus::new(user_data, on_demand)? }; + let mut state = State::lock(); + let demand = state.publish.group_request_demand(request)?; + state.publish.demand(demand, on_demand) + }) +} + /// Accept a group request, resolving the waiting fetches with the group you then fill. /// /// Consumes the request handle. The returned producer starts at diff --git a/rs/moq-c/src/publish.rs b/rs/moq-c/src/publish.rs index 9c8b5832059..f8f3f083700 100644 --- a/rs/moq-c/src/publish.rs +++ b/rs/moq-c/src/publish.rs @@ -1,4 +1,5 @@ use std::collections::BTreeMap; +use std::future::Future; use moq_mux::catalog::hang::Extra; use moq_mux::import; @@ -17,6 +18,32 @@ struct TaskEntry { callback: OnStatus, } +/// A watch-only demand handle: a track's subscribers or a group request's waiting fetches. +pub(crate) trait Demand: Send + Sync + 'static { + fn used(&self) -> impl Future> + Send; + fn unused(&self) -> impl Future> + Send; +} + +impl Demand for moq_net::track::Demand { + fn used(&self) -> impl Future> + Send { + moq_net::track::Demand::used(self) + } + + fn unused(&self) -> impl Future> + Send { + moq_net::track::Demand::unused(self) + } +} + +impl Demand for moq_net::group::Demand { + fn used(&self) -> impl Future> + Send { + moq_net::group::Demand::used(self) + } + + fn unused(&self) -> impl Future> + Send { + moq_net::group::Demand::unused(self) + } +} + /// A subscriber's request for a track the broadcast has not declared, kept with the /// broadcast it was made on so media can be published onto it under that broadcast's catalog. struct TrackRequest { @@ -421,13 +448,13 @@ impl Publish { Ok(self.media.get(media).ok_or(Error::MediaNotFound)?.demand()) } - /// Watch a track's subscriber demand, reporting the current state and every change. + /// Watch a track's or group request's demand, reporting the current state and every change. /// /// `on_demand` fires with a [`moq_demand`] value immediately and again on each change, then - /// once with a terminal code: `0` when the track ends or the watcher is closed, negative when - /// the track aborts. Seeding with the current state is what makes a late registration safe: a - /// track that went unused before the watcher existed still reports it. - pub fn demand(&mut self, demand: moq_net::track::Demand, on_demand: OnStatus) -> Result { + /// once with a terminal code: `0` when the track or request ends or the watcher is closed, + /// negative when it aborts. Seeding with the current state is what makes a late registration + /// safe: a track that went unused before the watcher existed still reports it. + pub fn demand(&mut self, demand: impl Demand, on_demand: OnStatus) -> Result { let channel = oneshot::channel(); let id = self.demand.insert(Some(TaskEntry { close: Some(channel.0), @@ -450,7 +477,7 @@ impl Publish { pub(crate) async fn run_demand( callback: OnStatus, - demand: moq_net::track::Demand, + demand: impl Demand, mut close: oneshot::Receiver<()>, ) -> Result<(), Error> { // Neither handle exposes the current state, only the level-triggered waits, and exactly @@ -664,6 +691,11 @@ impl Publish { Ok((request.sequence(), request.priority(), request.frame_start())) } + /// A watch-only handle to the fetches still waiting on a group request. + pub fn group_request_demand(&self, request: Id) -> Result { + Ok(self.group_request.get(request).ok_or(Error::NotFound)?.demand()) + } + /// Accept a group request, returning a group handle like [`Self::track_group`]. /// /// The producer is positioned at the request's `frame_start` so frames keep the diff --git a/rs/moq-c/src/test.rs b/rs/moq-c/src/test.rs index b9068fce036..3f2b70620a1 100644 --- a/rs/moq-c/src/test.rs +++ b/rs/moq-c/src/test.rs @@ -2752,6 +2752,60 @@ fn track_dynamic_serves_a_fetch_miss() { assert_eq!(moq_origin_close(origin), 0); } +#[test] +fn group_request_demand_follows_the_fetch() { + let origin = id(moq_origin_create()); + let broadcast = publish_broadcast(origin, b"group-request-demand"); + let name = b"data"; + let track = + id(unsafe { moq_publish_track(broadcast, name.as_ptr() as *const c_char, name.len(), std::ptr::null()) }); + let group_cb = Callback::new(); + let dynamic = id(unsafe { moq_publish_track_dynamic(track, Some(channel_callback), group_cb.ptr) }); + + // An abandoned fetch withdraws its demand, and freeing the request ends the watcher cleanly. + let consumer = State::lock() + .publish + .producer(Id::try_from(broadcast).unwrap()) + .unwrap() + .consume() + .track("data") + .unwrap(); + let fetch = crate::ffi::RUNTIME.spawn(async move { consumer.fetch_group(5, None).await }); + let request = id(group_cb.recv()); + let demand_cb = Callback::new(); + let _watcher = id(unsafe { moq_group_request_demand(request, Some(channel_callback), demand_cb.ptr) }); + assert_eq!(demand_cb.recv(), moq_demand::MOQ_DEMAND_USED as i32); + fetch.abort(); + assert_eq!(demand_cb.recv(), moq_demand::MOQ_DEMAND_UNUSED as i32); + assert_eq!(moq_group_request_free(request), 0); + assert_eq!(demand_cb.recv_terminal(), 0); + + // Rejecting the request ends the watcher with an error. The failed fetch may leave first. + let fetch = fetch_group(broadcast, "data", 6, 0, 0); + let request = id(group_cb.recv()); + let demand_cb = Callback::new(); + let _watcher = id(unsafe { moq_group_request_demand(request, Some(channel_callback), demand_cb.ptr) }); + assert_eq!(demand_cb.recv(), moq_demand::MOQ_DEMAND_USED as i32); + assert_eq!(moq_group_request_abort(request, 9), 0); + let mut code = demand_cb.recv(); + if code == moq_demand::MOQ_DEMAND_UNUSED as i32 { + code = demand_cb.recv(); + } + assert!(code < 0, "a rejected request is a negative terminal, got {code}"); + assert!(fetch.recv_timeout(TIMEOUT).unwrap().is_err()); + + assert!( + unsafe { moq_group_request_demand(request, Some(channel_callback), demand_cb.ptr) } < 0, + "an answered request has no demand" + ); + + assert_eq!(moq_publish_dynamic_cancel(dynamic), 0); + assert_eq!(group_cb.recv_terminal(), 0); + assert_eq!(moq_publish_track_finish(track), 0); + assert_eq!(moq_publish_close(broadcast), 0); + assert_eq!(moq_origin_close(origin), 0); +} + #[test] fn track_dynamic_serves_a_fetch_from_frame_start() { let origin = id(moq_origin_create()); From 3587ae3e5631b7916632c940ecedc7a6c1b2d9ee Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 06:08:10 -0700 Subject: [PATCH 4/6] docs(ffi): a group demand fails with the answer's error Co-Authored-By: Claude Opus 5.5 --- go/wrapper/publish.go | 4 +++- py/moq-rs/moq/publish.py | 5 +++-- rs/moq-ffi/src/demand.rs | 3 ++- swift/Sources/Moq/Dynamic.swift | 3 ++- 4 files changed, 10 insertions(+), 5 deletions(-) diff --git a/go/wrapper/publish.go b/go/wrapper/publish.go index 7f0f57f94a7..e7e043a3fa0 100644 --- a/go/wrapper/publish.go +++ b/go/wrapper/publish.go @@ -400,7 +400,9 @@ func (r *GroupRequest) Abort(errorCode uint16) error { // // 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 return ErrClosed once the request is released. +// 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 } diff --git a/py/moq-rs/moq/publish.py b/py/moq-rs/moq/publish.py index f655c55d58a..84606b1071b 100644 --- a/py/moq-rs/moq/publish.py +++ b/py/moq-rs/moq/publish.py @@ -200,8 +200,9 @@ class GroupDemand: 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 - ``moq.Error.Closed`` once the request is released. + 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: diff --git a/rs/moq-ffi/src/demand.rs b/rs/moq-ffi/src/demand.rs index d26a496ef29..72223a44532 100644 --- a/rs/moq-ffi/src/demand.rs +++ b/rs/moq-ffi/src/demand.rs @@ -49,7 +49,8 @@ impl MoqTrackDemand { /// /// 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 with `Closed` once the request is released. +/// 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, diff --git a/swift/Sources/Moq/Dynamic.swift b/swift/Sources/Moq/Dynamic.swift index 31101e8764c..9ab9445c60f 100644 --- a/swift/Sources/Moq/Dynamic.swift +++ b/swift/Sources/Moq/Dynamic.swift @@ -36,7 +36,8 @@ public final class TrackRequest: Sendable { /// /// 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 `MoqError.Closed` once the request is released. +/// 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 From 3df7c3e6a5cda448516b5749843784ff119701f0 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 06:48:34 -0700 Subject: [PATCH 5/6] test(ffi): pin how an accepted group request's demand ends Co-Authored-By: Claude Opus 5.5 --- rs/moq-c/src/test.rs | 21 +++++++++++++++++++++ rs/moq-ffi/src/test.rs | 29 +++++++++++++++++++++++++++++ 2 files changed, 50 insertions(+) diff --git a/rs/moq-c/src/test.rs b/rs/moq-c/src/test.rs index 3f2b70620a1..11e78e3bf13 100644 --- a/rs/moq-c/src/test.rs +++ b/rs/moq-c/src/test.rs @@ -2799,6 +2799,27 @@ fn group_request_demand_follows_the_fetch() { "an answered request has no demand" ); + // Accepting the request also ends the watcher with an error: the accept fails any joined + // fetch it can't cover, and the watcher reads that as the request's end. + let fetch = fetch_group(broadcast, "data", 7, 0, 0); + let request = id(group_cb.recv()); + let demand_cb = Callback::new(); + let _watcher = id(unsafe { moq_group_request_demand(request, Some(channel_callback), demand_cb.ptr) }); + assert_eq!(demand_cb.recv(), moq_demand::MOQ_DEMAND_USED as i32); + let group = id(moq_group_request_accept(request)); + let mut code = demand_cb.recv(); + if code == moq_demand::MOQ_DEMAND_UNUSED as i32 { + code = demand_cb.recv(); + } + assert!(code < 0, "an accepted request is a negative terminal, got {code}"); + let payload = b"fetched"; + assert_eq!( + unsafe { moq_publish_group_frame(group, payload.as_ptr(), payload.len(), 100_000) }, + 0 + ); + assert_eq!(moq_publish_group_finish(group), 0); + assert_eq!(fetch.recv_timeout(TIMEOUT).unwrap().unwrap(), (0, payload.to_vec())); + assert_eq!(moq_publish_dynamic_cancel(dynamic), 0); assert_eq!(group_cb.recv_terminal(), 0); assert_eq!(moq_publish_track_finish(track), 0); diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index 9f8a7bc99f7..84b10be010d 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -1392,6 +1392,35 @@ async fn group_request_demand_ends_when_the_fetcher_leaves() { 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(); From 68571747751665dcdd01c4b60e963372adf67e9c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 07:12:00 -0700 Subject: [PATCH 6/6] revert(moq-c): keep the hand-written C frozen The hand-written moq-c gets no feature work (quest/m1/c/README.md, #5186); generated C inherits group request demand from moq-ffi. Co-Authored-By: Claude Opus 5.5 --- rs/moq-c/README.md | 1 - rs/moq-c/src/api.rs | 31 +---------------- rs/moq-c/src/publish.rs | 44 ++++-------------------- rs/moq-c/src/test.rs | 75 ----------------------------------------- 4 files changed, 7 insertions(+), 144 deletions(-) diff --git a/rs/moq-c/README.md b/rs/moq-c/README.md index df90debee24..b4eabd8fff9 100644 --- a/rs/moq-c/README.md +++ b/rs/moq-c/README.md @@ -105,7 +105,6 @@ int32_t moq_track_request_free(uint32_t request); int32_t moq_group_request_sequence(uint32_t request, uint64_t *dst); int32_t moq_group_request_priority(uint32_t request, uint8_t *dst); int32_t moq_group_request_frame_start(uint32_t request, uint64_t *dst); -int32_t moq_group_request_demand(uint32_t request, moq_status_callback on_demand, void *user_data); int32_t moq_group_request_accept(uint32_t request); int32_t moq_group_request_abort(uint32_t request, uint16_t error_code); int32_t moq_group_request_free(uint32_t request); diff --git a/rs/moq-c/src/api.rs b/rs/moq-c/src/api.rs index 35199ea5f64..87fb0872011 100644 --- a/rs/moq-c/src/api.rs +++ b/rs/moq-c/src/api.rs @@ -2142,8 +2142,7 @@ pub unsafe extern "C" fn moq_publish_media_demand( } /// Stop a demand watcher from [moq_publish_track_demand], [moq_publish_media_demand], -/// [moq_group_request_demand], [`crate::moq_encode_video_demand`], or -/// [`crate::moq_encode_audio_demand`]. +/// [`crate::moq_encode_video_demand`], or [`crate::moq_encode_audio_demand`]. /// /// Returns immediately: zero on success, or a negative code if already closed. The /// watcher's `on_demand` callback still fires once more with a terminal `0`, and @@ -2891,34 +2890,6 @@ pub unsafe extern "C" fn moq_group_request_frame_start(request: u32, dst: *mut u }) } -/// Watch whether any fetch still waits on a group request. See [moq_publish_media_demand] -/// for the callback contract. -/// -/// The last fetch to leave withdraws the request and a later fetch of the group starts a -/// fresh one, so once `MOQ_DEMAND_UNUSED` fires, demand never returns: free the request. -/// The watcher does not keep the request alive. It ends with `0` once the request is -/// freed, or negative once it is accepted or rejected. -/// -/// Returns a non-zero watcher handle on success, or a negative code on failure. -/// -/// # Safety -/// - `on_demand` must be non-NULL. -/// - The caller must keep `user_data` valid until the terminal (`<= 0`) `on_demand` callback. -#[unsafe(no_mangle)] -pub unsafe extern "C" fn moq_group_request_demand( - request: u32, - on_demand: ffi::moq_status_callback, - user_data: *mut c_void, -) -> i32 { - ffi::enter(move || { - let request = ffi::parse_id(request)?; - let on_demand = unsafe { ffi::OnStatus::new(user_data, on_demand)? }; - let mut state = State::lock(); - let demand = state.publish.group_request_demand(request)?; - state.publish.demand(demand, on_demand) - }) -} - /// Accept a group request, resolving the waiting fetches with the group you then fill. /// /// Consumes the request handle. The returned producer starts at diff --git a/rs/moq-c/src/publish.rs b/rs/moq-c/src/publish.rs index f8f3f083700..9c8b5832059 100644 --- a/rs/moq-c/src/publish.rs +++ b/rs/moq-c/src/publish.rs @@ -1,5 +1,4 @@ use std::collections::BTreeMap; -use std::future::Future; use moq_mux::catalog::hang::Extra; use moq_mux::import; @@ -18,32 +17,6 @@ struct TaskEntry { callback: OnStatus, } -/// A watch-only demand handle: a track's subscribers or a group request's waiting fetches. -pub(crate) trait Demand: Send + Sync + 'static { - fn used(&self) -> impl Future> + Send; - fn unused(&self) -> impl Future> + Send; -} - -impl Demand for moq_net::track::Demand { - fn used(&self) -> impl Future> + Send { - moq_net::track::Demand::used(self) - } - - fn unused(&self) -> impl Future> + Send { - moq_net::track::Demand::unused(self) - } -} - -impl Demand for moq_net::group::Demand { - fn used(&self) -> impl Future> + Send { - moq_net::group::Demand::used(self) - } - - fn unused(&self) -> impl Future> + Send { - moq_net::group::Demand::unused(self) - } -} - /// A subscriber's request for a track the broadcast has not declared, kept with the /// broadcast it was made on so media can be published onto it under that broadcast's catalog. struct TrackRequest { @@ -448,13 +421,13 @@ impl Publish { Ok(self.media.get(media).ok_or(Error::MediaNotFound)?.demand()) } - /// Watch a track's or group request's demand, reporting the current state and every change. + /// Watch a track's subscriber demand, reporting the current state and every change. /// /// `on_demand` fires with a [`moq_demand`] value immediately and again on each change, then - /// once with a terminal code: `0` when the track or request ends or the watcher is closed, - /// negative when it aborts. Seeding with the current state is what makes a late registration - /// safe: a track that went unused before the watcher existed still reports it. - pub fn demand(&mut self, demand: impl Demand, on_demand: OnStatus) -> Result { + /// once with a terminal code: `0` when the track ends or the watcher is closed, negative when + /// the track aborts. Seeding with the current state is what makes a late registration safe: a + /// track that went unused before the watcher existed still reports it. + pub fn demand(&mut self, demand: moq_net::track::Demand, on_demand: OnStatus) -> Result { let channel = oneshot::channel(); let id = self.demand.insert(Some(TaskEntry { close: Some(channel.0), @@ -477,7 +450,7 @@ impl Publish { pub(crate) async fn run_demand( callback: OnStatus, - demand: impl Demand, + demand: moq_net::track::Demand, mut close: oneshot::Receiver<()>, ) -> Result<(), Error> { // Neither handle exposes the current state, only the level-triggered waits, and exactly @@ -691,11 +664,6 @@ impl Publish { Ok((request.sequence(), request.priority(), request.frame_start())) } - /// A watch-only handle to the fetches still waiting on a group request. - pub fn group_request_demand(&self, request: Id) -> Result { - Ok(self.group_request.get(request).ok_or(Error::NotFound)?.demand()) - } - /// Accept a group request, returning a group handle like [`Self::track_group`]. /// /// The producer is positioned at the request's `frame_start` so frames keep the diff --git a/rs/moq-c/src/test.rs b/rs/moq-c/src/test.rs index 11e78e3bf13..b9068fce036 100644 --- a/rs/moq-c/src/test.rs +++ b/rs/moq-c/src/test.rs @@ -2752,81 +2752,6 @@ fn track_dynamic_serves_a_fetch_miss() { assert_eq!(moq_origin_close(origin), 0); } -#[test] -fn group_request_demand_follows_the_fetch() { - let origin = id(moq_origin_create()); - let broadcast = publish_broadcast(origin, b"group-request-demand"); - let name = b"data"; - let track = - id(unsafe { moq_publish_track(broadcast, name.as_ptr() as *const c_char, name.len(), std::ptr::null()) }); - let group_cb = Callback::new(); - let dynamic = id(unsafe { moq_publish_track_dynamic(track, Some(channel_callback), group_cb.ptr) }); - - // An abandoned fetch withdraws its demand, and freeing the request ends the watcher cleanly. - let consumer = State::lock() - .publish - .producer(Id::try_from(broadcast).unwrap()) - .unwrap() - .consume() - .track("data") - .unwrap(); - let fetch = crate::ffi::RUNTIME.spawn(async move { consumer.fetch_group(5, None).await }); - let request = id(group_cb.recv()); - let demand_cb = Callback::new(); - let _watcher = id(unsafe { moq_group_request_demand(request, Some(channel_callback), demand_cb.ptr) }); - assert_eq!(demand_cb.recv(), moq_demand::MOQ_DEMAND_USED as i32); - fetch.abort(); - assert_eq!(demand_cb.recv(), moq_demand::MOQ_DEMAND_UNUSED as i32); - assert_eq!(moq_group_request_free(request), 0); - assert_eq!(demand_cb.recv_terminal(), 0); - - // Rejecting the request ends the watcher with an error. The failed fetch may leave first. - let fetch = fetch_group(broadcast, "data", 6, 0, 0); - let request = id(group_cb.recv()); - let demand_cb = Callback::new(); - let _watcher = id(unsafe { moq_group_request_demand(request, Some(channel_callback), demand_cb.ptr) }); - assert_eq!(demand_cb.recv(), moq_demand::MOQ_DEMAND_USED as i32); - assert_eq!(moq_group_request_abort(request, 9), 0); - let mut code = demand_cb.recv(); - if code == moq_demand::MOQ_DEMAND_UNUSED as i32 { - code = demand_cb.recv(); - } - assert!(code < 0, "a rejected request is a negative terminal, got {code}"); - assert!(fetch.recv_timeout(TIMEOUT).unwrap().is_err()); - - assert!( - unsafe { moq_group_request_demand(request, Some(channel_callback), demand_cb.ptr) } < 0, - "an answered request has no demand" - ); - - // Accepting the request also ends the watcher with an error: the accept fails any joined - // fetch it can't cover, and the watcher reads that as the request's end. - let fetch = fetch_group(broadcast, "data", 7, 0, 0); - let request = id(group_cb.recv()); - let demand_cb = Callback::new(); - let _watcher = id(unsafe { moq_group_request_demand(request, Some(channel_callback), demand_cb.ptr) }); - assert_eq!(demand_cb.recv(), moq_demand::MOQ_DEMAND_USED as i32); - let group = id(moq_group_request_accept(request)); - let mut code = demand_cb.recv(); - if code == moq_demand::MOQ_DEMAND_UNUSED as i32 { - code = demand_cb.recv(); - } - assert!(code < 0, "an accepted request is a negative terminal, got {code}"); - let payload = b"fetched"; - assert_eq!( - unsafe { moq_publish_group_frame(group, payload.as_ptr(), payload.len(), 100_000) }, - 0 - ); - assert_eq!(moq_publish_group_finish(group), 0); - assert_eq!(fetch.recv_timeout(TIMEOUT).unwrap().unwrap(), (0, payload.to_vec())); - - assert_eq!(moq_publish_dynamic_cancel(dynamic), 0); - assert_eq!(group_cb.recv_terminal(), 0); - assert_eq!(moq_publish_track_finish(track), 0); - assert_eq!(moq_publish_close(broadcast), 0); - assert_eq!(moq_origin_close(origin), 0); -} - #[test] fn track_dynamic_serves_a_fetch_from_frame_start() { let origin = id(moq_origin_create());