diff --git a/cpp/moq/include/moq/moq.hpp b/cpp/moq/include/moq/moq.hpp index df911cce4f5..ae4236d7187 100644 --- a/cpp/moq/include/moq/moq.hpp +++ b/cpp/moq/include/moq/moq.hpp @@ -70,8 +70,8 @@ using JsonStreamConsumer = MoqJsonStreamConsumer; using JsonStreamProducer = MoqJsonStreamProducer; using MediaCatalogConsumer = MoqMediaCatalogConsumer; using MediaCatalogProducer = MoqMediaCatalogProducer; -using MediaContainerConsumer = MoqMediaContainerConsumer; -using MediaContainerGroupConsumer = MoqMediaContainerGroupConsumer; +using MediaTrackConsumer = MoqMediaTrackConsumer; +using MediaGroupConsumer = MoqMediaGroupConsumer; using MediaContainerProducer = MoqMediaContainerProducer; using MediaContainerStreamProducer = MoqMediaContainerStreamProducer; using MediaTrackProducer = MoqMediaTrackProducer; @@ -111,8 +111,8 @@ using FlateConfig = MoqFlateConfig; using Frame = MoqFrame; using JsonSnapshotConfig = MoqJsonSnapshotConfig; using JsonStreamConfig = MoqJsonStreamConfig; -using MediaContainerConfig = MoqMediaContainerConfig; -using MediaContainerGroupConfig = MoqMediaContainerGroupConfig; +using MediaTrackConfig = MoqMediaTrackConfig; +using MediaGroupConfig = MoqMediaGroupConfig; using MediaFrame = MoqMediaFrame; using OriginConfig = MoqOriginConfig; using ProtocolError = MoqProtocolError; diff --git a/cpp/moq/test/doc-samples.hpp b/cpp/moq/test/doc-samples.hpp index 0dea9f1f5c2..c2efa484856 100644 --- a/cpp/moq/test/doc-samples.hpp +++ b/cpp/moq/test/doc-samples.hpp @@ -28,7 +28,7 @@ inline void ok(moq::expected result) { } inline std::shared_ptr session; -inline std::shared_ptr media; +inline std::shared_ptr media; inline std::vector opus_init; inline std::vector packet; inline std::vector rgba; diff --git a/cpp/obs/src/moq-source.cpp b/cpp/obs/src/moq-source.cpp index 7042d97dc93..06255fc482c 100644 --- a/cpp/obs/src/moq-source.cpp +++ b/cpp/obs/src/moq-source.cpp @@ -209,7 +209,7 @@ struct Track { std::unique_ptr audio_decoder; moq::Container container; - std::shared_ptr consumer; + std::shared_ptr consumer; std::optional call; }; @@ -584,7 +584,7 @@ static std::string moq_source_track_name(const std::pair &conn, const std::shared_ptr &track, - moq::expected> result); + moq::expected> result); static void moq_source_read_track(struct moq_source *ctx, const std::shared_ptr &conn, const std::shared_ptr &track); static void moq_source_on_frame(struct moq_source *ctx, const std::shared_ptr &conn, @@ -636,16 +636,15 @@ static void moq_source_on_resolved(struct moq_source *ctx, const std::shared_ptr moq_source_discard_candidate(conn, track); return; } - track->call = - moq::MediaContainerConsumer::subscribe(*result, {track->name, track->container}) - .then(ctx->worker.Executor(), - moq_source_track_current>>( - ctx, conn, track, moq_source_on_subscribed)); + track->call = moq::MediaTrackConsumer::subscribe(*result, {track->name, track->container}) + .then(ctx->worker.Executor(), + moq_source_track_current>>( + ctx, conn, track, moq_source_on_subscribed)); } static void moq_source_on_subscribed(struct moq_source *ctx, const std::shared_ptr &conn, const std::shared_ptr &track, - moq::expected> result) + moq::expected> result) { if (!result) { LOG_ERROR("Failed to subscribe to %s track: %s", track->kind, result.error().to_string().c_str()); diff --git a/cpp/obs/test/moq-output-test.cpp b/cpp/obs/test/moq-output-test.cpp index fa3b53bd6d8..02edc9773c3 100644 --- a/cpp/obs/test/moq-output-test.cpp +++ b/cpp/obs/test/moq-output-test.cpp @@ -432,9 +432,9 @@ int main() CHECK(catalog.has_value()); if (catalog) { const auto &[name, rendition] = *catalog->video.begin(); - auto media = TestOk( - moq::MediaContainerConsumer::subscribe(broadcast, {name, rendition.container}).get(), - "container consumer"); + auto media = + TestOk(moq::MediaTrackConsumer::subscribe(broadcast, {name, rendition.container}).get(), + "container consumer"); // Read a frame first, so the relay is subscribed before the last one is written. auto next = media->next(); CHECK(next.wait_for(std::chrono::seconds(10)) == std::future_status::ready); diff --git a/dart/moq/lib/media.dart b/dart/moq/lib/media.dart index 33ecdb3953e..cdacce18930 100644 --- a/dart/moq/lib/media.dart +++ b/dart/moq/lib/media.dart @@ -16,10 +16,10 @@ typedef ContainerProducer = MoqMediaContainerProducer; typedef ContainerStreamProducer = MoqMediaContainerStreamProducer; /// The read side of a media track: yields frames with codec metadata in decode order. -typedef ContainerConsumer = MoqMediaContainerConsumer; +typedef TrackConsumer = MoqMediaTrackConsumer; /// A finite fetched media group: yields container-decoded frames until the group ends. -typedef ContainerGroupConsumer = MoqMediaContainerGroupConsumer; +typedef GroupConsumer = MoqMediaGroupConsumer; /// The read side of a broadcast's catalog: yields updates as the set of tracks changes. typedef CatalogConsumer = MoqMediaCatalogConsumer; @@ -76,10 +76,10 @@ typedef Named = NamedMoqMediaTarget; typedef Requested = RequestedMoqMediaTarget; /// A track name, container, and live delivery options. -typedef ContainerConfig = MoqMediaContainerConfig; +typedef TrackConfig = MoqMediaTrackConfig; /// A fetched group name, sequence, container, and delivery options. -typedef ContainerGroupConfig = MoqMediaContainerGroupConfig; +typedef GroupConfig = MoqMediaGroupConfig; /// The packaging advertised for a media track. typedef Container = MoqContainer; diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index ed98abf3a0a..86f382c8533 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/dart/moq_ffi/lib/src/moq.dart @@ -8,9 +8,7 @@ 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"; @@ -1031,91 +1029,91 @@ class FfiConverterMoqFrame { } } -class MoqMediaContainerConfig { - final String name; - final MoqContainer container; - final MoqSubscription? subscription; - MoqMediaContainerConfig({ - required this.name, - required this.container, - this.subscription = null, +class MoqMediaFrame { + final Uint8List payload; + final int timestampUs; + final bool keyframe; + MoqMediaFrame({ + required this.payload, + required this.timestampUs, + required this.keyframe, }); } -class FfiConverterMoqMediaContainerConfig { - static MoqMediaContainerConfig lift(RustBuffer buf) { - return FfiConverterMoqMediaContainerConfig.read(buf.asUint8List()).value; +class FfiConverterMoqMediaFrame { + static MoqMediaFrame lift(RustBuffer buf) { + return FfiConverterMoqMediaFrame.read(buf.asUint8List()).value; } - static LiftRetVal read(Uint8List buf) { + static LiftRetVal read(Uint8List buf) { int new_offset = buf.offsetInBytes; - final name_lifted = FfiConverterString.read( + final payload_lifted = FfiConverterUint8List.read( Uint8List.view(buf.buffer, new_offset), ); - final name = name_lifted.value; - new_offset += name_lifted.bytesRead; - final container_lifted = FfiConverterMoqContainer.read( + final payload = payload_lifted.value; + new_offset += payload_lifted.bytesRead; + final timestampUs_lifted = FfiConverterUInt64.read( Uint8List.view(buf.buffer, new_offset), ); - final container = container_lifted.value; - new_offset += container_lifted.bytesRead; - final subscription_lifted = FfiConverterOptionalMoqSubscription.read( + final timestampUs = timestampUs_lifted.value; + new_offset += timestampUs_lifted.bytesRead; + final keyframe_lifted = FfiConverterBool.read( Uint8List.view(buf.buffer, new_offset), ); - final subscription = subscription_lifted.value; - new_offset += subscription_lifted.bytesRead; + final keyframe = keyframe_lifted.value; + new_offset += keyframe_lifted.bytesRead; return LiftRetVal( - MoqMediaContainerConfig( - name: name, - container: container, - subscription: subscription, + MoqMediaFrame( + payload: payload, + timestampUs: timestampUs, + keyframe: keyframe, ), new_offset - buf.offsetInBytes, ); } - static RustBuffer lower(MoqMediaContainerConfig value) { + static RustBuffer lower(MoqMediaFrame value) { final total_length = - FfiConverterString.allocationSize(value.name) + - FfiConverterMoqContainer.allocationSize(value.container) + - FfiConverterOptionalMoqSubscription.allocationSize(value.subscription) + + FfiConverterUint8List.allocationSize(value.payload) + + FfiConverterUInt64.allocationSize(value.timestampUs) + + FfiConverterBool.allocationSize(value.keyframe) + 0; final buf = Uint8List(total_length); write(value, buf); return toRustBuffer(buf); } - static int write(MoqMediaContainerConfig value, Uint8List buf) { + static int write(MoqMediaFrame value, Uint8List buf) { int new_offset = buf.offsetInBytes; - new_offset += FfiConverterString.write( - value.name, + new_offset += FfiConverterUint8List.write( + value.payload, Uint8List.view(buf.buffer, new_offset), ); - new_offset += FfiConverterMoqContainer.write( - value.container, + new_offset += FfiConverterUInt64.write( + value.timestampUs, Uint8List.view(buf.buffer, new_offset), ); - new_offset += FfiConverterOptionalMoqSubscription.write( - value.subscription, + new_offset += FfiConverterBool.write( + value.keyframe, Uint8List.view(buf.buffer, new_offset), ); return new_offset - buf.offsetInBytes; } - static int allocationSize(MoqMediaContainerConfig value) { - return FfiConverterString.allocationSize(value.name) + - FfiConverterMoqContainer.allocationSize(value.container) + - FfiConverterOptionalMoqSubscription.allocationSize(value.subscription) + + static int allocationSize(MoqMediaFrame value) { + return FfiConverterUint8List.allocationSize(value.payload) + + FfiConverterUInt64.allocationSize(value.timestampUs) + + FfiConverterBool.allocationSize(value.keyframe) + 0; } } -class MoqMediaContainerGroupConfig { +class MoqMediaGroupConfig { final String name; final int sequence; final MoqContainer container; final MoqFetchGroupOptions? options; - MoqMediaContainerGroupConfig({ + MoqMediaGroupConfig({ required this.name, required this.sequence, required this.container, @@ -1123,14 +1121,12 @@ class MoqMediaContainerGroupConfig { }); } -class FfiConverterMoqMediaContainerGroupConfig { - static MoqMediaContainerGroupConfig lift(RustBuffer buf) { - return FfiConverterMoqMediaContainerGroupConfig.read( - buf.asUint8List(), - ).value; +class FfiConverterMoqMediaGroupConfig { + static MoqMediaGroupConfig lift(RustBuffer buf) { + return FfiConverterMoqMediaGroupConfig.read(buf.asUint8List()).value; } - static LiftRetVal read(Uint8List buf) { + static LiftRetVal read(Uint8List buf) { int new_offset = buf.offsetInBytes; final name_lifted = FfiConverterString.read( Uint8List.view(buf.buffer, new_offset), @@ -1153,7 +1149,7 @@ class FfiConverterMoqMediaContainerGroupConfig { final options = options_lifted.value; new_offset += options_lifted.bytesRead; return LiftRetVal( - MoqMediaContainerGroupConfig( + MoqMediaGroupConfig( name: name, sequence: sequence, container: container, @@ -1163,7 +1159,7 @@ class FfiConverterMoqMediaContainerGroupConfig { ); } - static RustBuffer lower(MoqMediaContainerGroupConfig value) { + static RustBuffer lower(MoqMediaGroupConfig value) { final total_length = FfiConverterString.allocationSize(value.name) + FfiConverterUInt64.allocationSize(value.sequence) + @@ -1175,7 +1171,7 @@ class FfiConverterMoqMediaContainerGroupConfig { return toRustBuffer(buf); } - static int write(MoqMediaContainerGroupConfig value, Uint8List buf) { + static int write(MoqMediaGroupConfig value, Uint8List buf) { int new_offset = buf.offsetInBytes; new_offset += FfiConverterString.write( value.name, @@ -1196,7 +1192,7 @@ class FfiConverterMoqMediaContainerGroupConfig { return new_offset - buf.offsetInBytes; } - static int allocationSize(MoqMediaContainerGroupConfig value) { + static int allocationSize(MoqMediaGroupConfig value) { return FfiConverterString.allocationSize(value.name) + FfiConverterUInt64.allocationSize(value.sequence) + FfiConverterMoqContainer.allocationSize(value.container) + @@ -1205,81 +1201,81 @@ class FfiConverterMoqMediaContainerGroupConfig { } } -class MoqMediaFrame { - final Uint8List payload; - final int timestampUs; - final bool keyframe; - MoqMediaFrame({ - required this.payload, - required this.timestampUs, - required this.keyframe, +class MoqMediaTrackConfig { + final String name; + final MoqContainer container; + final MoqSubscription? subscription; + MoqMediaTrackConfig({ + required this.name, + required this.container, + this.subscription = null, }); } -class FfiConverterMoqMediaFrame { - static MoqMediaFrame lift(RustBuffer buf) { - return FfiConverterMoqMediaFrame.read(buf.asUint8List()).value; +class FfiConverterMoqMediaTrackConfig { + static MoqMediaTrackConfig lift(RustBuffer buf) { + return FfiConverterMoqMediaTrackConfig.read(buf.asUint8List()).value; } - static LiftRetVal read(Uint8List buf) { + static LiftRetVal read(Uint8List buf) { int new_offset = buf.offsetInBytes; - final payload_lifted = FfiConverterUint8List.read( + final name_lifted = FfiConverterString.read( Uint8List.view(buf.buffer, new_offset), ); - final payload = payload_lifted.value; - new_offset += payload_lifted.bytesRead; - final timestampUs_lifted = FfiConverterUInt64.read( + final name = name_lifted.value; + new_offset += name_lifted.bytesRead; + final container_lifted = FfiConverterMoqContainer.read( Uint8List.view(buf.buffer, new_offset), ); - final timestampUs = timestampUs_lifted.value; - new_offset += timestampUs_lifted.bytesRead; - final keyframe_lifted = FfiConverterBool.read( + final container = container_lifted.value; + new_offset += container_lifted.bytesRead; + final subscription_lifted = FfiConverterOptionalMoqSubscription.read( Uint8List.view(buf.buffer, new_offset), ); - final keyframe = keyframe_lifted.value; - new_offset += keyframe_lifted.bytesRead; + final subscription = subscription_lifted.value; + new_offset += subscription_lifted.bytesRead; return LiftRetVal( - MoqMediaFrame( - payload: payload, - timestampUs: timestampUs, - keyframe: keyframe, + MoqMediaTrackConfig( + name: name, + container: container, + subscription: subscription, ), new_offset - buf.offsetInBytes, ); } - static RustBuffer lower(MoqMediaFrame value) { + static RustBuffer lower(MoqMediaTrackConfig value) { final total_length = - FfiConverterUint8List.allocationSize(value.payload) + - FfiConverterUInt64.allocationSize(value.timestampUs) + - FfiConverterBool.allocationSize(value.keyframe) + + FfiConverterString.allocationSize(value.name) + + FfiConverterMoqContainer.allocationSize(value.container) + + FfiConverterOptionalMoqSubscription.allocationSize(value.subscription) + 0; final buf = Uint8List(total_length); write(value, buf); return toRustBuffer(buf); } - static int write(MoqMediaFrame value, Uint8List buf) { + static int write(MoqMediaTrackConfig value, Uint8List buf) { int new_offset = buf.offsetInBytes; - new_offset += FfiConverterUint8List.write( - value.payload, + new_offset += FfiConverterString.write( + value.name, Uint8List.view(buf.buffer, new_offset), ); - new_offset += FfiConverterUInt64.write( - value.timestampUs, + new_offset += FfiConverterMoqContainer.write( + value.container, Uint8List.view(buf.buffer, new_offset), ); - new_offset += FfiConverterBool.write( - value.keyframe, + new_offset += FfiConverterOptionalMoqSubscription.write( + value.subscription, Uint8List.view(buf.buffer, new_offset), ); return new_offset - buf.offsetInBytes; } - static int allocationSize(MoqMediaFrame value) { - return FfiConverterUint8List.allocationSize(value.payload) + - FfiConverterUInt64.allocationSize(value.timestampUs) + - FfiConverterBool.allocationSize(value.keyframe) + + static int allocationSize(MoqMediaTrackConfig value) { + return FfiConverterString.allocationSize(value.name) + + FfiConverterMoqContainer.allocationSize(value.container) + + FfiConverterOptionalMoqSubscription.allocationSize(value.subscription) + 0; } } @@ -6539,432 +6535,436 @@ class FfiConverterMoqMediaCatalogProducer { } } -abstract class MoqMediaContainerConsumerInterface { - void cancel(); - Future next(); +abstract class MoqMediaContainerProducerInterface { + void cut(); + void finish(); + void seek({required int sequence}); + void write({required Uint8List payload}); } -final _MoqMediaContainerConsumerFinalizer = Finalizer>((ptr) { +final _MoqMediaContainerProducerFinalizer = Finalizer>((ptr) { rustCall( - (status) => uniffi_moq_ffi_fn_free_moqmediacontainerconsumer(ptr, status), + (status) => uniffi_moq_ffi_fn_free_moqmediacontainerproducer(ptr, status), ); }); -class MoqMediaContainerConsumer implements MoqMediaContainerConsumerInterface { +class MoqMediaContainerProducer implements MoqMediaContainerProducerInterface { late final Pointer _ptr; - MoqMediaContainerConsumer._(this._ptr) { - _MoqMediaContainerConsumerFinalizer.attach(this, _ptr, detach: this); + MoqMediaContainerProducer._(this._ptr) { + _MoqMediaContainerProducerFinalizer.attach(this, _ptr, detach: this); } - static Future subscribe({ - required MoqBroadcastConsumer broadcast, - required MoqMediaContainerConfig config, - }) { - return uniffiRustCallAsync( - () => uniffi_moq_ffi_fn_constructor_moqmediacontainerconsumer_subscribe( - FfiConverterMoqBroadcastConsumer.lower(broadcast), - FfiConverterMoqMediaContainerConfig.lower(config), - ), - ffi_moq_ffi_rust_future_poll_u64, - ffi_moq_ffi_rust_future_complete_u64, - ffi_moq_ffi_rust_future_free_u64, - (int handle) => - MoqMediaContainerConsumer._(Pointer.fromAddress(handle)), - moqExceptionErrorHandler, - ); + MoqMediaContainerProducer({ + required MoqBroadcastProducer broadcast, + required MoqContainerInit init, + }) : _ptr = rustCall( + (status) => + uniffi_moq_ffi_fn_constructor_moqmediacontainerproducer_new( + FfiConverterMoqBroadcastProducer.lower(broadcast), + FfiConverterMoqContainerInit.lower(init), + status, + ), + moqExceptionErrorHandler, + ) { + _MoqMediaContainerProducerFinalizer.attach(this, _ptr, detach: this); } - - factory MoqMediaContainerConsumer.lift(Pointer ptr) { - return MoqMediaContainerConsumer._(ptr); + factory MoqMediaContainerProducer.lift(Pointer ptr) { + return MoqMediaContainerProducer._(ptr); } Pointer uniffiClonePointer() { return rustCall( (status) => - uniffi_moq_ffi_fn_clone_moqmediacontainerconsumer(_ptr, status), + uniffi_moq_ffi_fn_clone_moqmediacontainerproducer(_ptr, status), ); } void dispose() { - _MoqMediaContainerConsumerFinalizer.detach(this); + _MoqMediaContainerProducerFinalizer.detach(this); rustCall( (status) => - uniffi_moq_ffi_fn_free_moqmediacontainerconsumer(_ptr, status), + uniffi_moq_ffi_fn_free_moqmediacontainerproducer(_ptr, status), ); } - void cancel() { + void cut() { return rustCall((status) { - uniffi_moq_ffi_fn_method_moqmediacontainerconsumer_cancel( + uniffi_moq_ffi_fn_method_moqmediacontainerproducer_cut( uniffiClonePointer(), status, ); - }, null); + }, moqExceptionErrorHandler); } - Future next() { - return uniffiRustCallAsync( - () => uniffi_moq_ffi_fn_method_moqmediacontainerconsumer_next( + void finish() { + return rustCall((status) { + uniffi_moq_ffi_fn_method_moqmediacontainerproducer_finish( uniffiClonePointer(), - ), - ffi_moq_ffi_rust_future_poll_rust_buffer, - ffi_moq_ffi_rust_future_complete_rust_buffer, - ffi_moq_ffi_rust_future_free_rust_buffer, - FfiConverterOptionalMoqMediaFrame.lift, - moqExceptionErrorHandler, - ); + status, + ); + }, moqExceptionErrorHandler); + } + + void seek({required int sequence}) { + return rustCall((status) { + uniffi_moq_ffi_fn_method_moqmediacontainerproducer_seek( + uniffiClonePointer(), + FfiConverterUInt64.lower(sequence), + status, + ); + }, moqExceptionErrorHandler); + } + + void write({required Uint8List payload}) { + return rustCall((status) { + uniffi_moq_ffi_fn_method_moqmediacontainerproducer_write( + uniffiClonePointer(), + FfiConverterUint8List.lower(payload), + status, + ); + }, moqExceptionErrorHandler); } } -class FfiConverterMoqMediaContainerConsumer { - static MoqMediaContainerConsumer lift(Pointer ptr) { - return MoqMediaContainerConsumer.lift(ptr); +class FfiConverterMoqMediaContainerProducer { + static MoqMediaContainerProducer lift(Pointer ptr) { + return MoqMediaContainerProducer.lift(ptr); } - static Pointer lower(MoqMediaContainerConsumer value) { + static Pointer lower(MoqMediaContainerProducer value) { return value.uniffiClonePointer(); } - static int allocationSize(MoqMediaContainerConsumer value) { + static int allocationSize(MoqMediaContainerProducer value) { return 8; } - static LiftRetVal read(Uint8List buf) { + static LiftRetVal read(Uint8List buf) { final handle = buf.buffer.asByteData(buf.offsetInBytes).getInt64(0); final pointer = Pointer.fromAddress(handle); - return LiftRetVal(MoqMediaContainerConsumer.lift(pointer), 8); + return LiftRetVal(MoqMediaContainerProducer.lift(pointer), 8); } - static int write(MoqMediaContainerConsumer value, Uint8List buf) { + static int write(MoqMediaContainerProducer value, Uint8List buf) { final handle = lower(value); buf.buffer.asByteData(buf.offsetInBytes).setInt64(0, handle.address); return 8; } } -abstract class MoqMediaContainerGroupConsumerInterface { - void cancel(); - Future next(); - int sequence(); +abstract class MoqMediaContainerStreamProducerInterface { + void finish(); + void write({required Uint8List payload}); } -final _MoqMediaContainerGroupConsumerFinalizer = Finalizer>(( +final _MoqMediaContainerStreamProducerFinalizer = Finalizer>(( ptr, ) { rustCall( (status) => - uniffi_moq_ffi_fn_free_moqmediacontainergroupconsumer(ptr, status), + uniffi_moq_ffi_fn_free_moqmediacontainerstreamproducer(ptr, status), ); }); -class MoqMediaContainerGroupConsumer - implements MoqMediaContainerGroupConsumerInterface { +class MoqMediaContainerStreamProducer + implements MoqMediaContainerStreamProducerInterface { late final Pointer _ptr; - MoqMediaContainerGroupConsumer._(this._ptr) { - _MoqMediaContainerGroupConsumerFinalizer.attach(this, _ptr, detach: this); + MoqMediaContainerStreamProducer._(this._ptr) { + _MoqMediaContainerStreamProducerFinalizer.attach(this, _ptr, detach: this); } - static Future fetch({ - required MoqBroadcastConsumer broadcast, - required MoqMediaContainerGroupConfig config, - }) { - return uniffiRustCallAsync( - () => uniffi_moq_ffi_fn_constructor_moqmediacontainergroupconsumer_fetch( - FfiConverterMoqBroadcastConsumer.lower(broadcast), - FfiConverterMoqMediaContainerGroupConfig.lower(config), - ), - ffi_moq_ffi_rust_future_poll_u64, - ffi_moq_ffi_rust_future_complete_u64, - ffi_moq_ffi_rust_future_free_u64, - (int handle) => - MoqMediaContainerGroupConsumer._(Pointer.fromAddress(handle)), - moqExceptionErrorHandler, - ); + MoqMediaContainerStreamProducer({ + required MoqBroadcastProducer broadcast, + required MoqContainerFormat format, + }) : _ptr = rustCall( + (status) => + uniffi_moq_ffi_fn_constructor_moqmediacontainerstreamproducer_new( + FfiConverterMoqBroadcastProducer.lower(broadcast), + FfiConverterMoqContainerFormat.lower(format), + status, + ), + moqExceptionErrorHandler, + ) { + _MoqMediaContainerStreamProducerFinalizer.attach(this, _ptr, detach: this); } - - factory MoqMediaContainerGroupConsumer.lift(Pointer ptr) { - return MoqMediaContainerGroupConsumer._(ptr); + factory MoqMediaContainerStreamProducer.lift(Pointer ptr) { + return MoqMediaContainerStreamProducer._(ptr); } Pointer uniffiClonePointer() { return rustCall( (status) => - uniffi_moq_ffi_fn_clone_moqmediacontainergroupconsumer(_ptr, status), + uniffi_moq_ffi_fn_clone_moqmediacontainerstreamproducer(_ptr, status), ); } void dispose() { - _MoqMediaContainerGroupConsumerFinalizer.detach(this); + _MoqMediaContainerStreamProducerFinalizer.detach(this); rustCall( (status) => - uniffi_moq_ffi_fn_free_moqmediacontainergroupconsumer(_ptr, status), + uniffi_moq_ffi_fn_free_moqmediacontainerstreamproducer(_ptr, status), ); } - void cancel() { + void finish() { return rustCall((status) { - uniffi_moq_ffi_fn_method_moqmediacontainergroupconsumer_cancel( + uniffi_moq_ffi_fn_method_moqmediacontainerstreamproducer_finish( uniffiClonePointer(), status, ); - }, null); + }, moqExceptionErrorHandler); } - Future next() { - return uniffiRustCallAsync( - () => uniffi_moq_ffi_fn_method_moqmediacontainergroupconsumer_next( + void write({required Uint8List payload}) { + return rustCall((status) { + uniffi_moq_ffi_fn_method_moqmediacontainerstreamproducer_write( uniffiClonePointer(), - ), - ffi_moq_ffi_rust_future_poll_rust_buffer, - ffi_moq_ffi_rust_future_complete_rust_buffer, - ffi_moq_ffi_rust_future_free_rust_buffer, - FfiConverterOptionalMoqMediaFrame.lift, - moqExceptionErrorHandler, - ); - } - - int sequence() { - return rustCallWithLifter( - (status) => - uniffi_moq_ffi_fn_method_moqmediacontainergroupconsumer_sequence( - uniffiClonePointer(), - status, - ), - FfiConverterUInt64.lift, - null, - ); + FfiConverterUint8List.lower(payload), + status, + ); + }, moqExceptionErrorHandler); } } -class FfiConverterMoqMediaContainerGroupConsumer { - static MoqMediaContainerGroupConsumer lift(Pointer ptr) { - return MoqMediaContainerGroupConsumer.lift(ptr); +class FfiConverterMoqMediaContainerStreamProducer { + static MoqMediaContainerStreamProducer lift(Pointer ptr) { + return MoqMediaContainerStreamProducer.lift(ptr); } - static Pointer lower(MoqMediaContainerGroupConsumer value) { + static Pointer lower(MoqMediaContainerStreamProducer value) { return value.uniffiClonePointer(); } - static int allocationSize(MoqMediaContainerGroupConsumer value) { + static int allocationSize(MoqMediaContainerStreamProducer value) { return 8; } - static LiftRetVal read(Uint8List buf) { + static LiftRetVal read(Uint8List buf) { final handle = buf.buffer.asByteData(buf.offsetInBytes).getInt64(0); final pointer = Pointer.fromAddress(handle); - return LiftRetVal(MoqMediaContainerGroupConsumer.lift(pointer), 8); + return LiftRetVal(MoqMediaContainerStreamProducer.lift(pointer), 8); } - static int write(MoqMediaContainerGroupConsumer value, Uint8List buf) { + static int write(MoqMediaContainerStreamProducer value, Uint8List buf) { final handle = lower(value); buf.buffer.asByteData(buf.offsetInBytes).setInt64(0, handle.address); return 8; } } -abstract class MoqMediaContainerProducerInterface { - void cut(); - void finish(); - void seek({required int sequence}); - void write({required Uint8List payload}); +abstract class MoqMediaGroupConsumerInterface { + void cancel(); + Future next(); + int sequence(); } -final _MoqMediaContainerProducerFinalizer = Finalizer>((ptr) { +final _MoqMediaGroupConsumerFinalizer = Finalizer>((ptr) { rustCall( - (status) => uniffi_moq_ffi_fn_free_moqmediacontainerproducer(ptr, status), + (status) => uniffi_moq_ffi_fn_free_moqmediagroupconsumer(ptr, status), ); }); -class MoqMediaContainerProducer implements MoqMediaContainerProducerInterface { +class MoqMediaGroupConsumer implements MoqMediaGroupConsumerInterface { late final Pointer _ptr; - MoqMediaContainerProducer._(this._ptr) { - _MoqMediaContainerProducerFinalizer.attach(this, _ptr, detach: this); + MoqMediaGroupConsumer._(this._ptr) { + _MoqMediaGroupConsumerFinalizer.attach(this, _ptr, detach: this); } - MoqMediaContainerProducer({ - required MoqBroadcastProducer broadcast, - required MoqContainerInit init, + MoqMediaGroupConsumer({ + required MoqGroupConsumer group, + required MoqContainer container, }) : _ptr = rustCall( - (status) => - uniffi_moq_ffi_fn_constructor_moqmediacontainerproducer_new( - FfiConverterMoqBroadcastProducer.lower(broadcast), - FfiConverterMoqContainerInit.lower(init), - status, - ), + (status) => uniffi_moq_ffi_fn_constructor_moqmediagroupconsumer_new( + FfiConverterMoqGroupConsumer.lower(group), + FfiConverterMoqContainer.lower(container), + status, + ), moqExceptionErrorHandler, ) { - _MoqMediaContainerProducerFinalizer.attach(this, _ptr, detach: this); + _MoqMediaGroupConsumerFinalizer.attach(this, _ptr, detach: this); } - factory MoqMediaContainerProducer.lift(Pointer ptr) { - return MoqMediaContainerProducer._(ptr); + static Future fetch({ + required MoqBroadcastConsumer broadcast, + required MoqMediaGroupConfig config, + }) { + return uniffiRustCallAsync( + () => uniffi_moq_ffi_fn_constructor_moqmediagroupconsumer_fetch( + FfiConverterMoqBroadcastConsumer.lower(broadcast), + FfiConverterMoqMediaGroupConfig.lower(config), + ), + ffi_moq_ffi_rust_future_poll_u64, + ffi_moq_ffi_rust_future_complete_u64, + ffi_moq_ffi_rust_future_free_u64, + (int handle) => + MoqMediaGroupConsumer._(Pointer.fromAddress(handle)), + moqExceptionErrorHandler, + ); + } + + factory MoqMediaGroupConsumer.lift(Pointer ptr) { + return MoqMediaGroupConsumer._(ptr); } Pointer uniffiClonePointer() { return rustCall( - (status) => - uniffi_moq_ffi_fn_clone_moqmediacontainerproducer(_ptr, status), + (status) => uniffi_moq_ffi_fn_clone_moqmediagroupconsumer(_ptr, status), ); } void dispose() { - _MoqMediaContainerProducerFinalizer.detach(this); + _MoqMediaGroupConsumerFinalizer.detach(this); rustCall( - (status) => - uniffi_moq_ffi_fn_free_moqmediacontainerproducer(_ptr, status), + (status) => uniffi_moq_ffi_fn_free_moqmediagroupconsumer(_ptr, status), ); } - void cut() { - return rustCall((status) { - uniffi_moq_ffi_fn_method_moqmediacontainerproducer_cut( - uniffiClonePointer(), - status, - ); - }, moqExceptionErrorHandler); - } - - void finish() { + void cancel() { return rustCall((status) { - uniffi_moq_ffi_fn_method_moqmediacontainerproducer_finish( + uniffi_moq_ffi_fn_method_moqmediagroupconsumer_cancel( uniffiClonePointer(), status, ); - }, moqExceptionErrorHandler); + }, null); } - void seek({required int sequence}) { - return rustCall((status) { - uniffi_moq_ffi_fn_method_moqmediacontainerproducer_seek( + Future next() { + return uniffiRustCallAsync( + () => uniffi_moq_ffi_fn_method_moqmediagroupconsumer_next( uniffiClonePointer(), - FfiConverterUInt64.lower(sequence), - status, - ); - }, moqExceptionErrorHandler); + ), + ffi_moq_ffi_rust_future_poll_rust_buffer, + ffi_moq_ffi_rust_future_complete_rust_buffer, + ffi_moq_ffi_rust_future_free_rust_buffer, + FfiConverterOptionalMoqMediaFrame.lift, + moqExceptionErrorHandler, + ); } - - void write({required Uint8List payload}) { - return rustCall((status) { - uniffi_moq_ffi_fn_method_moqmediacontainerproducer_write( + + int sequence() { + return rustCallWithLifter( + (status) => uniffi_moq_ffi_fn_method_moqmediagroupconsumer_sequence( uniffiClonePointer(), - FfiConverterUint8List.lower(payload), status, - ); - }, moqExceptionErrorHandler); + ), + FfiConverterUInt64.lift, + null, + ); } } -class FfiConverterMoqMediaContainerProducer { - static MoqMediaContainerProducer lift(Pointer ptr) { - return MoqMediaContainerProducer.lift(ptr); +class FfiConverterMoqMediaGroupConsumer { + static MoqMediaGroupConsumer lift(Pointer ptr) { + return MoqMediaGroupConsumer.lift(ptr); } - static Pointer lower(MoqMediaContainerProducer value) { + static Pointer lower(MoqMediaGroupConsumer value) { return value.uniffiClonePointer(); } - static int allocationSize(MoqMediaContainerProducer value) { + static int allocationSize(MoqMediaGroupConsumer value) { return 8; } - static LiftRetVal read(Uint8List buf) { + static LiftRetVal read(Uint8List buf) { final handle = buf.buffer.asByteData(buf.offsetInBytes).getInt64(0); final pointer = Pointer.fromAddress(handle); - return LiftRetVal(MoqMediaContainerProducer.lift(pointer), 8); + return LiftRetVal(MoqMediaGroupConsumer.lift(pointer), 8); } - static int write(MoqMediaContainerProducer value, Uint8List buf) { + static int write(MoqMediaGroupConsumer value, Uint8List buf) { final handle = lower(value); buf.buffer.asByteData(buf.offsetInBytes).setInt64(0, handle.address); return 8; } } -abstract class MoqMediaContainerStreamProducerInterface { - void finish(); - void write({required Uint8List payload}); +abstract class MoqMediaTrackConsumerInterface { + void cancel(); + Future next(); } -final _MoqMediaContainerStreamProducerFinalizer = Finalizer>(( - ptr, -) { +final _MoqMediaTrackConsumerFinalizer = Finalizer>((ptr) { rustCall( - (status) => - uniffi_moq_ffi_fn_free_moqmediacontainerstreamproducer(ptr, status), + (status) => uniffi_moq_ffi_fn_free_moqmediatrackconsumer(ptr, status), ); }); -class MoqMediaContainerStreamProducer - implements MoqMediaContainerStreamProducerInterface { +class MoqMediaTrackConsumer implements MoqMediaTrackConsumerInterface { late final Pointer _ptr; - MoqMediaContainerStreamProducer._(this._ptr) { - _MoqMediaContainerStreamProducerFinalizer.attach(this, _ptr, detach: this); + MoqMediaTrackConsumer._(this._ptr) { + _MoqMediaTrackConsumerFinalizer.attach(this, _ptr, detach: this); } - MoqMediaContainerStreamProducer({ - required MoqBroadcastProducer broadcast, - required MoqContainerFormat format, - }) : _ptr = rustCall( - (status) => - uniffi_moq_ffi_fn_constructor_moqmediacontainerstreamproducer_new( - FfiConverterMoqBroadcastProducer.lower(broadcast), - FfiConverterMoqContainerFormat.lower(format), - status, - ), - moqExceptionErrorHandler, - ) { - _MoqMediaContainerStreamProducerFinalizer.attach(this, _ptr, detach: this); + static Future subscribe({ + required MoqBroadcastConsumer broadcast, + required MoqMediaTrackConfig config, + }) { + return uniffiRustCallAsync( + () => uniffi_moq_ffi_fn_constructor_moqmediatrackconsumer_subscribe( + FfiConverterMoqBroadcastConsumer.lower(broadcast), + FfiConverterMoqMediaTrackConfig.lower(config), + ), + ffi_moq_ffi_rust_future_poll_u64, + ffi_moq_ffi_rust_future_complete_u64, + ffi_moq_ffi_rust_future_free_u64, + (int handle) => + MoqMediaTrackConsumer._(Pointer.fromAddress(handle)), + moqExceptionErrorHandler, + ); } - factory MoqMediaContainerStreamProducer.lift(Pointer ptr) { - return MoqMediaContainerStreamProducer._(ptr); + + factory MoqMediaTrackConsumer.lift(Pointer ptr) { + return MoqMediaTrackConsumer._(ptr); } Pointer uniffiClonePointer() { return rustCall( - (status) => - uniffi_moq_ffi_fn_clone_moqmediacontainerstreamproducer(_ptr, status), + (status) => uniffi_moq_ffi_fn_clone_moqmediatrackconsumer(_ptr, status), ); } void dispose() { - _MoqMediaContainerStreamProducerFinalizer.detach(this); + _MoqMediaTrackConsumerFinalizer.detach(this); rustCall( - (status) => - uniffi_moq_ffi_fn_free_moqmediacontainerstreamproducer(_ptr, status), + (status) => uniffi_moq_ffi_fn_free_moqmediatrackconsumer(_ptr, status), ); } - void finish() { + void cancel() { return rustCall((status) { - uniffi_moq_ffi_fn_method_moqmediacontainerstreamproducer_finish( + uniffi_moq_ffi_fn_method_moqmediatrackconsumer_cancel( uniffiClonePointer(), status, ); - }, moqExceptionErrorHandler); + }, null); } - void write({required Uint8List payload}) { - return rustCall((status) { - uniffi_moq_ffi_fn_method_moqmediacontainerstreamproducer_write( + Future next() { + return uniffiRustCallAsync( + () => uniffi_moq_ffi_fn_method_moqmediatrackconsumer_next( uniffiClonePointer(), - FfiConverterUint8List.lower(payload), - status, - ); - }, moqExceptionErrorHandler); + ), + ffi_moq_ffi_rust_future_poll_rust_buffer, + ffi_moq_ffi_rust_future_complete_rust_buffer, + ffi_moq_ffi_rust_future_free_rust_buffer, + FfiConverterOptionalMoqMediaFrame.lift, + moqExceptionErrorHandler, + ); } } -class FfiConverterMoqMediaContainerStreamProducer { - static MoqMediaContainerStreamProducer lift(Pointer ptr) { - return MoqMediaContainerStreamProducer.lift(ptr); +class FfiConverterMoqMediaTrackConsumer { + static MoqMediaTrackConsumer lift(Pointer ptr) { + return MoqMediaTrackConsumer.lift(ptr); } - static Pointer lower(MoqMediaContainerStreamProducer value) { + static Pointer lower(MoqMediaTrackConsumer value) { return value.uniffiClonePointer(); } - static int allocationSize(MoqMediaContainerStreamProducer value) { + static int allocationSize(MoqMediaTrackConsumer value) { return 8; } - static LiftRetVal read(Uint8List buf) { + static LiftRetVal read(Uint8List buf) { final handle = buf.buffer.asByteData(buf.offsetInBytes).getInt64(0); final pointer = Pointer.fromAddress(handle); - return LiftRetVal(MoqMediaContainerStreamProducer.lift(pointer), 8); + return LiftRetVal(MoqMediaTrackConsumer.lift(pointer), 8); } - static int write(MoqMediaContainerStreamProducer value, Uint8List buf) { + static int write(MoqMediaTrackConsumer value, Uint8List buf) { final handle = lower(value); buf.buffer.asByteData(buf.offsetInBytes).setInt64(0, handle.address); return 8; @@ -11116,7 +11116,7 @@ uniffi_moq_ffi_fn_method_moqmediacatalogproducer_set_video_properties( @Native Function(Pointer, Pointer)>( assetId: _uniffiAssetId, ) -external Pointer uniffi_moq_ffi_fn_clone_moqmediacontainerconsumer( +external Pointer uniffi_moq_ffi_fn_clone_moqmediacontainerproducer( Pointer handle, Pointer uniffiStatus, ); @@ -11124,37 +11124,59 @@ external Pointer uniffi_moq_ffi_fn_clone_moqmediacontainerconsumer( @Native, Pointer)>( assetId: _uniffiAssetId, ) -external void uniffi_moq_ffi_fn_free_moqmediacontainerconsumer( +external void uniffi_moq_ffi_fn_free_moqmediacontainerproducer( Pointer handle, Pointer uniffiStatus, ); -@Native Function(Pointer, RustBuffer)>( - assetId: _uniffiAssetId, -) +@Native< + Pointer Function(Pointer, RustBuffer, Pointer) +>(assetId: _uniffiAssetId) external Pointer -uniffi_moq_ffi_fn_constructor_moqmediacontainerconsumer_subscribe( +uniffi_moq_ffi_fn_constructor_moqmediacontainerproducer_new( Pointer broadcast, - RustBuffer config, + RustBuffer init, + Pointer uniffiStatus, ); @Native, Pointer)>( assetId: _uniffiAssetId, ) -external void uniffi_moq_ffi_fn_method_moqmediacontainerconsumer_cancel( +external void uniffi_moq_ffi_fn_method_moqmediacontainerproducer_cut( Pointer ptr, Pointer uniffiStatus, ); -@Native Function(Pointer)>(assetId: _uniffiAssetId) -external Pointer uniffi_moq_ffi_fn_method_moqmediacontainerconsumer_next( +@Native, Pointer)>( + assetId: _uniffiAssetId, +) +external void uniffi_moq_ffi_fn_method_moqmediacontainerproducer_finish( + Pointer ptr, + Pointer uniffiStatus, +); + +@Native, Uint64, Pointer)>( + assetId: _uniffiAssetId, +) +external void uniffi_moq_ffi_fn_method_moqmediacontainerproducer_seek( + Pointer ptr, + int sequence, + Pointer uniffiStatus, +); + +@Native, RustBuffer, Pointer)>( + assetId: _uniffiAssetId, +) +external void uniffi_moq_ffi_fn_method_moqmediacontainerproducer_write( Pointer ptr, + RustBuffer payload, + Pointer uniffiStatus, ); @Native Function(Pointer, Pointer)>( assetId: _uniffiAssetId, ) -external Pointer uniffi_moq_ffi_fn_clone_moqmediacontainergroupconsumer( +external Pointer uniffi_moq_ffi_fn_clone_moqmediacontainerstreamproducer( Pointer handle, Pointer uniffiStatus, ); @@ -11162,44 +11184,42 @@ external Pointer uniffi_moq_ffi_fn_clone_moqmediacontainergroupconsumer( @Native, Pointer)>( assetId: _uniffiAssetId, ) -external void uniffi_moq_ffi_fn_free_moqmediacontainergroupconsumer( +external void uniffi_moq_ffi_fn_free_moqmediacontainerstreamproducer( Pointer handle, Pointer uniffiStatus, ); -@Native Function(Pointer, RustBuffer)>( - assetId: _uniffiAssetId, -) +@Native< + Pointer Function(Pointer, RustBuffer, Pointer) +>(assetId: _uniffiAssetId) external Pointer -uniffi_moq_ffi_fn_constructor_moqmediacontainergroupconsumer_fetch( +uniffi_moq_ffi_fn_constructor_moqmediacontainerstreamproducer_new( Pointer broadcast, - RustBuffer config, + RustBuffer format, + Pointer uniffiStatus, ); @Native, Pointer)>( assetId: _uniffiAssetId, ) -external void uniffi_moq_ffi_fn_method_moqmediacontainergroupconsumer_cancel( +external void uniffi_moq_ffi_fn_method_moqmediacontainerstreamproducer_finish( Pointer ptr, Pointer uniffiStatus, ); -@Native Function(Pointer)>(assetId: _uniffiAssetId) -external Pointer -uniffi_moq_ffi_fn_method_moqmediacontainergroupconsumer_next(Pointer ptr); - -@Native, Pointer)>( +@Native, RustBuffer, Pointer)>( assetId: _uniffiAssetId, ) -external int uniffi_moq_ffi_fn_method_moqmediacontainergroupconsumer_sequence( +external void uniffi_moq_ffi_fn_method_moqmediacontainerstreamproducer_write( Pointer ptr, + RustBuffer payload, Pointer uniffiStatus, ); @Native Function(Pointer, Pointer)>( assetId: _uniffiAssetId, ) -external Pointer uniffi_moq_ffi_fn_clone_moqmediacontainerproducer( +external Pointer uniffi_moq_ffi_fn_clone_moqmediagroupconsumer( Pointer handle, Pointer uniffiStatus, ); @@ -11207,59 +11227,54 @@ external Pointer uniffi_moq_ffi_fn_clone_moqmediacontainerproducer( @Native, Pointer)>( assetId: _uniffiAssetId, ) -external void uniffi_moq_ffi_fn_free_moqmediacontainerproducer( +external void uniffi_moq_ffi_fn_free_moqmediagroupconsumer( Pointer handle, Pointer uniffiStatus, ); -@Native< - Pointer Function(Pointer, RustBuffer, Pointer) ->(assetId: _uniffiAssetId) +@Native Function(Pointer, RustBuffer)>( + assetId: _uniffiAssetId, +) external Pointer -uniffi_moq_ffi_fn_constructor_moqmediacontainerproducer_new( +uniffi_moq_ffi_fn_constructor_moqmediagroupconsumer_fetch( Pointer broadcast, - RustBuffer init, - Pointer uniffiStatus, + RustBuffer config, ); -@Native, Pointer)>( - assetId: _uniffiAssetId, -) -external void uniffi_moq_ffi_fn_method_moqmediacontainerproducer_cut( - Pointer ptr, +@Native< + Pointer Function(Pointer, RustBuffer, Pointer) +>(assetId: _uniffiAssetId) +external Pointer uniffi_moq_ffi_fn_constructor_moqmediagroupconsumer_new( + Pointer group, + RustBuffer container, Pointer uniffiStatus, ); @Native, Pointer)>( assetId: _uniffiAssetId, ) -external void uniffi_moq_ffi_fn_method_moqmediacontainerproducer_finish( +external void uniffi_moq_ffi_fn_method_moqmediagroupconsumer_cancel( Pointer ptr, Pointer uniffiStatus, ); -@Native, Uint64, Pointer)>( - assetId: _uniffiAssetId, -) -external void uniffi_moq_ffi_fn_method_moqmediacontainerproducer_seek( +@Native Function(Pointer)>(assetId: _uniffiAssetId) +external Pointer uniffi_moq_ffi_fn_method_moqmediagroupconsumer_next( Pointer ptr, - int sequence, - Pointer uniffiStatus, ); -@Native, RustBuffer, Pointer)>( +@Native, Pointer)>( assetId: _uniffiAssetId, ) -external void uniffi_moq_ffi_fn_method_moqmediacontainerproducer_write( +external int uniffi_moq_ffi_fn_method_moqmediagroupconsumer_sequence( Pointer ptr, - RustBuffer payload, Pointer uniffiStatus, ); @Native Function(Pointer, Pointer)>( assetId: _uniffiAssetId, ) -external Pointer uniffi_moq_ffi_fn_clone_moqmediacontainerstreamproducer( +external Pointer uniffi_moq_ffi_fn_clone_moqmediatrackconsumer( Pointer handle, Pointer uniffiStatus, ); @@ -11267,36 +11282,31 @@ external Pointer uniffi_moq_ffi_fn_clone_moqmediacontainerstreamproducer( @Native, Pointer)>( assetId: _uniffiAssetId, ) -external void uniffi_moq_ffi_fn_free_moqmediacontainerstreamproducer( +external void uniffi_moq_ffi_fn_free_moqmediatrackconsumer( Pointer handle, Pointer uniffiStatus, ); -@Native< - Pointer Function(Pointer, RustBuffer, Pointer) ->(assetId: _uniffiAssetId) +@Native Function(Pointer, RustBuffer)>( + assetId: _uniffiAssetId, +) external Pointer -uniffi_moq_ffi_fn_constructor_moqmediacontainerstreamproducer_new( +uniffi_moq_ffi_fn_constructor_moqmediatrackconsumer_subscribe( Pointer broadcast, - RustBuffer format, - Pointer uniffiStatus, + RustBuffer config, ); @Native, Pointer)>( assetId: _uniffiAssetId, ) -external void uniffi_moq_ffi_fn_method_moqmediacontainerstreamproducer_finish( +external void uniffi_moq_ffi_fn_method_moqmediatrackconsumer_cancel( Pointer ptr, Pointer uniffiStatus, ); -@Native, RustBuffer, Pointer)>( - assetId: _uniffiAssetId, -) -external void uniffi_moq_ffi_fn_method_moqmediacontainerstreamproducer_write( +@Native Function(Pointer)>(assetId: _uniffiAssetId) +external Pointer uniffi_moq_ffi_fn_method_moqmediatrackconsumer_next( Pointer ptr, - RustBuffer payload, - Pointer uniffiStatus, ); @Native Function(Pointer, Pointer)>( @@ -12853,42 +12863,39 @@ external int uniffi_moq_ffi_checksum_method_moqmediacatalogproducer_set_video_properties(); @Native(assetId: _uniffiAssetId) -external int uniffi_moq_ffi_checksum_method_moqmediacontainerconsumer_cancel(); +external int uniffi_moq_ffi_checksum_method_moqmediacontainerproducer_cut(); @Native(assetId: _uniffiAssetId) -external int uniffi_moq_ffi_checksum_method_moqmediacontainerconsumer_next(); +external int uniffi_moq_ffi_checksum_method_moqmediacontainerproducer_finish(); @Native(assetId: _uniffiAssetId) -external int -uniffi_moq_ffi_checksum_method_moqmediacontainergroupconsumer_cancel(); +external int uniffi_moq_ffi_checksum_method_moqmediacontainerproducer_seek(); @Native(assetId: _uniffiAssetId) -external int -uniffi_moq_ffi_checksum_method_moqmediacontainergroupconsumer_next(); +external int uniffi_moq_ffi_checksum_method_moqmediacontainerproducer_write(); @Native(assetId: _uniffiAssetId) external int -uniffi_moq_ffi_checksum_method_moqmediacontainergroupconsumer_sequence(); +uniffi_moq_ffi_checksum_method_moqmediacontainerstreamproducer_finish(); @Native(assetId: _uniffiAssetId) -external int uniffi_moq_ffi_checksum_method_moqmediacontainerproducer_cut(); +external int +uniffi_moq_ffi_checksum_method_moqmediacontainerstreamproducer_write(); @Native(assetId: _uniffiAssetId) -external int uniffi_moq_ffi_checksum_method_moqmediacontainerproducer_finish(); +external int uniffi_moq_ffi_checksum_method_moqmediagroupconsumer_cancel(); @Native(assetId: _uniffiAssetId) -external int uniffi_moq_ffi_checksum_method_moqmediacontainerproducer_seek(); +external int uniffi_moq_ffi_checksum_method_moqmediagroupconsumer_next(); @Native(assetId: _uniffiAssetId) -external int uniffi_moq_ffi_checksum_method_moqmediacontainerproducer_write(); +external int uniffi_moq_ffi_checksum_method_moqmediagroupconsumer_sequence(); @Native(assetId: _uniffiAssetId) -external int -uniffi_moq_ffi_checksum_method_moqmediacontainerstreamproducer_finish(); +external int uniffi_moq_ffi_checksum_method_moqmediatrackconsumer_cancel(); @Native(assetId: _uniffiAssetId) -external int -uniffi_moq_ffi_checksum_method_moqmediacontainerstreamproducer_write(); +external int uniffi_moq_ffi_checksum_method_moqmediatrackconsumer_next(); @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_method_moqmediatrackproducer_cut(); @@ -13175,19 +13182,21 @@ external int uniffi_moq_ffi_checksum_constructor_moqmediacatalogproducer_new(); @Native(assetId: _uniffiAssetId) external int -uniffi_moq_ffi_checksum_constructor_moqmediacontainerconsumer_subscribe(); +uniffi_moq_ffi_checksum_constructor_moqmediacontainerproducer_new(); @Native(assetId: _uniffiAssetId) external int -uniffi_moq_ffi_checksum_constructor_moqmediacontainergroupconsumer_fetch(); +uniffi_moq_ffi_checksum_constructor_moqmediacontainerstreamproducer_new(); @Native(assetId: _uniffiAssetId) -external int -uniffi_moq_ffi_checksum_constructor_moqmediacontainerproducer_new(); +external int uniffi_moq_ffi_checksum_constructor_moqmediagroupconsumer_fetch(); + +@Native(assetId: _uniffiAssetId) +external int uniffi_moq_ffi_checksum_constructor_moqmediagroupconsumer_new(); @Native(assetId: _uniffiAssetId) external int -uniffi_moq_ffi_checksum_constructor_moqmediacontainerstreamproducer_new(); +uniffi_moq_ffi_checksum_constructor_moqmediatrackconsumer_subscribe(); @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_constructor_moqmediatrackproducer_audio(); @@ -13247,7 +13256,7 @@ void _checkApiChecksums() { 18633) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } - if (uniffi_moq_ffi_checksum_method_moqbroadcastconsumer_resolve() != 65350) { + if (uniffi_moq_ffi_checksum_method_moqbroadcastconsumer_resolve() != 65499) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqbroadcastconsumer_subscribe_track() != @@ -13279,7 +13288,7 @@ void _checkApiChecksums() { 17412) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } - if (uniffi_moq_ffi_checksum_method_moqtrackconsumer_recv_group() != 60887) { + if (uniffi_moq_ffi_checksum_method_moqtrackconsumer_recv_group() != 28712) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqtrackconsumer_update() != 24851) { @@ -13376,26 +13385,6 @@ void _checkApiChecksums() { 6577) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } - if (uniffi_moq_ffi_checksum_method_moqmediacontainerconsumer_cancel() != - 27429) { - throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); - } - if (uniffi_moq_ffi_checksum_method_moqmediacontainerconsumer_next() != - 17029) { - throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); - } - if (uniffi_moq_ffi_checksum_method_moqmediacontainergroupconsumer_cancel() != - 8253) { - throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); - } - if (uniffi_moq_ffi_checksum_method_moqmediacontainergroupconsumer_next() != - 11213) { - throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); - } - if (uniffi_moq_ffi_checksum_method_moqmediacontainergroupconsumer_sequence() != - 15687) { - throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); - } if (uniffi_moq_ffi_checksum_method_moqmediacontainerproducer_cut() != 35477) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } @@ -13418,6 +13407,22 @@ void _checkApiChecksums() { 13648) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } + if (uniffi_moq_ffi_checksum_method_moqmediagroupconsumer_cancel() != 4852) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } + if (uniffi_moq_ffi_checksum_method_moqmediagroupconsumer_next() != 3173) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } + if (uniffi_moq_ffi_checksum_method_moqmediagroupconsumer_sequence() != + 57049) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } + if (uniffi_moq_ffi_checksum_method_moqmediatrackconsumer_cancel() != 17311) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } + if (uniffi_moq_ffi_checksum_method_moqmediatrackconsumer_next() != 44214) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } if (uniffi_moq_ffi_checksum_method_moqmediatrackproducer_cut() != 54122) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } @@ -13713,14 +13718,6 @@ void _checkApiChecksums() { 4684) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } - if (uniffi_moq_ffi_checksum_constructor_moqmediacontainerconsumer_subscribe() != - 56807) { - throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); - } - if (uniffi_moq_ffi_checksum_constructor_moqmediacontainergroupconsumer_fetch() != - 48900) { - throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); - } if (uniffi_moq_ffi_checksum_constructor_moqmediacontainerproducer_new() != 7951) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); @@ -13729,6 +13726,18 @@ void _checkApiChecksums() { 32133) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } + if (uniffi_moq_ffi_checksum_constructor_moqmediagroupconsumer_fetch() != + 43671) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } + if (uniffi_moq_ffi_checksum_constructor_moqmediagroupconsumer_new() != + 14424) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } + if (uniffi_moq_ffi_checksum_constructor_moqmediatrackconsumer_subscribe() != + 24751) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } if (uniffi_moq_ffi_checksum_constructor_moqmediatrackproducer_audio() != 8819) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); diff --git a/doc/lib/cpp/index.md b/doc/lib/cpp/index.md index a2ee4b0579f..b2eba24224e 100644 --- a/doc/lib/cpp/index.md +++ b/doc/lib/cpp/index.md @@ -100,12 +100,16 @@ if (auto catalog = ok(catalogs->next().get())) { // std::nullopt once the cata for (const auto &[name, video] : catalog->video) { // The track is the rendition's `track` when set, otherwise its catalog key. auto track = video.track.value_or(name); - auto media = ok(moq::MediaContainerConsumer::subscribe(broadcast, {track, video.container}).get()); + auto media = ok(moq::MediaTrackConsumer::subscribe(broadcast, {track, video.container}).get()); auto frame = ok(media->next().get()); // std::nullopt once the track ends } } ``` +`MediaTrackConsumer` reads in decode order. For arrival-order reads, decode each group +from `TrackConsumer::recv_group` with `moq::MediaGroupConsumer::init(group, container)`; +frames carry container media time as they land. + ```cpp // Publish encoded frames, or raw pixels with the codec inside the binding. // session is connected as above; opus_init, packet, and rgba come from your diff --git a/doc/lib/dart/index.md b/doc/lib/dart/index.md index 6e50c44da31..08090415cc8 100644 --- a/doc/lib/dart/index.md +++ b/doc/lib/dart/index.md @@ -17,8 +17,10 @@ Import `package:moq/media.dart` with a prefix for catalogs, encoded-media import container consumers. `TrackProducer.audio` / `.video` take a broadcast, an init record, and a `Named` or `Requested` target. `CatalogProducer(broadcast: ...)` owns catalog properties and sections; it holds the broadcast weakly and fails after the broadcast closes. -`CatalogConsumer.subscribe`, `ContainerConsumer.subscribe`, and `ContainerGroupConsumer.fetch` -construct the read side from a broadcast. +`media.CatalogConsumer.subscribe`, `media.TrackConsumer.subscribe`, and `media.GroupConsumer.fetch` +construct the read side from a broadcast. For arrival-order reads, decode each group from +`TrackConsumer.recvGroup()` with `media.GroupConsumer(group: ..., container: ...)`; frames +carry container media time as they land. ```bash dart pub add moq # or: flutter pub add moq @@ -70,7 +72,7 @@ await moq.close(); The rest of the [shared feature list](/lib/#what-every-binding-can-do) maps one to one; the API reference has the names. -- **No codecs.** Unlike the other bindings, the published Dart binaries carry no encoder or decoder. Already-encoded frames flow through `media.TrackProducer` and `media.ContainerConsumer`; encoding is up to `package:camera`, platform channels, or another codec package. +- **No codecs.** Unlike the other bindings, the published Dart binaries carry no encoder or decoder. Already-encoded frames flow through `media.TrackProducer` and `media.TrackConsumer`; encoding is up to `package:camera`, platform channels, or another codec package. - **Names.** Types drop the `Moq` prefix (`Session`, `BroadcastProducer`) as aliases, so the generated names still work. `Container`, `Route`, and the exceptions keep it, since Flutter owns those names. `dynamic` is reserved, so the origin method is `dynamic_`. - **Audio needs cuts.** Video groups at its keyframes, but audio forms a group only where you call `cut()`: after every frame, or at a segment cadence to align with video. - **Live encoder timing.** After writing a frame you encoded yourself, call `flush(timestampUs: ...)` with the same timestamp so the catalog advertises your jitter. Skip it for file and network imports. On a seek or pause, call `discontinuity()`, then keep timestamps moving forward and resume video on a keyframe. diff --git a/doc/lib/go/index.md b/doc/lib/go/index.md index a9540a3512b..92c4053dd8a 100644 --- a/doc/lib/go/index.md +++ b/doc/lib/go/index.md @@ -17,6 +17,9 @@ linux/amd64, linux/arm64, darwin/arm64 (macOS 12.3+), windows/amd64. Construct `NewAudioTrackProducer` / `NewVideoTrackProducer` from a broadcast, an init record, and a `Named` or `Requested` target. `NewCatalogProducer` holds its broadcast weakly; catalog writes fail with `ErrClosed` after the broadcast closes. Media timestamps use `time.Duration`. +For arrival-order reads, decode each group from `moq.TrackConsumer.RecvGroup` with +`media.NewGroupConsumer`; frames carry container media time as they land. `media.FetchGroup` +fetches and decodes one past group. ```bash go get moq.dev/moq@latest diff --git a/doc/lib/kt/index.md b/doc/lib/kt/index.md index 6edf5c36199..98cc4f5be7e 100644 --- a/doc/lib/kt/index.md +++ b/doc/lib/kt/index.md @@ -16,8 +16,11 @@ x86\_64) and desktop JVM (Linux x86\_64/aarch64, macOS arm64, Windows x64). `dev.moq.media` owns catalogs, encoded imports, and container consumers. `TrackProducer.audio` / `.video` take a broadcast, a `Named` or `Requested` target, and an init record. `CatalogProducer(broadcast)` holds its broadcast weakly and fails with `Closed` after it closes. -Construct readers with `CatalogConsumer.subscribe`, `ContainerConsumer.subscribe`, or -`ContainerGroupConsumer.fetch`. +Construct readers with `CatalogConsumer.subscribe`, `TrackConsumer.subscribe`, or +`GroupConsumer.fetch`. For arrival-order reads, decode each group from +`dev.moq.TrackConsumer.recvGroup()` with `dev.moq.media.GroupConsumer(group, container)`; frames +carry container media time as they land. Like `TrackProducer`, these names also exist at the +root for moq-net handles, so qualify them when both packages are star-imported. ```kotlin ignore dependencies { diff --git a/doc/lib/py/index.md b/doc/lib/py/index.md index 004293a236e..9a8e257ba1f 100644 --- a/doc/lib/py/index.md +++ b/doc/lib/py/index.md @@ -18,6 +18,8 @@ Use `Named(name)` or `Requested(request)` as the target for `TrackProducer.audio `CatalogProducer(broadcast)` holds the broadcast weakly and owns catalog properties and sections; its writes fail with `Error.Closed` after the broadcast closes or is released. `MediaFrame.timestamp`, media writes, and `flush` use `datetime.timedelta`. +For arrival-order reads, decode each group from `TrackConsumer.recv_group()` with +`media.GroupConsumer(group, container)`; frames carry container media time as they land. ```bash pip install moq-rs # or: uv add moq-rs @@ -40,7 +42,7 @@ async def main(): name, audio = next(iter(catalog.audio.items())) # The track is the rendition's `track` when set, otherwise its catalog key. track = audio.track if audio.track is not None else name - async for frame in await moq.media.ContainerConsumer.subscribe(broadcast, track, audio.container): + async for frame in await moq.media.TrackConsumer.subscribe(broadcast, track, audio.container): print(frame.timestamp, len(frame.payload)) asyncio.run(main()) diff --git a/doc/lib/swift/index.md b/doc/lib/swift/index.md index 2a8fffabd3b..beb0f08bcf0 100644 --- a/doc/lib/swift/index.md +++ b/doc/lib/swift/index.md @@ -17,6 +17,8 @@ slices for iOS 16+, the iOS Simulator, and macOS 12.3+. `.named(name:)` or `.requested(request)` target. `Media.CatalogProducer(broadcast:)` updates catalog properties and sections without keeping the broadcast open. Its writes fail after closing or releasing the broadcast. +For arrival-order reads, decode each group from `TrackConsumer.recvGroup()` with +`Media.GroupConsumer(group:container:)`; frames carry container media time as they land. ```swift ignore dependencies: [ diff --git a/doc/setup/upgrade.md b/doc/setup/upgrade.md index 99818f5879a..7513e46d881 100644 --- a/doc/setup/upgrade.md +++ b/doc/setup/upgrade.md @@ -112,11 +112,13 @@ These land with the next breaking release, not the 2026-09-23 train. Construct `MoqMediaTrackProducer::audio` / `video` (from the broadcast, a `MoqMediaTarget::Named` or `Requested` target, and the init record), `MoqMediaContainerProducer`, `MoqMediaCatalogProducer`, - `MoqMediaCatalogConsumer::subscribe`, `MoqMediaContainerConsumer::subscribe`, - or `MoqMediaContainerGroupConsumer::fetch` instead. Producers drop `name`, + `MoqMediaCatalogConsumer::subscribe`, `MoqMediaTrackConsumer::subscribe`, + or `MoqMediaGroupConsumer::fetch` instead. Producers drop `name`, `used`, and `unused`; read them through `demand()`. The wrappers expose these as `moq.media` in Python, `moq.dev/moq/media` in Go, `dev.moq.media` in - Kotlin, `package:moq/media.dart` in Dart, and `Media` in Swift. + Kotlin, `package:moq/media.dart` in Dart, and `Media` in Swift. Go fetches a + group with `media.FetchGroup`; its `NewGroupConsumer` decodes + a raw group instead. - **moq-ffi error payloads are named.** `MoqError` cases that carried a positional string (`Transport`, `Internal`, `Connect`, ...) name it `reason` (`Reason` in Go, the `reason:` label in Swift). The Display text is diff --git a/go/wrapper/README.md b/go/wrapper/README.md index 8e70ffb5368..575da3c7180 100644 --- a/go/wrapper/README.md +++ b/go/wrapper/README.md @@ -81,9 +81,10 @@ target takes over a subscriber's pending request. A duplicate name fails. `VideoInit.Hint` seeds fields the bitstream cannot supply. `NewContainerProducer` / `NewContainerStreamProducer` demux whole chunks or a byte stream into their own tracks. `NewCatalogProducer` holds its broadcast weakly and owns catalog properties and sections. -`NewCatalogConsumer`, `NewContainerConsumer`, and `NewContainerGroupConsumer` read -catalog snapshots, live media, and one fetched media group. Frames and `Flush` use -`time.Duration`; `Demand()` owns track names and subscriber waits. +`NewCatalogConsumer`, `NewTrackConsumer`, and `FetchGroup` read +catalog snapshots, live media, and one fetched media group. `NewGroupConsumer` +decodes a raw group, such as one from `moq.TrackConsumer.RecvGroup` for arrival-order reads. +Frames and `Flush` use `time.Duration`; `Demand()` owns track names and subscriber waits. `BroadcastProducer.Dynamic()` accepts subscriber-requested tracks. Call `TrackRequest.Accept()` for raw tracks, or pass the request as `media.Requested` diff --git a/go/wrapper/bridge.go b/go/wrapper/bridge.go index 525733cef8e..eb16247a280 100644 --- a/go/wrapper/bridge.go +++ b/go/wrapper/bridge.go @@ -19,5 +19,6 @@ func init() { bridge.BroadcastProducer = func(v any) *ffi.MoqBroadcastProducer { return v.(*BroadcastProducer).inner } bridge.TrackProducer = func(v any) *ffi.MoqTrackProducer { return v.(*TrackProducer).inner } bridge.TrackConsumer = func(v any) *ffi.MoqTrackConsumer { return v.(*TrackConsumer).inner } + bridge.GroupConsumer = func(v any) *ffi.MoqGroupConsumer { return v.(*GroupConsumer).inner } bridge.TrackDemand = func(inner *ffi.MoqTrackDemand) any { return &TrackDemand{inner: inner} } } diff --git a/go/wrapper/internal/bridge/handles.go b/go/wrapper/internal/bridge/handles.go index b37eaeeeec7..b5de8b5eee1 100644 --- a/go/wrapper/internal/bridge/handles.go +++ b/go/wrapper/internal/bridge/handles.go @@ -17,6 +17,8 @@ var ( TrackProducer func(any) *ffi.MoqTrackProducer // TrackConsumer returns the handle behind a *moq.TrackConsumer. TrackConsumer func(any) *ffi.MoqTrackConsumer + // GroupConsumer returns the handle behind a *moq.GroupConsumer. + GroupConsumer func(any) *ffi.MoqGroupConsumer // TrackDemand wraps a handle as a *moq.TrackDemand. TrackDemand func(*ffi.MoqTrackDemand) any ) diff --git a/go/wrapper/media/media.go b/go/wrapper/media/media.go index f0cb6fdc416..12eb5cc3405 100644 --- a/go/wrapper/media/media.go +++ b/go/wrapper/media/media.go @@ -185,41 +185,55 @@ func NewCatalogConsumer(ctx context.Context, broadcast *moq.BroadcastConsumer) ( return &CatalogConsumer{inner: inner}, nil } -// ContainerConfig names a media track, its container, and live delivery options. -type ContainerConfig struct { +// TrackConfig names a media track, its container, and live delivery options. +type TrackConfig struct { Name string Container Container Subscription *moq.Subscription } -// NewContainerConsumer subscribes to and decodes a media track. -func NewContainerConsumer(ctx context.Context, broadcast *moq.BroadcastConsumer, config ContainerConfig) (*ContainerConsumer, error) { +// NewTrackConsumer subscribes to and decodes a media track. +func NewTrackConsumer(ctx context.Context, broadcast *moq.BroadcastConsumer, config TrackConfig) (*TrackConsumer, error) { sub, err := bridge.Subscription(config.Subscription) if err != nil { return nil, err } - inner, err := ffi.MoqMediaContainerConsumerSubscribe(ctx, bridge.BroadcastConsumer(broadcast), ffi.MoqMediaContainerConfig{Name: config.Name, Container: config.Container, Subscription: sub}) + inner, err := ffi.MoqMediaTrackConsumerSubscribe(ctx, bridge.BroadcastConsumer(broadcast), ffi.MoqMediaTrackConfig{Name: config.Name, Container: config.Container, Subscription: sub}) if err != nil { return nil, err } - return &ContainerConsumer{inner: inner}, nil + return &TrackConsumer{inner: inner}, nil } -// ContainerGroupConfig names a group, its container, and fetch delivery options. -type ContainerGroupConfig struct { +// GroupConfig names a group, its container, and fetch delivery options. +type GroupConfig struct { Name string Sequence uint64 Container Container Options *moq.FetchGroupOptions } -// NewContainerGroupConsumer fetches and decodes exactly one media group. -func NewContainerGroupConsumer(ctx context.Context, broadcast *moq.BroadcastConsumer, config ContainerGroupConfig) (*ContainerGroupConsumer, error) { - inner, err := ffi.MoqMediaContainerGroupConsumerFetch(ctx, bridge.BroadcastConsumer(broadcast), ffi.MoqMediaContainerGroupConfig{Name: config.Name, Sequence: config.Sequence, Container: config.Container, Options: config.Options}) +// FetchGroup fetches and decodes exactly one media group. +func FetchGroup(ctx context.Context, broadcast *moq.BroadcastConsumer, config GroupConfig) (*GroupConsumer, error) { + inner, err := ffi.MoqMediaGroupConsumerFetch(ctx, bridge.BroadcastConsumer(broadcast), ffi.MoqMediaGroupConfig{Name: config.Name, Sequence: config.Sequence, Container: config.Container, Options: config.Options}) if err != nil { return nil, err } - return &ContainerGroupConsumer{inner: inner}, nil + return &GroupConsumer{inner: inner}, nil +} + +// NewGroupConsumer decodes a raw group with the track's container, as each frame lands. +// +// 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 +// 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) + if err != nil { + return nil, err + } + return &GroupConsumer{inner: inner}, nil } // LegacyContainer selects the legacy hang container for a media subscription. @@ -414,13 +428,13 @@ func (m *TrackStreamProducer) Finish() error { return m.inner.Finish() } -// ContainerConsumer is a stream of decoded media frames. -type ContainerConsumer struct { - inner *ffi.MoqMediaContainerConsumer +// TrackConsumer is a stream of decoded media frames. +type TrackConsumer struct { + inner *ffi.MoqMediaTrackConsumer } // Next returns the next frame, or (nil, nil) when the track ends. -func (m *ContainerConsumer) Next(ctx context.Context) (*MediaFrame, error) { +func (m *TrackConsumer) Next(ctx context.Context) (*MediaFrame, error) { frame, err := bridge.Call(ctx, m.inner.Cancel, m.inner.Next) if frame == nil || err != nil { return nil, err @@ -429,28 +443,28 @@ func (m *ContainerConsumer) Next(ctx context.Context) (*MediaFrame, error) { } // All ranges over frames until the track ends or the loop breaks. -func (m *ContainerConsumer) All(ctx context.Context) iter.Seq2[*MediaFrame, error] { +func (m *TrackConsumer) All(ctx context.Context) iter.Seq2[*MediaFrame, error] { return bridge.Seq(ctx, m.Next) } // Cancel stops the stream. -func (m *ContainerConsumer) Cancel() { +func (m *TrackConsumer) Cancel() { m.inner.Cancel() } -// ContainerGroupConsumer is a finite stream of decoded media frames from one fetched -// group. It ends after the group's last frame. -type ContainerGroupConsumer struct { - inner *ffi.MoqMediaContainerGroupConsumer +// GroupConsumer is a finite stream of decoded media frames from one group, +// fetched or decoded from a raw group. It ends after the group's last frame. +type GroupConsumer struct { + inner *ffi.MoqMediaGroupConsumer } // Sequence is this group's sequence number within the track. -func (m *ContainerGroupConsumer) Sequence() uint64 { +func (m *GroupConsumer) Sequence() uint64 { return m.inner.Sequence() } // Next returns the next decoded frame, or (nil, nil) when the group ends. -func (m *ContainerGroupConsumer) Next(ctx context.Context) (*MediaFrame, error) { +func (m *GroupConsumer) Next(ctx context.Context) (*MediaFrame, error) { frame, err := bridge.Call(ctx, m.inner.Cancel, m.inner.Next) if frame == nil || err != nil { return nil, err @@ -459,12 +473,12 @@ func (m *ContainerGroupConsumer) Next(ctx context.Context) (*MediaFrame, error) } // All ranges over decoded frames until the group ends or the loop breaks. -func (m *ContainerGroupConsumer) All(ctx context.Context) iter.Seq2[*MediaFrame, error] { +func (m *GroupConsumer) All(ctx context.Context) iter.Seq2[*MediaFrame, error] { return bridge.Seq(ctx, m.Next) } // Cancel stops the stream. -func (m *ContainerGroupConsumer) Cancel() { +func (m *GroupConsumer) Cancel() { m.inner.Cancel() } diff --git a/go/wrapper/moq_test.go b/go/wrapper/moq_test.go index 2be3a9de5a2..594f511e27d 100644 --- a/go/wrapper/moq_test.go +++ b/go/wrapper/moq_test.go @@ -544,7 +544,7 @@ func TestLocalPublishConsumeAudio(t *testing.T) { t.Fatalf("audio = %+v, want opus/48000/2", audio) } - mediaConsumer, err := moqmedia.NewContainerConsumer(ctx, bc, moqmedia.ContainerConfig{Name: trackName, Container: audio.Container, Subscription: nil}) + mediaConsumer, err := moqmedia.NewTrackConsumer(ctx, bc, moqmedia.TrackConfig{Name: trackName, Container: audio.Container, Subscription: nil}) if err != nil { t.Fatal(err) } @@ -565,6 +565,32 @@ func TestLocalPublishConsumeAudio(t *testing.T) { if string(frame.Payload) != string(payload) || uint64(frame.Timestamp.Microseconds()) != 1_000_000 { t.Fatalf("frame = %+v, want payload=%q ts=1000000", frame, payload) } + + // The same group, received raw in arrival order and decoded with the container. + track, err := bc.SubscribeTrack(ctx, trackName, nil) + if err != nil { + t.Fatal(err) + } + defer track.Cancel() + group, err := track.RecvGroup(ctx) + if err != nil { + t.Fatal(err) + } + if group == nil { + t.Fatal("expected a group") + } + decoded, err := moqmedia.NewGroupConsumer(group, audio.Container) + if err != nil { + t.Fatal(err) + } + defer decoded.Cancel() + frame, err = decoded.Next(ctx) + if err != nil { + t.Fatal(err) + } + if frame == nil || string(frame.Payload) != string(payload) || uint64(frame.Timestamp.Microseconds()) != 1_000_000 { + t.Fatalf("decoded frame = %+v, want payload=%q ts=1000000", frame, payload) + } } func TestTrackPublishConsume(t *testing.T) { @@ -800,12 +826,12 @@ func TestDynamicTrackRequestCanPublishAudio(t *testing.T) { } type subscribeResult struct { - media *moqmedia.ContainerConsumer + media *moqmedia.TrackConsumer err error } subscribe := make(chan subscribeResult, 1) go func() { - media, err := moqmedia.NewContainerConsumer(ctx, consumer, moqmedia.ContainerConfig{Name: "requested-audio", Container: moqmedia.LegacyContainer(), Subscription: nil}) + media, err := moqmedia.NewTrackConsumer(ctx, consumer, moqmedia.TrackConfig{Name: "requested-audio", Container: moqmedia.LegacyContainer(), Subscription: nil}) subscribe <- subscribeResult{media: media, err: err} }() @@ -837,7 +863,7 @@ func TestDynamicTrackRequestCanPublishAudio(t *testing.T) { t.Fatalf("request name after accept error = %v, want ErrClosed", err) } - var mediaConsumer *moqmedia.ContainerConsumer + var mediaConsumer *moqmedia.TrackConsumer select { case res := <-subscribe: if res.err != nil { diff --git a/go/wrapper/subscribe.go b/go/wrapper/subscribe.go index 1de453ed239..e4707880671 100644 --- a/go/wrapper/subscribe.go +++ b/go/wrapper/subscribe.go @@ -49,8 +49,8 @@ func (b *BroadcastConsumer) FetchGroup( // Resolve returns the broadcast serving a catalog rendition, honoring its broadcast // reference: ffi.MoqVideo.Broadcast / ffi.MoqAudio.Broadcast. A nil or empty reference names this // broadcast, anything else names a sibling relative to it (e.g. "./source"). Call it on a -// rendition that carries one before media.NewContainerConsumer, SubscribeTrack, FetchGroup, or -// media.NewContainerGroupConsumer, which take a track name rather than a rendition: the +// rendition that carries one before media.NewTrackConsumer, SubscribeTrack, FetchGroup, or +// media.FetchGroup, which take a track name rather than a rendition: the // rendition's Track, or its catalog key when nil. DecodeAudio and DecodeVideo resolve both themselves. // // Errors if this broadcast came from a local producer rather than an origin, since a diff --git a/kt/README.md b/kt/README.md index 65eca11cb0c..e88b3e7dd32 100644 --- a/kt/README.md +++ b/kt/README.md @@ -57,7 +57,7 @@ The `dev.moq` package is intentionally thin: Kotlin has extension functions, so - **Typealiases** (`Aliases.kt`): re-export the `Moq*`-prefixed FFI types under clean `dev.moq` names (`OriginProducer`, `BroadcastConsumer`, `Frame`, ...), so you import `dev.moq.*` only. `MoqException` is not aliased because Kotlin can't resolve its subtypes through a typealias; use `uniffi.moq.*` for it. - **Flow extensions** (`Flows.kt`): `updates()`, `groups()`, and `frames()` turn the pull-based consumers into coroutine `Flow`s with cancellation wired through. - **Media** (`dev.moq.media`): catalog, importers, and container consumers, with `catalog(broadcast)` for a one-shot catalog read and `frames()` for container-decoded media. -- **Fetched media**: `ContainerGroupConsumer.fetch(...).frames()` streams the decoded frames of one retained group, then completes. +- **Fetched media**: `dev.moq.media.GroupConsumer.fetch(...).frames()` streams the decoded frames of one retained group, then completes. `dev.moq.media.GroupConsumer(group, container)` decodes a raw group instead, such as one from `recvGroup()` for arrival-order reads. - **Duration extensions** (`Durations.kt`): the FFI carries microseconds as integers, so `stats.rtt`, `backoff.initial`, `frame.timestamp`, and their siblings read back as a `kotlin.time.Duration`. - **`logLevel(...)`**: configures native Rust tracing without importing the raw bindings package. - **Raw datagrams**: `TrackProducer.appendDatagram(Frame(payload, timestampUs))` sends one best-effort frame and returns its sequence; `TrackConsumer.recvDatagram()` and `datagrams()` receive them. Payloads are capped at 1200 bytes, require a datagram-capable transport plus lite-05 or newer moq-lite or moq-transport, and have no stream fallback. 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 9f393d19721..e2977cf0721 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/media/Media.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/media/Media.kt @@ -22,9 +22,9 @@ typealias ContainerProducer = uniffi.moq.MoqMediaContainerProducer /** The write side of a container fed a raw byte stream. */ typealias ContainerStreamProducer = uniffi.moq.MoqMediaContainerStreamProducer /** The read side of a media track: yields frames with codec metadata in decode order. */ -typealias ContainerConsumer = uniffi.moq.MoqMediaContainerConsumer +typealias TrackConsumer = uniffi.moq.MoqMediaTrackConsumer /** A finite fetched media group: yields container-decoded frames until the group ends. */ -typealias ContainerGroupConsumer = uniffi.moq.MoqMediaContainerGroupConsumer +typealias GroupConsumer = uniffi.moq.MoqMediaGroupConsumer /** The read side of a broadcast's catalog: yields updates as the set of tracks changes. */ typealias CatalogConsumer = uniffi.moq.MoqMediaCatalogConsumer /** A broadcast's catalog: its tracks and their properties, plus any application sections. */ @@ -63,9 +63,9 @@ typealias Named = uniffi.moq.MoqMediaTarget.Named /** A subscriber-requested track. */ typealias Requested = uniffi.moq.MoqMediaTarget.Requested /** A track name, container, and live delivery options. */ -typealias ContainerConfig = uniffi.moq.MoqMediaContainerConfig +typealias TrackConfig = uniffi.moq.MoqMediaTrackConfig /** A fetched group name, sequence, container, and delivery options. */ -typealias ContainerGroupConfig = uniffi.moq.MoqMediaContainerGroupConfig +typealias GroupConfig = uniffi.moq.MoqMediaGroupConfig /** The packaging advertised for a media track. */ typealias Container = uniffi.moq.MoqContainer @@ -100,7 +100,7 @@ suspend fun catalog(broadcast: dev.moq.BroadcastConsumer): Catalog { } /** Stream of decoded media frames in decode order. */ -fun ContainerConsumer.frames(): Flow = flow { +fun TrackConsumer.frames(): Flow = flow { while (true) { currentCoroutineContext().ensureActive() emit(next() ?: break) @@ -109,8 +109,8 @@ fun ContainerConsumer.frames(): Flow = flow { if (cause is CancellationException) cancel() } -/** Stream of decoded frames from one finite, fetched media group. */ -fun ContainerGroupConsumer.frames(): Flow = flow { +/** Stream of decoded frames from one finite media group. */ +fun GroupConsumer.frames(): Flow = flow { while (true) { currentCoroutineContext().ensureActive() emit(next() ?: break) diff --git a/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt b/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt index 783d555331b..05c85f66047 100644 --- a/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt +++ b/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt @@ -245,7 +245,7 @@ class SmokeTest { // Fetch while the track is still published: finishing the media producer // unpublishes it, and the fetch would then miss with NotFound. - val fetched: ContainerGroupConsumer = ContainerGroupConsumer.fetch(consumer, ContainerGroupConfig(name = name, sequence = 0uL, container = audio.container, options = FetchGroupOptions(priority = 3u))) + val fetched = dev.moq.media.GroupConsumer.fetch(consumer, dev.moq.media.GroupConfig(name = name, sequence = 0uL, container = audio.container, options = FetchGroupOptions(priority = 3u))) // Close the group so the fetched stream terminates instead of waiting for more. media.finish() diff --git a/py/moq-rs/README.md b/py/moq-rs/README.md index 5364a2e4538..dec2b5364e0 100644 --- a/py/moq-rs/README.md +++ b/py/moq-rs/README.md @@ -39,7 +39,7 @@ async def main(): for name, audio in catalog.audio.items(): # The track is the rendition's `track` when set, otherwise its catalog key. track = audio.track if audio.track is not None else name - frames = await moq.media.ContainerConsumer.subscribe(broadcast, track, audio.container) + frames = await moq.media.TrackConsumer.subscribe(broadcast, track, audio.container) async with frames: async for frame in frames: print(f"Got frame: {len(frame.payload)} bytes, ts={frame.timestamp}") @@ -190,10 +190,13 @@ client = moq.Client( or releasing the broadcast. - `await media.CatalogConsumer.subscribe(broadcast)` streams catalog snapshots; `await media.catalog(broadcast)` reads one snapshot and releases its subscription. -- `await media.ContainerConsumer.subscribe(broadcast, name, container, *, subscription=None)` +- `await media.TrackConsumer.subscribe(broadcast, name, container, *, subscription=None)` decodes live media. Pass the catalog rendition's `track` (or its key when unset) and `container`. -- `await media.ContainerGroupConsumer.fetch(broadcast, name, sequence, container, *, options=None)` +- `await media.GroupConsumer.fetch(broadcast, name, sequence, container, *, options=None)` fetches and decodes exactly one group. +- `media.GroupConsumer(group, container)` decodes a raw group, such as one from + `TrackConsumer.recv_group()` for arrival-order reads, with container media time on each frame + as it lands. `media.TrackProducer.write_frame(payload, timestamp=timedelta(0))` and `.flush(timestamp)` use `timedelta`; returned `media.MediaFrame.timestamp` does too. `demand()` owns the track name and @@ -214,7 +217,7 @@ subscriber waits. `.cut()` / `.seek(sequence)` draw group boundaries; `.finish() - **`GroupConsumer`**. Async iterator of timestamped `Frame`s. - `.read_frame() -> Frame | None` returns a timestamped raw frame. -Every handle whose cleanup is `cancel()` is an async context manager, so exiting `async with` releases it: the consumers (`media.CatalogConsumer`, `media.ContainerConsumer`, `media.ContainerGroupConsumer`, `TrackConsumer`, `AudioConsumer`, `GroupConsumer`, `json.SnapshotConsumer`, `json.StreamConsumer`, `AnnounceConsumer`, `AnnouncedBroadcast`) and the dynamic sources (`OriginDynamic`, `BroadcastDynamic`, `TrackDynamic`). +Every handle whose cleanup is `cancel()` is an async context manager, so exiting `async with` releases it: the consumers (`media.CatalogConsumer`, `media.TrackConsumer`, `media.GroupConsumer`, `TrackConsumer`, `AudioConsumer`, `GroupConsumer`, `json.SnapshotConsumer`, `json.StreamConsumer`, `AnnounceConsumer`, `AnnouncedBroadcast`) and the dynamic sources (`OriginDynamic`, `BroadcastDynamic`, `TrackDynamic`). ### Origin (advanced) diff --git a/py/moq-rs/docs/index.md b/py/moq-rs/docs/index.md index f2f28724214..7837a93e066 100644 --- a/py/moq-rs/docs/index.md +++ b/py/moq-rs/docs/index.md @@ -98,8 +98,8 @@ Every importer takes a broadcast; single-track imports choose a `Named` or `Requ TrackStreamProducer ContainerProducer ContainerStreamProducer - ContainerConsumer - ContainerGroupConsumer + TrackConsumer + GroupConsumer Named Requested MediaFrame diff --git a/py/moq-rs/moq/media.py b/py/moq-rs/moq/media.py index c5420aacc58..1a6038a6994 100644 --- a/py/moq-rs/moq/media.py +++ b/py/moq-rs/moq/media.py @@ -29,14 +29,14 @@ MoqError, MoqMediaCatalogConsumer, MoqMediaCatalogProducer, - MoqMediaContainerConfig, - MoqMediaContainerConsumer, - MoqMediaContainerGroupConfig, - MoqMediaContainerGroupConsumer, MoqMediaContainerProducer, MoqMediaContainerStreamProducer, MoqMediaFrame, + MoqMediaGroupConfig, + MoqMediaGroupConsumer, MoqMediaTarget, + MoqMediaTrackConfig, + MoqMediaTrackConsumer, MoqMediaTrackProducer, MoqMediaTrackStreamProducer, ) @@ -50,6 +50,7 @@ MoqVideoProperties as VideoProperties, ) +from . import subscribe as _subscribe from ._records import Frame, Subscription, _from_us, _subscription, _to_us from .publish import BroadcastProducer, TrackDemand, TrackRequest from .subscribe import BroadcastConsumer @@ -254,7 +255,7 @@ def finish(self) -> None: self._inner.finish() -class ContainerConsumer: +class TrackConsumer: """Async-iterable stream of decoded :class:`MediaFrame` in decode order. Construct with :meth:`subscribe`. Iterate with ``async for``; @@ -264,12 +265,12 @@ class ContainerConsumer: @classmethod async def subscribe( cls, broadcast: BroadcastConsumer, name: str, container: Container, *, subscription: Subscription | None = None - ) -> ContainerConsumer: + ) -> TrackConsumer: """Subscribe to a media track and decode its container.""" - config = MoqMediaContainerConfig(name=name, container=container, subscription=_subscription(subscription)) - return cls(await MoqMediaContainerConsumer.subscribe(broadcast._inner, config)) + config = MoqMediaTrackConfig(name=name, container=container, subscription=_subscription(subscription)) + return cls(await MoqMediaTrackConsumer.subscribe(broadcast._inner, config)) - def __init__(self, inner: MoqMediaContainerConsumer) -> None: + def __init__(self, inner: MoqMediaTrackConsumer) -> None: self._inner = inner async def __aenter__(self): @@ -292,14 +293,32 @@ def cancel(self) -> None: self._inner.cancel() -class ContainerGroupConsumer: - """Async iterator of decoded :class:`MediaFrame` within a single fetched group. +class GroupConsumer: + """Async iterator of decoded :class:`MediaFrame` within a single group. - Construct with :meth:`fetch`. Finite: iteration ends - after the group's last frame. Usable as an async context manager that cancels - on exit. + Construct with :meth:`fetch`, or decode a raw :class:`~moq.GroupConsumer`, + such as one from :meth:`~moq.TrackConsumer.recv_group` for arrival-order + reads. Finite: iteration ends after the group's last frame. Usable as an + async context manager that cancels on exit. """ + def __init__(self, group: _subscribe.GroupConsumer, container: Container) -> None: + """Decode ``group`` with the track's container, as each frame lands. + + 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; raises ``Error.AlreadyCommitted`` if it has already + read a frame. + """ + self._inner = MoqMediaGroupConsumer(group._inner, container) + + @classmethod + def _from_inner(cls, inner: MoqMediaGroupConsumer) -> GroupConsumer: + self = cls.__new__(cls) + self._inner = inner + return self + @classmethod async def fetch( cls, @@ -309,13 +328,10 @@ async def fetch( container: Container, *, options: FetchGroupOptions | None = None, - ) -> ContainerGroupConsumer: + ) -> GroupConsumer: """Fetch and decode exactly one media group.""" - config = MoqMediaContainerGroupConfig(name=name, sequence=sequence, container=container, options=options) - return cls(await MoqMediaContainerGroupConsumer.fetch(broadcast._inner, config)) - - def __init__(self, inner: MoqMediaContainerGroupConsumer) -> None: - self._inner = inner + config = MoqMediaGroupConfig(name=name, sequence=sequence, container=container, options=options) + return cls._from_inner(await MoqMediaGroupConsumer.fetch(broadcast._inner, config)) @property def sequence(self) -> int: @@ -399,8 +415,8 @@ async def catalog(broadcast: BroadcastConsumer) -> Catalog: "TrackStreamProducer", "ContainerProducer", "ContainerStreamProducer", - "ContainerConsumer", - "ContainerGroupConsumer", + "TrackConsumer", + "GroupConsumer", "Audio", "Video", "Catalog", diff --git a/py/moq-rs/moq/subscribe.py b/py/moq-rs/moq/subscribe.py index 409e0e015f5..a5000fe6f1e 100644 --- a/py/moq-rs/moq/subscribe.py +++ b/py/moq-rs/moq/subscribe.py @@ -266,8 +266,8 @@ async def resolve(self, reference: str | None) -> "BroadcastConsumer": ``reference`` is ``catalog.video[name].broadcast`` / ``catalog.audio[name].broadcast``: ``None`` or empty names this broadcast, anything else names a sibling relative to it (e.g. ``./source``). Call it on - a rendition that carries one before :meth:`~moq.media.ContainerConsumer.subscribe`, - :meth:`subscribe_track`, :meth:`fetch_group`, or :meth:`~moq.media.ContainerGroupConsumer.fetch`, + a rendition that carries one before :meth:`~moq.media.TrackConsumer.subscribe`, + :meth:`subscribe_track`, :meth:`fetch_group`, or :meth:`~moq.media.GroupConsumer.fetch`, which take a track name rather than a rendition: the rendition's ``track``, or its catalog key when unset. :meth:`decode_audio` and :meth:`decode_video` resolve both themselves. diff --git a/py/moq-rs/tests/test_local.py b/py/moq-rs/tests/test_local.py index 0fc98a27879..5873e6bbf37 100644 --- a/py/moq-rs/tests/test_local.py +++ b/py/moq-rs/tests/test_local.py @@ -222,7 +222,7 @@ async def test_local_publish_consume_audio(): assert audio.sample_rate == 48000 assert audio.channel_count == 2 - media_consumer = await moq.media.ContainerConsumer.subscribe(broadcast_consumer, track_name, audio.container) + media_consumer = await moq.media.TrackConsumer.subscribe(broadcast_consumer, track_name, audio.container) payload = b"opus audio payload data" media.write_frame(payload, timedelta(microseconds=1_000_000)) @@ -258,7 +258,7 @@ async def test_video_publish_consume(): assert video.coded.width == 1280 assert video.coded.height == 720 - media_consumer = await moq.media.ContainerConsumer.subscribe(broadcast_consumer, track_name, video.container) + media_consumer = await moq.media.TrackConsumer.subscribe(broadcast_consumer, track_name, video.container) keyframe = bytes([0x00, 0x00, 0x00, 0x01, 0x65, 0xAA, 0xBB, 0xCC]) media.write_frame(keyframe, timedelta(microseconds=0)) @@ -304,7 +304,7 @@ async def test_multiple_frames_ordering(): catalog = await moq.media.catalog(broadcast_consumer) track_name = list(catalog.audio.keys())[0] audio = catalog.audio[track_name] - media_consumer = await moq.media.ContainerConsumer.subscribe(broadcast_consumer, track_name, audio.container) + media_consumer = await moq.media.TrackConsumer.subscribe(broadcast_consumer, track_name, audio.container) timestamps = [0, 20_000, 40_000, 60_000, 80_000] for i, ts in enumerate(timestamps): @@ -567,9 +567,9 @@ async def test_dynamic_track_request_can_publish_media(): catalog_consumer = await moq.media.CatalogConsumer.subscribe(consumer) # TrackProducer.audio with Requested accepts the request (at the media timescale), which is what - # unblocks ContainerConsumer.subscribe, so run the subscribe concurrently until then. + # unblocks TrackConsumer.subscribe, so run the subscribe concurrently until then. subscribe = asyncio.create_task( - moq.media.ContainerConsumer.subscribe( + moq.media.TrackConsumer.subscribe( consumer, "requested-audio", cast(moq.media.Container, moq.media.Container.LEGACY()) ) ) @@ -776,7 +776,7 @@ def test_public_api_exports(): async def test_container_consumer_default_latency_and_context_manager(): - """ContainerConsumer.subscribe defaults the + """TrackConsumer.subscribe defaults the latency; the returned consumer is also an async context manager.""" origin = moq.OriginProducer() broadcast = create_announced(origin, "live") @@ -795,7 +795,7 @@ async def test_container_consumer_default_latency_and_context_manager(): payload = b"opus audio payload data" media.write_frame(payload, timedelta(microseconds=1_000_000)) - async with await moq.media.ContainerConsumer.subscribe( + async with await moq.media.TrackConsumer.subscribe( broadcast_consumer, track_name, audio.container ) as media_consumer: async for frame in media_consumer: @@ -979,6 +979,98 @@ async def _take(iterator, count: int): return +def _legacy(timestamp_us: int, payload: bytes) -> bytes: + """Encode a hang legacy frame: a 4-byte QUIC varint µs timestamp, then the payload.""" + assert timestamp_us < 1 << 30 + return (timestamp_us | 0x8000_0000).to_bytes(4, "big") + payload + + +_LEGACY = cast(moq.media.Container, moq.media.Container.LEGACY()) + + +async def _recv_raw(track: moq.TrackConsumer) -> moq.GroupConsumer: + group = await asyncio.wait_for(track.recv_group(), timeout=5.0) + assert group is not None + return group + + +async def _recv_media(track: moq.TrackConsumer) -> moq.media.GroupConsumer: + return moq.media.GroupConsumer(await _recv_raw(track), _LEGACY) + + +async def test_container_group_decodes_arrival_order(): + """Raw groups received in arrival order decode as their frames land: a + stalled group is neither skipped nor blocks a newer one, and each timestamp + is the container's media time, not the transport's.""" + broadcast = moq.BroadcastProducer() + track = broadcast.publish_track("audio") + consumer = await broadcast.consume().subscribe_track("audio") + + first = track.append_group() + first.write_frame(_legacy(1_000_000, b"a"), timedelta(milliseconds=1)) + first_media = await _recv_media(consumer) + frame = await asyncio.wait_for(anext(first_media), timeout=5.0) + assert (frame.payload, frame.timestamp, frame.keyframe) == (b"a", timedelta(seconds=1), True) + + # A newer group lands while the first is stalled mid-way. + second = track.append_group() + second.write_frame(_legacy(1_040_000, b"c"), timedelta(milliseconds=2)) + second_media = await _recv_media(consumer) + assert second_media.sequence == 1 + frame = await asyncio.wait_for(anext(second_media), timeout=5.0) + assert (frame.payload, frame.timestamp) == (b"c", timedelta(microseconds=1_040_000)) + + # The stalled group resumes and is still delivered. + first.write_frame(_legacy(1_020_000, b"b"), timedelta(milliseconds=3)) + frame = await asyncio.wait_for(anext(first_media), timeout=5.0) + assert (frame.payload, frame.timestamp, frame.keyframe) == (b"b", timedelta(microseconds=1_020_000), False) + + first.finish() + second.finish() + assert [frame async for frame in first_media] == [] + assert [frame async for frame in second_media] == [] + + +async def test_container_group_takes_over_an_unread_group(): + """Decoding refuses a raw group that already read a frame, and closes the one it takes.""" + broadcast = moq.BroadcastProducer() + track = broadcast.publish_track("audio") + consumer = await broadcast.consume().subscribe_track("audio") + + group = track.append_group() + group.write_frame(_legacy(1_000_000, b"a"), timedelta(milliseconds=1)) + raw = await _recv_raw(consumer) + assert await raw.read_frame() is not None + with pytest.raises(moq.Error.AlreadyCommitted): # type: ignore[attr-defined] + moq.media.GroupConsumer(raw, _LEGACY) + + group = track.append_group() + group.write_frame(_legacy(1_020_000, b"b"), timedelta(milliseconds=2)) + raw = await _recv_raw(consumer) + media = moq.media.GroupConsumer(raw, _LEGACY) + with pytest.raises(moq.Error.Cancelled): # type: ignore[attr-defined] + await raw.read_frame() + frame = await asyncio.wait_for(anext(media), timeout=5.0) + assert frame.timestamp == timedelta(microseconds=1_020_000) + + +async def test_container_group_fails_loudly(): + """An unconfigurable container fails before taking the group; an undecodable frame fails the read.""" + broadcast = moq.BroadcastProducer() + track = broadcast.publish_track("audio") + consumer = await broadcast.consume().subscribe_track("audio") + + group = track.append_group() + # A truncated varint timestamp: the 2-byte form with only its first byte. + group.write_frame(b"\x40", timedelta(milliseconds=1)) + raw = await _recv_raw(consumer) + with pytest.raises(moq.Error.Codec): # type: ignore[attr-defined] + moq.media.GroupConsumer(raw, cast(moq.media.Container, moq.media.Container.CMAF(init=b""))) + media = moq.media.GroupConsumer(raw, _LEGACY) + with pytest.raises(moq.Error): # type: ignore[attr-defined] + await asyncio.wait_for(anext(media), timeout=5.0) + + async def test_raw_multi_frame_group(): """A single group can carry multiple frames, not just one per group.""" origin = moq.OriginProducer() diff --git a/py/moq-rs/tests/test_server.py b/py/moq-rs/tests/test_server.py index 37d6df0feca..90d877bead8 100644 --- a/py/moq-rs/tests/test_server.py +++ b/py/moq-rs/tests/test_server.py @@ -68,7 +68,7 @@ async def accept_loop() -> None: track_name, audio = next(iter(catalog.audio.items())) assert audio.codec == "opus" - media_consumer = await moq.media.ContainerConsumer.subscribe( + media_consumer = await moq.media.TrackConsumer.subscribe( broadcast_consumer, track_name, audio.container ) diff --git a/quest/m1/ffi-crate-split.md b/quest/m1/ffi-crate-split.md index a197bfbff42..442b7a748d9 100644 --- a/quest/m1/ffi-crate-split.md +++ b/quest/m1/ffi-crate-split.md @@ -5,7 +5,7 @@ The generated bindings carry the layers themselves. moq-ffi becomes one UniFFI crate per layer (`net`, `media`, `json`, `flate`, `audio`, `video`), all linked into the one moq-ffi library, so a generated type is -`media.ContainerGroupConsumer` instead of `MoqMediaContainerGroupConsumer`. +`media.GroupConsumer` instead of `MoqMediaGroupConsumer`. The separate feasibility spike decides whether all generators can carry this shape. A no-go removes this migration from the release and retains wrapper namespaces; do not hold the release for an infeasible generator. @@ -15,11 +15,13 @@ Out of scope: the hand-written moq-c, the wire, and Rust type names. ## Plan Raised on #5356, where the maintainer found `MoqMediaContainerGroupConsumer` -too long and asked how to namespace the modules better. The long name shows -up in moq-ffi's source, the raw generated packages (`go/ffi`, -`dart/moq_ffi`), and C++ (`moq::MediaContainerGroupConsumer`). The Python, -Go, Swift, Kotlin, and Dart wrappers already shorten it to -`media.ContainerGroupConsumer`. +too long and asked how to namespace the modules better. #5356 renamed the +media consumers to `MoqMediaTrackConsumer` and `MoqMediaGroupConsumer`, but +every generated name still carries `Moq` and its layer, as in +`MoqMediaContainerStreamProducer`. They show up in moq-ffi's source, the raw +generated packages (`go/ffi`, `dart/moq_ffi`), and C++ +(`moq::MediaGroupConsumer`). The Python, Go, Swift, Kotlin, and Dart +wrappers already shorten them to `media.GroupConsumer`. ✅ The maintainer reopened (2026-10-13) the 2026-10-09 decision in [The bindings mirror Rust's layers](/quest/m1/ffi-shape/README.md) that diff --git a/quest/m1/ffi-shape/README.md b/quest/m1/ffi-shape/README.md index 90a702cfe5a..60b9d37793b 100644 --- a/quest/m1/ffi-shape/README.md +++ b/quest/m1/ffi-shape/README.md @@ -37,6 +37,11 @@ Settled shape: consumers); `json`, `flate`, `audio`, and `video` own their producers and consumers. `flate` holds the opaque snapshot and stream tracks, named after the `moq-flate` crate. +- A layer reuses the net noun it decodes, and its namespace tells them apart: + `media.TrackConsumer` and `media.GroupConsumer` read what + `media.TrackProducer` writes, a level above the raw `TrackConsumer` and + `GroupConsumer` (decided 2026-10-13 on #5356, replacing the 5-word + `MoqMediaContainerGroupConsumer`). - A layer's type is constructed from the handles its Rust constructor takes, not reached through an accessor on the broadcast: JSON wraps a track (`moq_json::snapshot::Producer::new(track, config)`), so it also works on a diff --git a/rs/moq-ffi/src/audio.rs b/rs/moq-ffi/src/audio.rs index f407c861866..f3edca887e1 100644 --- a/rs/moq-ffi/src/audio.rs +++ b/rs/moq-ffi/src/audio.rs @@ -1,7 +1,7 @@ //! Raw-audio import/export via [`moq_audio`]. //! //! Sibling to [`media::MoqMediaTrackProducer`](crate::media::MoqMediaTrackProducer) -//! and [`media::MoqMediaContainerConsumer`](crate::media::MoqMediaContainerConsumer): +//! and [`media::MoqMediaTrackConsumer`](crate::media::MoqMediaTrackConsumer): //! those deal in already-encoded frames, these deal in PCM and run //! Opus encode/decode inside the FFI boundary. diff --git a/rs/moq-ffi/src/consumer.rs b/rs/moq-ffi/src/consumer.rs index bebe90ee48d..b9e93db1137 100644 --- a/rs/moq-ffi/src/consumer.rs +++ b/rs/moq-ffi/src/consumer.rs @@ -32,7 +32,7 @@ pub struct MoqSubscription { /// `0` skips immediately; a larger value tolerates that much reordering. /// /// Enforced both by the publisher's cache (sent on the wire) and by any local - /// buffering, such as [`MoqMediaContainerConsumer`]'s jitter buffer. + /// buffering, such as [`MoqMediaTrackConsumer`]'s jitter buffer. #[uniffi(default = 0)] pub max_delay_us: u64, /// The lowest group to deliver (a floor), or null for none. A floor is not a @@ -136,8 +136,8 @@ impl MoqBroadcastConsumer { /// /// `reference` is [`MoqVideo::broadcast`] / [`MoqAudio::broadcast`]: absent or empty names /// this broadcast, anything else names a sibling relative to it (e.g. `./source`). Call it on a - /// rendition that carries one before [`crate::media::MoqMediaContainerConsumer::subscribe`], [`Self::subscribe_track`], - /// [`Self::fetch_group`], or [`crate::media::MoqMediaContainerGroupConsumer::fetch`], which take a track name rather than a + /// rendition that carries one before [`crate::media::MoqMediaTrackConsumer::subscribe`], [`Self::subscribe_track`], + /// [`Self::fetch_group`], or [`crate::media::MoqMediaGroupConsumer::fetch`], which take a track name rather than a /// rendition; `decode_video` and `decode_audio` resolve it themselves. /// /// Errors if this broadcast came from a local producer rather than an origin, since a @@ -476,6 +476,8 @@ 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_us` + /// 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 /// ([`Self::next_group`], [`Self::read_frame`]) fail with @@ -595,6 +597,18 @@ impl MoqGroupConsumer { task: Task::new(GroupInner { group }), } } + + /// Hand the group to a typed reader (a media group consumer, say), closing this handle. + /// + /// Errors with [`MoqError::AlreadyCommitted`] once a frame has been read, since the reader + /// would start mid-group, and [`MoqError::Busy`] while a read is in flight. + pub(crate) fn take(&self) -> Result { + let inner = self.task.take(|inner| match inner.group.index() { + 0 => Ok(()), + _ => Err(MoqError::AlreadyCommitted), + })?; + Ok(inner.group) + } } #[uniffi::export] diff --git a/rs/moq-ffi/src/error.rs b/rs/moq-ffi/src/error.rs index cfad3580683..f2d6ab00291 100644 --- a/rs/moq-ffi/src/error.rs +++ b/rs/moq-ffi/src/error.rs @@ -255,13 +255,14 @@ pub enum MoqError { #[error("unsupported")] Unsupported, - /// This track already committed to another way of reading it. + /// This track or group already committed to another way of reading it. /// /// A track is read in arrival order, in sequence order, or by a typed reader such as a - /// JSON consumer, never two of them, and the first group read picks the order. Reaching + /// JSON consumer, never two of them, and the first group read picks the order. A group is + /// read raw or by a media group consumer, and the first frame read picks raw. Reaching /// for another one afterwards is this error rather than [`Self::Unsupported`]: each works - /// fine here, the track just isn't reading the way you asked for. Read the track through - /// a second consumer if you genuinely need both. + /// fine here, the handle just isn't reading the way you asked for. Read through a second + /// consumer if you genuinely need both. #[error("already committed to another way of reading")] AlreadyCommitted, diff --git a/rs/moq-ffi/src/ffi.rs b/rs/moq-ffi/src/ffi.rs index 1b1a7b6c8cf..a60b71a077b 100644 --- a/rs/moq-ffi/src/ffi.rs +++ b/rs/moq-ffi/src/ffi.rs @@ -366,6 +366,27 @@ impl Task { } } + /// Move the state to a new owner, ending this handle as [Self::cancel] would. + /// + /// `check` sees the state first, and an error there leaves this handle untouched. + /// [`MoqError::Busy`] while a [Self::run] owns the state, [`MoqError::Cancelled`] after + /// [Self::cancel]. + pub fn take(&self, check: impl FnOnce(&T) -> Result<(), MoqError>) -> Result { + if *self.cancel.borrow() { + return Err(MoqError::Cancelled); + } + let mut state = self.state.clone().try_lock_owned().map_err(|_| MoqError::Busy)?; + // Re-read under the lock, as `lock` does: a cancel we don't see here has not taken the state. + if *self.cancel.borrow() { + return Err(MoqError::Cancelled); + } + check(state.as_ref().ok_or(MoqError::Cancelled)?)?; + let inner = state.take().ok_or(MoqError::Cancelled)?; + // The state is already gone, so this only fails later runs; nothing is left to drop. + self.cancel.send_replace(true); + Ok(inner) + } + /// Cancel all current and future [Self::run] calls, causing them to return [MoqError::Cancelled], /// and drop the state. /// @@ -589,6 +610,51 @@ mod tests { assert!(matches!(err, MoqError::Cancelled)); } + #[tokio::test] + async fn take_is_busy_while_a_run_holds_the_lock() { + let task = Task::new(7); + let (started, running) = tokio::sync::oneshot::channel(); + let run = task.run(move |state| async move { + let _state = state; + started.send(()).expect("test should still be waiting"); + std::future::pending::>().await + }); + tokio::pin!(run); + + soon(async { + tokio::select! { + _ = &mut run => panic!("a pending run should not resolve"), + started = running => started.expect("the run should start"), + } + }) + .await; + assert!(matches!(task.take(|_| Ok(())), Err(MoqError::Busy))); + } + + #[tokio::test] + async fn take_moves_the_state_unless_refused() { + let task = Task::new(7); + assert!(matches!( + task.take(|_| Err(MoqError::AlreadyCommitted)), + Err(MoqError::AlreadyCommitted) + )); + assert_eq!( + task.take(|state| { + assert_eq!(*state, 7); + Ok(()) + }) + .unwrap(), + 7 + ); + + let err = task + .run(|_| async { Ok(()) }) + .await + .expect_err("taken state cannot run"); + assert!(matches!(err, MoqError::Cancelled)); + assert!(matches!(task.take(|_| Ok(())), Err(MoqError::Cancelled))); + } + #[test] fn configure_is_cancelled_from_the_moment_cancel_returns() { let task = Task::new(()); diff --git a/rs/moq-ffi/src/media.rs b/rs/moq-ffi/src/media.rs index 6782c05fc62..df1102fdcc3 100644 --- a/rs/moq-ffi/src/media.rs +++ b/rs/moq-ffi/src/media.rs @@ -4,7 +4,9 @@ use std::sync::{Arc, Weak}; use bytes::Buf; use moq_mux::catalog::hang::Extra; -use crate::consumer::{MoqBroadcastConsumer, MoqFetchGroupOptions, MoqSubscription, map_fetch_error, timestamp_us}; +use crate::consumer::{ + MoqBroadcastConsumer, MoqFetchGroupOptions, MoqGroupConsumer, MoqSubscription, map_fetch_error, timestamp_us, +}; use crate::demand::MoqTrackDemand; use crate::error::MoqError; use crate::ffi::Task; @@ -1034,7 +1036,7 @@ fn reserve_track( /// A media track's name, container, and live delivery options. #[derive(Clone, uniffi::Record)] -pub struct MoqMediaContainerConfig { +pub struct MoqMediaTrackConfig { /// The track name. pub name: String, /// How to decode the track's frames. @@ -1046,7 +1048,7 @@ pub struct MoqMediaContainerConfig { /// A fetched media group's name, sequence, container, and delivery options. #[derive(Clone, uniffi::Record)] -pub struct MoqMediaContainerGroupConfig { +pub struct MoqMediaGroupConfig { /// The track name. pub name: String, /// The group sequence. @@ -1101,7 +1103,7 @@ impl Catalog { /// Reads container-decoded frames from a live media track. #[derive(uniffi::Object)] -pub struct MoqMediaContainerConsumer { +pub struct MoqMediaTrackConsumer { task: Task, } @@ -1132,15 +1134,14 @@ impl MediaGroupInner { } } -/// A finite, container-decoded media group returned by -/// [`MoqMediaContainerGroupConsumer::fetch`]. +/// A finite, container-decoded media group, fetched or decoded from a raw group. #[derive(uniffi::Object)] -pub struct MoqMediaContainerGroupConsumer { +pub struct MoqMediaGroupConsumer { sequence: u64, task: Task, } -impl MoqMediaContainerGroupConsumer { +impl MoqMediaGroupConsumer { fn from_inner(group: moq_net::group::Consumer, container: moq_mux::catalog::hang::Container) -> Self { let inner = moq_mux::container::GroupConsumer::new(group, container); Self { @@ -1151,7 +1152,7 @@ impl MoqMediaContainerGroupConsumer { } #[uniffi::export] -impl MoqMediaContainerGroupConsumer { +impl MoqMediaGroupConsumer { /// The sequence number of this group within the track. pub fn sequence(&self) -> u64 { self.sequence @@ -1190,7 +1191,7 @@ impl MoqMediaCatalogConsumer { // ---- Media Consumer ---- #[uniffi::export] -impl MoqMediaContainerConsumer { +impl MoqMediaTrackConsumer { /// Get the next frame. Returns `None` when the track ends or is closed. pub async fn next(&self) -> Result, MoqError> { self.task.run(|mut state| async move { state.next().await }).await @@ -1222,14 +1223,14 @@ impl MoqMediaCatalogConsumer { } #[uniffi::export] -impl MoqMediaContainerConsumer { +impl MoqMediaTrackConsumer { /// Subscribe to a media track and decode its container. #[uniffi::constructor] pub async fn subscribe( broadcast: &MoqBroadcastConsumer, - config: MoqMediaContainerConfig, - ) -> Result, MoqError> { - let MoqMediaContainerConfig { + config: MoqMediaTrackConfig, + ) -> Result, MoqError> { + let MoqMediaTrackConfig { name, container, subscription, @@ -1240,21 +1241,35 @@ impl MoqMediaContainerConsumer { let subscription = subscription.map(moq_net::track::Subscription::from).unwrap_or_default(); let track = broadcast.inner.track(&name)?.subscribe(subscription).await?; let consumer = moq_mux::container::Consumer::new(track, media); - Ok(Arc::new(MoqMediaContainerConsumer { + Ok(Arc::new(MoqMediaTrackConsumer { task: Task::new(Media { inner: consumer }), })) } } #[uniffi::export] -impl MoqMediaContainerGroupConsumer { +impl MoqMediaGroupConsumer { + /// Decode `group` with the track's container, as each frame lands. + /// + /// 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_us`. 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> { + // Parse the container first, so a bad init leaves the raw group readable. + let media = media_container(container)?; + Ok(Arc::new(Self::from_inner(group.take()?, media))) + } + /// Fetch and decode exactly one media group. #[uniffi::constructor] pub async fn fetch( broadcast: &MoqBroadcastConsumer, - config: MoqMediaContainerGroupConfig, - ) -> Result, MoqError> { - let MoqMediaContainerGroupConfig { + config: MoqMediaGroupConfig, + ) -> Result, MoqError> { + let MoqMediaGroupConfig { name, sequence, container, @@ -1266,7 +1281,7 @@ impl MoqMediaContainerGroupConsumer { let options = options.map(moq_net::group::Fetch::from); let track = broadcast.inner.track(&name).map_err(map_fetch_error)?; let group = track.fetch_group(sequence, options).await.map_err(map_fetch_error)?; - Ok(Arc::new(MoqMediaContainerGroupConsumer::from_inner(group, media))) + Ok(Arc::new(MoqMediaGroupConsumer::from_inner(group, media))) } } diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index 4ec8609ca5f..ff7e6dce576 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -4,6 +4,7 @@ use super::server::{MoqServer, MoqServerConfig, MoqServerTls}; use super::session::{MoqClient, MoqClientConfig, MoqClientTls, MoqSession}; use crate::consumer::MoqBroadcastConsumer; use crate::consumer::MoqFetchGroupOptions; +use crate::consumer::MoqGroupConsumer; use crate::consumer::MoqSubscription; use crate::consumer::MoqTrackConsumer; use crate::error::{MoqError, MoqProtocolError, MoqProtocolKind}; @@ -529,9 +530,9 @@ async fn raw_audio_activity() { .unwrap() .expect("expected a raw audio catalog"); let container = catalog.audio.get("microphone").unwrap().container.clone(); - let subscription = MoqMediaContainerConsumer::subscribe( + let subscription = MoqMediaTrackConsumer::subscribe( &consumer, - MoqMediaContainerConfig { + MoqMediaTrackConfig { name: "microphone".into(), container: container.clone(), subscription: None, @@ -562,9 +563,9 @@ async fn raw_audio_activity() { .expect("timed out waiting for raw audio to become unused") .unwrap(); - let subscription = MoqMediaContainerConsumer::subscribe( + let subscription = MoqMediaTrackConsumer::subscribe( &consumer, - MoqMediaContainerConfig { + MoqMediaTrackConfig { name: "microphone".into(), container, subscription: None, @@ -1324,9 +1325,9 @@ async fn fetches_cached_media_group_and_decodes_container() { .unwrap(); media.finish().unwrap(); - let fetched = MoqMediaContainerGroupConsumer::fetch( + let fetched = MoqMediaGroupConsumer::fetch( &consumer, - MoqMediaContainerGroupConfig { + MoqMediaGroupConfig { name: "media".into(), sequence: 0, container: crate::media::MoqContainer::Legacy, @@ -1354,9 +1355,9 @@ async fn fetch_media_group_rejects_invalid_container_before_fetching() { let _track = broadcast.create_track("media", timed()).unwrap(); let consumer = MoqBroadcastConsumer::new(broadcast.consume()); - let result = MoqMediaContainerGroupConsumer::fetch( + let result = MoqMediaGroupConsumer::fetch( &consumer, - MoqMediaContainerGroupConfig { + MoqMediaGroupConfig { name: "media".into(), sequence: 0, container: crate::media::MoqContainer::Cmaf { init: Vec::new() }, @@ -1403,9 +1404,9 @@ async fn fetch_media_group_decodes_multiple_cmaf_samples() { let fetched = tokio::time::timeout( TIMEOUT, - MoqMediaContainerGroupConsumer::fetch( + MoqMediaGroupConsumer::fetch( &consumer, - MoqMediaContainerGroupConfig { + MoqMediaGroupConfig { name: "video".into(), sequence: 0, container: crate::media::MoqContainer::Cmaf { init: init.to_vec() }, @@ -1442,6 +1443,155 @@ async fn fetch_media_group_decodes_multiple_cmaf_samples() { ); } +/// Write one legacy-container frame whose transport timestamp disagrees with its media time, +/// the way a receiver that stamps arrival time would see it. +fn write_legacy(group: &mut moq_net::group::Producer, transport_us: u64, media_us: u64, payload: &'static [u8]) { + let mut encoded = bytes::BytesMut::new(); + hang::container::Frame { + timestamp: moq_net::Timestamp::from_micros(media_us).unwrap(), + payload: bytes::Bytes::from_static(payload), + } + .encode(&mut encoded) + .unwrap(); + write_raw(group, transport_us, encoded.freeze()); +} + +fn write_raw(group: &mut moq_net::group::Producer, transport_us: u64, payload: bytes::Bytes) { + let mut frame = group + .create_frame(moq_net::frame::Info { + size: payload.len() as u64, + timestamp: Some(moq_net::Timestamp::from_micros(transport_us).unwrap()), + }) + .unwrap(); + frame.write(payload).unwrap(); + frame.finish().unwrap(); +} + +/// A broadcast with one legacy-container track, and a raw consumer subscribed to it. +async fn legacy_track() -> ( + moq_net::broadcast::Producer, + moq_net::track::Producer, + Arc, +) { + let broadcast = moq_net::broadcast::Info::new().produce(); + let track = broadcast.create_track("audio", hang::container::track_info(0)).unwrap(); + let consumer = MoqBroadcastConsumer::new(broadcast.consume()) + .subscribe_track("audio".into(), None) + .await + .unwrap(); + (broadcast, track, consumer) +} + +async fn recv_raw_group(track: &MoqTrackConsumer) -> Arc { + tokio::time::timeout(TIMEOUT, track.recv_group()) + .await + .expect("timed out waiting for a group") + .unwrap() + .expect("expected a group") +} + +async fn recv_media_group(track: &MoqTrackConsumer) -> Arc { + let group = recv_raw_group(track).await; + MoqMediaGroupConsumer::new(&group, MoqContainer::Legacy).unwrap() +} + +async fn next_media(group: &MoqMediaGroupConsumer) -> Result, MoqError> { + tokio::time::timeout(TIMEOUT, group.next()) + .await + .expect("timed out reading a media frame") +} + +/// Raw groups received in arrival order decode as their frames land: a stalled group is +/// neither skipped nor blocks a newer one, and each timestamp is the container's media time, +/// not the transport's. +#[tokio::test] +async fn media_group_decodes_arrival_order_groups() { + let (_broadcast, track, consumer) = legacy_track().await; + + let mut first = track.append_group().unwrap(); + write_legacy(&mut first, 1, 1_000_000, b"a"); + let first_media = recv_media_group(&consumer).await; + assert_eq!(first_media.sequence(), 0); + let frame = next_media(&first_media).await.unwrap().expect("expected a frame"); + assert_eq!(frame.payload, b"a"); + assert_eq!(frame.timestamp_us, 1_000_000); + assert!(frame.keyframe); + + // A newer group lands while the first is stalled mid-way. + let mut second = track.append_group().unwrap(); + write_legacy(&mut second, 2, 1_040_000, b"c"); + let second_media = recv_media_group(&consumer).await; + assert_eq!(second_media.sequence(), 1); + let frame = next_media(&second_media).await.unwrap().expect("expected a frame"); + assert_eq!(frame.payload, b"c"); + assert_eq!(frame.timestamp_us, 1_040_000); + + // The stalled group resumes and is still delivered. + write_legacy(&mut first, 3, 1_020_000, b"b"); + let frame = next_media(&first_media) + .await + .unwrap() + .expect("expected the stalled group's frame"); + assert_eq!(frame.payload, b"b"); + assert_eq!(frame.timestamp_us, 1_020_000); + assert!(!frame.keyframe); + + first.finish().unwrap(); + second.finish().unwrap(); + assert!(next_media(&first_media).await.unwrap().is_none()); + assert!(next_media(&second_media).await.unwrap().is_none()); +} + +/// Decoding takes over the raw group, and refuses one that already read a frame, since the +/// decoder would start mid-group. A refusal leaves the raw handle readable. +#[tokio::test] +async fn media_group_takes_over_an_unread_group() { + let (_broadcast, track, consumer) = legacy_track().await; + let mut group = track.append_group().unwrap(); + write_legacy(&mut group, 1, 1_000_000, b"a"); + write_legacy(&mut group, 2, 1_020_000, b"b"); + group.finish().unwrap(); + + let raw = recv_raw_group(&consumer).await; + assert!(raw.read_frame().await.unwrap().is_some()); + assert!(matches!( + MoqMediaGroupConsumer::new(&raw, MoqContainer::Legacy), + Err(MoqError::AlreadyCommitted) + )); + assert!( + raw.read_frame().await.unwrap().is_some(), + "the refusal kept the raw handle" + ); + + let mut group = track.append_group().unwrap(); + write_legacy(&mut group, 3, 1_040_000, b"c"); + let raw = recv_raw_group(&consumer).await; + let media = MoqMediaGroupConsumer::new(&raw, MoqContainer::Legacy).unwrap(); + assert!(matches!(raw.read_frame().await, Err(MoqError::Cancelled))); + assert_eq!( + next_media(&media).await.unwrap().expect("expected a frame").payload, + b"c" + ); +} + +/// A container that can't be configured fails before taking the group, and a frame the +/// container can't decode fails the read rather than being passed through. +#[tokio::test] +async fn media_group_fails_loudly_on_bad_containers() { + let (_broadcast, track, consumer) = legacy_track().await; + let mut group = track.append_group().unwrap(); + // A truncated varint timestamp: the 2-byte form with only its first byte. + write_raw(&mut group, 1, bytes::Bytes::from_static(&[0x40])); + + let raw = recv_raw_group(&consumer).await; + assert!(matches!( + MoqMediaGroupConsumer::new(&raw, MoqContainer::Cmaf { init: Vec::new() }), + Err(MoqError::Codec { .. }) + )); + let media = MoqMediaGroupConsumer::new(&raw, MoqContainer::Legacy).unwrap(); + assert!(next_media(&media).await.is_err()); +} + #[tokio::test] async fn dynamic_track_serves_fetch_miss_and_priority() { let broadcast = MoqBroadcastProducer::new().unwrap(); @@ -1680,9 +1830,9 @@ async fn requested_track_keeps_its_name() { let dynamic = broadcast.dynamic().unwrap(); let consumer = broadcast.consume().unwrap(); let subscribe = tokio::spawn(async move { - MoqMediaContainerConsumer::subscribe( + MoqMediaTrackConsumer::subscribe( &consumer, - MoqMediaContainerConfig { + MoqMediaTrackConfig { name: "requested".into(), container: crate::media::MoqContainer::Legacy, subscription: None, @@ -1720,9 +1870,9 @@ async fn dynamic_track_request_can_publish_media() { let subscribe = { let consumer = consumer.clone(); tokio::spawn(async move { - MoqMediaContainerConsumer::subscribe( + MoqMediaTrackConsumer::subscribe( &consumer, - MoqMediaContainerConfig { + MoqMediaTrackConfig { name: "requested-audio".into(), container: crate::media::MoqContainer::Legacy, subscription: None, @@ -2252,9 +2402,9 @@ async fn local_publish_consume_audio() { assert_eq!(audio.channel_count, 2); assert!(catalog.video.is_empty()); - let media_consumer = MoqMediaContainerConsumer::subscribe( + let media_consumer = MoqMediaTrackConsumer::subscribe( &broadcast_consumer, - MoqMediaContainerConfig { + MoqMediaTrackConfig { name: track_name.clone(), container: audio.container.clone(), subscription: None, @@ -2321,9 +2471,9 @@ async fn video_publish_consume() { assert_eq!(coded.height, 720); assert!(catalog.audio.is_empty()); - let media_consumer = MoqMediaContainerConsumer::subscribe( + let media_consumer = MoqMediaTrackConsumer::subscribe( &broadcast_consumer, - MoqMediaContainerConfig { + MoqMediaTrackConfig { name: track_name.clone(), container: video.container.clone(), subscription: None, @@ -2434,9 +2584,9 @@ async fn video_raw_publish_consume() { assert_eq!(coded.height, 240); assert!(catalog.audio.is_empty()); - let media_consumer = MoqMediaContainerConsumer::subscribe( + let media_consumer = MoqMediaTrackConsumer::subscribe( &broadcast_consumer, - MoqMediaContainerConfig { + MoqMediaTrackConfig { name: track_name.clone(), container: rendition.container.clone(), subscription: None, @@ -2774,9 +2924,9 @@ async fn multiple_frames_ordering() { .unwrap(); let (track_name, audio) = catalog.audio.iter().next().unwrap(); - let media_consumer = MoqMediaContainerConsumer::subscribe( + let media_consumer = MoqMediaTrackConsumer::subscribe( &broadcast_consumer, - MoqMediaContainerConfig { + MoqMediaTrackConfig { name: track_name.clone(), container: audio.container.clone(), subscription: None, @@ -3860,9 +4010,9 @@ async fn server_client_roundtrip() { .unwrap() .expect("expected a catalog"); let (track_name, audio) = catalog.audio.iter().next().unwrap(); - let media_consumer = MoqMediaContainerConsumer::subscribe( + let media_consumer = MoqMediaTrackConsumer::subscribe( &bc, - MoqMediaContainerConfig { + MoqMediaTrackConfig { name: track_name.clone(), container: audio.container.clone(), subscription: None, diff --git a/rs/moq-mux/src/container/group.rs b/rs/moq-mux/src/container/group.rs index 31fcd7d84e4..d550f22de95 100644 --- a/rs/moq-mux/src/container/group.rs +++ b/rs/moq-mux/src/container/group.rs @@ -1,5 +1,5 @@ use std::collections::VecDeque; -use std::task::{Poll, ready}; +use std::task::Poll; use super::{Container, Frame}; @@ -8,8 +8,8 @@ use super::{Container, Frame}; /// This is the group-scoped counterpart to [`Consumer`](super::Consumer). Where that one /// subscribes to a track and juggles group ordering, age skipping, and rewinds, this one /// reads exactly the group it was handed, in arrival order, and ends. That is what a caller -/// wants after a FETCH: a group already chosen by sequence, with no live subscription and no -/// max delay budget that could skip the very group being asked for. +/// wants after a FETCH, a group already chosen by sequence, or for a live group taken from +/// [`recv_group`](moq_net::track::Subscriber::recv_group), whose frames return as they land. /// /// A batch of frames decoded from one wire frame (a CMAF fragment carrying several samples) is /// handed back one frame at a time. @@ -47,12 +47,17 @@ impl GroupConsumer { /// Poll for the next frame, without blocking. pub fn poll_read(&mut self, waiter: &kio::Waiter) -> Poll, F::Error>> { - // Hold the latest media frame until a marker or FIN times it. FETCH groups are - // already finished, so this look-ahead does not add latency there. + // Peek past an untimed video frame for a marker or successor that times it, but only + // at what has already landed: a live group returns the frame now rather than waiting. while self.pending.front().is_none_or(|frame| { self.format.kind() == super::Kind::Video && frame.duration.is_none() && self.pending.len() < 2 }) { - match ready!(self.format.poll_read(&mut self.group, waiter)?) { + let next = match self.format.poll_read(&mut self.group, waiter)? { + Poll::Ready(next) => next, + Poll::Pending if self.pending.is_empty() => return Poll::Pending, + Poll::Pending => break, + }; + match next { Some(frames) => { for frame in frames { if let Some(bound) = self.format.end(&frame) { @@ -186,6 +191,34 @@ mod tests { assert!(group.read().await.unwrap().is_none()); } + /// A live video group hands back each frame as it lands, rather than holding it until a + /// successor or the group's end times it. + #[tokio::test] + async fn a_live_video_frame_is_not_held_for_its_successor() { + use futures::FutureExt; + + let broadcast = moq_net::broadcast::Info::new().produce(); + let track = broadcast.create_track("media", timed()).unwrap(); + let format = Hang::Legacy(crate::container::Kind::Video); + let mut group = track.append_group().unwrap(); + let mut live = GroupConsumer::new(group.consume(), Hang::Legacy(crate::container::Kind::Video)); + + format + .write(&mut group, &[frame(1_000_000, b"keyframe", true)]) + .unwrap(); + let first = live.read().now_or_never().expect("held the frame").unwrap().unwrap(); + assert_eq!(first.payload, b"keyframe".as_slice()); + assert_eq!(first.duration, None); + + format.write(&mut group, &[frame(1_020_000, b"delta", false)]).unwrap(); + let second = live.read().now_or_never().expect("held the frame").unwrap().unwrap(); + assert_eq!(second.payload, b"delta".as_slice()); + assert!(live.read().now_or_never().is_none(), "the open group has nothing more"); + + group.finish().unwrap(); + assert!(live.read().await.unwrap().is_none()); + } + /// One CMAF fragment decodes to several samples, which are handed back one at a time. #[tokio::test] async fn hands_back_a_cmaf_batch_one_frame_at_a_time() { diff --git a/rs/moq-mux/src/container/mod.rs b/rs/moq-mux/src/container/mod.rs index 26602015cdc..0725b7e1928 100644 --- a/rs/moq-mux/src/container/mod.rs +++ b/rs/moq-mux/src/container/mod.rs @@ -90,7 +90,8 @@ pub struct Frame { /// Sample duration in the frame's own scale, when the container reports it. /// /// CMAF carries a per-sample duration (trun sample-duration). Legacy and LOC - /// can fill it from a duration marker when reading a fetched group. Streaming + /// can fill it from a duration marker that has already landed when a + /// [`GroupConsumer`] reads the frame, as in a fetched group. Streaming /// muxers receive the later endpoint separately, so media stays immediately available. /// The [`Consumer`] adds it to `timestamp` to learn how far a group has /// presented, so it can advance to a newer group as soon as the gap is diff --git a/swift/Sources/Moq/Broadcast.swift b/swift/Sources/Moq/Broadcast.swift index b94f49ec19d..6a28016d163 100644 --- a/swift/Sources/Moq/Broadcast.swift +++ b/swift/Sources/Moq/Broadcast.swift @@ -30,8 +30,8 @@ public final class BroadcastConsumer: Sendable { /// /// `reference` is `Media.Video.broadcast` / `Media.Audio.broadcast`: `nil` or empty names this /// broadcast, anything else names a sibling relative to it (e.g. `./source`). Call it on a - /// rendition that carries one before `Media.ContainerConsumer.subscribe`, `subscribeTrack`, `fetchGroup`, or - /// `Media.ContainerGroupConsumer.fetch`, which take a track name rather than a rendition: the rendition's + /// rendition that carries one before `Media.TrackConsumer.subscribe`, `subscribeTrack`, `fetchGroup`, or + /// `Media.GroupConsumer.fetch`, which take a track name rather than a rendition: the rendition's /// `track`, or its catalog key when `nil`. `decodeAudio` and `decodeVideo` resolve both themselves. /// /// Throws if this broadcast came from a local producer rather than an origin, since a diff --git a/swift/Sources/Moq/Media.swift b/swift/Sources/Moq/Media.swift index c496d2bb88c..50fe1dcd506 100644 --- a/swift/Sources/Moq/Media.swift +++ b/swift/Sources/Moq/Media.swift @@ -51,9 +51,9 @@ public enum Media { /// A container format and its leading bytes. public typealias ContainerInit = MoqContainerInit /// A track name, its container, and live delivery options. - public typealias ContainerConfig = MoqMediaContainerConfig + public typealias TrackConfig = MoqMediaTrackConfig /// A fetched group's name, sequence, container, and delivery options. - public typealias ContainerGroupConfig = MoqMediaContainerGroupConfig + public typealias GroupConfig = MoqMediaGroupConfig /// Updates a catalog without keeping its broadcast open. public final class CatalogProducer: Sendable { @@ -103,17 +103,17 @@ public enum Media { } /// Read side of a media track. Iterating yields decoded frames in decode order. - public final class ContainerConsumer: AsyncSequence, Sendable { + public final class TrackConsumer: AsyncSequence, Sendable { /// The decoded media frame emitted by this sequence. public typealias Element = MediaFrame - let ffi: MoqMediaContainerConsumer + let ffi: MoqMediaTrackConsumer /// Subscribe to a track and decode its container. - public static func subscribe(broadcast: BroadcastConsumer, config: ContainerConfig) async throws -> ContainerConsumer { - ContainerConsumer(try await MoqMediaContainerConsumer.subscribe(broadcast: broadcast.ffi, config: config)) + public static func subscribe(broadcast: BroadcastConsumer, config: TrackConfig) async throws -> TrackConsumer { + TrackConsumer(try await MoqMediaTrackConsumer.subscribe(broadcast: broadcast.ffi, config: config)) } - init(_ ffi: MoqMediaContainerConsumer) { + init(_ ffi: MoqMediaTrackConsumer) { self.ffi = ffi } @@ -135,21 +135,32 @@ public enum Media { } } - /// A finite, container-decoded media group returned by ``ContainerGroupConsumer/fetch(broadcast:config:)``. - public final class ContainerGroupConsumer: AsyncSequence, Sendable { + /// A finite, container-decoded media group, from ``GroupConsumer/fetch(broadcast:config:)`` + /// or decoded from a raw ``Moq/GroupConsumer``. + public final class GroupConsumer: AsyncSequence, Sendable { /// The decoded media frame emitted by this sequence. public typealias Element = MediaFrame - let ffi: MoqMediaContainerGroupConsumer + let ffi: MoqMediaGroupConsumer /// Fetch and decode exactly one media group. - public static func fetch(broadcast: BroadcastConsumer, config: ContainerGroupConfig) async throws -> ContainerGroupConsumer { - ContainerGroupConsumer(try await MoqMediaContainerGroupConsumer.fetch(broadcast: broadcast.ffi, config: config)) + public static func fetch(broadcast: BroadcastConsumer, config: GroupConfig) async throws -> GroupConsumer { + GroupConsumer(try await MoqMediaGroupConsumer.fetch(broadcast: broadcast.ffi, config: config)) } - init(_ ffi: MoqMediaContainerGroupConsumer) { + init(_ ffi: MoqMediaGroupConsumer) { self.ffi = ffi } + /// Decode `group` with the track's container, as each frame lands. + /// + /// 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 `maxDelayUs`. 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)) + } + /// The sequence number of this group within the track. public var sequence: UInt64 { ffi.sequence() diff --git a/test/interop/clients/cpp/main.cpp b/test/interop/clients/cpp/main.cpp index 4af0aaee292..06b10cc6f43 100644 --- a/test/interop/clients/cpp/main.cpp +++ b/test/interop/clients/cpp/main.cpp @@ -196,7 +196,7 @@ int subscribe(const std::string &url, const std::string &path, double timeout) { moq::Subscription subscription; subscription.max_delay_us = MAX_DELAY_US; auto media = until( - moq::MediaContainerConsumer::subscribe(consumer, {name, video.container, subscription}), deadline, + moq::MediaTrackConsumer::subscribe(consumer, {name, video.container, subscription}), deadline, "container consumer" ); diff --git a/test/interop/clients/go/main.go b/test/interop/clients/go/main.go index 17df78e2fc3..06a24a5c625 100644 --- a/test/interop/clients/go/main.go +++ b/test/interop/clients/go/main.go @@ -29,7 +29,7 @@ import ( const readChunk = 64 * 1024 -// ContainerConsumer max delay: how much reordering the jitter buffer tolerates. +// media.TrackConsumer max delay: how much reordering the jitter buffer tolerates. const maxDelay = time.Second // Synthetic audio: a 48 kHz mono tone, encoded as Opus. @@ -202,7 +202,7 @@ func subscribe(ctx context.Context, url, broadcast string, timeout time.Duration break } - media, err := moqmedia.NewContainerConsumer(ctx, consumer, moqmedia.ContainerConfig{Name: name, Container: video.Container, Subscription: &moq.Subscription{MaxDelay: maxDelay}}) + media, err := moqmedia.NewTrackConsumer(ctx, consumer, moqmedia.TrackConfig{Name: name, Container: video.Container, Subscription: &moq.Subscription{MaxDelay: maxDelay}}) if err != nil { return err } diff --git a/test/interop/clients/python/interop.py b/test/interop/clients/python/interop.py index 3ef42d909dc..65f4ee64d03 100644 --- a/test/interop/clients/python/interop.py +++ b/test/interop/clients/python/interop.py @@ -107,7 +107,7 @@ async def subscribe(url: str, broadcast: str, timeout: float) -> None: track_name = next(iter(catalog.video)) video = catalog.video[track_name] - media = await moq.media.ContainerConsumer.subscribe( + media = await moq.media.TrackConsumer.subscribe( consumer, track_name, video.container, subscription=moq.Subscription(max_delay=MAX_DELAY) )