diff --git a/doc/lib/kt/moq.md b/doc/lib/kt/moq.md index 12c01a9b7f..e3136303dc 100644 --- a/doc/lib/kt/moq.md +++ b/doc/lib/kt/moq.md @@ -228,24 +228,28 @@ Call `request.abort(code)` when the requested group cannot be produced. Fetch is ### Fetching media groups -`fetchGroup` hands back raw payloads. `fetchMediaGroup` decodes the same group through the rendition's advertised container, so you get timestamped frames without opening a live subscription: +Catalog audio and video renditions include an optional timeline that maps presentation timestamps to retained group sequences. Subscribe to that index, then fetch and decode a specific group with the rendition's advertised container: ```kotlin -val (name, audio) = consumer.catalog().audio.entries.first() - -consumer.fetchMediaGroup( - name, - 42uL, - audio.container, - FetchGroupOptions(priority = 10u), -).use { group -> - group.frames().collect { frame -> - println("${frame.timestampUs}: ${frame.payload.size} bytes") +val catalog = consumer.catalog() +val (name, audio) = catalog.audio.entries.first() +val timeline = requireNotNull(audio.timeline) + +consumer.subscribeTimeline(timeline).entries().collect { entry -> + consumer.fetchMediaGroup( + name, + entry.group, + audio.container, + FetchGroupOptions(priority = 10u), + ).use { group -> + group.frames().collect { frame -> + println("${frame.timestampUs}: ${frame.payload.size} bytes") + } } } ``` -`frames()` is a cancellation-aware `Flow`. A fetched media group is finite: it completes after the group's last decoded frame, unlike the live `subscribeMedia` stream. Latency-based group skipping does not apply, so you always get every frame in the group. +`entries()` and `frames()` are cancellation-aware `Flow` helpers. A fetched media group is finite: it completes after the group's last decoded frame, unlike the live `subscribeMedia` stream. Latency-based group skipping does not apply, so you always get every frame in the group. ### On-demand raw tracks diff --git a/doc/lib/swift/moq.md b/doc/lib/swift/moq.md index 0e4701a275..b2038b1a28 100644 --- a/doc/lib/swift/moq.md +++ b/doc/lib/swift/moq.md @@ -174,23 +174,27 @@ Call `request.abort(errorCode:)` when the requested group cannot be produced. Fe ### Fetching media groups -`fetchGroup` hands back raw payloads. `fetchMediaGroup` decodes the same group through the rendition's advertised container, so you get timestamped frames without opening a live subscription: +Catalog audio and video renditions include an optional timeline that maps presentation timestamps to retained group sequences. Subscribe to that index, then fetch and decode a specific group with the rendition's advertised container: ```swift let catalog = try await consumer.subscribeCatalog().next()! let (name, audio) = catalog.audio.first! -let group = try await consumer.fetchMediaGroup( - name: name, - sequence: 42, - container: audio.container -) -for try await frame in group { - print(frame.timestampUs, frame.payload.count) +if let timeline = audio.timeline { + for try await entry in try await consumer.subscribeTimeline(timeline) { + let group = try await consumer.fetchMediaGroup( + name: name, + sequence: entry.group, + container: audio.container + ) + for try await frame in group { + print(frame.timestampUs, frame.payload.count) + } + } } ``` -`MediaGroupConsumer` is an `AsyncSequence`, and cancels the native read when iteration ends. A fetched media group is finite: it completes after the group's last decoded frame, unlike the live `subscribeMedia` stream. Latency-based group skipping does not apply, so you always get every frame in the group. +Both timeline and fetched media consumers are `AsyncSequence` values and cancel native reads when iteration ends. A fetched media group is finite: it completes after the group's last decoded frame, unlike the live `subscribeMedia` stream. Latency-based group skipping does not apply, so you always get every frame in the group. ### On-demand raw tracks diff --git a/kt/README.md b/kt/README.md index d8f76cf6c4..abbd06f036 100644 --- a/kt/README.md +++ b/kt/README.md @@ -47,7 +47,7 @@ The `dev.moq` package is intentionally thin: Kotlin has extension functions, so - **`Moq.connect(...)`**: a connection facade (`Moq.kt`), so you never hand-wire a `MoqClient`. - **Typealiases** (`Aliases.kt`): re-export the `Moq*`-prefixed FFI types under clean `dev.moq` names (`OriginProducer`, `BroadcastConsumer`, `Catalog`, `Frame`, ...), so you import `dev.moq.*` only. A couple of sealed types (`Container`, `MoqException`) are not aliased because Kotlin can't resolve their subtypes through a typealias; use `uniffi.moq.*` for those. - **Flow extensions** (`Flows.kt`): `updates()`, `groups()`, `frames()`, `announcements()`, `catalog()` turn the pull-based consumers into coroutine `Flow`s with cancellation wired through. -- **Fetched media**: `fetchMediaGroup(...).frames()` streams the decoded frames of one retained group, then completes. +- **Fetched media and timelines**: `fetchMediaGroup(...).frames()` streams decoded frames from one retained group, while `subscribeTimeline(...).entries()` streams its timestamp-to-group index. - **`logLevel(...)`**: configures native Rust tracing without importing the raw bindings package. - **Raw datagrams**: `TrackProducer.appendDatagram(Frame(payload, timestampUs))` sends one best-effort frame and returns its sequence; `TrackConsumer.recvDatagram()` and `datagrams()` receive them. Payloads are capped at 1200 bytes, require a datagram-capable transport plus lite-05 or newer moq-lite, and have no stream fallback. - **`MoqException.isShutdown`** (`Errors.kt`): true for the graceful `Cancelled`/`Closed` cases. diff --git a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt index bee9bf2b81..ff7aaebf5d 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt @@ -67,6 +67,8 @@ typealias MediaStreamProducer = uniffi.moq.MoqMediaStreamProducer typealias MediaConsumer = uniffi.moq.MoqMediaConsumer /** A finite fetched media group: yields container-decoded frames until the group ends. */ typealias MediaGroupConsumer = uniffi.moq.MoqMediaGroupConsumer +/** A media timeline subscription: yields timestamp-to-group index entries. */ +typealias TimelineConsumer = uniffi.moq.MoqTimelineConsumer /** The write side of a raw-audio track; PCM written here is encoded inside the FFI boundary. */ typealias AudioProducer = uniffi.moq.MoqAudioProducer /** The read side of a raw-audio track: yields decoded PCM frames. */ @@ -111,6 +113,10 @@ typealias Route = uniffi.moq.MoqRoute typealias Subscription = uniffi.moq.MoqSubscription /** Options for fetching one past group by sequence. */ typealias FetchGroupOptions = uniffi.moq.MoqFetchGroupOptions +/** A media track's companion timestamp-to-group index. */ +typealias Timeline = uniffi.moq.MoqTimeline +/** One timestamp-to-group mapping read from a media timeline track. */ +typealias TimelineEntry = uniffi.moq.MoqTimelineEntry /** Delivery settings for a raw track: priority, ordering, latency budget, and timescale. */ typealias TrackInfo = uniffi.moq.MoqTrackInfo /** One audio frame: PCM payload bytes plus a presentation timestamp. */ diff --git a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Flows.kt b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Flows.kt index b17ac90d9c..e602f38cac 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Flows.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Flows.kt @@ -27,6 +27,8 @@ import uniffi.moq.MoqMediaGroupConsumer import uniffi.moq.MoqOriginConsumer import uniffi.moq.MoqOriginDynamic import uniffi.moq.MoqRoute +import uniffi.moq.MoqTimelineConsumer +import uniffi.moq.MoqTimelineEntry import uniffi.moq.MoqTrackConsumer import uniffi.moq.MoqTrackDynamic import uniffi.moq.MoqTrackRequest @@ -81,6 +83,16 @@ fun MoqMediaGroupConsumer.frames(): Flow = flow { if (cause is CancellationException) cancel() } +/** Stream of indexed group boundaries from a media timeline track. */ +fun MoqTimelineConsumer.entries(): Flow = flow { + while (true) { + currentCoroutineContext().ensureActive() + emit(next() ?: break) + } +}.onCompletion { cause -> + if (cause is CancellationException) cancel() +} + /** * Stream of decoded audio frames in the layout declared by the * `MoqAudioDecoderConfig` the consumer was created with. diff --git a/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt b/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt index e3f83712ba..301a4be913 100644 --- a/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt +++ b/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt @@ -104,15 +104,16 @@ class SmokeTest { } } - /** A fetched media group streams its decoded frames and then completes. */ @Test - fun `media group helper streams fetched frames`() = runTest { + fun `media group and timeline helpers stream fetched data`() = runTest { BroadcastProducer().use { broadcast -> val media = broadcast.publishMedia( Init(format = "opus", data = opusHead(), video = null), ) val consumer = broadcast.consume() val (name, audio) = consumer.catalog().audio.entries.single() + val timeline = checkNotNull(audio.timeline) + val timelineConsumer: TimelineConsumer = consumer.subscribeTimeline(timeline) media.writeFrame(Frame(payload = "opus frame".encodeToByteArray(), timestampUs = 5_000_000uL)) @@ -124,7 +125,6 @@ class SmokeTest { audio.container, FetchGroupOptions(priority = 3u), ) - // Close the group so the fetched stream terminates instead of waiting for more. media.finish() @@ -136,6 +136,15 @@ class SmokeTest { assertEquals("opus frame", frame.payload.decodeToString()) assertEquals(5_000_000uL, frame.timestampUs) } + + broadcast.finish() + timelineConsumer.use { + val entries = it.entries().toList() + assertEquals(1, entries.size) + val entry = entries.single() + assertEquals(0uL, entry.group) + assertEquals(5_000_000uL, entry.timestampUs) + } } } diff --git a/rs/moq-ffi/src/audio.rs b/rs/moq-ffi/src/audio.rs index a115e5cd24..cb873957b2 100644 --- a/rs/moq-ffi/src/audio.rs +++ b/rs/moq-ffi/src/audio.rs @@ -289,6 +289,7 @@ mod tests { channel_count: 2, bitrate: None, container: MoqContainer::Legacy, + timeline: None, } } diff --git a/rs/moq-ffi/src/consumer.rs b/rs/moq-ffi/src/consumer.rs index 51c6bceccb..c2caf536ff 100644 --- a/rs/moq-ffi/src/consumer.rs +++ b/rs/moq-ffi/src/consumer.rs @@ -153,6 +153,27 @@ pub struct MoqCatalogConsumer { task: Task, } +#[derive(uniffi::Object)] +pub struct MoqTimelineConsumer { + task: Task, +} + +struct Timeline { + inner: moq_mux::timeline::Consumer, +} + +impl Timeline { + async fn next(&mut self) -> Result, MoqError> { + let Some(entry) = self.inner.next().await? else { + return Ok(None); + }; + Ok(Some(MoqTimelineEntry { + group: entry.group, + timestamp_us: timestamp_us(entry.pts)?, + })) + } +} + struct Catalog { // Consume with the untyped `Extra` extension so application sections survive into // `MoqCatalog.sections` instead of being dropped. @@ -218,6 +239,15 @@ impl MoqBroadcastConsumer { })) } + /// Subscribe to a media track's timestamp-to-group timeline index. + pub async fn subscribe_timeline(&self, timeline: MoqTimeline) -> Result, MoqError> { + let timeline: hang::catalog::Timeline = timeline.into(); + let consumer = moq_mux::timeline::Consumer::subscribe(&self.inner, &timeline).await?; + Ok(Arc::new(MoqTimelineConsumer { + task: Task::new(Timeline { inner: consumer }), + })) + } + /// Subscribe to a track by name, the same pattern as moq-boy's command/status tracks. /// /// Frames are returned as plain byte payloads with no codec or container parsing. @@ -296,6 +326,18 @@ impl MoqBroadcastConsumer { } } +#[uniffi::export] +impl MoqTimelineConsumer { + /// Return the next indexed group boundary, or `None` when the timeline ends. + pub async fn next(&self) -> Result, MoqError> { + self.task.run(|mut state| async move { state.next().await }).await + } + + pub fn cancel(&self) { + self.task.cancel(); + } +} + fn map_fetch_error(err: moq_net::Error) -> MoqError { match err { moq_net::Error::NotFound => MoqError::NotFound, diff --git a/rs/moq-ffi/src/media.rs b/rs/moq-ffi/src/media.rs index 67934e1efd..adcdd3497d 100644 --- a/rs/moq-ffi/src/media.rs +++ b/rs/moq-ffi/src/media.rs @@ -24,6 +24,33 @@ pub struct MoqVideoProperties { pub flip: Option, } +/// A media track's companion timestamp-to-group index. +#[derive(Clone, uniffi::Record)] +pub struct MoqTimeline { + pub track: String, + pub timescale: u32, + pub wall: Option, +} + +impl From for MoqTimeline { + fn from(timeline: hang::catalog::Timeline) -> Self { + Self { + track: timeline.track, + timescale: timeline.timescale, + wall: timeline.wall, + } + } +} + +impl From for hang::catalog::Timeline { + fn from(timeline: MoqTimeline) -> Self { + let mut out = Self::new(timeline.track); + out.timescale = timeline.timescale; + out.wall = timeline.wall; + out + } +} + /// How a track's frames are packaged, as advertised in the catalog. #[derive(Clone, uniffi::Enum)] pub enum MoqContainer { @@ -89,6 +116,7 @@ pub struct MoqVideo { pub stalled: bool, pub framerate: Option, pub container: MoqContainer, + pub timeline: Option, } #[derive(Clone, uniffi::Record)] @@ -99,6 +127,14 @@ pub struct MoqAudio { pub channel_count: u32, pub bitrate: Option, pub container: MoqContainer, + pub timeline: Option, +} + +/// One timestamp-to-group mapping read from a media timeline track. +#[derive(Clone, uniffi::Record)] +pub struct MoqTimelineEntry { + pub group: u64, + pub timestamp_us: u64, } /// A payload and the time it should be presented. @@ -223,6 +259,7 @@ pub(crate) fn convert_catalog(catalog: &moq_mux::catalog::hang::Catalog TimelineConsumer { + TimelineConsumer(try await ffi.subscribeTimeline(timeline: timeline)) + } + /// Subscribe to a media track, delivering frames in decode order. `container` /// comes from the catalog. `subscription` tunes delivery priority, group ordering /// priority, group range, and the latency budget; omit for defaults. Raise diff --git a/swift/Sources/Moq/Media.swift b/swift/Sources/Moq/Media.swift index 427ef93cbc..ffa317373c 100644 --- a/swift/Sources/Moq/Media.swift +++ b/swift/Sources/Moq/Media.swift @@ -94,6 +94,35 @@ public final class MediaGroupConsumer: AsyncSequence, Sendable { } } +/// A media track's timestamp-to-group timeline subscription. +public final class TimelineConsumer: AsyncSequence, Sendable { + /// The indexed group boundary emitted by this sequence. + public typealias Element = TimelineEntry + + let ffi: MoqTimelineConsumer + + init(_ ffi: MoqTimelineConsumer) { + self.ffi = ffi + } + + /// The next indexed group boundary, or `nil` once the timeline ends. + public func next() async throws -> TimelineEntry? { + try await ffi.next() + } + + /// Cancel all current and future reads. + public func cancel() { + ffi.cancel() + } + + /// Create an iterator that cancels native reads when iteration ends. + public func makeAsyncIterator() -> AsyncThrowingStream.Iterator { + moqStream(cancel: { [ffi] in ffi.cancel() }) { [ffi] in + try await ffi.next() + }.makeAsyncIterator() + } +} + /// Write side of a media track fed pre-framed payloads. public final class MediaProducer: Sendable { let ffi: MoqMediaProducer