feat(ffi): a group request reports its demand

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;

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;
 

dart/moq_ffi/lib/src/moq.dart

@@ -5460,6 +5460,108 @@ class FfiConverterMoqTrackConsumer {
   }
 }
 
+abstract class MoqGroupDemandInterface {
+  bool isUsed();
+  int sequence();
+  Future<void> unused();
+  Future<void> used();
+}
+
+final _MoqGroupDemandFinalizer = Finalizer<Pointer<Void>>((ptr) {
+  rustCall((status) => uniffi_moq_ffi_fn_free_moqgroupdemand(ptr, status));
+});
+
+class MoqGroupDemand implements MoqGroupDemandInterface {
+  late final Pointer<Void> _ptr;
+  MoqGroupDemand._(this._ptr) {
+    _MoqGroupDemandFinalizer.attach(this, _ptr, detach: this);
+  }
+  factory MoqGroupDemand.lift(Pointer<Void> ptr) {
+    return MoqGroupDemand._(ptr);
+  }
+  Pointer<Void> 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<void> 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<void> 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<Void> ptr) {
+    return MoqGroupDemand.lift(ptr);
+  }
+
+  static Pointer<Void> lower(MoqGroupDemand value) {
+    return value.uniffiClonePointer();
+  }
+
+  static int allocationSize(MoqGroupDemand value) {
+    return 8;
+  }
+
+  static LiftRetVal<MoqGroupDemand> read(Uint8List buf) {
+    final handle = buf.buffer.asByteData(buf.offsetInBytes).getInt64(0);
+    final pointer = Pointer<Void>.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<RustCallStatus> uniffiStatus,
 );
 
+@Native<Pointer<Void> Function(Pointer<Void>, Pointer<RustCallStatus>)>(
+  assetId: _uniffiAssetId,
+)
+external Pointer<Void> uniffi_moq_ffi_fn_clone_moqgroupdemand(
+  Pointer<Void> handle,
+  Pointer<RustCallStatus> uniffiStatus,
+);
+
+@Native<Void Function(Pointer<Void>, Pointer<RustCallStatus>)>(
+  assetId: _uniffiAssetId,
+)
+external void uniffi_moq_ffi_fn_free_moqgroupdemand(
+  Pointer<Void> handle,
+  Pointer<RustCallStatus> uniffiStatus,
+);
+
+@Native<Int8 Function(Pointer<Void>, Pointer<RustCallStatus>)>(
+  assetId: _uniffiAssetId,
+)
+external int uniffi_moq_ffi_fn_method_moqgroupdemand_is_used(
+  Pointer<Void> ptr,
+  Pointer<RustCallStatus> uniffiStatus,
+);
+
+@Native<Uint64 Function(Pointer<Void>, Pointer<RustCallStatus>)>(
+  assetId: _uniffiAssetId,
+)
+external int uniffi_moq_ffi_fn_method_moqgroupdemand_sequence(
+  Pointer<Void> ptr,
+  Pointer<RustCallStatus> uniffiStatus,
+);
+
+@Native<Pointer<Void> Function(Pointer<Void>)>(assetId: _uniffiAssetId)
+external Pointer<Void> uniffi_moq_ffi_fn_method_moqgroupdemand_unused(
+  Pointer<Void> ptr,
+);
+
+@Native<Pointer<Void> Function(Pointer<Void>)>(assetId: _uniffiAssetId)
+external Pointer<Void> uniffi_moq_ffi_fn_method_moqgroupdemand_used(
+  Pointer<Void> ptr,
+);
+
 @Native<Pointer<Void> Function(Pointer<Void>, Pointer<RustCallStatus>)>(
   assetId: _uniffiAssetId,
 )
@@ -11624,6 +11780,14 @@ external Pointer<Void> uniffi_moq_ffi_fn_method_moqgrouprequest_accept(
   Pointer<RustCallStatus> uniffiStatus,
 );
 
+@Native<Pointer<Void> Function(Pointer<Void>, Pointer<RustCallStatus>)>(
+  assetId: _uniffiAssetId,
+)
+external Pointer<Void> uniffi_moq_ffi_fn_method_moqgrouprequest_demand(
+  Pointer<Void> ptr,
+  Pointer<RustCallStatus> uniffiStatus,
+);
+
 @Native<Uint8 Function(Pointer<Void>, Pointer<RustCallStatus>)>(
   assetId: _uniffiAssetId,
 )
@@ -12454,6 +12618,18 @@ external int uniffi_moq_ffi_checksum_method_moqtrackconsumer_recv_group();
 @Native<Uint16 Function()>(assetId: _uniffiAssetId)
 external int uniffi_moq_ffi_checksum_method_moqtrackconsumer_update();
 
+@Native<Uint16 Function()>(assetId: _uniffiAssetId)
+external int uniffi_moq_ffi_checksum_method_moqgroupdemand_is_used();
+
+@Native<Uint16 Function()>(assetId: _uniffiAssetId)
+external int uniffi_moq_ffi_checksum_method_moqgroupdemand_sequence();
+
+@Native<Uint16 Function()>(assetId: _uniffiAssetId)
+external int uniffi_moq_ffi_checksum_method_moqgroupdemand_unused();
+
+@Native<Uint16 Function()>(assetId: _uniffiAssetId)
+external int uniffi_moq_ffi_checksum_method_moqgroupdemand_used();
+
 @Native<Uint16 Function()>(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<Uint16 Function()>(assetId: _uniffiAssetId)
 external int uniffi_moq_ffi_checksum_method_moqgrouprequest_accept();
 
+@Native<Uint16 Function()>(assetId: _uniffiAssetId)
+external int uniffi_moq_ffi_checksum_method_moqgrouprequest_demand();
+
 @Native<Uint16 Function()>(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");
   }

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)

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

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. */

py/moq-rs/docs/index.md

@@ -60,6 +60,7 @@ asyncio.run(main())
    TrackRequest
    GroupProducer
    GroupRequest
+   GroupDemand
    AudioProducer
    VideoProducer
 ```

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",