diff --git a/cpp/moq/test/probe.cpp b/cpp/moq/test/probe.cpp index 930ab5a1133..fb119acc372 100644 --- a/cpp/moq/test/probe.cpp +++ b/cpp/moq/test/probe.cpp @@ -132,8 +132,19 @@ void audio_roundtrip() { size_t frames = 0; double real[2][2] = {}, imag[2][2] = {}; for (;;) { - auto next = ok(consumer->next().get(), "decoded audio"); - if (!next) break; + auto read = ok(consumer->next().get(), "decoded audio"); + if (!read) break; + CHECK(!std::holds_alternative(read->get_variant())); + // The finish ends the encoder where its eight 20 ms frames stop. + if (const auto *end = std::get_if(&read->get_variant())) { + CHECK(end->end_us == 250000 + 8 * 20000); + continue; + } + const auto *variant = std::get_if(&read->get_variant()); + CHECK(variant != nullptr); + + const auto *next = &variant->frame; + CHECK(next->timestamp_us == 250000 + frames * 1000000 / 48000); CHECK(next->data.size() % (2 * sizeof(float)) == 0); for (size_t sample = 0; sample < next->data.size() / (2 * sizeof(float)); sample++) { diff --git a/cpp/obs/src/moq-source.cpp b/cpp/obs/src/moq-source.cpp index b8de7a1122d..56d0a820627 100644 --- a/cpp/obs/src/moq-source.cpp +++ b/cpp/obs/src/moq-source.cpp @@ -699,13 +699,14 @@ static void moq_source_stop_audio(const std::shared_ptr &conn, const static void moq_source_read_audio(struct moq_source *ctx, const std::shared_ptr &conn, const std::shared_ptr &track); -// Takes one decoded frame under the source lock. A candidate's first frame commits it, and -// any failure stops the track. On success `audio` points into the frame, for output once -// the lock is released. -static bool moq_source_on_audio_frame(struct moq_source *ctx, const std::shared_ptr &conn, - const std::shared_ptr &track, - const moq::expected> &result, - obs_source_audio &audio) +// Takes one decoded read under the source lock. A candidate's first frame commits it, and +// any failure stops the track. On a frame `audio` points into it, for output once the lock +// is released. A gap or encoder end needs nothing here: OBS buffers by timestamp, so the +// next frame resumes wherever its timestamp lands. +static bool moq_source_on_audio_read(struct moq_source *ctx, const std::shared_ptr &conn, + const std::shared_ptr &track, + const moq::expected> &result, + obs_source_audio &audio) { if (!result) { LOG_ERROR("Audio track error: %s", result.error().to_string().c_str()); @@ -718,8 +719,14 @@ static bool moq_source_on_audio_frame(struct moq_source *ctx, const std::shared_ return false; } + const auto *read = std::get_if(&(*result)->get_variant()); + if (!read) { + moq_source_read_audio(ctx, conn, track); + return false; + } + // moq-audio delivers whole planar float samples in the layout the track asked for. - const auto &frame = **result; + const auto &frame = read->frame; const auto &layout = track->audio_layout; const size_t plane = frame.data.size() / layout.channels; if (plane == 0 || plane % sizeof(float) != 0 || plane * layout.channels != frame.data.size()) { @@ -754,7 +761,7 @@ static void moq_source_read_audio(struct moq_source *ctx, const std::shared_ptr< track->call = track->audio_consumer->next().then( ctx->worker.Executor(), [ctx, weak_conn = std::weak_ptr(conn), - weak_track = std::weak_ptr(track)](moq::expected> result) { + weak_track = std::weak_ptr(track)](moq::expected> result) { obs_source_audio audio = {}; { std::lock_guard lock(ctx->mutex); @@ -762,7 +769,7 @@ static void moq_source_read_audio(struct moq_source *ctx, const std::shared_ptr< if (!conn || conn != ctx->connection) return; auto track = moq_source_track(*conn, weak_track); - if (!track || !moq_source_on_audio_frame(ctx, conn, track, result, audio)) + if (!track || !moq_source_on_audio_read(ctx, conn, track, result, audio)) return; } // `result` owns the samples, and OBS copies them before returning. Destroy diff --git a/cpp/obs/test/moq-audio-encoder-test.cpp b/cpp/obs/test/moq-audio-encoder-test.cpp index 582e586186e..ea229cea72b 100644 --- a/cpp/obs/test/moq-audio-encoder-test.cpp +++ b/cpp/obs/test/moq-audio-encoder-test.cpp @@ -15,6 +15,7 @@ #include #include #include +#include #include #include @@ -270,9 +271,15 @@ void CheckSubscriberDecodes(const char *preset, size_t channels) CHECK(!"decoded audio stalled"); break; } - auto frame = TestOk(next.get(), "audio next"); - if (!frame) + auto read = TestOk(next.get(), "audio next"); + if (!read) break; + // Raw packets carry no encoder end, and nothing was lost. + const auto *variant = std::get_if(&read->get_variant()); + CHECK(variant != nullptr); + if (!variant) + break; + const auto *frame = &variant->frame; if (!first_decoded_us) first_decoded_us = frame->timestamp_us; const auto *samples = reinterpret_cast(frame->data.data()); diff --git a/cpp/obs/test/moq-source-test.cpp b/cpp/obs/test/moq-source-test.cpp index bb75410cbc6..5943f3f7f30 100644 --- a/cpp/obs/test/moq-source-test.cpp +++ b/cpp/obs/test/moq-source-test.cpp @@ -1156,8 +1156,9 @@ int main() auto active = conn->audio; CHECK(active != nullptr); obs_source_audio audio = {}; - moq::expected> misaligned(moq::AudioFrame{1, {0, 0, 0}}); - CHECK(!moq_source_on_audio_frame(ctx, conn, active, misaligned, audio)); + moq::expected> misaligned( + moq::AudioRead::kFrame{moq::AudioFrame{1, {0, 0, 0}}}); + CHECK(!moq_source_on_audio_read(ctx, conn, active, misaligned, audio)); CHECK(!conn->audio); } destroySource(source); diff --git a/dart/moq/MIGRATE.md b/dart/moq/MIGRATE.md index c8aca9bdb96..19e56003949 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 +### Decoders read gaps and ends + +`audio.Consumer.next()` and `video.Consumer.next()` return a `Read` instead of a frame. Take `read.frame` from a `ReadFrame` where you used the frame; `ReadGap` marks lost media and `ReadEnd` where the publisher stopped its encoder, and what follows either does not continue what came before. + ### Durations are `Duration` fields The records carry durations as `Duration` fields instead of `*Us` microsecond integers, so the duration extension getters are gone: read `backoff.initial`, `stats.rtt`, `grant.expires`, `info.maxAge`, `fill.maxDelay()`, and `consumer.delay()` directly. diff --git a/dart/moq/lib/audio.dart b/dart/moq/lib/audio.dart index 074c5294064..ac4ffeadcbc 100644 --- a/dart/moq/lib/audio.dart +++ b/dart/moq/lib/audio.dart @@ -16,9 +16,21 @@ typedef Producer = MoqAudioProducer; /// Encodes PCM without publishing it. typedef Encoder = MoqAudioEncoder; -/// Subscribes to a catalog audio rendition and yields decoded PCM; open one with `Consumer.subscribe`. +/// Subscribes to a catalog audio rendition and yields each [Read]; open one with `Consumer.subscribe`. typedef Consumer = MoqAudioConsumer; +/// One item a [Consumer] reads, in delivery order: [ReadFrame], [ReadGap], or [ReadEnd]. +typedef Read = MoqAudioRead; + +/// Decoded PCM. +typedef ReadFrame = FrameMoqAudioRead; + +/// Media after what was already read was lost; what follows does not continue it. +typedef ReadGap = GapMoqAudioRead; + +/// The publisher stopped its encoder; frames starting before `endUs` are its remaining media. +typedef ReadEnd = EndMoqAudioRead; + /// The PCM layout written to a [Producer]. typedef EncoderInput = MoqAudioEncoderInput; @@ -49,5 +61,17 @@ extension AudioFrameDuration on MoqAudioFrame { Duration get timestamp => Duration(microseconds: timestampUs); } +/// Duration view over a gap. +extension AudioReadGapDuration on GapMoqAudioRead { + /// The furthest presentation point already read. + Duration get timestamp => Duration(microseconds: timestampUs); +} + +/// Duration view over an encoder end. +extension AudioReadEndDuration on EndMoqAudioRead { + /// Where the stopped encoder's media stops. + Duration get end => Duration(microseconds: endUs); +} + /// Input, codec settings, and publication ownership for a producer. typedef ProducerOptions = MoqAudioProducerOptions; diff --git a/dart/moq/lib/video.dart b/dart/moq/lib/video.dart index c30e9f3713b..b7171c06451 100644 --- a/dart/moq/lib/video.dart +++ b/dart/moq/lib/video.dart @@ -16,9 +16,21 @@ typedef Producer = MoqVideoProducer; /// Encodes raw pictures without publishing them. typedef Encoder = MoqVideoEncoder; -/// Subscribes to a catalog video rendition and yields decoded frames; open one with `Consumer.subscribe`. +/// Subscribes to a catalog video rendition and yields each [Read]; open one with `Consumer.subscribe`. typedef Consumer = MoqVideoConsumer; +/// One item a [Consumer] reads, in delivery order: [ReadFrame], [ReadGap], or [ReadEnd]. +typedef Read = MoqVideoRead; + +/// A decoded picture; `dispose()` its frame promptly, since held frames stall the decoder. +typedef ReadFrame = FrameMoqVideoRead; + +/// Media after what was already read was lost; what follows does not continue it. +typedef ReadGap = GapMoqVideoRead; + +/// The publisher stopped its encoder; what follows comes from a fresh encoder. +typedef ReadEnd = EndMoqVideoRead; + /// The pixel layout, resolution, and framerate written to a [Producer]. typedef EncoderInput = MoqVideoEncoderInput; @@ -82,6 +94,18 @@ extension VideoDecodedFrameDuration on MoqVideoDecodedFrame { Duration get timestamp => Duration(microseconds: timestampUs()); } +/// Duration view over a gap. +extension VideoReadGapDuration on GapMoqVideoRead { + /// The furthest presentation point already read. + Duration get timestamp => Duration(microseconds: timestampUs); +} + +/// Duration view over an encoder end. +extension VideoReadEndDuration on EndMoqVideoRead { + /// Where the stopped encoder's media stops. + Duration get end => Duration(microseconds: endUs); +} + /// Input, codec settings, and publication ownership for a producer. typedef ProducerOptions = MoqVideoProducerOptions; diff --git a/dart/moq/test/moq_test.dart b/dart/moq/test/moq_test.dart index 7bcb0d57e88..cddf1ef3635 100644 --- a/dart/moq/test/moq_test.dart +++ b/dart/moq/test/moq_test.dart @@ -640,10 +640,11 @@ void main() { ); } - final frame = await decoder.next().timeout(timeout); + final read = await decoder.next().timeout(timeout); + final frame = (read as moq_video.ReadFrame).frame; // The frame outlives its consumer's cancellation. decoder.cancel(); - final pixels = frame!.width() * frame.height(); + final pixels = frame.width() * frame.height(); expect( frame.pixels(format: moq_video.PixelFormat.i420).length, pixels * 3 ~/ 2, @@ -711,8 +712,8 @@ void main() { ), ); } - expect(await fixed.next().timeout(timeout), isNotNull); - expect(await estimated.next().timeout(timeout), isNotNull); + expect(await fixed.next().timeout(timeout), isA()); + expect(await estimated.next().timeout(timeout), isA()); expect(fixed.delay(), const Duration(milliseconds: 80)); expect(estimated.delay(), isNot(initial)); diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index e0b7f111a40..8c9a51f9f13 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/dart/moq_ffi/lib/src/moq.dart @@ -4806,6 +4806,189 @@ class FfiConverterMoqVideoProducerOptions { } } +abstract class MoqAudioRead { + RustBuffer lower(); + int allocationSize(); + int write(Uint8List buf); +} + +class FfiConverterMoqAudioRead { + static MoqAudioRead lift(RustBuffer buffer) { + return FfiConverterMoqAudioRead.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 = FrameMoqAudioRead.read(subview); + return LiftRetVal( + lifted.value, + lifted.bytesRead - subview.offsetInBytes + 4, + ); + case 2: + final lifted = GapMoqAudioRead.read(subview); + return LiftRetVal( + lifted.value, + lifted.bytesRead - subview.offsetInBytes + 4, + ); + case 3: + final lifted = EndMoqAudioRead.read(subview); + return LiftRetVal( + lifted.value, + lifted.bytesRead - subview.offsetInBytes + 4, + ); + default: + throw UniffiInternalError( + UniffiInternalError.unexpectedEnumCase, + "Unable to determine enum variant", + ); + } + } + + static RustBuffer lower(MoqAudioRead value) { + return value.lower(); + } + + static int allocationSize(MoqAudioRead value) { + return value.allocationSize(); + } + + static int write(MoqAudioRead value, Uint8List buf) { + return value.write(buf) - buf.offsetInBytes; + } +} + +class FrameMoqAudioRead extends MoqAudioRead { + final MoqAudioFrame frame; + FrameMoqAudioRead(MoqAudioFrame this.frame); + FrameMoqAudioRead._(MoqAudioFrame this.frame); + static LiftRetVal read(Uint8List buf) { + int new_offset = buf.offsetInBytes; + final frame_lifted = FfiConverterMoqAudioFrame.read( + Uint8List.view(buf.buffer, new_offset), + ); + final frame = frame_lifted.value; + new_offset += frame_lifted.bytesRead; + return LiftRetVal(FrameMoqAudioRead._(frame), new_offset); + } + + @override + RustBuffer lower() { + final buf = Uint8List(allocationSize()); + write(buf); + return toRustBuffer(buf); + } + + @override + int allocationSize() { + return FfiConverterMoqAudioFrame.allocationSize(frame) + 4; + } + + @override + int write(Uint8List buf) { + buf.buffer.asByteData(buf.offsetInBytes).setInt32(0, 1); + int new_offset = buf.offsetInBytes + 4; + new_offset += FfiConverterMoqAudioFrame.write( + frame, + Uint8List.view(buf.buffer, new_offset), + ); + return new_offset; + } +} + +class GapMoqAudioRead extends MoqAudioRead { + final int timestampUs; + final Duration duration; + GapMoqAudioRead({ + required int this.timestampUs, + required Duration this.duration, + }); + GapMoqAudioRead._(int this.timestampUs, Duration this.duration); + static LiftRetVal read(Uint8List buf) { + int new_offset = buf.offsetInBytes; + final timestampUs_lifted = FfiConverterUInt64.read( + Uint8List.view(buf.buffer, new_offset), + ); + final timestampUs = timestampUs_lifted.value; + new_offset += timestampUs_lifted.bytesRead; + final duration_lifted = FfiConverterDuration.read( + Uint8List.view(buf.buffer, new_offset), + ); + final duration = duration_lifted.value; + new_offset += duration_lifted.bytesRead; + return LiftRetVal(GapMoqAudioRead._(timestampUs, duration), new_offset); + } + + @override + RustBuffer lower() { + final buf = Uint8List(allocationSize()); + write(buf); + return toRustBuffer(buf); + } + + @override + int allocationSize() { + return FfiConverterUInt64.allocationSize(timestampUs) + + FfiConverterDuration.allocationSize(duration) + + 4; + } + + @override + int write(Uint8List buf) { + buf.buffer.asByteData(buf.offsetInBytes).setInt32(0, 2); + int new_offset = buf.offsetInBytes + 4; + new_offset += FfiConverterUInt64.write( + timestampUs, + Uint8List.view(buf.buffer, new_offset), + ); + new_offset += FfiConverterDuration.write( + duration, + Uint8List.view(buf.buffer, new_offset), + ); + return new_offset; + } +} + +class EndMoqAudioRead extends MoqAudioRead { + final int endUs; + EndMoqAudioRead(int this.endUs); + EndMoqAudioRead._(int this.endUs); + static LiftRetVal read(Uint8List buf) { + int new_offset = buf.offsetInBytes; + final endUs_lifted = FfiConverterUInt64.read( + Uint8List.view(buf.buffer, new_offset), + ); + final endUs = endUs_lifted.value; + new_offset += endUs_lifted.bytesRead; + return LiftRetVal(EndMoqAudioRead._(endUs), new_offset); + } + + @override + RustBuffer lower() { + final buf = Uint8List(allocationSize()); + write(buf); + return toRustBuffer(buf); + } + + @override + int allocationSize() { + return FfiConverterUInt64.allocationSize(endUs) + 4; + } + + @override + int write(Uint8List buf) { + buf.buffer.asByteData(buf.offsetInBytes).setInt32(0, 3); + int new_offset = buf.offsetInBytes + 4; + new_offset += FfiConverterUInt64.write( + endUs, + Uint8List.view(buf.buffer, new_offset), + ); + return new_offset; + } +} + enum MoqAudioSampleFormat { u8, s16, @@ -8032,6 +8215,189 @@ class FfiConverterMoqVideoPixelFormat { } } +abstract class MoqVideoRead { + RustBuffer lower(); + int allocationSize(); + int write(Uint8List buf); +} + +class FfiConverterMoqVideoRead { + static MoqVideoRead lift(RustBuffer buffer) { + return FfiConverterMoqVideoRead.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 = FrameMoqVideoRead.read(subview); + return LiftRetVal( + lifted.value, + lifted.bytesRead - subview.offsetInBytes + 4, + ); + case 2: + final lifted = GapMoqVideoRead.read(subview); + return LiftRetVal( + lifted.value, + lifted.bytesRead - subview.offsetInBytes + 4, + ); + case 3: + final lifted = EndMoqVideoRead.read(subview); + return LiftRetVal( + lifted.value, + lifted.bytesRead - subview.offsetInBytes + 4, + ); + default: + throw UniffiInternalError( + UniffiInternalError.unexpectedEnumCase, + "Unable to determine enum variant", + ); + } + } + + static RustBuffer lower(MoqVideoRead value) { + return value.lower(); + } + + static int allocationSize(MoqVideoRead value) { + return value.allocationSize(); + } + + static int write(MoqVideoRead value, Uint8List buf) { + return value.write(buf) - buf.offsetInBytes; + } +} + +class FrameMoqVideoRead extends MoqVideoRead { + final MoqVideoDecodedFrame frame; + FrameMoqVideoRead(MoqVideoDecodedFrame this.frame); + FrameMoqVideoRead._(MoqVideoDecodedFrame this.frame); + static LiftRetVal read(Uint8List buf) { + int new_offset = buf.offsetInBytes; + final frame_lifted = FfiConverterMoqVideoDecodedFrame.read( + Uint8List.view(buf.buffer, new_offset), + ); + final frame = frame_lifted.value; + new_offset += frame_lifted.bytesRead; + return LiftRetVal(FrameMoqVideoRead._(frame), new_offset); + } + + @override + RustBuffer lower() { + final buf = Uint8List(allocationSize()); + write(buf); + return toRustBuffer(buf); + } + + @override + int allocationSize() { + return FfiConverterMoqVideoDecodedFrame.allocationSize(frame) + 4; + } + + @override + int write(Uint8List buf) { + buf.buffer.asByteData(buf.offsetInBytes).setInt32(0, 1); + int new_offset = buf.offsetInBytes + 4; + new_offset += FfiConverterMoqVideoDecodedFrame.write( + frame, + Uint8List.view(buf.buffer, new_offset), + ); + return new_offset; + } +} + +class GapMoqVideoRead extends MoqVideoRead { + final int timestampUs; + final Duration duration; + GapMoqVideoRead({ + required int this.timestampUs, + required Duration this.duration, + }); + GapMoqVideoRead._(int this.timestampUs, Duration this.duration); + static LiftRetVal read(Uint8List buf) { + int new_offset = buf.offsetInBytes; + final timestampUs_lifted = FfiConverterUInt64.read( + Uint8List.view(buf.buffer, new_offset), + ); + final timestampUs = timestampUs_lifted.value; + new_offset += timestampUs_lifted.bytesRead; + final duration_lifted = FfiConverterDuration.read( + Uint8List.view(buf.buffer, new_offset), + ); + final duration = duration_lifted.value; + new_offset += duration_lifted.bytesRead; + return LiftRetVal(GapMoqVideoRead._(timestampUs, duration), new_offset); + } + + @override + RustBuffer lower() { + final buf = Uint8List(allocationSize()); + write(buf); + return toRustBuffer(buf); + } + + @override + int allocationSize() { + return FfiConverterUInt64.allocationSize(timestampUs) + + FfiConverterDuration.allocationSize(duration) + + 4; + } + + @override + int write(Uint8List buf) { + buf.buffer.asByteData(buf.offsetInBytes).setInt32(0, 2); + int new_offset = buf.offsetInBytes + 4; + new_offset += FfiConverterUInt64.write( + timestampUs, + Uint8List.view(buf.buffer, new_offset), + ); + new_offset += FfiConverterDuration.write( + duration, + Uint8List.view(buf.buffer, new_offset), + ); + return new_offset; + } +} + +class EndMoqVideoRead extends MoqVideoRead { + final int endUs; + EndMoqVideoRead(int this.endUs); + EndMoqVideoRead._(int this.endUs); + static LiftRetVal read(Uint8List buf) { + int new_offset = buf.offsetInBytes; + final endUs_lifted = FfiConverterUInt64.read( + Uint8List.view(buf.buffer, new_offset), + ); + final endUs = endUs_lifted.value; + new_offset += endUs_lifted.bytesRead; + return LiftRetVal(EndMoqVideoRead._(endUs), new_offset); + } + + @override + RustBuffer lower() { + final buf = Uint8List(allocationSize()); + write(buf); + return toRustBuffer(buf); + } + + @override + int allocationSize() { + return FfiConverterUInt64.allocationSize(endUs) + 4; + } + + @override + int write(Uint8List buf) { + buf.buffer.asByteData(buf.offsetInBytes).setInt32(0, 3); + int new_offset = buf.offsetInBytes + 4; + new_offset += FfiConverterUInt64.write( + endUs, + Uint8List.view(buf.buffer, new_offset), + ); + return new_offset; + } +} + abstract class MoqVideoSurface { RustBuffer lower(); int allocationSize(); @@ -8189,7 +8555,7 @@ class FfiConverterMoqAudioCodec { abstract class MoqAudioConsumerInterface { void cancel(); Duration delay(); - UniffiRustFuture next(); + UniffiRustFuture next(); } final _MoqAudioConsumerFinalizer = Finalizer>((ptr) { @@ -8265,7 +8631,7 @@ class MoqAudioConsumer implements MoqAudioConsumerInterface { ); } - UniffiRustFuture next() { + UniffiRustFuture next() { return uniffiRustCallAsync( () => uniffi_moq_ffi_fn_method_moqaudioconsumer_next(uniffiClonePointer()), @@ -8273,7 +8639,7 @@ class MoqAudioConsumer implements MoqAudioConsumerInterface { ffi_moq_ffi_rust_future_complete_rust_buffer, ffi_moq_ffi_rust_future_cancel_rust_buffer, ffi_moq_ffi_rust_future_free_rust_buffer, - FfiConverterOptionalMoqAudioFrame.lift, + FfiConverterOptionalMoqAudioRead.lift, moqExceptionErrorHandler, ); } @@ -13644,7 +14010,7 @@ class FfiConverterMoqTokens { abstract class MoqVideoConsumerInterface { void cancel(); - UniffiRustFuture next(); + UniffiRustFuture next(); } final _MoqVideoConsumerFinalizer = Finalizer>((ptr) { @@ -13709,7 +14075,7 @@ class MoqVideoConsumer implements MoqVideoConsumerInterface { }, null); } - UniffiRustFuture next() { + UniffiRustFuture next() { return uniffiRustCallAsync( () => uniffi_moq_ffi_fn_method_moqvideoconsumer_next(uniffiClonePointer()), @@ -13717,7 +14083,7 @@ class MoqVideoConsumer implements MoqVideoConsumerInterface { ffi_moq_ffi_rust_future_complete_rust_buffer, ffi_moq_ffi_rust_future_cancel_rust_buffer, ffi_moq_ffi_rust_future_free_rust_buffer, - FfiConverterOptionalMoqVideoDecodedFrame.lift, + FfiConverterOptionalMoqVideoRead.lift, moqExceptionErrorHandler, ); } @@ -14681,46 +15047,46 @@ class FfiConverterOptionalMoqAnnounceEvent { } } -class FfiConverterOptionalMoqAudioFrame { - static MoqAudioFrame? lift(RustBuffer buf) { - return FfiConverterOptionalMoqAudioFrame.read(buf.asUint8List()).value; +class FfiConverterOptionalMoqAudioRead { + static MoqAudioRead? lift(RustBuffer buf) { + return FfiConverterOptionalMoqAudioRead.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 = FfiConverterMoqAudioFrame.read( + final result = FfiConverterMoqAudioRead.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([MoqAudioFrame? value]) { + static int allocationSize([MoqAudioRead? value]) { if (value == null) { return 1; } - return FfiConverterMoqAudioFrame.allocationSize(value) + 1; + return FfiConverterMoqAudioRead.allocationSize(value) + 1; } - static RustBuffer lower(MoqAudioFrame? value) { + static RustBuffer lower(MoqAudioRead? value) { if (value == null) { return toRustBuffer(Uint8List.fromList([0])); } final buf = Uint8List( - FfiConverterOptionalMoqAudioFrame.allocationSize(value), + FfiConverterOptionalMoqAudioRead.allocationSize(value), ); - FfiConverterOptionalMoqAudioFrame.write(value, buf); + FfiConverterOptionalMoqAudioRead.write(value, buf); return toRustBuffer(buf); } - static int write(MoqAudioFrame? value, Uint8List buf) { + static int write(MoqAudioRead? value, Uint8List buf) { if (value == null) { buf[0] = 0; return 1; } buf[0] = 1; - return FfiConverterMoqAudioFrame.write( + return FfiConverterMoqAudioRead.write( value, Uint8List.view(buf.buffer, buf.offsetInBytes + 1), ) + @@ -15720,51 +16086,46 @@ class FfiConverterOptionalMoqTrackInfo { } } -class FfiConverterOptionalMoqVideoDecodedFrame { - static MoqVideoDecodedFrame? lift(RustBuffer buf) { - return FfiConverterOptionalMoqVideoDecodedFrame.read( - buf.asUint8List(), - ).value; +class FfiConverterOptionalMoqVideoGop { + static MoqVideoGop? lift(RustBuffer buf) { + return FfiConverterOptionalMoqVideoGop.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 = FfiConverterMoqVideoDecodedFrame.read( + final result = FfiConverterMoqVideoGop.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([MoqVideoDecodedFrame? value]) { + static int allocationSize([MoqVideoGop? value]) { if (value == null) { return 1; } - return FfiConverterMoqVideoDecodedFrame.allocationSize(value) + 1; + return FfiConverterMoqVideoGop.allocationSize(value) + 1; } - static RustBuffer lower(MoqVideoDecodedFrame? value) { + static RustBuffer lower(MoqVideoGop? value) { if (value == null) { return toRustBuffer(Uint8List.fromList([0])); } final buf = Uint8List( - FfiConverterOptionalMoqVideoDecodedFrame.allocationSize(value), + FfiConverterOptionalMoqVideoGop.allocationSize(value), ); - FfiConverterOptionalMoqVideoDecodedFrame.write(value, buf); + FfiConverterOptionalMoqVideoGop.write(value, buf); return toRustBuffer(buf); } - static int write(MoqVideoDecodedFrame? value, Uint8List buf) { + static int write(MoqVideoGop? value, Uint8List buf) { if (value == null) { buf[0] = 0; return 1; } buf[0] = 1; - return FfiConverterMoqVideoDecodedFrame.write( + return FfiConverterMoqVideoGop.write( value, Uint8List.view(buf.buffer, buf.offsetInBytes + 1), ) + @@ -15772,46 +16133,46 @@ class FfiConverterOptionalMoqVideoDecodedFrame { } } -class FfiConverterOptionalMoqVideoGop { - static MoqVideoGop? lift(RustBuffer buf) { - return FfiConverterOptionalMoqVideoGop.read(buf.asUint8List()).value; +class FfiConverterOptionalMoqVideoHint { + static MoqVideoHint? lift(RustBuffer buf) { + return FfiConverterOptionalMoqVideoHint.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 = FfiConverterMoqVideoGop.read( + final result = FfiConverterMoqVideoHint.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([MoqVideoGop? value]) { + static int allocationSize([MoqVideoHint? value]) { if (value == null) { return 1; } - return FfiConverterMoqVideoGop.allocationSize(value) + 1; + return FfiConverterMoqVideoHint.allocationSize(value) + 1; } - static RustBuffer lower(MoqVideoGop? value) { + static RustBuffer lower(MoqVideoHint? value) { if (value == null) { return toRustBuffer(Uint8List.fromList([0])); } final buf = Uint8List( - FfiConverterOptionalMoqVideoGop.allocationSize(value), + FfiConverterOptionalMoqVideoHint.allocationSize(value), ); - FfiConverterOptionalMoqVideoGop.write(value, buf); + FfiConverterOptionalMoqVideoHint.write(value, buf); return toRustBuffer(buf); } - static int write(MoqVideoGop? value, Uint8List buf) { + static int write(MoqVideoHint? value, Uint8List buf) { if (value == null) { buf[0] = 0; return 1; } buf[0] = 1; - return FfiConverterMoqVideoGop.write( + return FfiConverterMoqVideoHint.write( value, Uint8List.view(buf.buffer, buf.offsetInBytes + 1), ) + @@ -15819,46 +16180,46 @@ class FfiConverterOptionalMoqVideoGop { } } -class FfiConverterOptionalMoqVideoHint { - static MoqVideoHint? lift(RustBuffer buf) { - return FfiConverterOptionalMoqVideoHint.read(buf.asUint8List()).value; +class FfiConverterOptionalMoqVideoRead { + static MoqVideoRead? lift(RustBuffer buf) { + return FfiConverterOptionalMoqVideoRead.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 = FfiConverterMoqVideoHint.read( + final result = FfiConverterMoqVideoRead.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([MoqVideoHint? value]) { + static int allocationSize([MoqVideoRead? value]) { if (value == null) { return 1; } - return FfiConverterMoqVideoHint.allocationSize(value) + 1; + return FfiConverterMoqVideoRead.allocationSize(value) + 1; } - static RustBuffer lower(MoqVideoHint? value) { + static RustBuffer lower(MoqVideoRead? value) { if (value == null) { return toRustBuffer(Uint8List.fromList([0])); } final buf = Uint8List( - FfiConverterOptionalMoqVideoHint.allocationSize(value), + FfiConverterOptionalMoqVideoRead.allocationSize(value), ); - FfiConverterOptionalMoqVideoHint.write(value, buf); + FfiConverterOptionalMoqVideoRead.write(value, buf); return toRustBuffer(buf); } - static int write(MoqVideoHint? value, Uint8List buf) { + static int write(MoqVideoRead? value, Uint8List buf) { if (value == null) { buf[0] = 0; return 1; } buf[0] = 1; - return FfiConverterMoqVideoHint.write( + return FfiConverterMoqVideoRead.write( value, Uint8List.view(buf.buffer, buf.offsetInBytes + 1), ) + @@ -20363,7 +20724,7 @@ void _checkApiChecksums() { if (uniffi_moq_ffi_checksum_method_moqaudioconsumer_delay() != 48052) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } - if (uniffi_moq_ffi_checksum_method_moqaudioconsumer_next() != 5941) { + if (uniffi_moq_ffi_checksum_method_moqaudioconsumer_next() != 7650) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqaudioencoder_encode() != 56589) { @@ -20921,7 +21282,7 @@ void _checkApiChecksums() { if (uniffi_moq_ffi_checksum_method_moqvideoconsumer_cancel() != 21122) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } - if (uniffi_moq_ffi_checksum_method_moqvideoconsumer_next() != 52826) { + if (uniffi_moq_ffi_checksum_method_moqvideoconsumer_next() != 21151) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqvideodecodedframe_height() != 7330) { diff --git a/doc/lib/c/index.md b/doc/lib/c/index.md index 0b00db91e70..096214f0a64 100644 --- a/doc/lib/c/index.md +++ b/doc/lib/c/index.md @@ -71,7 +71,7 @@ inside. The header is the reference: each function carries a doc comment. - **Ending cleanly.** When the input ends, finish every track, then call `moq_publish_catalog_finish()` to finish the catalog, which ends the content. The broadcast stays announced and keeps serving the finished tracks, so a subscriber still catching up can drain them, until `moq_publish_close()` takes it offline. How long to keep it up is yours to decide. A track still open at the close ends with `UNROUTABLE`, which readers take as the publisher leaving, and its handle is released. - **Timeline breaks.** On a seek or pause, call `moq_publish_media_discontinuity`, then keep timestamps moving forward and resume video on a keyframe. - **JSON values.** `moq_consume_json_value` fills `moq_json_value` with the document and its frame's `timestamp_us`, rounded down to a whole microsecond; `timestamp_present` is false on an untimed track. A snapshot consumer's callback fires for each state in order, but a group superseded before it is read is skipped. -- **Decoded frames hold decoder buffers.** Free each frame id with `moq_decode_video_frame_free` promptly, or the decoder stalls. +- **Decoded reads.** `moq_decode_audio` and `moq_decode_video` deliver read ids in order: a frame, a gap where media was lost, or an end where the publisher stopped its encoder. Check `kind` on what `moq_decode_audio_read` or `moq_decode_video_read` writes. Free each id with `moq_decode_audio_read_free` or `moq_decode_video_read_free` promptly: a held video frame holds a decoder buffer, and the decoder stalls. - **Stats.** `moq_session_stats()` fills `moq_connection_stats` with `rtt_us`, `estimated_send_rate_bps`, and the byte and packet counters (`bytes_sent`, `bytes_received`, `bytes_lost`, `packets_sent`, `packets_received`, `packets_lost`). Each has a `_valid` flag, false when the transport does not report it, which is not the same as zero. `moq_session_snapshot()` adds the negotiated `protocol`. A failed dial or dropped connection reaches the session's `on_state` with its `failure` code, and `moq_error()` gives the reason during that callback. Hold the handle returned by `moq_session_recv_bandwidth()` to request receive estimates. `moq_bandwidth_consumer_peek()` reads the estimate and `moq_bandwidth_consumer_changed()` waits for a change. `moq_bandwidth_consumer_close()` releases demand; the last consumer stops PROBE. Stats snapshots do not request it. - **Demand.** A watcher from `moq_publish_track_demand()` (or the media, video, and audio variants) calls back `MOQ_DEMAND_USED` or `MOQ_DEMAND_UNUSED`, again whenever the wanted group moves. `moq_publish_demand_wanted()` reads that group: start a new one once it passes your largest. Set `refresh` in `moq_track_info` to have a joining subscriber ask for a new group once the newest has grown past `refresh` times its first frame. - **Disabled renditions.** Audio and video convenience decoders reject a catalog configuration with `enabled: false`. Select an enabled rendition from the latest catalog before opening a decoder. Among the video renditions you can decode, prefer the highest `preference`; `0` is the default and `-1` marks a compatibility transcode. diff --git a/doc/lib/dart/index.md b/doc/lib/dart/index.md index 71d38e2f432..94354f927d1 100644 --- a/doc/lib/dart/index.md +++ b/doc/lib/dart/index.md @@ -110,9 +110,12 @@ final decoder = await video.Consumer.subscribe( catalogVideo: (await renditions.next())!.video['camera']!, output: video.DecoderOutput(), ); -final frame = (await decoder.next())!; -print(frame.pixels(format: video.PixelFormat.rgba).length); -frame.dispose(); +// A read is a frame, a gap where media was lost, or an end where the publisher stopped its encoder. +final read = await decoder.next(); +if (read is video.ReadFrame) { + print(read.frame.pixels(format: video.PixelFormat.rgba).length); + read.frame.dispose(); +} ``` ```dart @@ -140,6 +143,7 @@ shows which module holds each. - **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. - **Timeline breaks.** On a seek or pause, call `discontinuity()`, then keep timestamps moving forward and resume video on a keyframe. +- **Decoder reads.** `audio.Consumer` and `video.Consumer` return a `Read` in order: a `ReadFrame`, a `ReadGap` where media was lost, or a `ReadEnd` where the publisher stopped its encoder. What follows a gap or an end does not continue what came before. - **Decoded frames hold decoder buffers.** `dispose()` each frame from a `video.Consumer` promptly rather than waiting for the garbage collector, or the decoder stalls. `resize` is best effort, so read each frame's `width()` and `height()`. - **Closing.** `await moq.close()` drains for up to one second and reports its result. `session.shutdown()` requests the drain synchronously; `await session.closed()` waits for its result. `session.cancel(code: 0)` closes at once. Finish or abort live tracks first. Awaiting cancellation of an announcement, grant, or request subscription releases its pending native read. Announcements also release their cursor; authorization and server listeners leave their shared owner usable. Application delivery requires explicitly selecting `moq-lite-07-wip` on both peers; older Lite versions (including default lite-06) and IETF wait only for transport ACKs. - **Ending cleanly.** When the input ends, finish every track, then call `finish()` on its `CatalogProducer` to finish the catalog, which ends the content. The broadcast stays announced and keeps serving the finished tracks, so a subscriber still catching up can drain them, until `broadcast.close()` takes it offline. How long to keep it up is yours to decide. A track still open at the close, or released without a finish, ends with `Unroutable`, which readers take as the publisher leaving. diff --git a/doc/lib/go/index.md b/doc/lib/go/index.md index 2a7f22fa20e..4b1163f1744 100644 --- a/doc/lib/go/index.md +++ b/doc/lib/go/index.md @@ -130,6 +130,7 @@ shows which module holds each. - **Bandwidth ownership.** An encoder you manage can reserve a share of the session's bandwidth. Reuse its reservation when the ceiling changes; before abandoning or replacing it, call `Update(0)` to return its share and automatic probe ceiling without waiting for garbage collection. The track can stay subscribed and the reservation can be reused. - **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. - **Timeline breaks.** On a seek or pause, call `Discontinuity()`, then keep timestamps moving forward and resume video on a keyframe. +- **Decoder reads.** `audio.Consumer` and `video.Consumer` return a `Read` in order: `ReadFrame`, a `ReadGap` where media was lost, or a `ReadEnd` where the publisher stopped its encoder. What follows a gap or an end does not continue what came before. - **Decoded frames hold decoder buffers.** `Close` each `VideoDecodedFrame` promptly, or the decoder stalls. `Resize` is best effort, so read each frame's `Width()` and `Height()`. - **Closing.** `client.Close()` drains for up to one second and returns an error on failure. `session.Shutdown()` requests that drain immediately; `session.Closed(ctx)` waits for its result, and `session.Cancel(code)` closes at once. Finish or abort live tracks first. Application delivery requires explicitly selecting `moq-lite-07-wip` on both peers; older Lite versions (including default lite-06) and IETF wait only for transport ACKs. - **Ending cleanly.** When the input ends, finish every track, then call `Finish()` on its `media.CatalogProducer` to finish the catalog, which ends the content. The broadcast stays announced and keeps serving the finished tracks, so a subscriber still catching up can drain them, until `broadcast.Close()` takes it offline. How long to keep it up is yours to decide. A track still open at the close, or released without a finish, ends with `Unroutable`, which readers take as the publisher leaving. diff --git a/doc/lib/kt/index.md b/doc/lib/kt/index.md index 6a40fa0c709..957330dd689 100644 --- a/doc/lib/kt/index.md +++ b/doc/lib/kt/index.md @@ -100,6 +100,7 @@ shows which module holds each. - **`end`, not `close`.** `broadcast.end()` ends a broadcast for good. `close()`, or leaving `use { }`, releases the handle, which ends the broadcast only once no `dynamic()` handle remains. - **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. - **Timeline breaks.** On a seek or pause, call `discontinuity()`, then keep timestamps moving forward and resume video on a keyframe. +- **Decoder reads.** `dev.moq.audio.Consumer` and `dev.moq.video.Consumer` return a `Read` in order: `ReadFrame`, a `ReadGap` where media was lost, or a `ReadEnd` where the publisher stopped its encoder. What follows a gap or an end does not continue what came before. - **Decoded frames hold decoder buffers.** `close()` each frame from a `dev.moq.video.Consumer` promptly (or `use { }` it), or the decoder stalls. `resize` is best effort, so read each frame's `width()` and `height()`. - **Closing.** `moq.shutdown()` requests a drain immediately; suspending `moq.session.closed()` waits up to one second and reports its result. `Moq.close()`, and so `use { }`, cancels at once. Finish or abort live tracks first. Application delivery requires explicitly selecting `moq-lite-07-wip` on both peers; older Lite versions (including default lite-06) and IETF wait only for transport ACKs. - **Ending cleanly.** When the input ends, finish every track, then call `finish()` on its `CatalogProducer` to finish the catalog, which ends the content. The broadcast stays announced and keeps serving the finished tracks, so a subscriber still catching up can drain them, until `broadcast.end()` takes it offline. How long to keep it up is yours to decide. A track still open at the end, or released without a finish, ends with `Unroutable`, which readers take as the publisher leaving. diff --git a/doc/lib/py/index.md b/doc/lib/py/index.md index 37420806652..dd040e4b347 100644 --- a/doc/lib/py/index.md +++ b/doc/lib/py/index.md @@ -114,6 +114,7 @@ shows which module holds each. - **Context managers.** `async with` on a client or session awaits a graceful shutdown when the body exits cleanly, and cancels at once when it raises, so your exception survives. - **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. - **Timeline breaks.** On a seek or pause, call `discontinuity()`, then keep timestamps moving forward and resume video on a keyframe. +- **Decoder reads.** `moq.audio.Consumer` and `moq.video.Consumer` yield, in order, frames, a `Gap` where media was lost, and an `End` where the publisher stopped its encoder. What follows a gap or an end does not continue what came before. - **Decoded frames hold decoder buffers.** Drop each frame from a `moq.video.Consumer` promptly, or the decoder stalls. `resize` is best effort, so read each frame's `width()` and `height()`. - **Closing.** `session.shutdown()` requests a drain immediately; `await session.closed()` waits up to one second and reports its result. `cancel(code)` closes at once. Finish or abort live tracks first. Application delivery requires explicitly selecting `moq-lite-07-wip` on both peers; older Lite versions (including default lite-06) and IETF wait only for transport ACKs. - **Ending cleanly.** When the input ends, finish every track, then call `finish()` on its `moq.media.CatalogProducer` to finish the catalog, which ends the content. The broadcast stays announced and keeps serving the finished tracks, so a subscriber still catching up can drain them, until `broadcast.close()` takes it offline. How long to keep it up is yours to decide. A track still open at the close, or released without a finish, ends with `Unroutable`, which readers take as the publisher leaving. diff --git a/doc/lib/swift/index.md b/doc/lib/swift/index.md index 566584e526a..acc00300fbf 100644 --- a/doc/lib/swift/index.md +++ b/doc/lib/swift/index.md @@ -100,6 +100,7 @@ shows which module holds each. - **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. - **Timeline breaks.** On a seek or pause, call `discontinuity()`, then keep timestamps moving forward and resume video on a keyframe. +- **Decoder reads.** `Audio.Consumer` and `Video.Consumer` yield `Read` cases in order: `.frame`, a `.gap` where media was lost, or an `.end` where the publisher stopped its encoder. What follows a gap or an end does not continue what came before. - **Decoded frames hold decoder buffers.** Release each frame from a `Video.Consumer` promptly, or the decoder stalls. `resize` is best effort, so read each frame's `width()` and `height()`. - **Closing.** `session.shutdown()` requests a drain immediately; `try await session.closed()` waits up to one second and reports its result. `session.cancel(code:)` closes at once. Finish or abort live tracks first. Application delivery requires explicitly selecting `moq-lite-07-wip` on both peers; older Lite versions (including default lite-06) and IETF wait only for transport ACKs. - **Ending cleanly.** When the input ends, finish every track, then call `try finish()` on its `Media.CatalogProducer` to finish the catalog, which ends the content. The broadcast stays announced and keeps serving the finished tracks, so a subscriber still catching up can drain them, until `broadcast.close()` takes it offline. How long to keep it up is yours to decide. A track still open at the close, or released without a finish, ends with `Unroutable`, which readers take as the publisher leaving. diff --git a/go/wrapper/MIGRATE.md b/go/wrapper/MIGRATE.md index cb43e263b63..0a9d505c875 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 +### Decoders read gaps and ends + +`audio.Consumer.Next` and `video.Consumer.Next` return a `Read` (and `All` ranges over them) instead of a frame pointer. Switch on `ReadFrame` and take its `Frame` where you used the frame; `ReadGap` marks lost media and `ReadEnd` where the publisher stopped its encoder, and what follows either does not continue what came before. A nil `Read` still means the track ended. + ### Durations cross as `time.Duration` The re-exported records carry `time.Duration` instead of microseconds: `FetchGroupOptions.MaxDelay`, `media.Audio.Jitter`, and the raw `AudioDecoderOutput.{MaxDelay, Delay}`, `AudioEncoderOutput.FrameDuration`, and `VideoDecoderOutput.MaxDelay`, each a `*time.Duration` where it was a `*uint64` `...Us` field. diff --git a/go/wrapper/audio/audio.go b/go/wrapper/audio/audio.go index 9da5dafa368..1f42c597fdb 100644 --- a/go/wrapper/audio/audio.go +++ b/go/wrapper/audio/audio.go @@ -3,6 +3,7 @@ package audio import ( "context" + "fmt" "iter" "time" @@ -78,6 +79,46 @@ func frame(raw ffi.MoqAudioFrame) Frame { return Frame{Timestamp: time.Duration(raw.TimestampUs) * time.Microsecond, Data: raw.Data} } +// Read is one item a Consumer reads, in delivery order: a ReadFrame, ReadGap, or ReadEnd. +type Read interface{ read() } + +// ReadFrame is decoded PCM. +type ReadFrame struct { + Frame Frame +} + +// ReadGap is media lost after what was already read: what follows does not continue it. +type ReadGap struct { + // Timestamp is the furthest presentation point already read. + Timestamp time.Duration + // Duration is an upper bound on how long until the next frame resumes. + Duration time.Duration +} + +// ReadEnd is where the publisher stopped its encoder: it paused, switched source, or finished. +// Frames that follow and start before End are its remaining media; the first at or past End +// comes from a fresh encoder. +type ReadEnd struct { + // End is where the stopped encoder's media stops. + End time.Duration +} + +func (ReadFrame) read() {} +func (ReadGap) read() {} +func (ReadEnd) read() {} + +func audioRead(raw ffi.MoqAudioRead) (Read, error) { + switch read := raw.(type) { + case ffi.MoqAudioReadFrame: + return ReadFrame{Frame: frame(read.Frame)}, nil + case ffi.MoqAudioReadGap: + return ReadGap{Timestamp: time.Duration(read.TimestampUs) * time.Microsecond, Duration: read.Duration}, nil + case ffi.MoqAudioReadEnd: + return ReadEnd{End: time.Duration(read.EndUs) * time.Microsecond}, nil + } + return nil, fmt.Errorf("unknown audio read: %T", raw) +} + // Producer encodes PCM and publishes it as a track in a broadcast's catalog. type Producer struct{ inner *ffi.MoqAudioProducer } @@ -194,18 +235,18 @@ func Subscribe(ctx context.Context, broadcast *moq.BroadcastConsumer, id string, return &Consumer{inner}, nil } -// Next returns the next decoded frame, or (nil, nil) when the track ends. -func (d *Consumer) Next(ctx context.Context) (*Frame, error) { +// Next returns the next decoded frame, gap, or encoder end, or (nil, nil) when the track ends. +func (d *Consumer) Next(ctx context.Context) (Read, error) { raw, err := bridge.Call(ctx, d.inner.Cancel, d.inner.Next) if err != nil || raw == nil { return nil, err } - value := frame(*raw) - return &value, nil + return audioRead(*raw) } -// All ranges over decoded frames until the track ends or the loop breaks. -func (d *Consumer) All(ctx context.Context) iter.Seq2[*Frame, error] { return bridge.Seq(ctx, d.Next) } +// All ranges over every read until the track ends or the loop breaks. +func (d *Consumer) All(ctx context.Context) iter.Seq2[Read, error] { return bridge.Seq(ctx, d.Next) } + // Delay is the playout delay to hold ahead of the playhead: the fixed one, or the estimate from every frame read so far. // The estimate measures when Next returns, so call it again as soon as the previous frame is handled. diff --git a/go/wrapper/internal/bridge/call.go b/go/wrapper/internal/bridge/call.go index 7e1f9a848ed..b2611ec6278 100644 --- a/go/wrapper/internal/bridge/call.go +++ b/go/wrapper/internal/bridge/call.go @@ -89,7 +89,15 @@ func CallHandle[T Handle](ctx context.Context, cancel func(), call func(context. return run(ctx, cancel, func() (T, error) { return call(objectCallContext(ctx, cancel)) }, releaseHandle[T]) } +// CallRelease is CallHandle for a result that may hold a native handle without +// being one, such as an enum with a handle in one variant: release disposes of +// whatever a result nobody received holds. +func CallRelease[T any](ctx context.Context, cancel func(), call func(context.Context) (T, error), release func(T)) (T, error) { + return run(ctx, cancel, func() (T, error) { return call(objectCallContext(ctx, cancel)) }, release) +} + // objectCallContext is the context the generated binding sees. Object-owned + // calls already abort via cancel; sharing the caller's ctx lets the binding // return ctx.Err() first and skip that abort, leaving the stream, session, or // request active. diff --git a/go/wrapper/moq_test.go b/go/wrapper/moq_test.go index 1b39e8a8e52..0f24deccc50 100644 --- a/go/wrapper/moq_test.go +++ b/go/wrapper/moq_test.go @@ -305,8 +305,10 @@ func TestAudioConsumerDelay(t *testing.T) { } } for _, consumer := range []*moqaudio.Consumer{fixed, estimated} { - if frame, err := consumer.Next(ctx); err != nil || frame == nil { - t.Fatalf("next = %v, %v", frame, err) + if read, err := consumer.Next(ctx); err != nil { + t.Fatal(err) + } else if _, ok := read.(moqaudio.ReadFrame); !ok { + t.Fatalf("next = %#v, want a frame", read) } } @@ -472,14 +474,17 @@ func testDecodeVideoFrame(t *testing.T, surface bool) { } } - frame, err := decoder.Next(ctx) + read, err := decoder.Next(ctx) if err != nil { t.Fatal(err) } - if frame == nil { - t.Fatal("expected a frame") + decoded, ok := read.(moqvideo.ReadFrame) + if !ok { + t.Fatalf("next = %#v, want a frame", read) } + frame := decoded.Frame defer frame.Close() + decoder.Cancel() if !surface { diff --git a/go/wrapper/video/video.go b/go/wrapper/video/video.go index bfbff85a324..15d6ec16348 100644 --- a/go/wrapper/video/video.go +++ b/go/wrapper/video/video.go @@ -3,9 +3,11 @@ package video import ( "context" + "fmt" "iter" "time" + "moq.dev/moq" ffi "moq.dev/moq-ffi/moq" "moq.dev/moq/internal/bridge" @@ -192,24 +194,66 @@ func Subscribe(ctx context.Context, broadcast *moq.BroadcastConsumer, id string, return &Consumer{inner}, nil } -// Next returns the next decoded frame, or (nil, nil) when the track ends. -// Close each frame when done with it. -func (d *Consumer) Next(ctx context.Context) (*moq.VideoDecodedFrame, error) { - raw, err := bridge.CallHandle(ctx, d.inner.Cancel, func(ctx context.Context) (*ffi.MoqVideoDecodedFrame, error) { - raw, err := d.inner.Next(ctx) - if err != nil || raw == nil { - return nil, err - } - return *raw, nil - }) +// Read is one item a Consumer reads, in delivery order: a ReadFrame, ReadGap, or ReadEnd. +type Read interface{ read() } + +// ReadFrame is a decoded picture. Close its frame when done with it, since held +// frames hold decoder buffers. +type ReadFrame struct { + Frame *moq.VideoDecodedFrame +} + +// ReadGap is media lost after what was already read: every picture from before the +// loss precedes it, and what follows does not continue it. +type ReadGap struct { + // Timestamp is the furthest presentation point already read. + Timestamp time.Duration + // Duration is an upper bound on how long until the next picture resumes. + Duration time.Duration +} + +// ReadEnd is where the publisher stopped its encoder: it paused, switched source, or +// finished. The stopped encoder's last pictures precede it; what follows comes from a +// fresh encoder. +type ReadEnd struct { + // End is where the stopped encoder's media stops. + End time.Duration +} + +func (ReadFrame) read() {} +func (ReadGap) read() {} +func (ReadEnd) read() {} + +// releaseRead destroys the decoded frame a read nobody received holds. +func releaseRead(raw *ffi.MoqVideoRead) { + if raw == nil { + return + } + if read, ok := (*raw).(ffi.MoqVideoReadFrame); ok { + read.Frame.Destroy() + } +} + +// Next returns the next decoded frame, gap, or encoder end, or (nil, nil) when the +// track ends. Close each frame when done with it. +func (d *Consumer) Next(ctx context.Context) (Read, error) { + raw, err := bridge.CallRelease(ctx, d.inner.Cancel, d.inner.Next, releaseRead) if err != nil || raw == nil { return nil, err } - return bridge.VideoDecodedFrame(raw).(*moq.VideoDecodedFrame), nil + switch read := (*raw).(type) { + case ffi.MoqVideoReadFrame: + return ReadFrame{Frame: bridge.VideoDecodedFrame(read.Frame).(*moq.VideoDecodedFrame)}, nil + case ffi.MoqVideoReadGap: + return ReadGap{Timestamp: time.Duration(read.TimestampUs) * time.Microsecond, Duration: read.Duration}, nil + case ffi.MoqVideoReadEnd: + return ReadEnd{End: time.Duration(read.EndUs) * time.Microsecond}, nil + } + return nil, fmt.Errorf("unknown video read: %T", *raw) } -// All ranges over decoded frames until the track ends or the loop breaks. -func (d *Consumer) All(ctx context.Context) iter.Seq2[*moq.VideoDecodedFrame, error] { +// All ranges over every read until the track ends or the loop breaks. +func (d *Consumer) All(ctx context.Context) iter.Seq2[Read, error] { return bridge.Seq(ctx, d.Next) } diff --git a/kt/moq/MIGRATE.md b/kt/moq/MIGRATE.md index 649b519154e..d6372846093 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 +### Decoders read gaps and ends + +`dev.moq.audio.Consumer.next()` and `dev.moq.video.Consumer.next()` return a `Read` instead of a frame, and their `frames()` flows become `reads()`. Take `(read as? ReadFrame)?.frame` where you used the frame; `ReadGap` marks lost media and `ReadEnd` where the publisher stopped its encoder, and what follows either does not continue what came before. + ### Durations are `java.time.Duration` The records carry durations as `java.time.Duration` fields instead of `*Us` microsecond integers, so the duration extension properties are gone: read `backoff.initial`, `stats.rtt`, `grant.expires`, `info.maxAge`, `fill.maxDelay()`, and `consumer.delay()` directly. diff --git a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Flows.kt b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Flows.kt index 3795a732399..366009c8ebd 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Flows.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Flows.kt @@ -9,16 +9,19 @@ import kotlinx.coroutines.flow.onCompletion import uniffi.moq.MoqException /** - * Stream of decoded audio frames in the layout declared by the - * [AudioDecoderOutput] the consumer was created with. + * Stream of reads: decoded audio frames in the layout declared by the + * [AudioDecoderOutput] the consumer was created with, and the gaps and encoder + * ends between them. */ -fun dev.moq.audio.Consumer.frames(): Flow = pull({ cancel() }) { next() } +fun dev.moq.audio.Consumer.reads(): Flow = pull({ cancel() }) { next() } /** - * Stream of decoded video frames. Each owns the decoder's surface: close it - * (or `use` it) when done, since held frames stall the decoder. + * Stream of reads: decoded video frames, and the gaps and encoder ends between + * them. Each frame owns the decoder's surface: close it (or `use` it) when done, + * since held frames stall the decoder. */ -fun dev.moq.video.Consumer.frames(): Flow = pull({ cancel() }) { next() } +fun dev.moq.video.Consumer.reads(): Flow = pull({ cancel() }) { next() } + /** Stream of groups in sequence order, skipping forward if the reader falls behind. */ fun TrackConsumer.groups(): Flow = pull({ cancel() }) { nextGroup() } diff --git a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/audio/Audio.kt b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/audio/Audio.kt index 9116cb197cb..ced12836d93 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/audio/Audio.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/audio/Audio.kt @@ -4,7 +4,7 @@ package dev.moq.audio typealias Producer = uniffi.moq.MoqAudioProducer /** Encodes PCM without publishing it. */ typealias Encoder = uniffi.moq.MoqAudioEncoder -/** Subscribes to a catalog audio rendition and yields decoded PCM; open one with `Consumer.subscribe`. */ +/** Subscribes to a catalog audio rendition and yields each [Read]; open one with `Consumer.subscribe`. */ typealias Consumer = uniffi.moq.MoqAudioConsumer /** The PCM layout written to a [Producer]. */ typealias EncoderInput = uniffi.moq.MoqAudioEncoderInput @@ -14,6 +14,15 @@ typealias EncoderOutput = uniffi.moq.MoqAudioEncoderOutput typealias DecoderOutput = uniffi.moq.MoqAudioDecoderOutput /** PCM samples in the configured layout with their presentation timestamp. */ typealias Frame = uniffi.moq.MoqAudioFrame +/** One item a [Consumer] reads, in delivery order: [ReadFrame], [ReadGap], or [ReadEnd]. */ +typealias Read = uniffi.moq.MoqAudioRead +/** Decoded PCM. */ +typealias ReadFrame = uniffi.moq.MoqAudioRead.Frame +/** Media after what was already read was lost; what follows does not continue it. */ +typealias ReadGap = uniffi.moq.MoqAudioRead.Gap +/** The publisher stopped its encoder; frames starting before `endUs` are its remaining media. */ +typealias ReadEnd = uniffi.moq.MoqAudioRead.End + /** One encoded payload from an [Encoder]. */ typealias Packet = uniffi.moq.MoqAudioPacket /** The packets an [Encoder] drains on finish and the padded samples to discard. */ diff --git a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/video/Video.kt b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/video/Video.kt index 86cbc5b29e3..fb99df34bbc 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/video/Video.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/video/Video.kt @@ -4,7 +4,7 @@ package dev.moq.video typealias Producer = uniffi.moq.MoqVideoProducer /** Encodes raw pictures without publishing them. */ typealias Encoder = uniffi.moq.MoqVideoEncoder -/** Subscribes to a catalog video rendition and yields decoded frames; open one with `Consumer.subscribe`. */ +/** Subscribes to a catalog video rendition and yields each [Read]; open one with `Consumer.subscribe`. */ typealias Consumer = uniffi.moq.MoqVideoConsumer /** The pixel layout, resolution, and framerate written to a [Producer]. */ typealias EncoderInput = uniffi.moq.MoqVideoEncoderInput @@ -14,6 +14,15 @@ typealias EncoderOutput = uniffi.moq.MoqVideoEncoderOutput typealias DecoderOutput = uniffi.moq.MoqVideoDecoderOutput /** One raw picture with its presentation timestamp. */ typealias Frame = uniffi.moq.MoqVideoFrame +/** One item a [Consumer] reads, in delivery order: [ReadFrame], [ReadGap], or [ReadEnd]. */ +typealias Read = uniffi.moq.MoqVideoRead +/** A decoded picture; `close()` its frame promptly, since held frames stall the decoder. */ +typealias ReadFrame = uniffi.moq.MoqVideoRead.Frame +/** Media after what was already read was lost; what follows does not continue it. */ +typealias ReadGap = uniffi.moq.MoqVideoRead.Gap +/** The publisher stopped its encoder; what follows comes from a fresh encoder. */ +typealias ReadEnd = uniffi.moq.MoqVideoRead.End + /** One encoded access unit from an [Encoder]. */ typealias Packet = uniffi.moq.MoqVideoPacket /** How pictures are split into independently decodable groups. */ diff --git a/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt b/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt index 11c19642951..716df28d00a 100644 --- a/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt +++ b/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt @@ -2,9 +2,11 @@ package dev.moq import dev.moq.media.* import dev.moq.audio.Consumer as AudioConsumer +import dev.moq.audio.ReadFrame as AudioReadFrame import dev.moq.audio.Producer as AudioProducer import dev.moq.audio.ProducerOptions as AudioProducerOptions import dev.moq.video.Consumer as VideoConsumer +import dev.moq.video.ReadFrame as VideoReadFrame import dev.moq.video.Producer as VideoProducer import dev.moq.video.ProducerOptions as VideoProducerOptions import dev.moq.video.FrameRate as VideoFrameRate @@ -399,9 +401,10 @@ class SmokeTest { * by the interop test (`just test interop`). */ @Test - fun `video consumer exposes a frames flow like audio does`() { - val video: (dev.moq.video.Consumer) -> Flow = dev.moq.video.Consumer::frames - val audio: (dev.moq.audio.Consumer) -> Flow = dev.moq.audio.Consumer::frames + fun `video consumer exposes a reads flow like audio does`() { + val video: (dev.moq.video.Consumer) -> Flow = dev.moq.video.Consumer::reads + val audio: (dev.moq.audio.Consumer) -> Flow = dev.moq.audio.Consumer::reads + assertNotNull(video) assertNotNull(audio) } @@ -745,8 +748,8 @@ class SmokeTest { for (i in 0 until 2) { audio.write(AudioFrame(timestampUs = i.toULong() * 20_000uL, data = ByteArray(960 * 4))) } - assertNotNull(fixed.next()) - assertNotNull(estimated.next()) + assertIs(fixed.next()) + assertIs(estimated.next()) assertEquals(java.time.Duration.ofMillis(80), fixed.delay()) assertTrue(estimated.delay() != initial, "the estimate should move as frames are read") @@ -791,7 +794,8 @@ class SmokeTest { video.write(VideoFrame(timestampUs = i.toULong() * 33_333uL, data = rgba)) } - decoder.next()!!.use { frame -> + assertIs(decoder.next()).frame.use { frame -> + // The frame outlives its consumer's cancellation. decoder.cancel() diff --git a/py/moq-rs/MIGRATE.md b/py/moq-rs/MIGRATE.md index 2dfbe197af7..615ba80caaa 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 +### Decoders read gaps and ends + +`audio.Consumer` yields `audio.Read` (`audio.Frame | audio.Gap | audio.End`) and `video.Consumer` yields `video.Read` (`VideoDecodedFrame | video.Gap | video.End`). Check `isinstance(read, audio.Frame)` (or `VideoDecodedFrame`) where you used each item as a frame; a `Gap` marks lost media and an `End` where the publisher stopped its encoder, and what follows either does not continue what came before. + ### Audio frame duration is optional `audio.EncoderOutput.frame_duration` defaults to `None` for the codec's own frame. Pass `None` where you passed `timedelta(0)`. diff --git a/py/moq-rs/moq/_read.py b/py/moq-rs/moq/_read.py new file mode 100644 index 00000000000..2eac1f4984f --- /dev/null +++ b/py/moq-rs/moq/_read.py @@ -0,0 +1,40 @@ +"""The breaks between decoded frames, shared by :mod:`moq.audio` and :mod:`moq.video`.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import timedelta + +from ._records import _from_us + + +@dataclass(frozen=True) +class Gap: + """Media after what was already read was lost: a group was skipped or never arrived. + + What follows does not continue what came before. ``duration`` is an upper bound + on how long until the next frame resumes, starting at ``timestamp``, the + furthest presentation point already read. + """ + + timestamp: timedelta + duration: timedelta + + +@dataclass(frozen=True) +class End: + """The publisher stopped its encoder: it paused, switched source, or finished. + + Frames that follow and start before ``end`` are the stopped encoder's + remaining media; the first at or past it comes from a fresh encoder. + """ + + end: timedelta + + +def _gap(timestamp_us: int, duration: timedelta) -> Gap: + return Gap(timestamp=_from_us(timestamp_us), duration=duration) + + +def _end(end_us: int) -> End: + return End(end=_from_us(end_us)) diff --git a/py/moq-rs/moq/audio.py b/py/moq-rs/moq/audio.py index 45049f910b3..38c54853ed2 100644 --- a/py/moq-rs/moq/audio.py +++ b/py/moq-rs/moq/audio.py @@ -15,8 +15,10 @@ MoqAudioPacket, MoqAudioProducer, MoqAudioProducerOptions, + MoqAudioRead, ) +from ._read import End, Gap, _end, _gap from ._records import _from_us, _to_us from ._stream import _Stream from .media import Audio, AudioInit, CatalogProducer @@ -49,6 +51,10 @@ def _from_ffi(value: MoqAudioFrame) -> Frame: return Frame(timestamp=_from_us(value.timestamp_us), data=value.data) +Read = Frame | Gap | End +"""One item a :class:`Consumer` reads, in delivery order: decoded PCM, or a break in it.""" + + @dataclass(frozen=True) class EncoderOutput: """Codec-side settings. @@ -147,11 +153,13 @@ def finish(self) -> None: self._inner.finish() -class Consumer(_Stream[Frame]): - """Async iterator of decoded PCM frames. Construct with :meth:`subscribe`. +class Consumer(_Stream[Read]): + """Async iterator of decoded PCM :class:`Frame` reads, and the :class:`Gap` and + :class:`End` breaks between them. Construct with :meth:`subscribe`. The layout is fixed by the :class:`DecoderOutput` passed at subscribe time; - each frame's ``data`` is raw bytes in that format. + each frame's ``data`` is raw bytes in that format. Restart playback after a + gap or an end, since what follows does not continue what came before. """ @classmethod @@ -164,9 +172,17 @@ async def subscribe( def __init__(self, inner: MoqAudioConsumer) -> None: self._inner = inner - async def _next(self) -> Frame | None: + async def _next(self) -> Read | None: value = await self._inner.next() - return None if value is None else Frame._from_ffi(value) + if value is None: + return None + if isinstance(value, MoqAudioRead.FRAME): + return Frame._from_ffi(value.frame) + if isinstance(value, MoqAudioRead.GAP): + return _gap(value.timestamp_us, value.duration) + if isinstance(value, MoqAudioRead.END): + return _end(value.end_us) + raise TypeError(f"unknown audio read: {value!r}") @property def delay(self) -> timedelta: @@ -217,9 +233,12 @@ def finish(self, pcm: list[float] | None = None) -> EncoderFinish: "EncoderInput", "EncoderFinish", "EncoderOutput", + "End", "Frame", + "Gap", "Codec", "Packet", "Producer", + "Read", "SampleFormat", ] diff --git a/py/moq-rs/moq/video.py b/py/moq-rs/moq/video.py index 6eae36746f8..dc3ddd352aa 100644 --- a/py/moq-rs/moq/video.py +++ b/py/moq-rs/moq/video.py @@ -4,6 +4,7 @@ from dataclasses import dataclass from datetime import timedelta +from typing import Union from moq_ffi import ( MoqVideoConsumer, @@ -13,9 +14,11 @@ MoqVideoPacket, MoqVideoProducer, MoqVideoProducerOptions, + MoqVideoRead, ) from moq_ffi import MoqVideoFrameRate as FrameRate +from ._read import End, Gap, _end, _gap from ._records import _from_us, _to_us from ._stream import _Stream from .media import CatalogProducer, Dimensions, Video @@ -30,6 +33,10 @@ from .types import VideoGop as Gop from .types import VideoPixelFormat as PixelFormat +# A Union rather than `|`: the generated frame class does not support `|` with other types. +Read = Union[VideoDecodedFrame, Gap, End] +"""One item a :class:`Consumer` reads, in delivery order: a decoded picture, or a break between them.""" + @dataclass(frozen=True) class Frame: @@ -135,8 +142,9 @@ def finish(self) -> None: self._inner.finish() -class Consumer(_Stream[VideoDecodedFrame]): - """Async iterator of decoded frames. Construct with :meth:`subscribe`. +class Consumer(_Stream[Read]): + """Async iterator of decoded frames, and the :class:`Gap` and :class:`End` breaks + between them. Construct with :meth:`subscribe`. Each frame owns its decoded picture until it is garbage collected; ``frame.pixels(format)`` converts it to tightly packed CPU pixels. @@ -154,8 +162,17 @@ async def subscribe( def __init__(self, inner: MoqVideoConsumer) -> None: self._inner = inner - async def _next(self) -> VideoDecodedFrame | None: - return await self._inner.next() + async def _next(self) -> Read | None: + value = await self._inner.next() + if value is None: + return None + if isinstance(value, MoqVideoRead.FRAME): + return value.frame + if isinstance(value, MoqVideoRead.GAP): + return _gap(value.timestamp_us, value.duration) + if isinstance(value, MoqVideoRead.END): + return _end(value.end_us) + raise TypeError(f"unknown video read: {value!r}") def cancel(self) -> None: """Cancel the subscription and stop delivering video frames.""" @@ -197,10 +214,13 @@ def finish(self) -> list[Packet]: "EncoderInput", "EncoderKind", "EncoderOutput", + "End", "Frame", + "Gap", "Gop", "Packet", "PixelFormat", "Producer", + "Read", "FrameRate", ] diff --git a/py/moq-rs/tests/test_local.py b/py/moq-rs/tests/test_local.py index 12ea0dd07d2..bfe350c6758 100644 --- a/py/moq-rs/tests/test_local.py +++ b/py/moq-rs/tests/test_local.py @@ -1490,8 +1490,8 @@ def output(delay: timedelta | None) -> moq.audio.DecoderOutput: # Two 20 ms Opus frames of silence at 48 kHz mono. for i in range(2): audio.write(moq.audio.Frame(timestamp=timedelta(milliseconds=20 * i), data=bytes(960 * 4))) - await asyncio.wait_for(anext(fixed), timeout=5.0) - await asyncio.wait_for(anext(estimated), timeout=5.0) + assert isinstance(await asyncio.wait_for(anext(fixed), timeout=5.0), moq.audio.Frame) + assert isinstance(await asyncio.wait_for(anext(estimated), timeout=5.0), moq.audio.Frame) assert fixed.delay == timedelta(milliseconds=80) assert estimated.delay != initial @@ -1544,8 +1544,10 @@ async def test_decode_video_frame(): frame = await asyncio.wait_for(anext(decoder), timeout=5.0) decoder.cancel() + assert isinstance(frame, moq.VideoDecodedFrame) i420 = frame.pixels(moq.VideoPixelFormat.I420) + assert len(i420) == frame.width() * frame.height() * 3 // 2 packed = frame.pixels(moq.VideoPixelFormat.RGBA) diff --git a/quest/m1/api-audit/README.md b/quest/m1/api-audit/README.md index f1e5976e555..d81f5d8ffb0 100644 --- a/quest/m1/api-audit/README.md +++ b/quest/m1/api-audit/README.md @@ -75,7 +75,6 @@ Rejected in planning: - [Two broadcast resolutions](/quest/m1/api-audit/resolve.md) - Rust, JS, and the bindings resolve a broadcast with `request(path, epoch)` or `follow(path, epoch)`, and every racy two-step caller is gone - [Binding publications](/quest/m1/api-audit/binding-publication.md) - moq-ffi, the C ABI, and every wrapper hand out a publication handle that announces, replacing the private adapters -- [Bindings read gaps and ends](/quest/m1/api-audit/decode-read-bindings.md) - moq-ffi and every wrapper surface `Gap` and `End` instead of discarding them - [Encoded consumers read gaps and ends](/quest/m1/api-audit/encoded-read-bindings.md) - moq-ffi, moq-c, and every wrapper read encoded media with the tagged-record shape the decoded consumers use - [One lifecycle vocabulary](/quest/m1/api-audit/lifecycle.md) - finish, abort, closed, and close mean one thing on every handle, and Rust and JS observers fire alike - [moq-net names say one thing](/quest/m1/api-audit/net-names.md) - demand says `wanted`, colliding stats and transport names are renamed, and Rust and JS info lookups mirror diff --git a/quest/m1/api-audit/decode-read-bindings.md b/quest/m1/api-audit/decode-read-bindings.md deleted file mode 100644 index 31847455ea6..00000000000 --- a/quest/m1/api-audit/decode-read-bindings.md +++ /dev/null @@ -1,26 +0,0 @@ -# [L] Bindings read decoder gaps and ends - -## Goal - -Every binding reads the same audio and video decode stream as Rust: -frames, a gap where groups were skipped, and the end. moq-ffi, the generated -C and C++, Python, Swift, Kotlin, Dart, and Go stop discarding `Gap` and -`End`. - -## Plan - -moq-audio and moq-video yield `decode::Read { Frame, Gap, End }`, but moq-ffi (`audio.rs`, -`video.rs`), the hand-written moq-c, and the interop audio client still -hand over frames only. - -Open for this quest: the binding shape. Options are a tagged record per read -(mirrors Rust, one call), or frames plus a separate gap and end signal -(smaller change for wrappers). Recommend the tagged record, mirroring Rust -names, and ask the maintainer before implementing. - -Breaking in moq-ffi and every wrapper: a migration note per package in the Impact bullet, and -the Cross-Package Sync table's docs (`doc/lib/*`). Run `just test interop --all`. - -Decided 2026-10-23 (maintainer): tracked under the API audit. - -Public API: breaking, every binding. Wire: none. diff --git a/rs/moq-c/MIGRATE.md b/rs/moq-c/MIGRATE.md index 27fed02ed2a..66c25f322b4 100644 --- a/rs/moq-c/MIGRATE.md +++ b/rs/moq-c/MIGRATE.md @@ -4,6 +4,10 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Decoders deliver reads + +`moq_decode_audio` and `moq_decode_video` call back with a read id per decoded frame, gap, or encoder end instead of a frame id. Replace `moq_decode_audio_frame` / `moq_decode_video_frame` with `moq_decode_audio_read` / `moq_decode_video_read`, which fill a `moq_audio_read` / `moq_video_read`: check `kind` (`MOQ_DECODE_READ_FRAME`, `MOQ_DECODE_READ_GAP`, `MOQ_DECODE_READ_END`) and take the frame from its `frame` field, the lost span from `gap`, or `end_us`. Free every id with `moq_decode_audio_read_free` / `moq_decode_video_read_free` where you called the `_frame_free` functions. Skip gap and end reads to keep the old behavior, or restart playback there, since what follows does not continue what came before. + ### Sessions report a lifecycle snapshot `moq_session_connect` and `moq_session_request_accept` take a `moq_state_callback` instead of a `moq_status_callback`: `void on_state(void *user_data, const moq_connection_state *state)`. diff --git a/rs/moq-c/c-tests/audio-decode-output.c b/rs/moq-c/c-tests/audio-decode-output.c index f24f0a77b59..cc881aadbb5 100644 --- a/rs/moq-c/c-tests/audio-decode-output.c +++ b/rs/moq-c/c-tests/audio-decode-output.c @@ -1,4 +1,4 @@ -// The installed C ABI owns every decoded frame until frame_free, independently +// The installed C ABI owns every decoded read until read_free, independently // of its consumer. PCM avoids a lossy-codec tolerance in this boundary test; // moq-audio's independent-reference suite covers the codec samples themselves. #include @@ -50,39 +50,45 @@ static void on_catalog(void *data, int32_t id) { if (id > 0) CHECK(moq_consume_catalog_free((uint32_t)id) == 0); } -static void on_frame(void *data, int32_t id) { +static void check_frame(Context *c, const moq_audio_frame *frame) { + CHECK(frame->timestamp_us == 123000 + c->frames * 1000000 / 48000); + size_t bytes = c->format == MOQ_AUDIO_SAMPLE_FORMAT_S16 ? 2 : 4; + CHECK(frame->data_size % (bytes * c->channels) == 0); + size_t count = frame->data_size / (bytes * c->channels); + CHECK(count > 0); + for (size_t sample = 0; sample < count * c->channels; sample++) { + size_t channel = c->format == MOQ_AUDIO_SAMPLE_FORMAT_F32_PLANAR ? sample / count : sample % c->channels; + // Stereo is FL=+1/4, FR=-1/2; the documented mono mix averages them. + float expected = c->channels == 1 ? -0.125f : channel == 0 ? 0.25f : -0.5f; + if (bytes == 2) { + int16_t value; + memcpy(&value, frame->data + sample * bytes, bytes); + CHECK(value == (c->channels == 1 ? -4096 : channel == 0 ? 8192 : -16384)); + } else { + float value; + memcpy(&value, frame->data + sample * bytes, bytes); + CHECK(value == expected); + } + } + c->frames += count; +} + +static void on_read(void *data, int32_t id) { Context *c = data; pthread_mutex_lock(&c->mutex); if (id <= 0) { CHECK(id == 0); c->decode_done = 1; } else { - moq_audio_frame frame; - CHECK(moq_decode_audio_frame((uint32_t)id, NULL) < 0); - CHECK(moq_decode_audio_frame((uint32_t)id, &frame) == 0); - CHECK(frame.timestamp_us == 123000 + c->frames * 1000000 / 48000); - size_t bytes = c->format == MOQ_AUDIO_SAMPLE_FORMAT_S16 ? 2 : 4; - CHECK(frame.data_size % (bytes * c->channels) == 0); - size_t count = frame.data_size / (bytes * c->channels); - CHECK(count > 0); - for (size_t sample = 0; sample < count * c->channels; sample++) { - size_t channel = c->format == MOQ_AUDIO_SAMPLE_FORMAT_F32_PLANAR ? sample / count : sample % c->channels; - // Stereo is FL=+1/4, FR=-1/2; the documented mono mix averages them. - float expected = c->channels == 1 ? -0.125f : channel == 0 ? 0.25f : -0.5f; - if (bytes == 2) { - int16_t value; - memcpy(&value, frame.data + sample * bytes, bytes); - CHECK(value == (c->channels == 1 ? -4096 : channel == 0 ? 8192 : -16384)); - } else { - float value; - memcpy(&value, frame.data + sample * bytes, bytes); - CHECK(value == expected); - } - } - c->frames += count; - CHECK(moq_decode_audio_frame_free((uint32_t)id) == 0); - CHECK(moq_decode_audio_frame((uint32_t)id, &frame) < 0); - CHECK(moq_decode_audio_frame_free((uint32_t)id) < 0); + moq_audio_read read; + CHECK(moq_decode_audio_read((uint32_t)id, NULL) < 0); + CHECK(moq_decode_audio_read((uint32_t)id, &read) == 0); + // Continuous PCM loses nothing, and a whole last frame needs no end marker. + CHECK(read.kind == MOQ_DECODE_READ_FRAME); + check_frame(c, &read.frame); + CHECK(moq_decode_audio_read_free((uint32_t)id) == 0); + CHECK(moq_decode_audio_read((uint32_t)id, &read) < 0); + CHECK(moq_decode_audio_read_free((uint32_t)id) < 0); } pthread_cond_broadcast(&c->changed); pthread_mutex_unlock(&c->mutex); @@ -91,8 +97,8 @@ static void on_frame(void *data, int32_t id) { int main(void) { Context c = {.mutex = PTHREAD_MUTEX_INITIALIZER, .changed = PTHREAD_COND_INITIALIZER}; moq_audio_decoder_output invalid = {999, 0, 0, 0}; - CHECK(moq_decode_audio(INT32_MAX, 0, &invalid, on_frame, &c) == -15); - CHECK(moq_decode_audio(INT32_MAX, 0, NULL, on_frame, &c) == -6); + CHECK(moq_decode_audio(INT32_MAX, 0, &invalid, on_read, &c) == -15); + CHECK(moq_decode_audio(INT32_MAX, 0, NULL, on_read, &c) == -6); int32_t origin = moq_origin_create(); CHECK(origin > 0); @@ -127,7 +133,7 @@ int main(void) { for (uint32_t channels = 1; channels <= 2; channels++) { c.format = formats[format]; c.channels = channels; c.frames = 0; c.decode_done = 0; moq_audio_decoder_output decoded = {c.format, 48000, channels, 30000000}; - int32_t decoder = moq_decode_audio((uint32_t)c.catalog, 0, &decoded, on_frame, &c); + int32_t decoder = moq_decode_audio((uint32_t)c.catalog, 0, &decoded, on_read, &c); CHECK(decoder > 0); pthread_mutex_lock(&c.mutex); wait_until(&c, &c.decode_done); @@ -145,6 +151,8 @@ int main(void) { wait_until(&c, &c.request_done); wait_until(&c, &c.catalog_done); pthread_mutex_unlock(&c.mutex); - fprintf(stderr, "C audio decode layout, remix, timeline and frame ownership passed\n"); + fprintf(stderr, "C audio decode layout, remix, timeline and read ownership passed\n"); + + return 0; } diff --git a/rs/moq-c/c-tests/decode-output.c b/rs/moq-c/c-tests/decode-output.c index 1293a21080b..e9402c1eae4 100644 --- a/rs/moq-c/c-tests/decode-output.c +++ b/rs/moq-c/c-tests/decode-output.c @@ -36,6 +36,9 @@ _Static_assert(offsetof(moq_video_decoder_output, format) == 8, "format moved"); _Static_assert(offsetof(moq_video_decoder_output, width) == 12, "width moved"); _Static_assert(offsetof(moq_video_decoder_output, height) == 16, "height moved"); _Static_assert(sizeof(moq_video_decoder_output) == 24, "decoder output size changed"); +_Static_assert(MOQ_DECODE_READ_FRAME == 0, "frame read discriminant moved"); +_Static_assert(MOQ_DECODE_READ_GAP == 1, "gap read discriminant moved"); +_Static_assert(MOQ_DECODE_READ_END == 2, "end read discriminant moved"); // Numeric codes from rs/moq-c/src/error.rs. They are the contract a C caller // matches on; if they drift, this fixture fails and the header is stale. @@ -133,18 +136,19 @@ static void on_catalog(void *ud, int32_t catalog) { if (catalog > 0) moq_consume_catalog_free((uint32_t)catalog); } -static void on_frame(void *ud, int32_t frame_id) { +static void on_read(void *ud, int32_t read_id) { ctx_t *c = (ctx_t *)ud; - if (frame_id <= 0) { + if (read_id <= 0) { pthread_mutex_lock(&c->mu); c->done_frame = 1; pthread_cond_broadcast(&c->cv); pthread_mutex_unlock(&c->mu); return; } - moq_video_frame frame; - memset(&frame, 0, sizeof(frame)); - if (moq_decode_video_frame((uint32_t)frame_id, &frame) == 0) { + moq_video_read read; + memset(&read, 0, sizeof(read)); + if (moq_decode_video_read((uint32_t)read_id, &read) == 0 && read.kind == MOQ_DECODE_READ_FRAME) { + moq_video_frame frame = read.frame; pthread_mutex_lock(&c->mu); if (!c->got_frame) { if (frame.width != c->want_width || frame.height != c->want_height || @@ -158,7 +162,7 @@ static void on_frame(void *ud, int32_t frame_id) { pthread_cond_broadcast(&c->cv); pthread_mutex_unlock(&c->mu); } - moq_decode_video_frame_free((uint32_t)frame_id); + moq_decode_video_read_free((uint32_t)read_id); } static void wait_for(pthread_mutex_t *mu, pthread_cond_t *cv, const int *flag, double timeout_s) { @@ -196,7 +200,8 @@ static void decode_once(ctx_t *c, int32_t catalog, const moq_video_decoder_outpu c->done_frame = 0; pthread_mutex_unlock(&c->mu); - int32_t consumer = moq_decode_video((uint32_t)catalog, 0, output, on_frame, c); + int32_t consumer = moq_decode_video((uint32_t)catalog, 0, output, on_read, c); + if (consumer <= 0) fail("error: moq_decode_video failed: %d (%s)\n", consumer, moq_error()); // Keep feeding the encoder so the subscriber has live frames whatever diff --git a/rs/moq-c/src/audio.rs b/rs/moq-c/src/audio.rs index fdbe3d20765..7bcdfab4fc9 100644 --- a/rs/moq-c/src/audio.rs +++ b/rs/moq-c/src/audio.rs @@ -18,7 +18,10 @@ use std::time::Duration; use bytes::Bytes; use tokio::sync::oneshot; +use crate::decode::Read; use crate::ffi::OnStatus; +use crate::{moq_decode_gap, moq_decode_read_kind}; + use crate::task::Task; use crate::{Error, Id, NonZeroSlab, Shared, State, ffi}; @@ -127,7 +130,7 @@ pub struct moq_audio_decoder_output { /// One audio frame: payload bytes plus a presentation timestamp. /// /// `data` is owned by the consume slab (see -/// [`moq_decode_audio_frame_free`]) or borrowed by the publish call +/// [`moq_decode_audio_read_free`]) or borrowed by the publish call /// (the publisher copies before returning). #[repr(C)] pub struct moq_audio_frame { @@ -139,6 +142,24 @@ pub struct moq_audio_frame { pub data_size: usize, } +/// One read from [`moq_decode_audio`]: decoded PCM, or a break in it. `kind` says which +/// field is filled; the others are zeroed. +#[repr(C)] +pub struct moq_audio_read { + /// Which kind of read this is. + pub kind: moq_decode_read_kind, + /// The decoded PCM, for [moq_decode_read_kind::MOQ_DECODE_READ_FRAME]. + pub frame: moq_audio_frame, + /// The lost span, for [moq_decode_read_kind::MOQ_DECODE_READ_GAP]. What follows does + /// not continue what came before. + pub gap: moq_decode_gap, + /// Where the stopped encoder's media stops in microseconds, for + /// [moq_decode_read_kind::MOQ_DECODE_READ_END]. Frames that follow and start before it + /// are the stopped encoder's remaining media; the first at or past it comes from a fresh + /// encoder. + pub end_us: u64, +} + // ---- State extensions (used internally by lib.rs) ---- /// An audio producer, shared so the Opus encode in `write` runs with the global @@ -164,7 +185,7 @@ impl crate::publish::Abort for AudioProducer { pub struct Audio { producers: NonZeroSlab, consumer_tasks: NonZeroSlab, - frames: NonZeroSlab, + reads: NonZeroSlab>, } impl Audio { @@ -224,7 +245,7 @@ impl Audio { self.producers.remove(id).ok_or(Error::MediaNotFound) } - /// Decode a rendition and deliver frame handles to the callback. + /// Decode a rendition and deliver read handles to the callback. pub fn consume( &mut self, rendition: crate::consume::Rendition, @@ -260,14 +281,14 @@ impl Audio { mut close: oneshot::Receiver<()>, ) -> Result<(), Error> { loop { - // `biased` so a pending close always wins over a ready frame. - let frame = tokio::select! { + // `biased` so a pending close always wins over a ready read. + let read = tokio::select! { biased; _ = &mut close => return Ok(()), - frame = consumer.read() => match frame { - Ok(Some(moq_audio::decode::Read::Frame(frame))) => frame, - // The C ABI hands over PCM only. - Ok(Some(moq_audio::decode::Read::Gap { .. } | moq_audio::decode::Read::End { .. })) => continue, + read = consumer.read() => match read { + Ok(Some(moq_audio::decode::Read::Frame(frame))) => Read::Frame(frame), + Ok(Some(moq_audio::decode::Read::Gap { timestamp, duration })) => Read::gap(timestamp, duration)?, + Ok(Some(moq_audio::decode::Read::End { end })) => Read::end(end)?, Ok(None) => return Ok(()), // One packet the codec rejected is that packet's problem: the // decoder stays usable, so drop it and keep the subscription rather @@ -280,9 +301,9 @@ impl Audio { }, }; - // Hold the lock only to buffer the frame; release it before the callback. - let frame_id = State::lock().audio.frames.insert(frame)?; - callback.call(Ok(frame_id)); + // Hold the lock only to buffer the read; release it before the callback. + let read_id = State::lock().audio.reads.insert(read)?; + callback.call(Ok(read_id)); } } @@ -291,24 +312,38 @@ impl Audio { self.consumer_tasks.close(id, Error::TrackNotFound) } - /// Borrow a decoded frame and its metadata until its handle is freed. - pub fn frame_info(&self, id: Id, dst: &mut moq_audio_frame) -> Result<(), Error> { - let frame = self.frames.get(id).ok_or(Error::FrameNotFound)?; - *dst = moq_audio_frame { - // The C ABI carries plain microseconds, so flatten the scaled - // `Timestamp` here at the boundary. Saturating rather than erroring: - // this is a getter on a frame we already decoded, and a u64 overflow - // needs a timestamp ~580,000 years out. - timestamp_us: u64::try_from(frame.timestamp.as_micros()).unwrap_or(u64::MAX), - data: frame.data.as_ptr(), - data_size: frame.data.len(), + /// Borrow a delivered read, and a frame's samples, until its handle is freed. + pub fn read_info(&self, id: Id, dst: &mut moq_audio_read) -> Result<(), Error> { + let read = self.reads.get(id).ok_or(Error::FrameNotFound)?; + let (kind, gap, end_us) = read.info(); + let frame = match read { + Read::Frame(frame) => moq_audio_frame { + // The C ABI carries plain microseconds, so flatten the scaled + // `Timestamp` here at the boundary. Saturating rather than erroring: + // this is a getter on a frame we already decoded, and a u64 overflow + // needs a timestamp ~580,000 years out. + timestamp_us: u64::try_from(frame.timestamp.as_micros()).unwrap_or(u64::MAX), + data: frame.data.as_ptr(), + data_size: frame.data.len(), + }, + _ => moq_audio_frame { + timestamp_us: 0, + data: std::ptr::null(), + data_size: 0, + }, + }; + *dst = moq_audio_read { + kind, + frame, + gap, + end_us, }; Ok(()) } - /// Release a decoded frame handle. - pub fn frame_free(&mut self, id: Id) -> Result<(), Error> { - self.frames.remove(id).ok_or(Error::FrameNotFound)?; + /// Release a delivered read handle. + pub fn read_free(&mut self, id: Id) -> Result<(), Error> { + self.reads.remove(id).ok_or(Error::FrameNotFound)?; Ok(()) } } @@ -496,11 +531,12 @@ pub extern "C" fn moq_encode_audio_finish(producer: u32) -> i32 { /// /// Returns a non-zero handle on success or a negative error code. /// -/// `on_frame` is called with a positive frame ID per frame, then exactly once -/// more with a terminal code: `0` (closed cleanly) or a negative error. After -/// the terminal (`<= 0`) callback, `on_frame` is never called again and -/// `user_data` is never touched again, so release `user_data` there. The -/// terminal callback fires even after [`moq_decode_audio_cancel`]. +/// `on_read` is called with a positive read id per decoded frame, gap, or encoder +/// end (see [`moq_decode_audio_read`]), then exactly once more with a terminal +/// code: `0` (closed cleanly) or a negative error. After the terminal (`<= 0`) +/// callback, `on_read` is never called again and `user_data` is never touched +/// again, so release `user_data` there. The terminal callback fires even after +/// [`moq_decode_audio_cancel`]. /// /// Starts at the newest cached group so reopening live playback skips the backlog. /// @@ -509,13 +545,13 @@ pub extern "C" fn moq_encode_audio_finish(producer: u32) -> i32 { /// /// # Safety /// - `output` must point to a valid [`moq_audio_decoder_output`]. -/// - `user_data` must stay valid until the terminal (`<= 0`) `on_frame` callback. +/// - `user_data` must stay valid until the terminal (`<= 0`) `on_read` callback. #[unsafe(no_mangle)] pub unsafe extern "C" fn moq_decode_audio( catalog: u32, index: u32, output: *const moq_audio_decoder_output, - on_frame: crate::moq_status_callback, + on_read: crate::moq_status_callback, user_data: *mut c_void, ) -> i32 { ffi::enter(move || { @@ -531,23 +567,23 @@ pub unsafe extern "C" fn moq_decode_audio( .transpose()?; config.max_delay = Duration::from_micros(raw.max_delay_us); - let on_frame = unsafe { OnStatus::new(user_data, on_frame)? }; + let on_read = unsafe { OnStatus::new(user_data, on_read)? }; let mut state = State::lock(); let rendition = state.consume.rendition(catalog, index as usize)?; let State { audio, .. } = &mut *state; - audio.consume(rendition, config, on_frame) + audio.consume(rendition, config, on_read) }) } /// Stop an audio (raw PCM) consumer's background task. /// /// Returns immediately: zero on success, or a negative code if already closed. -/// Does NOT free `user_data`; the on-frame callback still fires once more with a +/// Does NOT free `user_data`; the on-read callback still fires once more with a /// terminal `0` (or a negative error), which is where `user_data` should be -/// released. Frame IDs already delivered to the callback are likewise not freed; -/// release each with [`moq_decode_audio_frame_free`]. +/// released. Read ids already delivered to the callback are likewise not freed; +/// release each with [`moq_decode_audio_read_free`]. #[unsafe(no_mangle)] pub extern "C" fn moq_decode_audio_cancel(consumer: u32) -> i32 { ffi::enter(move || { @@ -556,29 +592,29 @@ pub extern "C" fn moq_decode_audio_cancel(consumer: u32) -> i32 { }) } -/// Copy a delivered frame's metadata into `dst`. +/// Copy a delivered read into `dst`: its kind, and the frame, gap, or end it carries. /// -/// The written `dst->data` pointer remains valid until the same `id` -/// is released with [`moq_decode_audio_frame_free`]. +/// The written `dst->frame.data` pointer remains valid until the same `id` +/// is released with [`moq_decode_audio_read_free`]. /// /// # Safety -/// - `dst` must point to a writable [`moq_audio_frame`]. +/// - `dst` must point to a writable [`moq_audio_read`]. #[unsafe(no_mangle)] -pub unsafe extern "C" fn moq_decode_audio_frame(id: u32, dst: *mut moq_audio_frame) -> i32 { +pub unsafe extern "C" fn moq_decode_audio_read(id: u32, dst: *mut moq_audio_read) -> i32 { ffi::enter(move || { let id = ffi::parse_id(id)?; let dst = unsafe { dst.as_mut() }.ok_or(Error::InvalidPointer)?; - State::lock().audio.frame_info(id, dst) + State::lock().audio.read_info(id, dst) }) } -/// Free a frame previously delivered through the consume callback. -/// Required for every delivered frame ID; closing the parent consumer +/// Free a read previously delivered through the decode callback. +/// Required for every delivered read id; closing the parent consumer /// is not enough. #[unsafe(no_mangle)] -pub extern "C" fn moq_decode_audio_frame_free(id: u32) -> i32 { +pub extern "C" fn moq_decode_audio_read_free(id: u32) -> i32 { ffi::enter(move || { let id = ffi::parse_id(id)?; - State::lock().audio.frame_free(id) + State::lock().audio.read_free(id) }) } diff --git a/rs/moq-c/src/decode.rs b/rs/moq-c/src/decode.rs new file mode 100644 index 00000000000..5b0868207a1 --- /dev/null +++ b/rs/moq-c/src/decode.rs @@ -0,0 +1,76 @@ +//! What [`crate::moq_decode_audio`] and [`crate::moq_decode_video`] deliver besides frames. +//! +//! Both decoders hand over the same three kinds of read as `moq_audio::decode::Read` and +//! `moq_video::decode::Read`, so a C player handles loss and ends alike for every media type. + +use crate::Error; + +/// Which kind of read a decoder delivered: a decoded frame, or a break between frames. +#[repr(C)] +#[expect(non_camel_case_types, reason = "C ABI type names match moq.h")] +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum moq_decode_read_kind { + /// A decoded frame. + MOQ_DECODE_READ_FRAME = 0, + /// Media after what was already read was lost: a group was skipped or never arrived. + /// What follows does not continue what came before. + MOQ_DECODE_READ_GAP = 1, + /// The publisher stopped its encoder: it paused, switched source, or finished. What + /// follows comes from a fresh encoder. + MOQ_DECODE_READ_END = 2, +} + +/// Lost media, for a [moq_decode_read_kind::MOQ_DECODE_READ_GAP] read. +#[repr(C)] +#[derive(Clone, Copy, Debug, Default)] +pub struct moq_decode_gap { + /// The furthest presentation point already read, in microseconds. + pub timestamp_us: u64, + /// An upper bound on how long until the next frame resumes, in microseconds. + pub duration_us: u64, +} + +/// A delivered read held by its id, with its timestamps already in the C ABI's microseconds. +#[derive(Clone)] +pub(crate) enum Read { + Frame(F), + Gap(moq_decode_gap), + End { end_us: u64 }, +} + +impl Read { + pub(crate) fn gap(timestamp: moq_net::Timestamp, duration: moq_net::Timestamp) -> Result { + Ok(Self::Gap(moq_decode_gap { + timestamp_us: micros(timestamp)?, + duration_us: micros(duration)?, + })) + } + + pub(crate) fn end(end: moq_net::Timestamp) -> Result { + Ok(Self::End { end_us: micros(end)? }) + } + + /// The C kind, plus the gap and end fields a non-frame read fills. + pub(crate) fn info(&self) -> (moq_decode_read_kind, moq_decode_gap, u64) { + match self { + Self::Frame(_) => ( + moq_decode_read_kind::MOQ_DECODE_READ_FRAME, + moq_decode_gap::default(), + 0, + ), + Self::Gap(gap) => (moq_decode_read_kind::MOQ_DECODE_READ_GAP, *gap, 0), + Self::End { end_us } => ( + moq_decode_read_kind::MOQ_DECODE_READ_END, + moq_decode_gap::default(), + *end_us, + ), + } + } +} + +fn micros(timestamp: moq_net::Timestamp) -> Result { + timestamp + .as_micros() + .try_into() + .map_err(|_| Error::TimestampOverflow(moq_net::TimeOverflow)) +} diff --git a/rs/moq-c/src/lib.rs b/rs/moq-c/src/lib.rs index fc4c6d8e66c..5a05731adf7 100644 --- a/rs/moq-c/src/lib.rs +++ b/rs/moq-c/src/lib.rs @@ -23,6 +23,7 @@ mod audio; mod bandwidth; mod client; mod consume; +mod decode; mod error; mod ffi; mod id; @@ -37,6 +38,7 @@ mod video; pub use api::*; pub use audio::*; pub use bandwidth::*; +pub use decode::{moq_decode_gap, moq_decode_read_kind}; pub use error::*; pub use ffi::{moq_state_callback, moq_status_callback}; pub use id::*; diff --git a/rs/moq-c/src/test.rs b/rs/moq-c/src/test.rs index 28a138d44e0..e4625223bf8 100644 --- a/rs/moq-c/src/test.rs +++ b/rs/moq-c/src/test.rs @@ -250,6 +250,32 @@ extern "C" fn state_callback(user_data: *mut c_void, state: *const moq_connectio /// FFI state callback for tests that only inspect the immediate return code. extern "C" fn ignore_state(_user_data: *mut c_void, _state: *const moq_connection_state) {} +/// Read a delivered video read id, returning the call's code and what it wrote. +fn video_read(id: u32) -> (i32, moq_video_read) { + let mut read = moq_video_read { + kind: moq_decode_read_kind::MOQ_DECODE_READ_END, + frame: moq_video_frame { + timestamp_us: 0, + width: 0, + height: 0, + data: std::ptr::null(), + data_size: 0, + }, + gap: moq_decode_gap::default(), + end_us: 0, + }; + let code = unsafe { moq_decode_video_read(id, &mut read) }; + (code, read) +} + +/// Read a delivered video read id that must be a decoded frame. +fn video_frame(id: u32) -> moq_video_frame { + let (code, read) = video_read(id); + assert_eq!(code, 0); + assert_eq!(read.kind, moq_decode_read_kind::MOQ_DECODE_READ_FRAME); + read.frame +} + /// Build a valid OpusHead init buffer (RFC 7845 ยง5.1). fn opus_head() -> Vec { let mut head = Vec::with_capacity(19); @@ -4231,6 +4257,95 @@ fn gray_rgba(width: u32, height: u32) -> Vec { vec![0x80u8; width as usize * height as usize * 4] } +/// A finished encoder reaches a C decoder as an end read in microseconds rather +/// than vanishing: Opus pads a partial last frame, so the finish publishes where +/// the source really stopped. +#[test] +fn audio_raw_decode_reads_the_encoder_end() { + const START_US: u64 = 123_000; + const SAMPLES: usize = 860; + + let origin = id(moq_origin_create()); + let path = b"audio-raw-end-test"; + let broadcast = publish_broadcast(origin, path); + let input = moq_audio_encoder_input { + format: moq_audio_sample_format::MOQ_AUDIO_SAMPLE_FORMAT_F32 as u32, + sample_rate: 48_000, + channels: 1, + }; + let output = moq_audio_encoder_output { + codec: c"opus".as_ptr(), + codec_len: 4, + sample_rate: 0, + channels: 0, + bitrate: 0, + frame_duration_us: 0, + }; + let producer = id(unsafe { moq_encode_audio(broadcast, c"audio".as_ptr(), 5, &input, &output, 0) }); + let pcm: Vec = (0..SAMPLES) + .flat_map(|i| (i as f32 / SAMPLES as f32).to_le_bytes()) + .collect(); + let frame = moq_audio_frame { + timestamp_us: START_US, + data: pcm.as_ptr(), + data_size: pcm.len(), + }; + assert_eq!(unsafe { moq_encode_audio_frame(producer, &frame) }, 0); + assert_eq!(moq_encode_audio_finish(producer), 0); + + let consume = request_broadcast(origin, path); + let catalog_cb = Callback::new(); + let catalog_task = id(unsafe { moq_consume_catalog(consume, Some(channel_callback), catalog_cb.ptr) }); + let catalog_id = id(catalog_cb.recv()); + let decoded = moq_audio_decoder_output { + format: moq_audio_sample_format::MOQ_AUDIO_SAMPLE_FORMAT_F32 as u32, + sample_rate: 0, + channels: 0, + max_delay_us: 30_000_000, + }; + let read_cb = Callback::new(); + id(unsafe { moq_decode_audio(catalog_id, 0, &decoded, Some(channel_callback), read_cb.ptr) }); + + let mut ends = Vec::new(); + loop { + let code = read_cb.recv(); + if code <= 0 { + assert_eq!(code, 0, "a finished track closes cleanly"); + break; + } + let mut read = moq_audio_read { + kind: moq_decode_read_kind::MOQ_DECODE_READ_GAP, + frame: moq_audio_frame { + timestamp_us: 0, + data: std::ptr::null(), + data_size: 0, + }, + gap: moq_decode_gap::default(), + end_us: 0, + }; + assert_eq!(unsafe { moq_decode_audio_read(id(code), &mut read) }, 0); + match read.kind { + moq_decode_read_kind::MOQ_DECODE_READ_FRAME => { + assert!(!read.frame.data.is_null()); + } + moq_decode_read_kind::MOQ_DECODE_READ_END => { + assert!(read.frame.data.is_null(), "an end carries no samples"); + ends.push(read.end_us); + } + moq_decode_read_kind::MOQ_DECODE_READ_GAP => panic!("nothing was lost"), + } + assert_eq!(moq_decode_audio_read_free(id(code)), 0); + } + assert_eq!(ends, [START_US + SAMPLES as u64 * 1_000_000 / 48_000]); + + assert_eq!(moq_consume_catalog_free(catalog_id), 0); + assert_eq!(moq_consume_catalog_cancel(catalog_task), 0); + assert_eq!(catalog_cb.recv_catalog_terminal(), 0); + assert_eq!(moq_consume_close(consume), 0); + assert_eq!(moq_publish_close(broadcast), 0); + assert_eq!(moq_origin_close(origin), 0); +} + /// The publish-side mirror of [`video_raw_decode`]: hand raw RGBA to /// `moq_encode_video` and read decoded I420 back out of /// `moq_decode_video`, so the encode and decode halves meet on the wire. @@ -4299,43 +4414,38 @@ fn video_raw_publish_consume() { } let frame_id = id(frame_cb.recv()); - let mut frame = moq_video_frame { - timestamp_us: 0, - width: 0, - height: 0, - data: std::ptr::null(), - data_size: 0, - }; // Hold the frame across cancellation: its id owns the decoded picture, so it // stays readable after the terminal callback and until it is freed. assert_eq!(moq_decode_video_cancel(consumer), 0); loop { let code = frame_cb.recv(); if code > 0 { - assert_eq!(moq_decode_video_frame_free(id(code)), 0); + assert_eq!(moq_decode_video_read_free(id(code)), 0); } else { assert_eq!(code, 0, "raw video close delivers terminal 0"); break; } } - assert_eq!(unsafe { moq_decode_video_frame(frame_id, &mut frame) }, 0); + let frame = video_frame(frame_id); assert_eq!(frame.width, 320); assert_eq!(frame.height, 240); assert_eq!(frame.data_size, 320 * 240 * 3 / 2, "tightly-packed I420"); // Pixels are produced once per id, so a second read hands back the same buffer. - let first = frame.data; - assert_eq!(unsafe { moq_decode_video_frame(frame_id, &mut frame) }, 0); - assert_eq!(frame.data, first, "the pixel pointer is stable until the id is freed"); + assert_eq!( + video_frame(frame_id).data, + frame.data, + "the pixel pointer is stable until the id is freed" + ); - assert_eq!(moq_decode_video_frame_free(frame_id), 0); + assert_eq!(moq_decode_video_read_free(frame_id), 0); assert_eq!( - unsafe { moq_decode_video_frame(frame_id, &mut frame) }, + video_read(frame_id).0, Error::FrameNotFound.code(), "a freed id is released, not left readable" ); - assert_eq!(moq_decode_video_frame_free(frame_id), Error::FrameNotFound.code()); + assert_eq!(moq_decode_video_read_free(frame_id), Error::FrameNotFound.code()); assert_eq!(moq_consume_catalog_free(catalog_id), 0); assert_eq!(moq_consume_catalog_cancel(catalog_task), 0); @@ -4456,22 +4566,15 @@ fn decode_first_frame(output: &moq_video_decoder_output) -> (u32, u32, usize) { } let frame_id = id(frame_cb.recv()); - let mut frame = moq_video_frame { - timestamp_us: 0, - width: 0, - height: 0, - data: std::ptr::null(), - data_size: 0, - }; - assert_eq!(unsafe { moq_decode_video_frame(frame_id, &mut frame) }, 0); + let frame = video_frame(frame_id); let result = (frame.width, frame.height, frame.data_size); - assert_eq!(moq_decode_video_frame_free(frame_id), 0); + assert_eq!(moq_decode_video_read_free(frame_id), 0); assert_eq!(moq_decode_video_cancel(consumer), 0); loop { let code = frame_cb.recv(); if code > 0 { - assert_eq!(moq_decode_video_frame_free(id(code)), 0); + assert_eq!(moq_decode_video_read_free(id(code)), 0); } else { assert_eq!(code, 0, "raw video close delivers terminal 0"); break; @@ -4931,27 +5034,20 @@ fn video_raw_decode() { // First decoded frame: packed I420 at the encoder resolution. let frame_id = id(frame_cb.recv()); - let mut frame = moq_video_frame { - timestamp_us: 0, - width: 0, - height: 0, - data: std::ptr::null(), - data_size: 0, - }; - assert_eq!(unsafe { moq_decode_video_frame(frame_id, &mut frame) }, 0); + let frame = video_frame(frame_id); assert_eq!(frame.width, 320); assert_eq!(frame.height, 240); assert_eq!(frame.data_size, 320 * 240 * 3 / 2, "tightly-packed I420"); assert!(!frame.data.is_null()); - assert_eq!(moq_decode_video_frame_free(frame_id), 0); + assert_eq!(moq_decode_video_read_free(frame_id), 0); assert_eq!(moq_decode_video_cancel(consumer), 0); - // Drain any other decoded frames already queued, then expect the terminal 0. + // Drain any other decoded reads already queued, then expect the terminal 0. loop { let code = frame_cb.recv(); if code > 0 { - assert_eq!(moq_decode_video_frame_free(id(code)), 0); + assert_eq!(moq_decode_video_read_free(id(code)), 0); } else { assert_eq!(code, 0, "raw video close delivers terminal 0"); break; diff --git a/rs/moq-c/src/video.rs b/rs/moq-c/src/video.rs index a878240f5dd..bb71cbf6001 100644 --- a/rs/moq-c/src/video.rs +++ b/rs/moq-c/src/video.rs @@ -18,9 +18,10 @@ use std::time::Duration; use tokio::sync::oneshot; +use crate::decode::Read; use crate::ffi::OnStatus; use crate::task::Task; -use crate::{Error, Id, NonZeroSlab, Shared, State, ffi}; +use crate::{Error, Id, NonZeroSlab, Shared, State, ffi, moq_decode_gap, moq_decode_read_kind}; // ---- C-visible types ---- @@ -168,9 +169,9 @@ pub struct moq_video_decoder_output { /// height/2` each), no row padding, BT.601 limited range; RGBA is tightly /// packed `width * height * 4` bytes, no row padding. /// -/// The frame id owns the decoded picture, and the pixels are produced from it -/// on the first [`moq_decode_video_frame`] call for that id. `data` stays valid -/// until the same id is released with [`moq_decode_video_frame_free`]. +/// The read id owns the decoded picture, and the pixels are produced from it +/// on the first [`moq_decode_video_read`] call for that id. `data` stays valid +/// until the same id is released with [`moq_decode_video_read_free`]. /// /// The publish side has its own [`moq_video_encoder_frame`], which carries no /// dimensions because the encoder already fixed them. @@ -188,15 +189,32 @@ pub struct moq_video_frame { pub data_size: usize, } +/// One read from [`moq_decode_video`]: a decoded picture, or a break between them. +/// `kind` says which field is filled; the others are zeroed. +#[repr(C)] +pub struct moq_video_read { + /// Which kind of read this is. + pub kind: moq_decode_read_kind, + /// The decoded picture, for [moq_decode_read_kind::MOQ_DECODE_READ_FRAME]. + pub frame: moq_video_frame, + /// The lost span, for [moq_decode_read_kind::MOQ_DECODE_READ_GAP]. Every picture from + /// before the loss precedes it, and what follows does not continue what came before. + pub gap: moq_decode_gap, + /// Where the stopped encoder's media stops in microseconds, for + /// [moq_decode_read_kind::MOQ_DECODE_READ_END]. The stopped encoder's last pictures + /// precede it; what follows comes from a fresh encoder. + pub end_us: u64, +} + // ---- State extension (used internally by lib.rs) ---- -/// Raw-video state: encoders being published, plus decoder tasks and their -/// buffered decoded frames. +/// Raw-video state: encoders being published, plus decoder tasks and the +/// reads they delivered. #[derive(Default)] pub struct Video { producers: NonZeroSlab>, consumer_tasks: NonZeroSlab, - frames: NonZeroSlab>, + reads: NonZeroSlab>>, } /// Wait out an encode-thread round trip from a C entry point. @@ -332,7 +350,7 @@ impl crate::publish::Abort for Shared { } /// A delivered frame: the decoded picture, retained until its id is freed, and -/// the packed pixels [`moq_decode_video_frame`] produces from it on first use. +/// the packed pixels [`moq_decode_video_read`] produces from it on first use. /// /// Converting on first use rather than at delivery keeps the work off the /// delivery task and skips it for a frame the caller frees unread. The pixels @@ -581,7 +599,7 @@ impl Video { self.producers.remove(id).ok_or(Error::MediaNotFound) } - /// Decode a rendition and deliver frame handles to the callback. + /// Decode a rendition and deliver read handles to the callback. pub fn consume( &mut self, rendition: crate::consume::Rendition, @@ -619,25 +637,24 @@ impl Video { output: DecoderOutput, ) -> Result<(), Error> { loop { - // `biased` so a pending close always wins over a ready frame. - let frame = tokio::select! { + // `biased` so a pending close always wins over a ready read. + let read = tokio::select! { biased; _ = &mut close => return Ok(()), read = consumer.read() => match read? { - Some(moq_video::decode::Read::Frame(frame)) => frame, - // The C ABI hands over pictures only. - Some(moq_video::decode::Read::Gap { .. } | moq_video::decode::Read::End { .. }) => continue, + Some(moq_video::decode::Read::Frame(frame)) => Read::Frame(Arc::new(VideoFrame { + frame, + output, + pixels: OnceLock::new(), + })), + Some(moq_video::decode::Read::Gap { timestamp, duration }) => Read::gap(timestamp, duration)?, + Some(moq_video::decode::Read::End { end }) => Read::end(end)?, None => return Ok(()), }, }; - let frame = Arc::new(VideoFrame { - frame, - output, - pixels: OnceLock::new(), - }); - let frame_id = State::lock().video.frames.insert(frame)?; - callback.call(Ok(frame_id)); + let read_id = State::lock().video.reads.insert(read)?; + callback.call(Ok(read_id)); } } @@ -646,15 +663,15 @@ impl Video { self.consumer_tasks.close(id, Error::TrackNotFound) } - /// Resolve a frame handle, so the caller can convert it with the global lock - /// released. - fn frame(&self, id: Id) -> Result, Error> { - self.frames.get(id).cloned().ok_or(Error::FrameNotFound) + /// Resolve a read handle, so the caller can convert its frame with the global + /// lock released. + fn read(&self, id: Id) -> Result>, Error> { + self.reads.get(id).cloned().ok_or(Error::FrameNotFound) } - /// Release a decoded frame handle. - pub fn frame_free(&mut self, id: Id) -> Result<(), Error> { - self.frames.remove(id).ok_or(Error::FrameNotFound)?; + /// Release a delivered read handle. + pub fn read_free(&mut self, id: Id) -> Result<(), Error> { + self.reads.remove(id).ok_or(Error::FrameNotFound)?; Ok(()) } } @@ -936,28 +953,29 @@ pub extern "C" fn moq_encode_video_finish(producer: u32) -> i32 { /// /// An unknown `output->format` or an invalid `output->width`/`height` fails /// here, before subscribing. Each frame is produced in the requested layout by -/// [`moq_decode_video_frame`], which fails for that frame rather than deliver +/// [`moq_decode_video_read`], which fails for that frame rather than deliver /// another layout. /// /// Returns a non-zero handle on success or a negative error code. /// -/// `on_frame` is called with a positive frame id per decoded frame, then exactly -/// once more with a terminal code: `0` (closed cleanly) or a negative error. -/// After the terminal (`<= 0`) callback, `on_frame` is never called again and -/// `user_data` is never touched again, so release `user_data` there. The terminal -/// callback fires even after [`moq_decode_video_cancel`]. +/// `on_read` is called with a positive read id per decoded frame, gap, or encoder +/// end (see [`moq_decode_video_read`]), then exactly once more with a terminal +/// code: `0` (closed cleanly) or a negative error. After the terminal (`<= 0`) +/// callback, `on_read` is never called again and `user_data` is never touched +/// again, so release `user_data` there. The terminal callback fires even after +/// [`moq_decode_video_cancel`]. /// /// Starts at the newest cached group so reopening live playback skips the backlog. /// /// # Safety /// - `output` must point to a valid [`moq_video_decoder_output`]. -/// - `user_data` must stay valid until the terminal (`<= 0`) `on_frame` callback. +/// - `user_data` must stay valid until the terminal (`<= 0`) `on_read` callback. #[unsafe(no_mangle)] pub unsafe extern "C" fn moq_decode_video( catalog: u32, index: u32, output: *const moq_video_decoder_output, - on_frame: crate::moq_status_callback, + on_read: crate::moq_status_callback, user_data: *mut c_void, ) -> i32 { ffi::enter(move || { @@ -980,24 +998,24 @@ pub unsafe extern "C" fn moq_decode_video( // delivery loop still enforces it, since other backends ignore it. options.decoder.scale_hint = size; let output = DecoderOutput { format, size }; - let on_frame = unsafe { OnStatus::new(user_data, on_frame)? }; + let on_read = unsafe { OnStatus::new(user_data, on_read)? }; let mut state = State::lock(); let rendition = state.consume.rendition(catalog, index as usize)?; let State { video, .. } = &mut *state; - video.consume(rendition, options, output, on_frame) + video.consume(rendition, options, output, on_read) }) } /// Stop a video (raw) consumer's background task. /// /// Returns immediately: zero on success, or a negative code if already closed. -/// Does NOT free `user_data`; the on-frame callback still fires once more with a +/// Does NOT free `user_data`; the on-read callback still fires once more with a /// terminal `0` (or a negative error), which is where `user_data` should be -/// released. Frame ids already delivered are likewise not freed: each stays +/// released. Read ids already delivered are likewise not freed: each stays /// readable after the terminal callback until released with -/// [`moq_decode_video_frame_free`]. +/// [`moq_decode_video_read_free`]. #[unsafe(no_mangle)] pub extern "C" fn moq_decode_video_cancel(consumer: u32) -> i32 { ffi::enter(move || { @@ -1006,51 +1024,71 @@ pub extern "C" fn moq_decode_video_cancel(consumer: u32) -> i32 { }) } -/// Write a delivered frame's pixels and metadata into `dst`. +/// Write a delivered read into `dst`: its kind, and the frame's pixels, the gap, +/// or the end it carries. /// -/// The first call for an `id` converts the decoded picture to the layout and -/// size [`moq_decode_video`] asked for; later calls return the same pixels. A +/// The first call for a frame's `id` converts the decoded picture to the layout +/// and size [`moq_decode_video`] asked for; later calls return the same pixels. A /// conversion failure returns a negative code for this frame only, leaving the /// id to be freed as usual. Safe to call from any thread, including inside the -/// frame callback. +/// read callback. /// -/// The written `dst->data` pointer remains valid until the same `id` is released -/// with [`moq_decode_video_frame_free`]. +/// The written `dst->frame.data` pointer remains valid until the same `id` is +/// released with [`moq_decode_video_read_free`]. /// /// # Safety -/// - `dst` must point to a writable [`moq_video_frame`]. +/// - `dst` must point to a writable [`moq_video_read`]. #[unsafe(no_mangle)] -pub unsafe extern "C" fn moq_decode_video_frame(id: u32, dst: *mut moq_video_frame) -> i32 { +pub unsafe extern "C" fn moq_decode_video_read(id: u32, dst: *mut moq_video_read) -> i32 { ffi::enter(move || { let id = ffi::parse_id(id)?; let dst = unsafe { dst.as_mut() }.ok_or(Error::InvalidPointer)?; - let frame = State::lock().video.frame(id)?; - let pixels = frame.pixels()?; - *dst = moq_video_frame { - // The decoded Timestamp is bounded by a QUIC varint, so its microseconds fit. - timestamp_us: frame.frame.timestamp.as_micros() as u64, - width: pixels.width, - height: pixels.height, - // The slab keeps its `Arc` until the id is freed, and the pixels are - // never replaced once set, so the pointer outlives this call. - data: pixels.data.as_ptr(), - data_size: pixels.data.len(), + let read = State::lock().video.read(id)?; + let (kind, gap, end_us) = read.info(); + let frame = match &read { + Read::Frame(frame) => { + let pixels = frame.pixels()?; + moq_video_frame { + // The decoded Timestamp is bounded by a QUIC varint, so its microseconds fit. + timestamp_us: frame.frame.timestamp.as_micros() as u64, + width: pixels.width, + height: pixels.height, + // The slab keeps its `Arc` until the id is freed, and the pixels are + // never replaced once set, so the pointer outlives this call. + data: pixels.data.as_ptr(), + data_size: pixels.data.len(), + } + } + _ => moq_video_frame { + timestamp_us: 0, + width: 0, + height: 0, + data: std::ptr::null(), + data_size: 0, + }, + }; + *dst = moq_video_read { + kind, + frame, + gap, + end_us, }; Ok(()) }) } -/// Free a frame previously delivered through the consume callback, releasing -/// its decoded picture back to the decoder. Required for every delivered frame -/// id; closing the parent consumer is not enough, and a decoder whose frames -/// are all held stalls until some are freed. +/// Free a read previously delivered through the decode callback, releasing a +/// frame's decoded picture back to the decoder. Required for every delivered read +/// id; closing the parent consumer is not enough, and a decoder whose frames are +/// all held stalls until some are freed. #[unsafe(no_mangle)] -pub extern "C" fn moq_decode_video_frame_free(id: u32) -> i32 { +pub extern "C" fn moq_decode_video_read_free(id: u32) -> i32 { ffi::enter(move || { let id = ffi::parse_id(id)?; - State::lock().video.frame_free(id) + State::lock().video.read_free(id) }) } + #[cfg(test)] mod tests { use super::*; diff --git a/rs/moq-ffi/MIGRATE.md b/rs/moq-ffi/MIGRATE.md index 7b159619a87..6d952ac589b 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 +### Decoders read gaps and ends + +`MoqAudioConsumer::next` returns `MoqAudioRead` and `MoqVideoConsumer::next` returns `MoqVideoRead`, mirroring `moq_audio::decode::Read` and `moq_video::decode::Read`, instead of a bare frame. Take the frame from the `Frame { frame }` variant where you used the return value before. `Gap { timestamp_us, duration }` marks lost media and `End { end_us }` where the publisher stopped its encoder; what follows either does not continue what came before, so restart playback there, or skip them to keep the old behavior. `None` still means the track ended. + ### `MoqClientConfig` gains `tokens` It defaults to empty in every generated binding; a Rust struct literal adds `tokens: vec![]` or ends with `..Default::default()`. diff --git a/rs/moq-ffi/src/audio.rs b/rs/moq-ffi/src/audio.rs index 49e9e95e5aa..ca65723c574 100644 --- a/rs/moq-ffi/src/audio.rs +++ b/rs/moq-ffi/src/audio.rs @@ -195,6 +195,55 @@ pub struct MoqAudioFrame { pub data: Vec, } +/// One item of a [`MoqAudioConsumer`] read, in delivery order: decoded PCM, or a break in it. +/// +/// Mirrors `moq_audio::decode::Read`, the same shape as +/// [`MoqVideoRead`](crate::video::MoqVideoRead), so a player handles loss and ends alike +/// for every media type. +#[derive(uniffi::Enum)] +pub enum MoqAudioRead { + /// Decoded PCM. + Frame { + /// The decoded samples. + frame: MoqAudioFrame, + }, + /// Media after what was already read was lost: a group was skipped or never arrived, + /// or a packet's timestamp did not continue the previous one. What follows does not + /// continue what came before. + Gap { + /// The furthest presentation point already read, in microseconds. + timestamp_us: u64, + /// An upper bound on how long until the next frame resumes. + duration: Duration, + }, + /// The publisher stopped its encoder: it paused, switched source, or finished. + /// + /// Frames that follow and start before `end_us` are the stopped encoder's remaining + /// media; the first frame at or past it comes from a fresh encoder. + End { + /// Where the stopped encoder's media stops, in microseconds. + end_us: u64, + }, +} + +impl TryFrom for MoqAudioRead { + type Error = MoqError; + + fn try_from(read: moq_audio::decode::Read) -> Result { + use crate::consumer::timestamp_us; + Ok(match read { + moq_audio::decode::Read::Frame(frame) => Self::Frame { frame: frame.into() }, + moq_audio::decode::Read::Gap { timestamp, duration } => Self::Gap { + timestamp_us: timestamp_us(timestamp)?, + duration: crate::ffi::duration(duration.into()), + }, + moq_audio::decode::Read::End { end } => Self::End { + end_us: timestamp_us(end)?, + }, + }) + } +} + /// One packet emitted by a codec-only audio encoder. #[derive(uniffi::Record)] pub struct MoqAudioPacket { @@ -514,16 +563,10 @@ struct ConsumerInner { } impl ConsumerInner { - async fn next(&mut self) -> Result, MoqError> { - loop { - let read = self.consumer.read().await; - self.delay.store(nanos(self.consumer.delay()), Ordering::Relaxed); - match read? { - Some(moq_audio::decode::Read::Frame(frame)) => return Ok(Some(frame.into())), - Some(moq_audio::decode::Read::Gap { .. } | moq_audio::decode::Read::End { .. }) => {} - None => return Ok(None), - } - } + async fn next(&mut self) -> Result, MoqError> { + let read = self.consumer.read().await; + self.delay.store(nanos(self.consumer.delay()), Ordering::Relaxed); + read?.map(MoqAudioRead::try_from).transpose() } } @@ -543,8 +586,8 @@ pub struct MoqAudioConsumer { #[uniffi::export] impl MoqAudioConsumer { - /// The next decoded frame, or `None` once the track ends. - pub async fn next(&self) -> Result, MoqError> { + /// The next decoded frame, gap, or encoder end, or `None` once the track ends. + pub async fn next(&self) -> Result, MoqError> { self.task.run(|mut state| async move { state.next().await }).await } diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index 0211c489fa1..64c0be82ddd 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -731,7 +731,15 @@ async fn audio_raw_publish_from_many_threads() { let closer = audio.clone(); std::thread::spawn(move || closer.finish()).join().unwrap().unwrap(); let mut pcm = Vec::new(); - while let Some(frame) = tokio::time::timeout(TIMEOUT, decoder.next()).await.unwrap().unwrap() { + while let Some(read) = tokio::time::timeout(TIMEOUT, decoder.next()).await.unwrap().unwrap() { + let frame = match read { + MoqAudioRead::Frame { frame } => frame, + MoqAudioRead::End { end_us } => { + assert_eq!(end_us, 250_000 + 8 * 20_000, "the finish ends after the last frame"); + continue; + } + MoqAudioRead::Gap { .. } => panic!("nothing was lost"), + }; assert_eq!(frame.timestamp_us, 250_000 + pcm.len() as u64 / 2 * 1_000_000 / 48_000); pcm.extend(frame.data.as_chunks::<4>().0.iter().map(|s| f32::from_le_bytes(*s))); } @@ -867,12 +875,14 @@ async fn raw_audio_decode_delay() { .unwrap(); } for read in [fixed_read, estimated_read] { - let frame = tokio::time::timeout(TIMEOUT, read) + let read = tokio::time::timeout(TIMEOUT, read) .await .expect("timed out waiting for decoded audio") .unwrap() - .unwrap() - .expect("expected a decoded frame"); + .unwrap(); + let Some(MoqAudioRead::Frame { frame }) = read else { + panic!("expected a decoded frame"); + }; let pcm: Vec = frame .data .as_chunks::<4>() @@ -894,6 +904,95 @@ async fn raw_audio_decode_delay() { broadcast.close().unwrap(); } +/// Each encoder the publisher stops, by a reanchor or a finish, reaches the caller as an +/// end in microseconds rather than vanishing between frames. +#[cfg(feature = "audio")] +#[tokio::test] +async fn raw_audio_decode_reports_encoder_ends() { + use crate::audio::*; + + const FRAME_SAMPLES: usize = 960; + const FRAME_US: u64 = 20_000; + const RESUMED_US: u64 = 1_000_000; + + let broadcast = MoqBroadcastProducer::new().unwrap(); + let consumer = broadcast.consume().unwrap(); + let catalogs = MoqMediaCatalogConsumer::subscribe(&consumer).await.unwrap(); + let audio = MoqAudioProducer::new( + &MoqMediaCatalogProducer::new(broadcast.clone()).unwrap(), + MoqAudioProducerOptions { + input: MoqAudioEncoderInput { + format: MoqAudioSampleFormat::F32, + sample_rate: 48_000, + channels: 1, + }, + output: MoqAudioEncoderOutput { + codec: MoqAudioCodec::opus(), + sample_rate: None, + channels: None, + bitrate: None, + frame_duration: Some(Duration::from_micros(FRAME_US)), + }, + track: Some("microphone".into()), + bandwidth: None, + }, + ) + .unwrap(); + let snapshot = catalogs.next().await.unwrap().unwrap(); + let decoder = MoqAudioConsumer::subscribe( + &consumer, + "microphone".into(), + snapshot.audio["microphone"].clone(), + MoqAudioDecoderOutput { + // Reading starts only after finish, so keep every group. + max_delay: Some(Duration::from_secs(30)), + ..audio_output() + }, + ) + .await + .unwrap(); + + for (i, timestamp_us) in [0, FRAME_US, RESUMED_US, RESUMED_US + FRAME_US].into_iter().enumerate() { + if timestamp_us == RESUMED_US { + audio.reanchor().unwrap(); + } + audio + .write(MoqAudioFrame { + timestamp_us, + data: audio_tone(i * FRAME_SAMPLES, FRAME_SAMPLES, 1), + }) + .unwrap(); + } + audio.finish().unwrap(); + + let mut ends = Vec::new(); + let mut frames = Vec::new(); + while let Some(read) = tokio::time::timeout(TIMEOUT, decoder.next()).await.unwrap().unwrap() { + match read { + MoqAudioRead::Frame { frame } => frames.push((ends.len(), frame.timestamp_us)), + MoqAudioRead::Gap { .. } => panic!("nothing was lost"), + MoqAudioRead::End { end_us } => ends.push(end_us), + } + } + // The reanchor drops whatever the first encoder's lookahead still held, while the + // finish drains the second one completely. + assert_eq!(ends.len(), 2, "{ends:?}"); + assert!(ends[0] > 0 && ends[0] <= 2 * FRAME_US, "{ends:?}"); + assert_eq!(ends[1], RESUMED_US + 2 * FRAME_US); + assert!( + frames.iter().any(|&(epoch, _)| epoch == 1), + "the reanchored encoder decodes after the first end: {frames:?}" + ); + for (epoch, timestamp_us) in frames { + if epoch == 0 { + assert!(timestamp_us < ends[0], "{timestamp_us} past the first end"); + } else { + assert!(timestamp_us >= RESUMED_US, "{timestamp_us} before the reanchor"); + } + } + broadcast.close().unwrap(); +} + /// Opus' 2.5 ms frame survives the trip to the encoder, and a duration Opus can't frame is /// refused. #[cfg(feature = "audio")] @@ -3747,11 +3846,14 @@ async fn video_decode_frame_ownership() { } let next = async |decoder: &MoqVideoConsumer| { - tokio::time::timeout(TIMEOUT, decoder.next()) + let read = tokio::time::timeout(TIMEOUT, decoder.next()) .await .expect("timed out") - .unwrap() - .expect("expected a frame") + .unwrap(); + match read { + Some(MoqVideoRead::Frame { frame }) => frame, + _ => panic!("expected a frame"), + } }; let frame = next(&portable).await; let retained = match &retaining { diff --git a/rs/moq-ffi/src/video.rs b/rs/moq-ffi/src/video.rs index 489a90f4937..2c4db9a5bf0 100644 --- a/rs/moq-ffi/src/video.rs +++ b/rs/moq-ffi/src/video.rs @@ -924,21 +924,65 @@ impl MoqVideoDecodedFrame { } } +/// One item of a [`MoqVideoConsumer`] read, in delivery order: a decoded picture, or a +/// break between them. +/// +/// Mirrors `moq_video::decode::Read`, the same shape as +/// [`MoqAudioRead`](crate::audio::MoqAudioRead), so a player handles loss and ends alike +/// for every media type. +#[derive(uniffi::Enum)] +pub enum MoqVideoRead { + /// A decoded picture. + Frame { + /// The decoded frame, holding a decoder buffer until it is released. + frame: Arc, + }, + /// Media after what was already read was lost: a group was skipped or never arrived. + /// Every picture from before the loss precedes it, and what follows does not continue + /// what came before. + Gap { + /// The furthest presentation point already read, in microseconds. + timestamp_us: u64, + /// An upper bound on how long until the next picture resumes. + duration: std::time::Duration, + }, + /// The publisher stopped its encoder: it paused, switched source, or finished. + /// + /// The stopped encoder's last pictures precede it; what follows comes from a fresh + /// encoder. + End { + /// Where the stopped encoder's media stops, in microseconds. + end_us: u64, + }, +} + +impl TryFrom for MoqVideoRead { + type Error = MoqError; + + fn try_from(read: moq_video::decode::Read) -> Result { + use crate::consumer::timestamp_us; + Ok(match read { + moq_video::decode::Read::Frame(frame) => Self::Frame { + frame: Arc::new(MoqVideoDecodedFrame { frame }), + }, + moq_video::decode::Read::Gap { timestamp, duration } => Self::Gap { + timestamp_us: timestamp_us(timestamp)?, + duration: crate::ffi::duration(duration.into()), + }, + moq_video::decode::Read::End { end } => Self::End { + end_us: timestamp_us(end)?, + }, + }) + } +} + struct VideoConsumerInner { consumer: moq_video::decode::Consumer, } impl VideoConsumerInner { - async fn next(&mut self) -> Result>, MoqError> { - loop { - match self.consumer.read().await? { - Some(moq_video::decode::Read::Frame(frame)) => { - return Ok(Some(Arc::new(MoqVideoDecodedFrame { frame }))); - } - Some(moq_video::decode::Read::Gap { .. } | moq_video::decode::Read::End { .. }) => {} - None => return Ok(None), - } - } + async fn next(&mut self) -> Result, MoqError> { + self.consumer.read().await?.map(MoqVideoRead::try_from).transpose() } } @@ -950,8 +994,8 @@ pub struct MoqVideoConsumer { #[uniffi::export] impl MoqVideoConsumer { - /// The next decoded frame, or `None` once the track ends. - pub async fn next(&self) -> Result>, MoqError> { + /// The next decoded frame, gap, or encoder end, or `None` once the track ends. + pub async fn next(&self) -> Result, MoqError> { self.task.run(|mut state| async move { state.next().await }).await } diff --git a/swift/MIGRATE.md b/swift/MIGRATE.md index 8d07ab5c2c6..3eb2a684b7f 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 +### Decoders read gaps and ends + +`Audio.Consumer` and `Video.Consumer` yield `Audio.Read` / `Video.Read` instead of frames. Match `case .frame(let frame)` where you used each element as a frame; `.gap(timestampUs:duration:)` marks lost media and `.end(endUs:)` where the publisher stopped its encoder, and what follows either does not continue what came before. + ### Durations are `TimeInterval` Every duration is a `TimeInterval` in seconds instead of a `UInt64` `...Us` field, and drops the suffix: `Client(timeout:race:resolutionDelay:)`, `Backoff(initial:max:timeout:)`, `QuicConfig(idleTimeout:keepAlive:)`, `WebSocketConfig(delay:)`, `ConnectionStats.rtt`, `Grant.expires`, `TrackInfo.maxAge`, `Subscription.maxDelay`, `Fill.maxDelay`, `Audio.Consumer.delay`, and the decoder outputs' `maxDelay` and `delay`. diff --git a/swift/Sources/Moq/Audio.swift b/swift/Sources/Moq/Audio.swift index 106b05ad9fe..a253c5d5381 100644 --- a/swift/Sources/Moq/Audio.swift +++ b/swift/Sources/Moq/Audio.swift @@ -11,6 +11,9 @@ public enum Audio { public typealias DecoderOutput = AudioDecoderOutput /// Interleaved PCM with its presentation timestamp. public typealias Frame = AudioFrame + /// One item a `Consumer` reads, in delivery order: `.frame`, a `.gap` where media + /// was lost, or an `.end` where the publisher stopped its encoder. + public typealias Read = MoqAudioRead /// One encoded payload from an `Encoder`. public typealias Packet = MoqAudioPacket /// The packets an `Encoder` drains on finish and the padded samples to discard. @@ -66,10 +69,11 @@ public enum Audio { } /// Read side of a raw-audio track. Iterating yields decoded PCM frames in the - /// layout declared by the `DecoderOutput` passed to `subscribe`. + /// layout declared by the `DecoderOutput` passed to `subscribe`, and the gaps + /// and encoder ends between them. public final class Consumer: AsyncSequence, Sendable { - /// The decoded PCM audio frame emitted by this sequence. - public typealias Element = Frame + /// The frame, gap, or end emitted by this sequence. + public typealias Element = Read let ffi: MoqAudioConsumer @@ -92,8 +96,8 @@ public enum Audio { )) } - /// The next frame, or `nil` once the track ends or is closed. - public func next() async throws -> Frame? { + /// The next frame, gap, or end, or `nil` once the track ends or is closed. + public func next() async throws -> Read? { try await ffi.next() } @@ -107,7 +111,8 @@ public enum Audio { } /// Create an iterator that cancels native reads when iteration ends. - public func makeAsyncIterator() -> AsyncThrowingStream.Iterator { + public func makeAsyncIterator() -> AsyncThrowingStream.Iterator { + moqStream(cancel: { [ffi] in ffi.cancel() }) { [ffi] in try await ffi.next() }.makeAsyncIterator() diff --git a/swift/Sources/Moq/Video.swift b/swift/Sources/Moq/Video.swift index 26f1df4c9e1..fe2bea90ac6 100644 --- a/swift/Sources/Moq/Video.swift +++ b/swift/Sources/Moq/Video.swift @@ -13,6 +13,9 @@ public enum Video { public typealias DecoderOutput = VideoDecoderOutput /// One raw picture with its presentation timestamp. public typealias Frame = VideoFrame + /// One item a `Consumer` reads, in delivery order: a decoded `.frame`, a `.gap` + /// where media was lost, or an `.end` where the publisher stopped its encoder. + public typealias Read = MoqVideoRead /// One encoded access unit from an `Encoder`. public typealias Packet = MoqVideoPacket /// How pictures are split into independently decodable groups. @@ -89,11 +92,12 @@ public enum Video { } /// Read side of a video track decoded inside the bindings. Iterating yields - /// frames that own their decoded picture; call `pixels(format:)` for tightly - /// packed CPU pixels. Release frames promptly: held frames hold decoder buffers. + /// frames that own their decoded picture, and the gaps and encoder ends between + /// them; call `pixels(format:)` for tightly packed CPU pixels. Release frames + /// promptly: held frames hold decoder buffers. public final class Consumer: AsyncSequence, Sendable { - /// The decoded video frame emitted by this sequence. - public typealias Element = VideoDecodedFrame + /// The frame, gap, or end emitted by this sequence. + public typealias Element = Read let ffi: MoqVideoConsumer @@ -116,8 +120,8 @@ public enum Video { )) } - /// The next frame, or `nil` once the track ends or is closed. - public func next() async throws -> VideoDecodedFrame? { + /// The next frame, gap, or end, or `nil` once the track ends or is closed. + public func next() async throws -> Read? { try await ffi.next() } @@ -127,7 +131,8 @@ public enum Video { } /// Create an iterator that cancels native reads when iteration ends. - public func makeAsyncIterator() -> AsyncThrowingStream.Iterator { + public func makeAsyncIterator() -> AsyncThrowingStream.Iterator { + moqStream(cancel: { [ffi] in ffi.cancel() }) { [ffi] in try await ffi.next() }.makeAsyncIterator() diff --git a/swift/Tests/MoqTests/SmokeTests.swift b/swift/Tests/MoqTests/SmokeTests.swift index 9a243f6fa92..6628c9369f7 100644 --- a/swift/Tests/MoqTests/SmokeTests.swift +++ b/swift/Tests/MoqTests/SmokeTests.swift @@ -655,9 +655,12 @@ final class SmokeTests: XCTestCase { } let next = try await decoder.next() - let frame = try XCTUnwrap(next) + guard case .frame(let frame) = try XCTUnwrap(next) else { + return XCTFail("expected a decoded frame") + } decoder.cancel() + let planar = try frame.pixels(format: .i420) XCTAssertEqual(planar.count, Int(frame.width()) * Int(frame.height()) * 3 / 2) diff --git a/test/interop/clients/c/subscribe.c b/test/interop/clients/c/subscribe.c index a224eda66c8..1f2c7b4b488 100644 --- a/test/interop/clients/c/subscribe.c +++ b/test/interop/clients/c/subscribe.c @@ -137,18 +137,29 @@ static void on_frame(void *ud, int32_t frame) { } // Writes each decoded audio frame to its rendition's files for tone.py, until enough has arrived. +// A gap or an encoder end before then fails the run, as the Rust client does. // Runs on moq-c's runtime thread, the only writer, so the files need no lock. -static void on_audio(void *ud, int32_t frame) { +static void on_audio(void *ud, int32_t id) { audio_t *a = (audio_t *)ud; ctx_t *c = a->ctx; - if (frame <= 0) { // 0 = ended, negative = error - if (frame < 0) fprintf(stderr, "audio decode error: %d\n", frame); + if (id <= 0) { // 0 = ended, negative = error + if (id < 0) fprintf(stderr, "audio decode error: %d\n", id); done(c, &a->done); return; } - moq_audio_frame f; - memset(&f, 0, sizeof(f)); - if (moq_decode_audio_frame((uint32_t)frame, &f) == 0 && a->samples < a->want) { + moq_audio_read read; + memset(&read, 0, sizeof(read)); + int32_t status = moq_decode_audio_read((uint32_t)id, &read); + if (status == 0 && read.kind != MOQ_DECODE_READ_FRAME && a->samples < a->want) { + pthread_mutex_lock(&c->mu); + if (!c->audio_error) { + c->audio_error = read.kind == MOQ_DECODE_READ_GAP ? "audio gap" : "audio track ended early"; + } + pthread_cond_broadcast(&c->cv); + pthread_mutex_unlock(&c->mu); + } + moq_audio_frame f = read.frame; + if (status == 0 && read.kind == MOQ_DECODE_READ_FRAME && a->samples < a->want) { uint64_t samples = f.data_size / a->frame_bytes; fprintf(a->log, "%llu %llu\n", (unsigned long long)f.timestamp_us, (unsigned long long)samples); fwrite(f.data, 1, f.data_size, a->pcm); @@ -162,7 +173,7 @@ static void on_audio(void *ud, int32_t frame) { pthread_mutex_unlock(&c->mu); } } - moq_decode_audio_frame_free((uint32_t)frame); + moq_decode_audio_read_free((uint32_t)id); } // Whether every audio rendition has decoded enough, once any has started. Called under the lock. diff --git a/test/interop/clients/cpp/main.cpp b/test/interop/clients/cpp/main.cpp index b5ad0d58ae0..65338dfbba7 100644 --- a/test/interop/clients/cpp/main.cpp +++ b/test/interop/clients/cpp/main.cpp @@ -32,6 +32,7 @@ #include #include #include +#include #include #ifdef _WIN32 @@ -298,10 +299,20 @@ void decode_audio( const size_t frame_bytes = sizeof(float) * audio.channel_count; uint64_t decoded = 0; while (decoded * 1000 < audio_ms * audio.sample_rate) { - auto frame = until(decoder->next(), deadline, "audio frame"); - if (!frame) { + auto read = until(decoder->next(), deadline, "audio frame"); + if (!read) { fail("audio track ended early"); } + // moq-audio reports a hole it saw directly; tone.py checks the timestamps too. + if (std::holds_alternative(read->get_variant())) { + fail("audio gap"); + } + const auto *variant = std::get_if(&read->get_variant()); + if (!variant) { + fail("audio track ended early"); + } + const auto *frame = &variant->frame; + const size_t samples = frame->data.size() / frame_bytes; log << frame->timestamp_us << " " << samples << "\n"; pcm.write(reinterpret_cast(frame->data.data()), static_cast(frame->data.size())); diff --git a/test/interop/clients/go/main.go b/test/interop/clients/go/main.go index af5dddeec9d..c1c5771b190 100644 --- a/test/interop/clients/go/main.go +++ b/test/interop/clients/go/main.go @@ -473,14 +473,22 @@ func decodeAudio(ctx context.Context, consumer *moq.BroadcastConsumer, id string frameBytes := 4 * int(rendition.ChannelCount) decoded := 0 for decoded*1000 < audioMs*int(rendition.SampleRate) { - frame, err := decoder.Next(ctx) + read, err := decoder.Next(ctx) if err != nil { return err } - if frame == nil { + var frame moqaudio.Frame + switch read := read.(type) { + case moqaudio.ReadFrame: + frame = read.Frame + // moq-audio reports a hole it saw directly; tone.py checks the timestamps too. + case moqaudio.ReadGap: + return fmt.Errorf("audio gap after %d samples", decoded) + default: // nil, the track's end, or a ReadEnd, the encoder's return errors.New("audio track ended early") } samples := len(frame.Data) / frameBytes + fmt.Fprintf(log, "%d %d\n", frame.Timestamp.Microseconds(), samples) if _, err := pcm.Write(frame.Data); err != nil { return err diff --git a/test/interop/clients/python/interop.py b/test/interop/clients/python/interop.py index 419278c3025..e1f201ae7d4 100644 --- a/test/interop/clients/python/interop.py +++ b/test/interop/clients/python/interop.py @@ -253,6 +253,11 @@ async def _decode_audio( with open(f"{out}.txt", "x") as log, open(f"{out}.pcm", "xb") as pcm: log.write(f"{audio.sample_rate} {audio.channel_count}\n") async for frame in decoder: + # moq-audio reports a hole it saw directly; tone.py checks the timestamps too. + if isinstance(frame, moq.audio.Gap): + raise RuntimeError(f"audio gap after {decoded} samples") + if isinstance(frame, moq.audio.End): + raise RuntimeError("audio track ended early") samples = len(frame.data) // frame_bytes log.write(f"{frame.timestamp // timedelta(microseconds=1)} {samples}\n") pcm.write(frame.data)