diff --git a/cpp/moq/MIGRATE.md b/cpp/moq/MIGRATE.md index 64741d2fcf5..5d21f4f59c0 100644 --- a/cpp/moq/MIGRATE.md +++ b/cpp/moq/MIGRATE.md @@ -2,6 +2,10 @@ ## Unreleased +### `moq::VideoInit::lower` takes either producer kind ([#7699](https://github.com/moq-dev/moq/pull/7699)) + +`lower` is `std::optional` instead of a `std::shared_ptr`. Assign `moq::MediaProducer(moq::MediaProducer::kTrack{producer})`, or `kStream` for a `moq::MediaTrackStreamProducer`. + ### Media setup refusals are typed Audio and video subscribe setup now reports `moq::Error::kUnsupportedConfig` when the decoder refuses a codec, layout, or description. Handle this variant to try another rendition instead of inspecting an audio/video error message or `Unsupported`. Decoder faults, such as a backend that fails to open, and later peer and decode failures keep their existing errors. Regenerate bindings and rebuild the matching native library together. diff --git a/cpp/obs/src/moq-output.cpp b/cpp/obs/src/moq-output.cpp index 29ad0eaab46..95bfb576647 100644 --- a/cpp/obs/src/moq-output.cpp +++ b/cpp/obs/src/moq-output.cpp @@ -850,7 +850,7 @@ void MoQOutput::VideoInit(obs_encoder_t *encoder) if (below == video_tracks.end()) return; if (below->second.producer) { - config.lower = below->second.producer; + config.lower = moq::MediaProducer(moq::MediaProducer::kTrack{below->second.producer}); lower_bps = (uint64_t)lower_kbps * 1000ULL; } } diff --git a/dart/moq/MIGRATE.md b/dart/moq/MIGRATE.md index c26f0a1b451..455dd1ca3a3 100644 --- a/dart/moq/MIGRATE.md +++ b/dart/moq/MIGRATE.md @@ -4,6 +4,10 @@ How to update your code for each breaking change, newest first. ## Unreleased +### `VideoInit.lower` takes either producer kind ([#7699](https://github.com/moq-dev/moq/pull/7699)) + +`VideoInit.lower` is a `Producer` (`MoqMediaProducer`) instead of a `TrackProducer`. Pass `TrackMoqMediaProducer(producer: ...)`, or `StreamMoqMediaProducer(producer: ...)` for a byte-stream rendition. + ### Media setup refusals are typed Audio and video subscribe setup now reports `UnsupportedConfigMoqException` when the decoder refuses a codec, layout, or description. Handle this variant to try another rendition instead of inspecting an audio/video error message or `Unsupported`. Decoder faults, such as a backend that fails to open, and later peer and decode failures keep their existing errors. Regenerate bindings and rebuild the matching native library together. diff --git a/dart/moq/lib/media.dart b/dart/moq/lib/media.dart index 83feaeee975..b2346265b31 100644 --- a/dart/moq/lib/media.dart +++ b/dart/moq/lib/media.dart @@ -9,6 +9,9 @@ typedef TrackProducer = MoqMediaTrackProducer; /// The write side of a media track fed a raw byte stream, with frame boundaries inferred. typedef TrackStreamProducer = MoqMediaTrackStreamProducer; +/// A single-track producer of either kind, such as a video rendition's `lower`. +typedef Producer = MoqMediaProducer; + /// The write side of a container, which publishes each track it describes. typedef ContainerProducer = MoqMediaContainerProducer; diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index aff33b413d1..9b1653add91 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/dart/moq_ffi/lib/src/moq.dart @@ -2548,7 +2548,7 @@ class MoqVideoInit { final Uint8List data; final String? label; final MoqVideoHint? hint; - final MoqMediaTrackProducer? lower; + final MoqMediaProducer? lower; MoqVideoInit({ required this.format, required this.data, @@ -2585,7 +2585,7 @@ class FfiConverterMoqVideoInit { ); final hint = hint_lifted.value; new_offset += hint_lifted.bytesRead; - final lower_lifted = FfiConverterOptionalMoqMediaTrackProducer.read( + final lower_lifted = FfiConverterOptionalMoqMediaProducer.read( Uint8List.view(buf.buffer, new_offset), ); final lower = lower_lifted.value; @@ -2608,7 +2608,7 @@ class FfiConverterMoqVideoInit { FfiConverterUint8List.allocationSize(value.data) + FfiConverterOptionalString.allocationSize(value.label) + FfiConverterOptionalMoqVideoHint.allocationSize(value.hint) + - FfiConverterOptionalMoqMediaTrackProducer.allocationSize(value.lower) + + FfiConverterOptionalMoqMediaProducer.allocationSize(value.lower) + 0; final buf = Uint8List(total_length); write(value, buf); @@ -2633,7 +2633,7 @@ class FfiConverterMoqVideoInit { value.hint, Uint8List.view(buf.buffer, new_offset), ); - new_offset += FfiConverterOptionalMoqMediaTrackProducer.write( + new_offset += FfiConverterOptionalMoqMediaProducer.write( value.lower, Uint8List.view(buf.buffer, new_offset), ); @@ -2645,7 +2645,7 @@ class FfiConverterMoqVideoInit { FfiConverterUint8List.allocationSize(value.data) + FfiConverterOptionalString.allocationSize(value.label) + FfiConverterOptionalMoqVideoHint.allocationSize(value.hint) + - FfiConverterOptionalMoqMediaTrackProducer.allocationSize(value.lower) + + FfiConverterOptionalMoqMediaProducer.allocationSize(value.lower) + 0; } } @@ -7614,6 +7614,130 @@ class FfiConverterMoqDecodeStart { } } +abstract class MoqMediaProducer { + RustBuffer lower(); + int allocationSize(); + int write(Uint8List buf); +} + +class FfiConverterMoqMediaProducer { + static MoqMediaProducer lift(RustBuffer buffer) { + return FfiConverterMoqMediaProducer.read(buffer.asUint8List()).value; + } + + static LiftRetVal read(Uint8List buf) { + final index = buf.buffer.asByteData(buf.offsetInBytes).getInt32(0); + final subview = Uint8List.view(buf.buffer, buf.offsetInBytes + 4); + switch (index) { + case 1: + final lifted = TrackMoqMediaProducer.read(subview); + return LiftRetVal( + lifted.value, + lifted.bytesRead - subview.offsetInBytes + 4, + ); + case 2: + final lifted = StreamMoqMediaProducer.read(subview); + return LiftRetVal( + lifted.value, + lifted.bytesRead - subview.offsetInBytes + 4, + ); + default: + throw UniffiInternalError( + UniffiInternalError.unexpectedEnumCase, + "Unable to determine enum variant", + ); + } + } + + static RustBuffer lower(MoqMediaProducer value) { + return value.lower(); + } + + static int allocationSize(MoqMediaProducer value) { + return value.allocationSize(); + } + + static int write(MoqMediaProducer value, Uint8List buf) { + return value.write(buf) - buf.offsetInBytes; + } +} + +class TrackMoqMediaProducer extends MoqMediaProducer { + final MoqMediaTrackProducer producer; + TrackMoqMediaProducer(MoqMediaTrackProducer this.producer); + TrackMoqMediaProducer._(MoqMediaTrackProducer this.producer); + static LiftRetVal read(Uint8List buf) { + int new_offset = buf.offsetInBytes; + final producer_lifted = FfiConverterMoqMediaTrackProducer.read( + Uint8List.view(buf.buffer, new_offset), + ); + final producer = producer_lifted.value; + new_offset += producer_lifted.bytesRead; + return LiftRetVal(TrackMoqMediaProducer._(producer), new_offset); + } + + @override + RustBuffer lower() { + final buf = Uint8List(allocationSize()); + write(buf); + return toRustBuffer(buf); + } + + @override + int allocationSize() { + return FfiConverterMoqMediaTrackProducer.allocationSize(producer) + 4; + } + + @override + int write(Uint8List buf) { + buf.buffer.asByteData(buf.offsetInBytes).setInt32(0, 1); + int new_offset = buf.offsetInBytes + 4; + new_offset += FfiConverterMoqMediaTrackProducer.write( + producer, + Uint8List.view(buf.buffer, new_offset), + ); + return new_offset; + } +} + +class StreamMoqMediaProducer extends MoqMediaProducer { + final MoqMediaTrackStreamProducer producer; + StreamMoqMediaProducer(MoqMediaTrackStreamProducer this.producer); + StreamMoqMediaProducer._(MoqMediaTrackStreamProducer this.producer); + static LiftRetVal read(Uint8List buf) { + int new_offset = buf.offsetInBytes; + final producer_lifted = FfiConverterMoqMediaTrackStreamProducer.read( + Uint8List.view(buf.buffer, new_offset), + ); + final producer = producer_lifted.value; + new_offset += producer_lifted.bytesRead; + return LiftRetVal(StreamMoqMediaProducer._(producer), new_offset); + } + + @override + RustBuffer lower() { + final buf = Uint8List(allocationSize()); + write(buf); + return toRustBuffer(buf); + } + + @override + int allocationSize() { + return FfiConverterMoqMediaTrackStreamProducer.allocationSize(producer) + 4; + } + + @override + int write(Uint8List buf) { + buf.buffer.asByteData(buf.offsetInBytes).setInt32(0, 2); + int new_offset = buf.offsetInBytes + 4; + new_offset += FfiConverterMoqMediaTrackStreamProducer.write( + producer, + Uint8List.view(buf.buffer, new_offset), + ); + return new_offset; + } +} + abstract class MoqMediaRead { RustBuffer lower(); int allocationSize(); @@ -12182,6 +12306,7 @@ class FfiConverterMoqMediaTrackProducer { abstract class MoqMediaTrackStreamProducerInterface { MoqTrackDemand demand(); void finish(); + void setEnabled({required bool enabled}); void write({required Uint8List payload}); } @@ -12262,6 +12387,16 @@ class MoqMediaTrackStreamProducer }, moqExceptionErrorHandler); } + void setEnabled({required bool enabled}) { + return rustCall((status) { + uniffi_moq_ffi_fn_method_moqmediatrackstreamproducer_set_enabled( + uniffiClonePointer(), + FfiConverterBool.lower(enabled), + status, + ); + }, moqExceptionErrorHandler); + } + void write({required Uint8List payload}) { return rustCall((status) { uniffi_moq_ffi_fn_method_moqmediatrackstreamproducer_write( @@ -16668,46 +16803,46 @@ class FfiConverterOptionalMoqMediaFrame { } } -class FfiConverterOptionalMoqMediaRead { - static MoqMediaRead? lift(RustBuffer buf) { - return FfiConverterOptionalMoqMediaRead.read(buf.asUint8List()).value; +class FfiConverterOptionalMoqMediaProducer { + static MoqMediaProducer? lift(RustBuffer buf) { + return FfiConverterOptionalMoqMediaProducer.read(buf.asUint8List()).value; } - static LiftRetVal read(Uint8List buf) { + static LiftRetVal read(Uint8List buf) { if (ByteData.view(buf.buffer, buf.offsetInBytes).getInt8(0) == 0) { return LiftRetVal(null, 1); } - final result = FfiConverterMoqMediaRead.read( + final result = FfiConverterMoqMediaProducer.read( Uint8List.view(buf.buffer, buf.offsetInBytes + 1), ); - return LiftRetVal(result.value, result.bytesRead + 1); + return LiftRetVal(result.value, result.bytesRead + 1); } - static int allocationSize([MoqMediaRead? value]) { + static int allocationSize([MoqMediaProducer? value]) { if (value == null) { return 1; } - return FfiConverterMoqMediaRead.allocationSize(value) + 1; + return FfiConverterMoqMediaProducer.allocationSize(value) + 1; } - static RustBuffer lower(MoqMediaRead? value) { + static RustBuffer lower(MoqMediaProducer? value) { if (value == null) { return toRustBuffer(Uint8List.fromList([0])); } final buf = Uint8List( - FfiConverterOptionalMoqMediaRead.allocationSize(value), + FfiConverterOptionalMoqMediaProducer.allocationSize(value), ); - FfiConverterOptionalMoqMediaRead.write(value, buf); + FfiConverterOptionalMoqMediaProducer.write(value, buf); return toRustBuffer(buf); } - static int write(MoqMediaRead? value, Uint8List buf) { + static int write(MoqMediaProducer? value, Uint8List buf) { if (value == null) { buf[0] = 0; return 1; } buf[0] = 1; - return FfiConverterMoqMediaRead.write( + return FfiConverterMoqMediaProducer.write( value, Uint8List.view(buf.buffer, buf.offsetInBytes + 1), ) + @@ -16715,46 +16850,46 @@ class FfiConverterOptionalMoqMediaRead { } } -class FfiConverterOptionalMoqMediaTimebase { - static MoqMediaTimebase? lift(RustBuffer buf) { - return FfiConverterOptionalMoqMediaTimebase.read(buf.asUint8List()).value; +class FfiConverterOptionalMoqMediaRead { + static MoqMediaRead? lift(RustBuffer buf) { + return FfiConverterOptionalMoqMediaRead.read(buf.asUint8List()).value; } - static LiftRetVal read(Uint8List buf) { + static LiftRetVal read(Uint8List buf) { if (ByteData.view(buf.buffer, buf.offsetInBytes).getInt8(0) == 0) { return LiftRetVal(null, 1); } - final result = FfiConverterMoqMediaTimebase.read( + final result = FfiConverterMoqMediaRead.read( Uint8List.view(buf.buffer, buf.offsetInBytes + 1), ); - return LiftRetVal(result.value, result.bytesRead + 1); + return LiftRetVal(result.value, result.bytesRead + 1); } - static int allocationSize([MoqMediaTimebase? value]) { + static int allocationSize([MoqMediaRead? value]) { if (value == null) { return 1; } - return FfiConverterMoqMediaTimebase.allocationSize(value) + 1; + return FfiConverterMoqMediaRead.allocationSize(value) + 1; } - static RustBuffer lower(MoqMediaTimebase? value) { + static RustBuffer lower(MoqMediaRead? value) { if (value == null) { return toRustBuffer(Uint8List.fromList([0])); } final buf = Uint8List( - FfiConverterOptionalMoqMediaTimebase.allocationSize(value), + FfiConverterOptionalMoqMediaRead.allocationSize(value), ); - FfiConverterOptionalMoqMediaTimebase.write(value, buf); + FfiConverterOptionalMoqMediaRead.write(value, buf); return toRustBuffer(buf); } - static int write(MoqMediaTimebase? value, Uint8List buf) { + static int write(MoqMediaRead? value, Uint8List buf) { if (value == null) { buf[0] = 0; return 1; } buf[0] = 1; - return FfiConverterMoqMediaTimebase.write( + return FfiConverterMoqMediaRead.write( value, Uint8List.view(buf.buffer, buf.offsetInBytes + 1), ) + @@ -16762,51 +16897,46 @@ class FfiConverterOptionalMoqMediaTimebase { } } -class FfiConverterOptionalMoqMediaTrackProducer { - static MoqMediaTrackProducer? lift(RustBuffer buf) { - return FfiConverterOptionalMoqMediaTrackProducer.read( - buf.asUint8List(), - ).value; +class FfiConverterOptionalMoqMediaTimebase { + static MoqMediaTimebase? lift(RustBuffer buf) { + return FfiConverterOptionalMoqMediaTimebase.read(buf.asUint8List()).value; } - static LiftRetVal read(Uint8List buf) { + static LiftRetVal read(Uint8List buf) { if (ByteData.view(buf.buffer, buf.offsetInBytes).getInt8(0) == 0) { return LiftRetVal(null, 1); } - final result = FfiConverterMoqMediaTrackProducer.read( + final result = FfiConverterMoqMediaTimebase.read( Uint8List.view(buf.buffer, buf.offsetInBytes + 1), ); - return LiftRetVal( - result.value, - result.bytesRead + 1, - ); + return LiftRetVal(result.value, result.bytesRead + 1); } - static int allocationSize([MoqMediaTrackProducer? value]) { + static int allocationSize([MoqMediaTimebase? value]) { if (value == null) { return 1; } - return FfiConverterMoqMediaTrackProducer.allocationSize(value) + 1; + return FfiConverterMoqMediaTimebase.allocationSize(value) + 1; } - static RustBuffer lower(MoqMediaTrackProducer? value) { + static RustBuffer lower(MoqMediaTimebase? value) { if (value == null) { return toRustBuffer(Uint8List.fromList([0])); } final buf = Uint8List( - FfiConverterOptionalMoqMediaTrackProducer.allocationSize(value), + FfiConverterOptionalMoqMediaTimebase.allocationSize(value), ); - FfiConverterOptionalMoqMediaTrackProducer.write(value, buf); + FfiConverterOptionalMoqMediaTimebase.write(value, buf); return toRustBuffer(buf); } - static int write(MoqMediaTrackProducer? value, Uint8List buf) { + static int write(MoqMediaTimebase? value, Uint8List buf) { if (value == null) { buf[0] = 0; return 1; } buf[0] = 1; - return FfiConverterMoqMediaTrackProducer.write( + return FfiConverterMoqMediaTimebase.write( value, Uint8List.view(buf.buffer, buf.offsetInBytes + 1), ) + @@ -19450,6 +19580,15 @@ external void uniffi_moq_ffi_fn_method_moqmediatrackstreamproducer_finish( Pointer uniffiStatus, ); +@Native, Int8, Pointer)>( + assetId: _uniffiAssetId, +) +external void uniffi_moq_ffi_fn_method_moqmediatrackstreamproducer_set_enabled( + Pointer ptr, + int enabled, + Pointer uniffiStatus, +); + @Native, RustBuffer, Pointer)>( assetId: _uniffiAssetId, ) @@ -21418,6 +21557,10 @@ uniffi_moq_ffi_checksum_method_moqmediatrackstreamproducer_demand(); external int uniffi_moq_ffi_checksum_method_moqmediatrackstreamproducer_finish(); +@Native(assetId: _uniffiAssetId) +external int +uniffi_moq_ffi_checksum_method_moqmediatrackstreamproducer_set_enabled(); + @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_method_moqmediatrackstreamproducer_write(); @@ -22145,6 +22288,10 @@ void _checkApiChecksums() { 43608) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } + if (uniffi_moq_ffi_checksum_method_moqmediatrackstreamproducer_set_enabled() != + 9270) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } if (uniffi_moq_ffi_checksum_method_moqmediatrackstreamproducer_write() != 44842) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); diff --git a/doc/lib/cpp/index.md b/doc/lib/cpp/index.md index f10a48386b8..7ca3caf14e1 100644 --- a/doc/lib/cpp/index.md +++ b/doc/lib/cpp/index.md @@ -149,7 +149,7 @@ ok(publication->announce({})); // each broadcast mints a fresh instance ok(broadcast->close()); // keep the producer alive while publishing, then close explicitly ``` -Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `lower` in its `VideoInit` too, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `set_enabled(false)` after a `discontinuity()`, and `set_enabled(true)` on the keyframe that resumes it. +Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `lower` in its `VideoInit` too, as a `moq::MediaProducer::kTrack` or `kStream`, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `set_enabled(false)` after a `discontinuity()`, and `set_enabled(true)` on the keyframe that resumes it. A `MediaTrackStreamProducer` finds its own frames: `set_enabled(false)` drops what is written to it, and after `set_enabled(true)` it resumes at the stream's next keyframe. ## Futures diff --git a/doc/lib/dart/index.md b/doc/lib/dart/index.md index 672e698798c..1eeca556ee1 100644 --- a/doc/lib/dart/index.md +++ b/doc/lib/dart/index.md @@ -140,7 +140,7 @@ mine.close(); await moq.close(); ``` -Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `lower` in its `VideoInit` too, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `setEnabled(false)` after a `discontinuity()`, and `setEnabled(true)` on the keyframe that resumes it. +Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `lower` in its `VideoInit` too, as a `TrackMoqMediaProducer` or `StreamMoqMediaProducer`, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `setEnabled(false)` after a `discontinuity()`, and `setEnabled(true)` on the keyframe that resumes it. A `TrackStreamProducer` finds its own frames: `setEnabled(false)` drops what is written to it, and after `setEnabled(true)` it resumes at the stream's next keyframe. ## Things to know diff --git a/doc/lib/go/index.md b/doc/lib/go/index.md index 3e162edc3c8..ba330f0ee0c 100644 --- a/doc/lib/go/index.md +++ b/doc/lib/go/index.md @@ -119,7 +119,7 @@ _ = video.Finish() broadcast.Close() // keep the producer reachable while publishing, then close explicitly ``` -Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `Lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `Lower` in its `media.VideoInit` too, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `SetEnabled(false)` after a `Discontinuity()`, and `SetEnabled(true)` on the keyframe that resumes it. +Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `Lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `Lower` in its `media.VideoInit` too, a `*media.TrackProducer` or `*media.TrackStreamProducer`, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `SetEnabled(false)` after a `Discontinuity()`, and `SetEnabled(true)` on the keyframe that resumes it. A `TrackStreamProducer` finds its own frames: `SetEnabled(false)` drops what is written to it, and after `SetEnabled(true)` it resumes at the stream's next keyframe. ## Things to know diff --git a/doc/lib/kt/index.md b/doc/lib/kt/index.md index 0b9d467fc99..627039d98da 100644 --- a/doc/lib/kt/index.md +++ b/doc/lib/kt/index.md @@ -95,7 +95,7 @@ Moq.connect("https://relay.example.com").use { moq -> } ``` -Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `lower` in its `VideoInit` too, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `setEnabled(false)` after a `discontinuity()`, and `setEnabled(true)` on the keyframe that resumes it. +Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `lower` in its `VideoInit` too, as a `uniffi.moq.MoqMediaProducer.Track` or `.Stream`, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `setEnabled(false)` after a `discontinuity()`, and `setEnabled(true)` on the keyframe that resumes it. A `TrackStreamProducer` finds its own frames: `setEnabled(false)` drops what is written to it, and after `setEnabled(true)` it resumes at the stream's next keyframe. ## Things to know diff --git a/doc/lib/py/index.md b/doc/lib/py/index.md index 9525af5ecc0..c254b20b60c 100644 --- a/doc/lib/py/index.md +++ b/doc/lib/py/index.md @@ -109,7 +109,7 @@ async def main(): asyncio.run(main()) ``` -Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Pass the publication track name to the producer, alongside its input and codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded `media.TrackProducer.video` takes `lower` too, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `set_enabled(False)` after a `discontinuity()`, and `set_enabled(True)` on the keyframe that resumes it. +Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Pass the publication track name to the producer, alongside its input and codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded `media.TrackProducer.video` or `media.TrackStreamProducer.video` takes `lower` too, of either kind, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `set_enabled(False)` after a `discontinuity()`, and `set_enabled(True)` on the keyframe that resumes it. A `TrackStreamProducer` finds its own frames: `set_enabled(False)` drops what is written to it, and after `set_enabled(True)` it resumes at the stream's next keyframe. ## Things to know diff --git a/doc/lib/swift/index.md b/doc/lib/swift/index.md index 2a965b9c834..94086cb8a15 100644 --- a/doc/lib/swift/index.md +++ b/doc/lib/swift/index.md @@ -96,7 +96,7 @@ session.shutdown() try await session.closed() ``` -Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Pass the publication track name to the producer, alongside its input and codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded `Media.TrackProducer.video` takes `lower` too, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `setEnabled(false)` after a `discontinuity()`, and `setEnabled(true)` on the keyframe that resumes it. +Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Pass the publication track name to the producer, alongside its input and codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded `Media.TrackProducer.video` or `Media.TrackStreamProducer.video` takes `lower` too, as a `Media.Producer` of either kind, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `setEnabled(false)` after a `discontinuity()`, and `setEnabled(true)` on the keyframe that resumes it. A `TrackStreamProducer` finds its own frames: `setEnabled(false)` drops what is written to it, and after `setEnabled(true)` it resumes at the stream's next keyframe. ## Things to know diff --git a/go/wrapper/MIGRATE.md b/go/wrapper/MIGRATE.md index 35ef0342b19..86416d17b97 100644 --- a/go/wrapper/MIGRATE.md +++ b/go/wrapper/MIGRATE.md @@ -4,6 +4,10 @@ How to update your code for each breaking change, newest first. ## Unreleased +### `media.VideoInit.Lower` is a `media.Producer` ([#7699](https://github.com/moq-dev/moq/pull/7699)) + +`Lower` is the `media.Producer` interface instead of `*media.TrackProducer`, so a `*media.TrackStreamProducer` fits too. Assigning a `*media.TrackProducer`, nil included, compiles unchanged; code that reads `Lower` back type-asserts it. + ### Media setup refusals are typed Audio and video subscribe setup now reports `ErrUnsupportedConfig` when the decoder refuses a codec, layout, or description. Handle this variant to try another rendition instead of inspecting an audio/video error message or `Unsupported`. Decoder faults, such as a backend that fails to open, and later peer and decode failures keep their existing errors. Regenerate bindings and rebuild the matching native library together. diff --git a/go/wrapper/media/media.go b/go/wrapper/media/media.go index 7a9ba2096a3..1a3a5e348c7 100644 --- a/go/wrapper/media/media.go +++ b/go/wrapper/media/media.go @@ -69,15 +69,25 @@ type VideoInit struct { Label *string // Hint carries catalog fields the stream cannot reveal itself. Hint *VideoHint - // Lower is the next lower rendition of the same source: this track then sits one - // priority below it, so a bandwidth allocator funds that one first. - Lower *TrackProducer + // Lower is the next lower rendition of the same source, a *TrackProducer or a + // *TrackStreamProducer: this track then sits one priority below it, so a bandwidth + // allocator funds that one first. + Lower Producer +} + +// Producer is a single-track producer of either kind, *TrackProducer or +// *TrackStreamProducer, such as a video rendition's Lower. +type Producer interface { + // producer is nil for a nil pointer, which means no producer, as a nil Lower does. + producer() ffi.MoqMediaProducer } func (init VideoInit) ffi() ffi.MoqVideoInit { - var lower **ffi.MoqMediaTrackProducer + var lower *ffi.MoqMediaProducer if init.Lower != nil { - lower = &init.Lower.inner + if producer := init.Lower.producer(); producer != nil { + lower = &producer + } } return ffi.MoqVideoInit{Format: init.Format, Data: init.Data, Label: init.Label, Hint: init.Hint, Lower: lower} } @@ -439,6 +449,13 @@ func (m *TrackProducer) Finish() error { return m.inner.Finish() } +func (m *TrackProducer) producer() ffi.MoqMediaProducer { + if m == nil { + return nil + } + return ffi.MoqMediaProducerTrack{Producer: m.inner} +} + // ContainerProducer publishes a container, which demuxes and publishes its own // tracks. Unlike TrackProducer there is no per-frame timestamp: a container // carries its tracks' timing itself. @@ -488,11 +505,27 @@ func (m *TrackStreamProducer) Write(payload []byte) error { return m.inner.Write(payload) } +// SetEnabled stops or resumes publishing this rendition, advertised as its catalog +// enabled flag. Disable it when its sender stops for lack of bandwidth: this marks a +// discontinuity and drops everything written until it is enabled again, so the stream +// may keep flowing. Enabled again, it resumes, and advertises itself enabled, at the +// next keyframe in the stream. +func (m *TrackStreamProducer) SetEnabled(enabled bool) error { + return m.inner.SetEnabled(enabled) +} + // Finish closes the stream. func (m *TrackStreamProducer) Finish() error { return m.inner.Finish() } +func (m *TrackStreamProducer) producer() ffi.MoqMediaProducer { + if m == nil { + return nil + } + return ffi.MoqMediaProducerStream{Producer: m.inner} +} + // Read is one encoded frame, a ReadGap, or a ReadEnd, in delivery order. type Read interface{ read() } diff --git a/kt/moq/MIGRATE.md b/kt/moq/MIGRATE.md index 05171ab7c20..4eda798883f 100644 --- a/kt/moq/MIGRATE.md +++ b/kt/moq/MIGRATE.md @@ -4,6 +4,10 @@ How to update your code for each breaking change, newest first. ## Unreleased +### `VideoInit.lower` takes either producer kind ([#7699](https://github.com/moq-dev/moq/pull/7699)) + +`VideoInit.lower` is a `dev.moq.media.Producer` (`uniffi.moq.MoqMediaProducer`) instead of a `TrackProducer`. Pass `MoqMediaProducer.Track(producer)`, or `MoqMediaProducer.Stream(producer)` for a byte-stream rendition. + ### Media setup refusals are typed Audio and video subscribe setup now reports `MoqException.UnsupportedConfig` when the decoder refuses a codec, layout, or description. Handle this variant to try another rendition instead of inspecting an audio/video error message or `Unsupported`. Decoder faults, such as a backend that fails to open, and later peer and decode failures keep their existing errors. Regenerate bindings and rebuild the matching native library together. 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 d1bbec3d374..7f2368dcf9d 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/media/Media.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/media/Media.kt @@ -13,6 +13,9 @@ typealias TrackProducer = uniffi.moq.MoqMediaTrackProducer /** The write side of a media track fed a raw byte stream, with frame boundaries inferred. */ typealias TrackStreamProducer = uniffi.moq.MoqMediaTrackStreamProducer +/** A single-track producer of either kind, such as a video rendition's `lower`. */ +typealias Producer = uniffi.moq.MoqMediaProducer + /** The write side of a container, which publishes each track it describes. */ typealias ContainerProducer = uniffi.moq.MoqMediaContainerProducer diff --git a/py/moq-rs/MIGRATE.md b/py/moq-rs/MIGRATE.md index d23a82c442f..dbcce1fe979 100644 --- a/py/moq-rs/MIGRATE.md +++ b/py/moq-rs/MIGRATE.md @@ -4,6 +4,10 @@ How to update your code for each breaking change, newest first. ## Unreleased +### `VideoInit.lower` takes either producer kind ([#7699](https://github.com/moq-dev/moq/pull/7699)) + +The `lower=` argument of `TrackProducer.video` and `TrackStreamProducer.video` accepts either producer and needs no change. Code that sets the raw `VideoInit.lower` field wraps the handle as `MoqMediaProducer.TRACK(producer=...)` or `MoqMediaProducer.STREAM(producer=...)`. + ### Media setup refusals are typed Audio and video subscribe setup now reports `UnsupportedConfig` when the decoder refuses a codec, layout, or description. Handle this variant to try another rendition instead of inspecting an audio/video error message or `Unsupported`. Decoder faults, such as a backend that fails to open, and later peer and decode failures keep their existing errors. Regenerate bindings and rebuild the matching native library together. diff --git a/py/moq-rs/moq/media.py b/py/moq-rs/moq/media.py index 737fdd1a53a..cd81c49955f 100644 --- a/py/moq-rs/moq/media.py +++ b/py/moq-rs/moq/media.py @@ -30,6 +30,7 @@ MoqMediaFrame, MoqMediaGroupConfig, MoqMediaGroupConsumer, + MoqMediaProducer, MoqMediaRead, MoqMediaTarget, MoqMediaTimebase, @@ -173,12 +174,13 @@ def video( init: VideoInit, *, target: Target = Named(), - lower: TrackProducer | None = None, + lower: TrackProducer | TrackStreamProducer | None = None, ) -> TrackProducer: """Import complete encoded video frames on a named or requested track. - Pass ``lower``, the next lower rendition of the same source, to publish this - track one priority below it, so a bandwidth allocator funds that one first. + Pass ``lower``, the next lower rendition of the same source and of either + producer kind, to publish this track one priority below it, so a bandwidth + allocator funds that one first. """ return cls(MoqMediaTrackProducer.video(broadcast._inner, target._ffi(), _lower(init, lower))) @@ -279,7 +281,7 @@ def video( init: VideoInit, *, target: Target = Named(), - lower: TrackProducer | None = None, + lower: TrackProducer | TrackStreamProducer | None = None, ) -> TrackStreamProducer: """Import a video byte stream, inferring frame boundaries. @@ -298,16 +300,29 @@ def write(self, payload: bytes) -> None: """Push raw stream bytes; whole frames are emitted as they complete.""" self._inner.write(payload) + def set_enabled(self, enabled: bool) -> None: + """Stop or resume publishing this rendition, advertised as its catalog ``enabled``. + + Disable it when its sender stops for lack of bandwidth: this marks a + discontinuity and drops everything written until it is enabled again, so + the stream may keep flowing. Enabled again, it resumes, and advertises + itself enabled, at the next keyframe in the stream. + """ + self._inner.set_enabled(enabled) + def finish(self) -> None: """Finish publishing and flush a clean end to subscribers.""" self._inner.finish() -def _lower(init: VideoInit, lower: TrackProducer | None) -> VideoInit: +def _lower(init: VideoInit, lower: TrackProducer | TrackStreamProducer | None) -> VideoInit: if lower is None: return init init = copy.copy(init) - init.lower = lower._inner + if isinstance(lower, TrackProducer): + init.lower = cast(MoqMediaProducer, MoqMediaProducer.TRACK(producer=lower._inner)) + else: + init.lower = cast(MoqMediaProducer, MoqMediaProducer.STREAM(producer=lower._inner)) return init diff --git a/quest/m1/README.md b/quest/m1/README.md index 13b46b1670d..db9dffdf5b4 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -142,7 +142,6 @@ per-frame CPU growth since #5926. - [Cluster routing](/quest/m1/cluster-routing/README.md) - prove shared routing contracts, implement per-origin routes, and connect outbound local clusters - [Routing guide](/quest/m1/routing-guide.md) - concepts and practical outbound federation setup - [moq-c at parity](/quest/m1/c/README.md) - the hand-written `moq-c` ships at the cut with every m1 binding feature moq-ffi gains -- [Byte-stream renditions form a ladder](/quest/m1/stream-ladder.md) - moq-ffi stream producers link as `lower` and disable unfundable rungs like frame producers - [A job's exact route continues its claim's readers](/quest/m1/announce-takeover.md) - a job's route inherits its claim's instance, replacing the ANNOUNCE_TAKEOVER predecessor design - [Subscribe ranges](/quest/m1/subscribe-ranges/README.md) - a lite-07 SUBSCRIBE asks for past and live ranges with absolute or relative ends, in either order, and replaces FETCH; merged ranges never starve a subscriber; relays fill misses by range, including over moq-transport - [Drop set](/quest/m1/drop-set.md) - each cursor judges missing groups against its own budget, so a narrow reader drops promptly and a widening re-offers groups already reported dropped on lite-07 diff --git a/quest/m1/stream-ladder.md b/quest/m1/stream-ladder.md deleted file mode 100644 index 0accc13b139..00000000000 --- a/quest/m1/stream-ladder.md +++ /dev/null @@ -1,35 +0,0 @@ -# [M] Byte-stream renditions form a ladder - -## Goal - -A simulcast ladder of moq-ffi byte-stream producers -(`MoqMediaTrackStreamProducer`, which imports fMP4, MKV, TS, or FLV bytes) -works like a ladder of frame producers (`MoqMediaTrackProducer`): - -- A stream producer can be passed as another rendition's `lower`, and sits - one priority below it. -- It disables itself when the bandwidth grant can't fund it, and resumes at - the next keyframe, as - [moq-dev/moq#7029](https://github.com/moq-dev/moq/pull/7029) does for - frame rungs. - -Every binding exposes this. - -## Plan - -Planned 2026-11-01 (maintainer, follow-up planning; m1 so the binding API -breaks before the cut). Raised by CodeRabbit on #7029 -([r4436515610](https://github.com/moq-dev/moq/pull/7029#discussion_r4436515610)). - -- The stream producer records its priority and exports `set_enabled`. -- `lower` (`MoqVideoInit.lower`, `MoqVideoProducerOptions.lower`) accepts - either producer kind. That changes a public UniFFI field type in every - binding (Python, Swift, Kotlin, Go, Dart). - - Pick the type that keeps one field, for example a small enum or a shared - rendition handle. - - Record the choice, and the migration in each `MIGRATE.md`. -- The C ABI follows when the C ABI parity work for #7029 does. -- Test a two-rung stream ladder in `rs/moq-ffi/src/test.rs`: the lower rung - is funded first, and the upper rung disables and resumes at a keyframe. - -Public API: breaking in every binding (`lower`'s type). Wire: none. diff --git a/rs/moq-ffi/MIGRATE.md b/rs/moq-ffi/MIGRATE.md index c16be9304df..16c160d0c05 100644 --- a/rs/moq-ffi/MIGRATE.md +++ b/rs/moq-ffi/MIGRATE.md @@ -5,6 +5,10 @@ These are the UniFFI names; each language wrapper has its own `MIGRATE.md` in it ## Unreleased +### `MoqVideoInit.lower` takes either producer kind ([#7699](https://github.com/moq-dev/moq/pull/7699)) + +`MoqVideoInit.lower` is `Option` instead of `Option`. Wrap a frame producer as `MoqMediaProducer::Track { producer }`; a `MoqMediaTrackStreamProducer` goes in as `MoqMediaProducer::Stream { producer }`. + ### Media setup refusals are typed Audio and video subscribe setup now reports `UnsupportedConfig` when the decoder refuses a codec, layout, or description. Handle this variant to try another rendition instead of inspecting an audio/video error message or `Unsupported`. Decoder faults, such as a backend that fails to open, and later peer and decode failures keep their existing errors. Regenerate bindings and rebuild the matching native library together. diff --git a/rs/moq-ffi/src/media.rs b/rs/moq-ffi/src/media.rs index feba781c4ec..c279aa5e878 100644 --- a/rs/moq-ffi/src/media.rs +++ b/rs/moq-ffi/src/media.rs @@ -379,7 +379,47 @@ pub struct MoqVideoInit { /// it. This track then sits one priority below it, so a bandwidth allocator funds the /// lower rendition first. #[uniffi(default = None)] - pub lower: Option>, + pub lower: Option, +} + +/// A single-track media producer of either kind, such as a rendition's [`lower`](MoqVideoInit::lower). +#[derive(Clone, uniffi::Enum)] +pub enum MoqMediaProducer { + /// Imports complete encoded frames. + Track { + /// The frame producer. + producer: Arc, + }, + /// Imports a byte stream. + Stream { + /// The byte-stream producer. + producer: Arc, + }, +} + +impl MoqMediaProducer { + /// The priority of the video track this producer publishes. + fn video_priority(&self) -> Result { + match self { + Self::Track { producer } => { + let guard = producer.inner.lock().unwrap(); + guard + .as_ref() + .ok_or(MoqError::Closed)? + .video + .ok_or_else(|| MoqError::Codec { + reason: "the lower rendition is not a video track".into(), + }) + } + Self::Stream { producer } => Ok(producer + .inner + .lock() + .unwrap() + .as_ref() + .ok_or(MoqError::Closed)? + .priority), + } + } } /// Shares one source's timestamp offset across containers without keeping its broadcast open. @@ -550,12 +590,7 @@ impl MoqVideoInit { out.hint = hint.into(); } if let Some(lower) = self.lower { - let guard = lower.inner.lock().unwrap(); - let below = guard.as_ref().ok_or(MoqError::Closed)?; - let priority = below.video.ok_or_else(|| MoqError::Codec { - reason: "the lower rendition is not a video track".into(), - })?; - out.priority = priority.saturating_sub(1); + out.priority = lower.video_priority()?.saturating_sub(1); } Ok(out) } @@ -870,6 +905,8 @@ struct ContainerProducer { /// A byte-stream importer for one codec track, where frame boundaries are inferred. struct MediaStreamProducer { import: Box, + /// The video track's priority, which a higher rung takes one below. + priority: i8, } /// Imports a container into a broadcast, from chunks split anywhere such as a raw byte stream. @@ -1044,6 +1081,7 @@ impl MoqMediaTrackStreamProducer { let _guard = crate::ffi::enter(); let init = init.resolve()?; broadcast.with_state(|state| { + let priority = init.priority; let request = target.resolve(&state.broadcast, &init.format)?; let import = @@ -1056,6 +1094,7 @@ impl MoqMediaTrackStreamProducer { state.children.adopt(Arc::new(MoqMediaTrackStreamProducer { inner: std::sync::Mutex::new(Some(MediaStreamProducer { import: Box::new(import), + priority, })), children: state.children.clone(), })) @@ -1302,6 +1341,23 @@ impl MoqMediaTrackStreamProducer { Ok(()) } + /// Stop or resume publishing this rendition, advertised as its catalog `enabled`. + /// + /// Disable it when its sender stops for lack of bandwidth (a + /// [`MoqRateTarget::Disabled`](crate::bandwidth::MoqRateTarget::Disabled)): this marks a + /// discontinuity and drops every access unit written until it is enabled again, so the + /// stream may keep flowing. Enabled again, it resumes, and advertises itself enabled, at + /// the next keyframe in the stream. + pub fn set_enabled(&self, enabled: bool) -> Result<(), MoqError> { + let _guard = crate::ffi::enter(); + let mut guard = self.inner.lock().unwrap(); + let media = guard.as_mut().ok_or(MoqError::Closed)?; + media.import.set_enabled(enabled).map_err(|err| MoqError::Codec { + reason: format!("set enabled failed: {err}"), + })?; + Ok(()) + } + /// Finalize the track, publishing the last access unit still waiting for a following start code. /// /// The broadcast keeps the finished track until it ends, so its finish still holds it. A diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index a0d8bfa4fba..cf97f7035a3 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -2940,7 +2940,7 @@ async fn video_rung_sits_below_its_lower_and_disables() { &broadcast, named("high"), MoqVideoInit { - lower: Some(low.clone()), + lower: Some(MoqMediaProducer::Track { producer: low.clone() }), ..video_init(MoqVideoFormat::Avc3, h264_init()) }, ) @@ -2968,6 +2968,99 @@ async fn video_rung_sits_below_its_lower_and_disables() { assert!(disabled.video["low"].enabled); } +/// A byte-stream rung above another sits one priority below it, so a grant short of both funds +/// the lower one first. Disabled, it drops what is written to it, and enabled again it resumes at +/// the next keyframe in the stream. +#[tokio::test] +async fn stream_rung_is_funded_after_its_lower_and_resumes_at_a_keyframe() { + let estimate = moq_net::bandwidth::Producer::new(); + let bandwidth = crate::bandwidth::MoqAllocator::new(moq_net::bandwidth::Allocator::new(estimate.consume())); + let broadcast = MoqBroadcastProducer::new().unwrap(); + let consumer = broadcast.consume().unwrap(); + let catalog_consumer = MoqMediaCatalogConsumer::subscribe(&consumer).await.unwrap(); + + let named = |name: &str| MoqMediaTarget::Named { + name: Some(name.into()), + }; + let low = + MoqMediaTrackStreamProducer::video(&broadcast, named("low"), video_init(MoqVideoFormat::Avc3, h264_init())) + .unwrap(); + let high = MoqMediaTrackStreamProducer::video( + &broadcast, + named("high"), + MoqVideoInit { + lower: Some(MoqMediaProducer::Stream { producer: low.clone() }), + ..video_init(MoqVideoFormat::Avc3, h264_init()) + }, + ) + .unwrap(); + + let low_track = consumer.subscribe_track("low".into(), None).await.unwrap(); + let high_track = consumer.subscribe_track("high".into(), None).await.unwrap(); + assert_eq!(low_track.info().priority, hang::catalog::PRIORITY.video); + assert_eq!(high_track.info().priority, hang::catalog::PRIORITY.video - 1); + + // The higher rung asks for more, but strict priority fills the lower one first. + let low_reservation = bandwidth.reserve(&low.demand().unwrap(), 2_000_000); + let high_reservation = bandwidth.reserve(&high.demand().unwrap(), 4_000_000); + estimate + .set(Some(moq_net::bandwidth::Rate::from_bps(3_000_000))) + .unwrap(); + assert_eq!(low_reservation.rate(), Some(2_000_000)); + assert_eq!(high_reservation.rate(), Some(1_000_000)); + + // Each keyframe ends in its own slice bytes, to tell which one went out. + let keyframe = |last: u8| { + let mut bytes = h264_init(); + bytes.extend_from_slice(&[0, 0, 0, 1, 0x65, 0x88, 0x84, last]); + bytes + }; + let delta = [0, 0, 0, 1, 0x41, 0x9a, 0x21, 0x22]; + // The splitter emits an access unit once the next one starts, hence the trailing deltas. + let write = |bytes: &[&[u8]]| { + for bytes in bytes { + high.write(bytes.to_vec()).unwrap(); + } + }; + let next_frame = async || { + loop { + let frame = kio::clock::timeout(TIMEOUT, high_track.read_frame()) + .await + .expect("timed out waiting for a frame") + .unwrap() + .expect("the track ended"); + // The discontinuity marker is empty. + if !frame.payload.is_empty() { + return frame.payload; + } + } + }; + let high_enabled = async |enabled: bool| loop { + let catalog = kio::clock::timeout(TIMEOUT, catalog_consumer.next()) + .await + .expect("timed out waiting for catalog") + .unwrap() + .expect("expected a catalog"); + if catalog.video.get("high").is_some_and(|high| high.enabled == enabled) { + assert!(catalog.video["low"].enabled); + break; + } + }; + + write(&[&keyframe(0x21), &delta, &delta]); + assert!(next_frame().await.ends_with(&[0x21])); + + high.set_enabled(false).unwrap(); + high_enabled(false).await; + // The stream keeps flowing, but none of it goes out, the keyframe included. + write(&[&delta, &keyframe(0x23), &delta, &delta]); + + high.set_enabled(true).unwrap(); + write(&[&delta, &keyframe(0x24), &delta, &delta]); + high_enabled(true).await; + assert!(next_frame().await.ends_with(&[0x24]), "resumed on another keyframe"); +} + /// A keyframe whose SPS names another level is a new decoder identity, so the track producer moves /// to a new track, and its demand follows rather than watching the one that ended. #[tokio::test] diff --git a/rs/moq-mux/src/codec/av1/import.rs b/rs/moq-mux/src/codec/av1/import.rs index d2ad9008e54..7041586337a 100644 --- a/rs/moq-mux/src/codec/av1/import.rs +++ b/rs/moq-mux/src/codec/av1/import.rs @@ -30,6 +30,9 @@ pub struct Import { track: crate::container::Producer, catalog: crate::codec::video::Catalog, last_seq: Option, + /// A sequence header only a [`dropped`](Self::dropped) keyframe carried, which the next + /// keyframe written takes if it carries none, so a subscriber can still decode it. + owed_seq: bool, } impl Import { @@ -52,6 +55,7 @@ impl Import { track, catalog, last_seq: None, + owed_seq: false, }; if let Some(config) = import.catalog.initial_config() { import.apply_config(config)?; @@ -232,26 +236,22 @@ impl Import { /// Write split frames to the track, resolving the config from the first /// keyframe's inline sequence header and refining the catalog jitter. fn write_frames(&mut self, frames: impl IntoIterator) -> Result<()> { - for frame in frames { - if frame.keyframe { - let seq = find_sequence_header(&frame.payload); - // A new sequence header starts a coded sequence whose HDR metadata replaces the last - // one's. It goes first, so a keyframe that also configures publishes once. - // The first header after an av1C is the record's own sequence, which sync samples repeat. - let sequence = seq - .as_ref() - .is_some_and(|seq| self.last_seq.as_ref().is_some_and(|last| last != seq)); - self.catalog.hdr(&mut self.track, find_hdr(&frame.payload), sequence)?; - if let Some(seq) = seq { - self.configure_from_seq(&seq)?; - } - } + for mut frame in frames { + self.configure(&frame)?; // A keyframe we couldn't configure (no sequence header) is undecodable. if frame.keyframe && !self.catalog.configured() { return Err(Error::MissingSequenceHeader.into()); } + if frame.keyframe + && std::mem::take(&mut self.owed_seq) + && let Some(seq) = &self.last_seq + && find_sequence_header(&frame.payload).is_none() + { + frame.payload = with_sequence_header(&frame.payload, seq)?; + } + // A pre-keyframe delta has no group to anchor it: the producer returns // MissingKeyframe, which a caller joining mid-stream skips. self.track.write(frame)?; @@ -259,6 +259,35 @@ impl Import { Ok(()) } + /// Account for a frame its caller drops rather than writes: a keyframe still configures the + /// rendition, and its sequence header goes out with the next keyframe written. + pub(crate) fn dropped(&mut self, frame: &Frame) -> Result<()> { + self.configure(frame)?; + if frame.keyframe && find_sequence_header(&frame.payload).is_some() { + self.owed_seq = true; + } + Ok(()) + } + + /// Resolve the config and HDR metadata a keyframe carries inline, without publishing it. + fn configure(&mut self, frame: &Frame) -> Result<()> { + if !frame.keyframe { + return Ok(()); + } + let seq = find_sequence_header(&frame.payload); + // A new sequence header starts a coded sequence whose HDR metadata replaces the last + // one's. It goes first, so a keyframe that also configures publishes once. + // The first header after an av1C is the record's own sequence, which sync samples repeat. + let sequence = seq + .as_ref() + .is_some_and(|seq| self.last_seq.as_ref().is_some_and(|last| last != seq)); + self.catalog.hdr(&mut self.track, find_hdr(&frame.payload), sequence)?; + if let Some(seq) = seq { + self.configure_from_seq(&seq)?; + } + Ok(()) + } + /// Publish split frames, resolving the config from the first keyframe's inline /// sequence header and refining the catalog jitter. pub fn decode(&mut self, frames: impl IntoIterator) -> Result<()> { @@ -266,6 +295,23 @@ impl Import { } } +/// `payload` with the sequence header `seq` after its leading temporal delimiter, where a temporal +/// unit carries it. +fn with_sequence_header(payload: &Bytes, seq: &Bytes) -> Result { + let mut buf = payload.clone(); + let first = ObuIterator::new(&mut buf).next().transpose()?; + let delimiter = first.filter(|obu| { + let mut reader = &obu[..]; + ObuHeader::parse(&mut reader).is_ok_and(|header| header.obu_type == ObuType::TemporalDelimiter) + }); + let at = delimiter.map_or(0, |obu| obu.len()); + let mut out = bytes::BytesMut::with_capacity(payload.len() + seq.len()); + out.extend_from_slice(&payload[..at]); + out.extend_from_slice(seq); + out.extend_from_slice(&payload[at..]); + Ok(out.freeze()) +} + fn is_sequence_header(obu: &[u8]) -> bool { let mut reader = obu; ObuHeader::parse(&mut reader) diff --git a/rs/moq-mux/src/codec/h264/import.rs b/rs/moq-mux/src/codec/h264/import.rs index ee1d5b678c8..120844aff48 100644 --- a/rs/moq-mux/src/codec/h264/import.rs +++ b/rs/moq-mux/src/codec/h264/import.rs @@ -215,19 +215,7 @@ impl Import { /// keyframe's inline SPS and refining the catalog jitter as it goes. fn write_frames(&mut self, frames: impl IntoIterator) -> Result<()> { for frame in frames { - if frame.keyframe { - let hdr = - crate::codec::color::access_unit(&frame.payload, self.length_size, 1, |h| h & 0x1f == NAL_TYPE_SEI); - // avc1 config arrives out-of-band via initialize(); avc3 carries the SPS inline. - let sps = self.length_size.is_none().then(|| find_sps(&frame.payload)).flatten(); - // A new SPS starts a coded sequence whose HDR metadata replaces the last one's. It goes - // first, so a keyframe that also configures publishes once. - let sequence = sps.as_ref().is_some_and(|sps| self.last_sps.as_ref() != Some(sps)); - self.catalog.hdr(&mut self.track, hdr, sequence)?; - if let Some(sps) = sps { - self.configure_from_sps(&sps)?; - } - } + self.configure(&frame)?; if !self.catalog.configured() { // A keyframe we still can't configure is undecodable, so bail @@ -244,6 +232,30 @@ impl Import { Ok(()) } + /// Account for a frame its caller drops rather than writes: a keyframe still configures the + /// rendition. The splitter re-injects the SPS and PPS into the next keyframe itself. + pub(crate) fn dropped(&mut self, frame: &Frame) -> Result<()> { + self.configure(frame) + } + + /// Resolve the config and HDR metadata a keyframe carries inline, without publishing it. + fn configure(&mut self, frame: &Frame) -> Result<()> { + if !frame.keyframe { + return Ok(()); + } + let hdr = crate::codec::color::access_unit(&frame.payload, self.length_size, 1, |h| h & 0x1f == NAL_TYPE_SEI); + // avc1 config arrives out-of-band via initialize(); avc3 carries the SPS inline. + let sps = self.length_size.is_none().then(|| find_sps(&frame.payload)).flatten(); + // A new SPS starts a coded sequence whose HDR metadata replaces the last one's. It goes + // first, so a keyframe that also configures publishes once. + let sequence = sps.as_ref().is_some_and(|sps| self.last_sps.as_ref() != Some(sps)); + self.catalog.hdr(&mut self.track, hdr, sequence)?; + if let Some(sps) = sps { + self.configure_from_sps(&sps)?; + } + Ok(()) + } + /// Publish split frames, resolving the avc3 config from the first keyframe's /// inline SPS and refining the catalog jitter as it goes. pub fn decode(&mut self, frames: impl IntoIterator) -> Result<()> { diff --git a/rs/moq-mux/src/codec/h265/import.rs b/rs/moq-mux/src/codec/h265/import.rs index 2565119a66c..927496ef689 100644 --- a/rs/moq-mux/src/codec/h265/import.rs +++ b/rs/moq-mux/src/codec/h265/import.rs @@ -218,22 +218,7 @@ impl Import { /// keyframe's inline SPS and refining the catalog jitter as it goes. fn write_frames(&mut self, frames: impl IntoIterator) -> Result<()> { for frame in frames { - if frame.keyframe { - let hdr = crate::codec::color::access_unit(&frame.payload, self.length_size, 2, |h| { - nal_unit_type(h) == scuffle_h265::NALUnitType::PrefixSeiNut - }); - // hvc1 config arrives out-of-band via initialize(); hev1 carries the SPS inline. - // Scanning an hvc1 payload as Annex-B would misread a length prefix as a start - // code (a 322-byte NAL prefixes as `00 00 01 42`, an SPS NAL header). - let sps = self.length_size.is_none().then(|| find_sps(&frame.payload)).flatten(); - // A new SPS starts a coded sequence whose HDR metadata replaces the last one's. It goes - // first, so a keyframe that also configures publishes once. - let sequence = sps.as_ref().is_some_and(|sps| self.last_sps.as_ref() != Some(sps)); - self.catalog.hdr(&mut self.track, hdr, sequence)?; - if let Some(sps) = sps { - self.configure_from_sps(&sps)?; - } - } + self.configure(&frame)?; // A keyframe we still can't configure (no SPS) is undecodable. if frame.keyframe && !self.catalog.configured() { @@ -247,6 +232,34 @@ impl Import { Ok(()) } + /// Account for a frame its caller drops rather than writes: a keyframe still configures the + /// rendition. The splitter re-injects the VPS, SPS, and PPS into the next keyframe itself. + pub(crate) fn dropped(&mut self, frame: &Frame) -> Result<()> { + self.configure(frame) + } + + /// Resolve the config and HDR metadata a keyframe carries inline, without publishing it. + fn configure(&mut self, frame: &Frame) -> Result<()> { + if !frame.keyframe { + return Ok(()); + } + let hdr = crate::codec::color::access_unit(&frame.payload, self.length_size, 2, |h| { + nal_unit_type(h) == scuffle_h265::NALUnitType::PrefixSeiNut + }); + // hvc1 config arrives out-of-band via initialize(); hev1 carries the SPS inline. + // Scanning an hvc1 payload as Annex-B would misread a length prefix as a start + // code (a 322-byte NAL prefixes as `00 00 01 42`, an SPS NAL header). + let sps = self.length_size.is_none().then(|| find_sps(&frame.payload)).flatten(); + // A new SPS starts a coded sequence whose HDR metadata replaces the last one's. It goes + // first, so a keyframe that also configures publishes once. + let sequence = sps.as_ref().is_some_and(|sps| self.last_sps.as_ref() != Some(sps)); + self.catalog.hdr(&mut self.track, hdr, sequence)?; + if let Some(sps) = sps { + self.configure_from_sps(&sps)?; + } + Ok(()) + } + /// Publish split frames, resolving the config from the first keyframe's inline /// SPS and refining the catalog jitter as it goes. pub fn decode(&mut self, frames: impl IntoIterator) -> Result<()> { diff --git a/rs/moq-mux/src/import/track.rs b/rs/moq-mux/src/import/track.rs index c25e621fdf9..6bad2f2d7e2 100644 --- a/rs/moq-mux/src/import/track.rs +++ b/rs/moq-mux/src/import/track.rs @@ -8,6 +8,7 @@ use crate::Result; use crate::catalog::VideoHint; use crate::catalog::hang::CatalogExt; +use crate::container::Frame; use super::{AudioFormat, VideoFormat}; pub use super::{AudioInit, VideoInit}; @@ -592,6 +593,66 @@ macro_rules! stream { /// H.264, an fMP4 reader, …); the importer infers them. pub struct TrackStream { kind: TrackStreamKind, + gate: Gate, +} + +/// Which of the access units a [`TrackStream`] splits it publishes. +/// +/// The stream's caller cannot see frame boundaries, so the importer, not the caller, drops a +/// disabled rendition's access units and finds the keyframe that resumes it. +#[derive(Default)] +struct Gate { + state: GateState, + /// Whether an access unit went out, so a disable has a timeline to break. + published: bool, +} + +#[derive(Clone, Copy, Default, PartialEq, Eq)] +enum GateState { + #[default] + Open, + /// Disabled: every access unit is dropped. + Closed, + /// Enabled again, but still disabled in the catalog until the next keyframe. + Resuming, +} + +impl Gate { + /// Split `frames` into the access units dropped and those to publish, and report whether + /// the first published one resumes the rendition, so the catalog enables it again. + fn admit(&mut self, mut frames: Vec) -> (Vec, Vec, bool) { + let (kept, resumed) = match self.state { + GateState::Open => (0, false), + GateState::Closed => (frames.len(), false), + GateState::Resuming => match frames.iter().position(|frame| frame.keyframe) { + Some(keyframe) => { + self.state = GateState::Open; + (keyframe, true) + } + None => (frames.len(), false), + }, + }; + let publish = frames.split_off(kept); + self.published |= !publish.is_empty(); + (frames, publish, resumed) + } +} + +/// Publish split `frames` through a [`TrackStream`]'s gate. +/// +/// The importer still sees what the gate drops: a stream may carry its config only in band (an +/// AV1 sequence header), and the keyframe that resumes need not repeat it. +macro_rules! publish { + ($gate:expr, $import:expr, $frames:expr) => {{ + let (dropped, frames, resumed) = $gate.admit($frames); + for frame in &dropped { + $import.dropped(frame)?; + } + if resumed { + $import.set_enabled(true)?; + } + $import.decode(frames) + }}; } impl TrackStream { @@ -627,7 +688,10 @@ impl TrackStream { format => return Err(crate::Error::NotSelfDescribing(format.to_string())), }; - let mut stream = Self { kind }; + let mut stream = Self { + kind, + gate: Gate::default(), + }; if !init.data.is_empty() { stream.initialize(&init.data)?; } @@ -644,16 +708,14 @@ impl TrackStream { ref mut import, } => { import.initialize(data)?; - let frames = split.initialize(data)?; - import.decode(frames)?; + publish!(self.gate, import, split.initialize(data)?)?; } TrackStreamKind::Hev1 { ref mut split, ref mut import, } => { import.initialize(data)?; - let frames = split.initialize(data)?; - import.decode(frames)?; + publish!(self.gate, import, split.initialize(data)?)?; } TrackStreamKind::Av01 { ref mut split, @@ -667,7 +729,7 @@ impl TrackStream { } else { split.decode(data, None)? }; - import.decode(frames)?; + publish!(self.gate, import, frames)?; } } @@ -677,8 +739,7 @@ impl TrackStream { /// Decode a chunk of the byte stream. pub fn decode(&mut self, data: &[u8]) -> Result<()> { stream!(&mut self.kind, split, import => { - let frames = split.decode(data, None)?; - import.decode(frames) + publish!(self.gate, import, split.decode(data, None)?) }) } @@ -687,12 +748,43 @@ impl TrackStream { // The splitter is taken out to flush, so its input storage ends with the input. The import // stays, keeping the finished track and its rendition published. stream!(&mut self.kind, split, import => { - let tail = std::mem::take(split).flush(None)?; - import.decode(tail)?; + publish!(self.gate, import, std::mem::take(split).flush(None)?)?; import.finish() }) } + /// Stop or resume publishing this rendition, advertised as the catalog's + /// [`enabled`](hang::catalog::VideoConfig::enabled). + /// + /// Disable it when its sender cannot be funded (a + /// [`rate::Target::Disabled`](crate::rate::Target::Disabled)). Once anything was published, + /// that marks a [`discontinuity`](crate::container::Producer::discontinuity). Every access + /// unit decoded until it is enabled again is dropped, so the caller may keep writing the + /// stream, though a dropped keyframe still updates the codec config. Enabled again, it + /// drops access units until the next keyframe and advertises itself enabled with that + /// keyframe. + pub fn set_enabled(&mut self, enabled: bool) -> Result<()> { + match (enabled, self.gate.state) { + (false, GateState::Open) => { + let published = self.gate.published; + // The splitter keeps its partial NAL, since the next chunk may continue it. The + // access unit in flight completes behind the gate, which drops it unless it is + // a keyframe that resumes the rendition. + stream!(&mut self.kind, _, import => { + import.set_enabled(false)?; + if published { + import.discontinuity()?; + } + }); + self.gate.state = GateState::Closed; + } + (false, GateState::Resuming) => self.gate.state = GateState::Closed, + (true, GateState::Closed) => self.gate.state = GateState::Resuming, + (false, GateState::Closed) | (true, GateState::Open | GateState::Resuming) => {} + } + Ok(()) + } + /// Abort the track with `err` instead of finishing it cleanly, so subscribers /// see the real cause rather than [`moq_net::Error::Dropped`]. Consumes the importer. pub fn abort(self, err: moq_net::Error) { @@ -1648,4 +1740,105 @@ mod tests { ); } } + + /// A disabled [`TrackStream`] drops what it splits and marks the break, so a caller that + /// keeps writing sends nothing. Enabled again, it resumes, and advertises itself enabled, + /// only at the next keyframe. + #[tokio::test(start_paused = true)] + async fn track_stream_disables_and_resumes_at_a_keyframe() { + const H264_P: &[u8] = &[0x41, 0x9a, 0x21, 0x22]; + + let (broadcast, catalog) = new_broadcast(); + let consumer = broadcast.consume(); + let request = broadcast.reserve_track("video").unwrap(); + let mut import = TrackStream::video( + request, + catalog.reserve(), + VideoInit::new(VideoFormat::Avc3, Vec::new()), + ) + .unwrap(); + // Every group is written before anything reads, which the default + // REAL_TIME budget would collapse to the live edge. + let subscriber = consumer + .track("video") + .unwrap() + .subscribe( + moq_net::track::Subscription::default() + .with_delay(moq_net::track::Delay::default().with_max(Duration::from_secs(30))), + ) + .await + .unwrap(); + let enabled = || catalog.snapshot().video.renditions["video"].enabled; + + let keyframe = [h264_init(), annexb(&[H264_IDR])].concat(); + let delta = annexb(&[H264_P]); + // One access unit per frame interval. The splitter holds the last NAL until the next + // start code, so an access unit goes out one write after its last NAL arrives. + async fn write(import: &mut TrackStream, bytes: &[u8]) { + import.decode(bytes).unwrap(); + kio::clock::sleep(Duration::from_millis(33)).await; + } + for bytes in [&keyframe, &delta, &delta] { + write(&mut import, bytes).await; + } + // A chunk may end mid-NAL, and the next one continue it across the disable. + let (head, tail) = delta.split_at(6); + write(&mut import, head).await; + + import.set_enabled(false).unwrap(); + assert!(!enabled()); + for bytes in [tail, &keyframe, &delta, &delta] { + write(&mut import, bytes).await; + } + + import.set_enabled(true).unwrap(); + for bytes in [&delta, &keyframe, &delta] { + write(&mut import, bytes).await; + } + assert!(!enabled(), "enabled before the keyframe that resumes it"); + write(&mut import, &delta).await; + assert!(enabled()); + import.finish().unwrap(); + + // The first group loses the access units still in the splitter at the disable, the + // marker breaks the timeline, and the resumed group opens on the keyframe: three + // access units and the end marker `finish` bounds the last one with. + assert_eq!(collect_groups(subscriber).await, vec![2, 1, 4]); + } + + /// A keyframe dropped while disabled still configures the importer: an AV1 stream may send + /// its sequence header once, and the keyframe that resumes need not repeat it. + #[tokio::test(start_paused = true)] + async fn track_stream_configures_from_a_dropped_keyframe() { + let (_, with_header) = av1_init_and_keyframe(); + // A temporal delimiter and a key frame, with no sequence header. + let bare = [0x12, 0x00, 0x32, 0x01, 0x10]; + + let (broadcast, catalog) = new_broadcast(); + let consumer = broadcast.consume(); + let request = broadcast.reserve_track("video").unwrap(); + let mut import = TrackStream::video( + request, + catalog.reserve(), + VideoInit::new(VideoFormat::Av01, Vec::new()), + ) + .unwrap(); + + import.set_enabled(false).unwrap(); + // The next temporal delimiter completes the one carrying the sequence header. + import.decode(&with_header).unwrap(); + kio::clock::sleep(Duration::from_millis(33)).await; + import.decode(&bare).unwrap(); + kio::clock::sleep(Duration::from_millis(33)).await; + + // Finish flushes the last temporal unit, the keyframe that resumes. + import.set_enabled(true).unwrap(); + import.finish().unwrap(); + + assert!(catalog.snapshot().video.renditions["video"].enabled); + let frame = first_frame(&consumer, "video").await; + assert!(frame.keyframe, "the resumed group opens on a keyframe"); + // The dropped sequence header goes out with it, after its temporal delimiter. + assert_eq!(frame.payload.as_ref(), with_header.as_slice()); + } } diff --git a/swift/MIGRATE.md b/swift/MIGRATE.md index 88dc21b6c2c..6e98d5c51a3 100644 --- a/swift/MIGRATE.md +++ b/swift/MIGRATE.md @@ -4,6 +4,10 @@ How to update your code for each breaking change, newest first. ## Unreleased +### `lower` is a `Media.Producer` ([#7699](https://github.com/moq-dev/moq/pull/7699)) + +`Media.TrackProducer.video` and `Media.TrackStreamProducer.video` take `lower: Media.Producer?` instead of `TrackProducer?`. Pass `lower: .track(producer)`, or `.stream(producer)` for a byte-stream rendition. + ### Media setup refusals are typed Audio and video subscribe setup now reports `UnsupportedConfig` when the decoder refuses a codec, layout, or description. Handle this variant to try another rendition instead of inspecting an audio/video error message or `Unsupported`. Decoder faults, such as a backend that fails to open, and later peer and decode failures keep their existing errors. Regenerate bindings and rebuild the matching native library together. diff --git a/swift/Sources/Moq/Media.swift b/swift/Sources/Moq/Media.swift index f9748be482c..12d807b4402 100644 --- a/swift/Sources/Moq/Media.swift +++ b/swift/Sources/Moq/Media.swift @@ -49,6 +49,19 @@ public enum Media { } } } + /// A single-track producer of either kind, such as a video rendition's `lower`. + public enum Producer { + /// Imports complete encoded frames. + case track(TrackProducer) + /// Imports a byte stream. + case stream(TrackStreamProducer) + var ffi: MoqMediaProducer { + switch self { + case .track(let producer): return .track(producer: producer.ffi) + case .stream(let producer): return .stream(producer: producer.ffi) + } + } + } /// An audio codec, its initialization bytes, and label. public typealias AudioInit = MoqAudioInit /// A video codec, its initialization bytes, label, and hints. @@ -234,11 +247,12 @@ public enum Media { } /// Import complete video frames on a named or requested track. /// - /// Pass `lower`, the next lower rendition of the same source, to publish this track one - /// priority below it, so a bandwidth allocator funds that one first. + /// Pass `lower`, the next lower rendition of the same source and of either producer + /// kind, to publish this track one priority below it, so a bandwidth allocator funds + /// that one first. public static func video( broadcast: BroadcastProducer, initData: VideoInit, target: Target = .named(name: nil), - lower: TrackProducer? = nil + lower: Producer? = nil ) throws -> TrackProducer { var initData = initData if let lower { initData.lower = lower.ffi } @@ -347,7 +361,7 @@ public enum Media { /// `lower` is as for ``TrackProducer/video(broadcast:initData:target:lower:)``. public static func video( broadcast: BroadcastProducer, initData: VideoInit, target: Target = .named(name: nil), - lower: TrackProducer? = nil + lower: Producer? = nil ) throws -> TrackStreamProducer { var initData = initData if let lower { initData.lower = lower.ffi } @@ -369,6 +383,16 @@ public enum Media { try ffi.write(payload: payload) } + /// Stop or resume publishing this rendition, advertised as its catalog `enabled`. + /// + /// Disable it when its sender stops for lack of bandwidth: this marks a discontinuity + /// and drops everything written until it is enabled again, so the stream may keep + /// flowing. Enabled again, it resumes, and advertises itself enabled, at the next + /// keyframe in the stream. + public func setEnabled(_ enabled: Bool) throws { + try ffi.setEnabled(enabled: enabled) + } + /// Finalize the track, publishing the last access unit still waiting for a /// following start code. public func finish() throws {