feat(bindings)!: decoders read gaps and ends

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<moq::AudioRead::kGap>(read->get_variant()));
+        // The finish ends the encoder where its eight 20 ms frames stop.
+        if (const auto *end = std::get_if<moq::AudioRead::kEnd>(&read->get_variant())) {
+            CHECK(end->end_us == 250000 + 8 * 20000);
+            continue;
+        }
+        const auto *variant = std::get_if<moq::AudioRead::kFrame>(&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++) {

cpp/obs/src/moq-source.cpp

@@ -699,13 +699,14 @@ static void moq_source_stop_audio(const std::shared_ptr<Connection> &conn, const
 static void moq_source_read_audio(struct moq_source *ctx, const std::shared_ptr<Connection> &conn,
 				  const std::shared_ptr<Track> &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<Connection> &conn,
-				      const std::shared_ptr<Track> &track,
-				      const moq::expected<std::optional<moq::AudioFrame>> &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<Connection> &conn,
+				     const std::shared_ptr<Track> &track,
+				     const moq::expected<std::optional<moq::AudioRead>> &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<moq::AudioRead::kFrame>(&(*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,15 +761,15 @@ 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<Connection>(conn),
-		 weak_track = std::weak_ptr<Track>(track)](moq::expected<std::optional<moq::AudioFrame>> result) {
+		 weak_track = std::weak_ptr<Track>(track)](moq::expected<std::optional<moq::AudioRead>> result) {
 			obs_source_audio audio = {};
 			{
 				std::lock_guard<std::mutex> lock(ctx->mutex);
 				auto conn = weak_conn.lock();
 				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

cpp/obs/test/moq-audio-encoder-test.cpp

@@ -15,6 +15,7 @@
 #include <cstring>
 #include <optional>
 #include <string>
+#include <variant>
 #include <vector>
 
 #include <obs-module.h>
@@ -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<moq::AudioRead::kFrame>(&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<const float *>(frame->data.data());

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<std::optional<moq::AudioFrame>> misaligned(moq::AudioFrame{1, {0, 0, 0}});
-			CHECK(!moq_source_on_audio_frame(ctx, conn, active, misaligned, audio));
+			moq::expected<std::optional<moq::AudioRead>> 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);

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.

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;

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;
 

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<moq_audio.ReadFrame>());
+    expect(await estimated.next().timeout(timeout), isA<moq_audio.ReadFrame>());
 
     expect(fixed.delay(), const Duration(milliseconds: 80));
     expect(estimated.delay(), isNot(initial));