From d11e7944c49b4872a5ed6507362f20ad5fec77ff Mon Sep 17 00:00:00 2001 From: Brad Greenway Date: Thu, 24 Sep 2026 17:27:23 -0500 Subject: [PATCH 1/2] feat(libmoq): advertise JSON tracks in the catalog, add binary data tracks moq_publish_json_snapshot and moq_publish_json_stream created a bare track and told the caller to advertise it with moq_publish_catalog_section, which refuses the reserved json section, so a C publisher could not make a JSON track discoverable at all. They now go through the catalog producer's json_snapshot/json_stream, writing json.tracks. (mode, compression) and retiring it when the track finishes or fails, as the Rust API already does. Adds moq_publish_binary_snapshot/_stream (+ _update/_append/_finish) with moq_binary_config {compression, mime}, advertised under binary.tracks. the same way. moq-mux json::Config gains an optional delta_ratio so the snapshot keeps the delta setting libmoq already exposed; it is not part of the catalog entry. --- doc/lib/c/index.md | 2 +- rs/libmoq/src/api.rs | 166 +++++++++++++++++++++++++++--- rs/libmoq/src/publish.rs | 108 +++++++++++++++----- rs/libmoq/src/test.rs | 211 +++++++++++++++++++++++++++++++++++++++ rs/moq-mux/src/json.rs | 16 +++ 5 files changed, 463 insertions(+), 40 deletions(-) diff --git a/doc/lib/c/index.md b/doc/lib/c/index.md index ce305408e1..ed93a94bd3 100644 --- a/doc/lib/c/index.md +++ b/doc/lib/c/index.md @@ -41,7 +41,7 @@ and `target/include/moq.h`. - **Server.** `moq_server_listen` binds before it returns (a bad address or certificate fails there) and hands each incoming session to `on_request` as a request handle. Read `moq_session_request_path` and `_query` to route and authenticate, then `moq_session_request_accept` (a session handle, with origins like `moq_session_connect`) or `moq_session_request_reject` with an HTTP-style code (401 and 403 become the protocol's unauthorized close). An accepted session reports `1` once SETUP completes and never reconnects. `moq_server_addr` reports an ephemeral port and `moq_server_fingerprints` the hashes a client pins for a `tls_generate` certificate. `moq_server_close` stops listening; its terminal callback fires once the sockets are released. - **Demand.** A watcher on a published track (`moq_publish_track_demand`, `moq_publish_media_demand`, `moq_encode_video_demand`, `moq_encode_audio_demand`) calls `on_demand` with `MOQ_DEMAND_USED` or `MOQ_DEMAND_UNUSED` right away and again on every change, so an encoder on a battery-powered device runs only while someone is watching. The first call is the current state, so a track that went unused before the watcher existed still reports it. `moq_publish_demand_cancel` stops it; the terminal callback still fires. A container has no single demand and is refused. Demand follows the last real subscriber: an origin that served the track drops its source copy on the unused edge and keeps only the finished groups it already cached warm for 30 seconds, so the cache linger does not delay the unused edge. - **Requests.** `moq_publish_dynamic` serves subscriptions to tracks the broadcast never declared: each arrives as a request handle, read its name with `moq_track_request_name`, then `moq_track_request_accept` (a raw track handle), `moq_track_request_video` / `_audio` (the media handle `moq_publish_video` / `_audio` return), or `moq_track_request_abort` with an application code the subscriber sees. Without a live handler an unknown name is refused. `moq_publish_track_dynamic` does the same for fetches of groups a track no longer has cached, delivered as `moq_group_request_*` (`sequence`, `priority`, `frame_start`); `moq_group_request_accept` starts the producer at `frame_start` so written frames keep their group indices. Register it with `moq_track_request_dynamic` before accepting a track that was itself requested by a fetch, so that pending group survives the transition. Both handlers stop with `moq_publish_dynamic_cancel`. -- **Everything the bindings can do** ([list](/lib/#what-every-binding-can-do)): media publish and consume with the catalog managed for you, raw pixels and PCM with the codec inside (`moq_encode_video`, `moq_encode_audio`, and the `moq_decode_*` mirrors), raw tracks with timestamps and datagrams, JSON snapshot and stream tracks, group fetch, catalog sections, shared video properties, and stalled hints. The three advertising operations are `moq_origin_create_broadcast` (unannounced producer, invisible to everyone), `moq_publish_announce` / `moq_publish_unannounce` (exact-path advertisement), and `moq_origin_dynamic` (a claim over a path prefix and everything beneath it; `""` for everything). A route is a capability, not an inventory. `moq_origin_announced` takes a literal prefix and an optional relative pattern filter; `moq_announce_update.prefix` stays relative to the origin, while `captures` reports what each wildcard matched when `has_captures` is true. +- **Everything the bindings can do** ([list](/lib/#what-every-binding-can-do)): media publish and consume with the catalog managed for you, raw pixels and PCM with the codec inside (`moq_encode_video`, `moq_encode_audio`, and the `moq_decode_*` mirrors), raw tracks with timestamps and datagrams, JSON and binary data tracks (snapshot or stream, each advertised in the catalog for as long as it lives), group fetch, catalog sections, shared video properties, and stalled hints. The three advertising operations are `moq_origin_create_broadcast` (unannounced producer, invisible to everyone), `moq_publish_announce` / `moq_publish_unannounce` (exact-path advertisement), and `moq_origin_dynamic` (a claim over a path prefix and everything beneath it; `""` for everything). A route is a capability, not an inventory. `moq_origin_announced` takes a literal prefix and an optional relative pattern filter; `moq_announce_update.prefix` stays relative to the origin, while `captures` reports what each wildcard matched when `has_captures` is true. ```c moq_client_config config; diff --git a/rs/libmoq/src/api.rs b/rs/libmoq/src/api.rs index 6d100c48e4..0ea8b2e3c5 100644 --- a/rs/libmoq/src/api.rs +++ b/rs/libmoq/src/api.rs @@ -433,6 +433,22 @@ pub struct moq_json_stream_config { pub compression: bool, } +/// Options for a binary data track, in either mode. +/// +/// The mode is fixed by which constructor is called ([moq_publish_binary_snapshot] or +/// [moq_publish_binary_stream]), so it is not in here. +#[repr(C)] +#[allow(non_camel_case_types)] +pub struct moq_binary_config { + /// DEFLATE-compress each payload, advertised in the catalog entry. + pub compression: bool, + + /// The payloads' media type (e.g. `image/jpeg`), or NULL to leave it unstated. + pub mime: *const c_char, + /// Length of `mime` in bytes. + pub mime_len: usize, +} + /// A JSON value delivered by a consumer callback. #[repr(C)] #[allow(non_camel_case_types)] @@ -2886,10 +2902,12 @@ pub extern "C" fn moq_publish_group_abort(group: u32, error_code: u16) -> i32 { /// Create a JSON snapshot track (lossy latest-value) on a broadcast. /// /// Values published via [moq_publish_json_snapshot_update] reach subscribers as a single latest -/// state; a late joiner only sees the newest. Advertise the track in the catalog with -/// [moq_publish_catalog_section] if consumers should discover it. +/// state; a late joiner only sees the newest. The track is advertised in the broadcast's catalog +/// under `json.tracks.` with `mode: snapshot` (and `compression: deflate` when set), and the +/// entry is retired when the track finishes or fails, so consumers discover it with no extra call. /// -/// Returns a non-zero handle to the JSON producer on success, or a negative code on failure. +/// Returns a non-zero handle to the JSON producer on success, or a negative code on failure, +/// including a mux error when the catalog already carries an entry named `name`. /// /// # Safety /// - The caller must ensure `name` is a valid pointer to `name_len` bytes and `config` a valid pointer. @@ -2904,13 +2922,9 @@ pub unsafe extern "C" fn moq_publish_json_snapshot( let broadcast = ffi::parse_id(broadcast)?; let name = unsafe { ffi::parse_str(name, name_len)? }; let config = unsafe { config.as_ref() }.ok_or(Error::InvalidPointer)?; - let mut producer = moq_json::snapshot::Config::default(); - producer.delta_ratio = config.delta_ratio; - producer.compression = if config.compression { - moq_json::Compression::Deflate - } else { - moq_json::Compression::None - }; + let producer = moq_mux::json::Config::default() + .with_compression(config.compression) + .with_delta_ratio(config.delta_ratio); State::lock().publish.json_snapshot(broadcast, name, producer) }) } @@ -2946,8 +2960,11 @@ pub extern "C" fn moq_publish_json_snapshot_finish(json: u32) -> i32 { /// Create a JSON stream track (lossless append-log) on a broadcast. /// /// Every record appended via [moq_publish_json_stream_append] is preserved and delivered in order. +/// The track is advertised in the broadcast's catalog under `json.tracks.` with +/// `mode: stream`, for as long as the track lives. /// -/// Returns a non-zero handle to the JSON stream producer on success, or a negative code on failure. +/// Returns a non-zero handle to the JSON stream producer on success, or a negative code on failure, +/// including a mux error when the catalog already carries an entry named `name`. /// /// # Safety /// - The caller must ensure `name` is a valid pointer to `name_len` bytes and `config` a valid pointer. @@ -2962,10 +2979,7 @@ pub unsafe extern "C" fn moq_publish_json_stream( let broadcast = ffi::parse_id(broadcast)?; let name = unsafe { ffi::parse_str(name, name_len)? }; let config = unsafe { config.as_ref() }.ok_or(Error::InvalidPointer)?; - let mut producer = moq_json::stream::Config::default(); - if config.compression { - producer.compression = moq_json::Compression::Deflate; - } + let producer = moq_mux::json::Config::default().with_compression(config.compression); State::lock().publish.json_stream(broadcast, name, producer) }) } @@ -2997,6 +3011,128 @@ pub extern "C" fn moq_publish_json_stream_finish(stream: u32) -> i32 { }) } +/// Parse a [moq_binary_config] into the mux's binary track config. +/// +/// # Safety +/// - `config` must be a valid pointer, and its `mime` a valid pointer to `mime_len` bytes when not NULL. +unsafe fn binary_config(config: *const moq_binary_config) -> Result { + let config = unsafe { config.as_ref() }.ok_or(Error::InvalidPointer)?; + let mut binary = moq_mux::binary::Config::default().with_compression(config.compression); + if let Some(mime) = unsafe { ffi::parse_str_optional(config.mime, config.mime_len)? } { + binary = binary.with_mime(mime); + } + Ok(binary) +} + +/// Create a binary snapshot track (lossy latest-value) on a broadcast: each payload supersedes the +/// last, and a late joiner only sees the newest, e.g. the latest thumbnail of a camera. +/// +/// The track is advertised in the broadcast's catalog under `binary.tracks.` with +/// `mode: snapshot` (plus `mime` and `compression` when set), and the entry is retired when the +/// track finishes or fails. +/// +/// Returns a non-zero handle to the binary producer on success, or a negative code on failure, +/// including a mux error when the catalog already carries an entry named `name`. +/// +/// # Safety +/// - The caller must ensure `name` is a valid pointer to `name_len` bytes and `config` a valid pointer. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn moq_publish_binary_snapshot( + broadcast: u32, + name: *const c_char, + name_len: usize, + config: *const moq_binary_config, +) -> i32 { + ffi::enter(move || { + let broadcast = ffi::parse_id(broadcast)?; + let name = unsafe { ffi::parse_str(name, name_len)? }; + let config = unsafe { binary_config(config)? }; + State::lock().publish.binary_snapshot(broadcast, name, config) + }) +} + +/// Publish a new payload to a binary snapshot track, superseding the last. +/// +/// Returns a zero on success, or a negative code on failure. +/// +/// # Safety +/// - The caller must ensure `payload` is a valid pointer to `payload_len` bytes. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn moq_publish_binary_snapshot_update( + binary: u32, + payload: *const u8, + payload_len: usize, +) -> i32 { + ffi::enter(move || { + let binary = ffi::parse_id(binary)?; + let payload = unsafe { ffi::parse_slice(payload, payload_len)? }; + State::lock().publish.binary_snapshot_update(binary, payload) + }) +} + +/// Finish a binary snapshot track and retire its catalog entry. No more payloads can be published. +/// +/// Returns a zero on success, or a negative code on failure. +#[unsafe(no_mangle)] +pub extern "C" fn moq_publish_binary_snapshot_finish(binary: u32) -> i32 { + ffi::enter(move || { + let binary = ffi::parse_id(binary)?; + State::lock().publish.binary_snapshot_finish(binary) + }) +} + +/// Create a binary stream track (lossless append-log) on a broadcast: every payload is preserved +/// and delivered in order. +/// +/// The track is advertised in the broadcast's catalog under `binary.tracks.` with +/// `mode: stream` (plus `mime` and `compression` when set), for as long as the track lives. +/// +/// Returns a non-zero handle to the binary stream producer on success, or a negative code on +/// failure, including a mux error when the catalog already carries an entry named `name`. +/// +/// # Safety +/// - The caller must ensure `name` is a valid pointer to `name_len` bytes and `config` a valid pointer. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn moq_publish_binary_stream( + broadcast: u32, + name: *const c_char, + name_len: usize, + config: *const moq_binary_config, +) -> i32 { + ffi::enter(move || { + let broadcast = ffi::parse_id(broadcast)?; + let name = unsafe { ffi::parse_str(name, name_len)? }; + let config = unsafe { binary_config(config)? }; + State::lock().publish.binary_stream(broadcast, name, config) + }) +} + +/// Append one payload to a binary stream track. +/// +/// Returns a zero on success, or a negative code on failure. +/// +/// # Safety +/// - The caller must ensure `payload` is a valid pointer to `payload_len` bytes. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn moq_publish_binary_stream_append(stream: u32, payload: *const u8, payload_len: usize) -> i32 { + ffi::enter(move || { + let stream = ffi::parse_id(stream)?; + let payload = unsafe { ffi::parse_slice(payload, payload_len)? }; + State::lock().publish.binary_stream_append(stream, payload) + }) +} + +/// Finish a binary stream track and retire its catalog entry. No more payloads can be appended. +/// +/// Returns a zero on success, or a negative code on failure. +#[unsafe(no_mangle)] +pub extern "C" fn moq_publish_binary_stream_finish(stream: u32) -> i32 { + ffi::enter(move || { + let stream = ffi::parse_id(stream)?; + State::lock().publish.binary_stream_finish(stream) + }) +} + /// Create a catalog consumer for a broadcast. /// /// `on_catalog` is invoked with a positive catalog ID for each catalog update diff --git a/rs/libmoq/src/publish.rs b/rs/libmoq/src/publish.rs index 18e5c28a2d..489a3e3634 100644 --- a/rs/libmoq/src/publish.rs +++ b/rs/libmoq/src/publish.rs @@ -70,11 +70,18 @@ pub struct Publish { /// Raw group producers, created from a raw track producer. groups: NonZeroSlab, - /// JSON snapshot producers (lossy latest-value tracks). - json_snapshot: NonZeroSlab>, + /// JSON snapshot producers (lossy latest-value tracks), each advertised in its broadcast's + /// catalog for as long as it lives. + json_snapshot: NonZeroSlab>, - /// JSON stream producers (lossless append-log tracks). - json_stream: NonZeroSlab>, + /// JSON stream producers (lossless append-log tracks), advertised the same way. + json_stream: NonZeroSlab>, + + /// Binary snapshot producers (lossy latest-value tracks of opaque bytes), advertised the same way. + binary_snapshot: NonZeroSlab>, + + /// Binary stream producers (lossless append-log tracks of opaque bytes), advertised the same way. + binary_stream: NonZeroSlab>, /// Demand watchers. Close signals shutdown; the task delivers a final callback, then removes itself. demand: NonZeroSlab>, @@ -132,6 +139,12 @@ impl Publish { Ok(&mut self.broadcasts.get_mut(id).ok_or(Error::BroadcastNotFound)?.catalog) } + /// The broadcast's current catalog, as consumers would receive it next. + #[cfg(test)] + pub fn catalog_snapshot(&mut self, id: Id) -> Result, Error> { + Ok(self.catalog(id)?.snapshot()) + } + /// Mutable access to both the broadcast and its catalog producer. /// Used by sibling modules (e.g. `audio`) that need to attach a new /// track to an existing publish. @@ -728,20 +741,26 @@ impl Publish { Ok(()) } - /// Create a JSON snapshot track (lossy latest-value) on a broadcast. - /// - /// Values published via [`Self::json_snapshot_update`] reach subscribers as a single latest - /// state; a late joiner only sees the newest value. Advertise the track in the catalog with - /// [`Self::catalog_section_set`] if consumers should discover it. - pub fn json_snapshot( + /// Create a track on a broadcast and hand it, with the broadcast's catalog, to `publish`, which + /// wraps it in a data producer that advertises the track in that catalog. + fn data_track( &mut self, broadcast: Id, name: &str, - config: moq_json::snapshot::Config, - ) -> Result { - let broadcast = self.producer(broadcast)?; - let track = broadcast.create_track(name, None)?; - let producer = moq_json::snapshot::Producer::new(track, config); + publish: impl FnOnce(&moq_mux::catalog::Producer, moq_net::track::Producer) -> moq_mux::Result, + ) -> Result { + let broadcast = self.broadcasts.get_mut(broadcast).ok_or(Error::BroadcastNotFound)?; + let track = broadcast.producer.create_track(name, None)?; + Ok(publish(&broadcast.catalog, track)?) + } + + /// Create a JSON snapshot track (lossy latest-value) on a broadcast, advertised in its catalog. + /// + /// Values published via [`Self::json_snapshot_update`] reach subscribers as a single latest + /// state; a late joiner only sees the newest value. The catalog entry (`json.tracks.`, + /// `mode: snapshot`) is written now and retired when the track finishes or fails. + pub fn json_snapshot(&mut self, broadcast: Id, name: &str, config: moq_mux::json::Config) -> Result { + let producer = self.data_track(broadcast, name, |catalog, track| catalog.json_snapshot(track, config))?; self.json_snapshot.insert(producer) } @@ -752,20 +771,19 @@ impl Publish { Ok(()) } - /// Finish a JSON snapshot track. No more values can be published. + /// Finish a JSON snapshot track and retire its catalog entry. No more values can be published. pub fn json_snapshot_finish(&mut self, json: Id) -> Result<(), Error> { - let mut producer = self.json_snapshot.remove(json).ok_or(Error::TrackNotFound)?; + let producer = self.json_snapshot.remove(json).ok_or(Error::TrackNotFound)?; producer.finish()?; Ok(()) } - /// Create a JSON stream track (lossless append-log) on a broadcast. + /// Create a JSON stream track (lossless append-log) on a broadcast, advertised in its catalog. /// /// Every record appended via [`Self::json_stream_append`] is preserved and delivered in order. - pub fn json_stream(&mut self, broadcast: Id, name: &str, config: moq_json::stream::Config) -> Result { - let broadcast = self.producer(broadcast)?; - let track = broadcast.create_track(name, None)?; - let producer = moq_json::stream::Producer::new(track, config); + /// The catalog entry (`json.tracks.`, `mode: stream`) lives as long as the track. + pub fn json_stream(&mut self, broadcast: Id, name: &str, config: moq_mux::json::Config) -> Result { + let producer = self.data_track(broadcast, name, |catalog, track| catalog.json_stream(track, config))?; self.json_stream.insert(producer) } @@ -776,9 +794,51 @@ impl Publish { Ok(()) } - /// Finish a JSON stream track. No more records can be appended. + /// Finish a JSON stream track and retire its catalog entry. No more records can be appended. pub fn json_stream_finish(&mut self, stream: Id) -> Result<(), Error> { - let mut producer = self.json_stream.remove(stream).ok_or(Error::TrackNotFound)?; + let producer = self.json_stream.remove(stream).ok_or(Error::TrackNotFound)?; + producer.finish()?; + Ok(()) + } + + /// Create a binary snapshot track (lossy latest-value) on a broadcast, advertised in its catalog + /// as `binary.tracks.`, `mode: snapshot`. + pub fn binary_snapshot(&mut self, broadcast: Id, name: &str, config: moq_mux::binary::Config) -> Result { + let producer = self.data_track(broadcast, name, |catalog, track| catalog.binary_snapshot(track, config))?; + self.binary_snapshot.insert(producer) + } + + /// Publish a new payload to a binary snapshot track, superseding the last. + pub fn binary_snapshot_update(&mut self, binary: Id, payload: &[u8]) -> Result<(), Error> { + let producer = self.binary_snapshot.get_mut(binary).ok_or(Error::TrackNotFound)?; + producer.update(bytes::Bytes::copy_from_slice(payload))?; + Ok(()) + } + + /// Finish a binary snapshot track and retire its catalog entry. + pub fn binary_snapshot_finish(&mut self, binary: Id) -> Result<(), Error> { + let producer = self.binary_snapshot.remove(binary).ok_or(Error::TrackNotFound)?; + producer.finish()?; + Ok(()) + } + + /// Create a binary stream track (lossless append-log) on a broadcast, advertised in its catalog + /// as `binary.tracks.`, `mode: stream`. + pub fn binary_stream(&mut self, broadcast: Id, name: &str, config: moq_mux::binary::Config) -> Result { + let producer = self.data_track(broadcast, name, |catalog, track| catalog.binary_stream(track, config))?; + self.binary_stream.insert(producer) + } + + /// Append one payload to a binary stream track. + pub fn binary_stream_append(&mut self, stream: Id, payload: &[u8]) -> Result<(), Error> { + let producer = self.binary_stream.get_mut(stream).ok_or(Error::TrackNotFound)?; + producer.append(bytes::Bytes::copy_from_slice(payload))?; + Ok(()) + } + + /// Finish a binary stream track and retire its catalog entry. + pub fn binary_stream_finish(&mut self, stream: Id) -> Result<(), Error> { + let producer = self.binary_stream.remove(stream).ok_or(Error::TrackNotFound)?; producer.finish()?; Ok(()) } diff --git a/rs/libmoq/src/test.rs b/rs/libmoq/src/test.rs index 985e77526f..baaf02007d 100644 --- a/rs/libmoq/src/test.rs +++ b/rs/libmoq/src/test.rs @@ -4830,3 +4830,214 @@ fn server_listen_refuses_bad_config() { MOQ_ERROR_INVALID_CONFIG ); } + +/// Subscribe to `name` on `consume` as a raw track and return the payloads of its first `count` +/// frames. +fn read_raw_frames(consume: u32, name: &[u8], count: usize) -> Vec> { + let frame_cb = Callback::new(); + let subscription = moq_subscription { + priority: 0, + max_age_us: 1_000_000, + group_start: 0, + group_start_present: false, + group_end: 0, + group_end_present: false, + }; + let track = id(unsafe { + moq_consume_track( + consume, + name.as_ptr() as *const c_char, + name.len(), + &subscription, + Some(channel_callback), + frame_cb.ptr, + ) + }); + let mut payloads = Vec::with_capacity(count); + for _ in 0..count { + let frame_id = id(frame_cb.recv()); + let mut frame = moq_frame { + payload: std::ptr::null(), + payload_size: 0, + timestamp_us: 0, + keyframe: false, + }; + assert_eq!(unsafe { moq_consume_track_frame(frame_id, &mut frame) }, 0); + payloads.push(unsafe { std::slice::from_raw_parts(frame.payload, frame.payload_size) }.to_vec()); + assert_eq!(moq_consume_track_frame_free(frame_id), 0); + } + assert_eq!(moq_consume_track_cancel(track), 0); + // The callback context must outlive libmoq's last call into it: wait for the terminal. + assert_eq!(frame_cb.recv_terminal(), 0, "clean cancel delivers terminal 0"); + payloads +} + +/// The broadcast's current catalog, read on the publish side. +fn published_catalog(broadcast: u32) -> moq_mux::catalog::hang::Catalog { + let id = crate::Id::try_from(broadcast).expect("valid broadcast id"); + crate::State::lock() + .publish + .catalog_snapshot(id) + .expect("broadcast exists") +} + +#[test] +fn json_tracks_are_advertised_in_the_catalog() { + let origin = id(moq_origin_create()); + let broadcast = publish_broadcast(origin, b"json-catalog"); + + let status = b"status"; + let snapshot = id(unsafe { + moq_publish_json_snapshot( + broadcast, + status.as_ptr() as *const c_char, + status.len(), + &moq_json_snapshot_config { + delta_ratio: 4, + compression: true, + }, + ) + }); + let events = b"events"; + let stream = id(unsafe { + moq_publish_json_stream( + broadcast, + events.as_ptr() as *const c_char, + events.len(), + &moq_json_stream_config { compression: false }, + ) + }); + + let catalog = published_catalog(broadcast); + let entry = catalog.json.tracks.get("status").expect("snapshot track advertised"); + assert_eq!(entry.mode, hang::catalog::Mode::Snapshot); + assert_eq!(entry.compression, Some(hang::catalog::Compression::Deflate)); + let entry = catalog.json.tracks.get("events").expect("stream track advertised"); + assert_eq!(entry.mode, hang::catalog::Mode::Stream); + assert_eq!(entry.compression, None); + + // Finishing a track retires its entry; the other stays. + assert_eq!(moq_publish_json_snapshot_finish(snapshot), 0); + let catalog = published_catalog(broadcast); + assert!( + !catalog.json.tracks.contains_key("status"), + "finished track still advertised" + ); + assert!(catalog.json.tracks.contains_key("events")); + + assert_eq!(moq_publish_json_stream_finish(stream), 0); + assert!(published_catalog(broadcast).json.tracks.is_empty()); + + assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_origin_close(origin), 0); +} + +#[test] +fn binary_snapshot_is_advertised_and_delivered() { + let origin = id(moq_origin_create()); + let path = b"binary-snapshot"; + let broadcast = publish_broadcast(origin, path); + + let name = b"thumbnail"; + let mime = b"image/jpeg"; + let producer = id(unsafe { + moq_publish_binary_snapshot( + broadcast, + name.as_ptr() as *const c_char, + name.len(), + &moq_binary_config { + compression: false, + mime: mime.as_ptr() as *const c_char, + mime_len: mime.len(), + }, + ) + }); + + let catalog = published_catalog(broadcast); + let entry = catalog.binary.tracks.get("thumbnail").expect("binary track advertised"); + assert_eq!(entry.mode, hang::catalog::Mode::Snapshot); + assert_eq!(entry.mime.as_deref(), Some("image/jpeg")); + + // The payload reaches a raw subscriber of the same track name, untouched. + let payload = [0xff_u8, 0xd8, 0xff, 0xe0, 1, 2, 3]; + assert_eq!( + unsafe { moq_publish_binary_snapshot_update(producer, payload.as_ptr(), payload.len()) }, + 0 + ); + let consume = request_broadcast(origin, path); + let frames = read_raw_frames(consume, name, 1); + assert_eq!(frames, vec![payload.to_vec()]); + + assert_eq!(moq_publish_binary_snapshot_finish(producer), 0); + assert!( + moq_publish_binary_snapshot_finish(producer) < 0, + "double-finish should fail" + ); + assert!(published_catalog(broadcast).binary.tracks.is_empty()); + + assert_eq!(moq_consume_close(consume), 0); + assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_origin_close(origin), 0); +} + +#[test] +fn data_track_names_cannot_collide() { + let origin = id(moq_origin_create()); + let broadcast = publish_broadcast(origin, b"data-collide"); + + let name = b"state"; + let first = id(unsafe { + moq_publish_json_snapshot( + broadcast, + name.as_ptr() as *const c_char, + name.len(), + &moq_json_snapshot_config { + delta_ratio: 0, + compression: false, + }, + ) + }); + // A second data track under the same name is refused rather than silently replacing the + // first entry, and a NULL mime is allowed (left unstated). + assert!( + unsafe { + moq_publish_binary_stream( + broadcast, + name.as_ptr() as *const c_char, + name.len(), + &moq_binary_config { + compression: false, + mime: std::ptr::null(), + mime_len: 0, + }, + ) + } < 0, + "a duplicate data track name should fail" + ); + assert_eq!( + published_catalog(broadcast) + .json + .tracks + .get("state") + .map(|e| e.mode.clone()), + Some(hang::catalog::Mode::Snapshot), + "the refused duplicate must leave the first entry in place" + ); + + // A NULL config is refused. + let other = b"other"; + assert!( + unsafe { + moq_publish_binary_stream( + broadcast, + other.as_ptr() as *const c_char, + other.len(), + std::ptr::null(), + ) + } < 0 + ); + + assert_eq!(moq_publish_json_snapshot_finish(first), 0); + assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_origin_close(origin), 0); +} diff --git a/rs/moq-mux/src/json.rs b/rs/moq-mux/src/json.rs index 026c489d88..26692fa745 100644 --- a/rs/moq-mux/src/json.rs +++ b/rs/moq-mux/src/json.rs @@ -74,6 +74,13 @@ pub struct Config { /// An optional identifier for the shape of each value, typically a JSON Schema URL. pub schema: Option, + + /// Override the snapshot encoder's [`delta_ratio`](moq_json::snapshot::Config::delta_ratio), + /// or `None` for its default. Only a [`Snapshot`] reads it: a stream has no deltas. + /// + /// Not part of the catalog entry: deltas are a property of the frames, which every consumer + /// decodes the same way, so a reader needs nothing from the entry to follow them. + pub delta_ratio: Option, } impl Config { @@ -89,6 +96,12 @@ impl Config { self } + /// Set [`delta_ratio`](Self::delta_ratio) (a builder, since the struct is `#[non_exhaustive]`). + pub fn with_delta_ratio(mut self, delta_ratio: u32) -> Self { + self.delta_ratio = Some(delta_ratio); + self + } + /// The catalog entry describing a track published under this config in `mode`. pub(crate) fn entry(&self, mode: Mode) -> JsonConfig { let mut entry = JsonConfig::new(mode); @@ -117,6 +130,9 @@ impl Snapshot { if config.compression { json.compression = moq_json::Compression::Deflate; } + if let Some(delta_ratio) = config.delta_ratio { + json.delta_ratio = delta_ratio; + } let inner = moq_json::snapshot::Producer::new(track, json); rendition.set(config.entry(Mode::Snapshot))?; Ok(Self { inner, rendition }) From 6663b09b889ce0fc873fe8cd04bae7fd13021163 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 24 Sep 2026 23:17:35 -0700 Subject: [PATCH 2/2] test(libmoq): cover binary stream delivery with a NULL mime; note C data tracks in hang doc Co-Authored-By: Claude Opus 5.5 --- doc/concept/hang.md | 3 ++- rs/libmoq/src/test.rs | 48 ++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 49 insertions(+), 2 deletions(-) diff --git a/doc/concept/hang.md b/doc/concept/hang.md index a2907e3e0a..22699656f8 100644 --- a/doc/concept/hang.md +++ b/doc/concept/hang.md @@ -112,7 +112,8 @@ In Rust the catalog owns the lifetime: `catalog.json_stream(track, config)` (or `json_snapshot` / `binary_snapshot` / `binary_stream`) writes the entry and retracts it when the producer drops. Read the config from `catalog.json.tracks` or `catalog.binary.tracks`, then pair its name and config with -`moq_mux::catalog::Entry::new` to subscribe. In the browser, read the same map, +`moq_mux::catalog::Entry::new` to subscribe. In C, `moq_publish_json_*` and +`moq_publish_binary_*` do the same, retracting on `_finish`. In the browser, read the same map, subscribe by name, and hand the track to `@moq/json` or `@moq/binary`. ## Container diff --git a/rs/libmoq/src/test.rs b/rs/libmoq/src/test.rs index 75773dd08d..535d3574d8 100644 --- a/rs/libmoq/src/test.rs +++ b/rs/libmoq/src/test.rs @@ -4983,6 +4983,52 @@ fn binary_snapshot_is_advertised_and_delivered() { assert_eq!(moq_origin_close(origin), 0); } +#[test] +fn binary_stream_is_advertised_and_delivered() { + let origin = id(moq_origin_create()); + let path = b"binary-stream"; + let broadcast = publish_broadcast(origin, path); + + // A NULL mime leaves the media type unstated. + let name = b"blobs"; + let producer = id(unsafe { + moq_publish_binary_stream( + broadcast, + name.as_ptr() as *const c_char, + name.len(), + &moq_binary_config { + compression: false, + mime: std::ptr::null(), + mime_len: 0, + }, + ) + }); + + let catalog = published_catalog(broadcast); + let entry = catalog.binary.tracks.get("blobs").expect("binary stream advertised"); + assert_eq!(entry.mode, hang::catalog::Mode::Stream); + assert_eq!(entry.mime, None); + + // Every appended payload is delivered, in order. + let payloads: [&[u8]; 2] = [b"first", b"second"]; + for payload in payloads { + assert_eq!( + unsafe { moq_publish_binary_stream_append(producer, payload.as_ptr(), payload.len()) }, + 0 + ); + } + let consume = request_broadcast(origin, path); + let frames = read_raw_frames(consume, name, payloads.len()); + assert_eq!(frames, payloads.map(<[u8]>::to_vec)); + + assert_eq!(moq_publish_binary_stream_finish(producer), 0); + assert!(published_catalog(broadcast).binary.tracks.is_empty()); + + assert_eq!(moq_consume_close(consume), 0); + assert_eq!(moq_publish_finish(broadcast), 0); + assert_eq!(moq_origin_close(origin), 0); +} + #[test] fn data_track_names_cannot_collide() { let origin = id(moq_origin_create()); @@ -5001,7 +5047,7 @@ fn data_track_names_cannot_collide() { ) }); // A second data track under the same name is refused rather than silently replacing the - // first entry, and a NULL mime is allowed (left unstated). + // first entry. assert!( unsafe { moq_publish_binary_stream(