Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

46 changes: 46 additions & 0 deletions rs/moq-json/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -224,6 +224,20 @@ pub struct Consumer<T> {
_marker: PhantomData<fn() -> T>,
}

// Manual impl so cloning doesn't require `T: Clone`; `T` only lives in PhantomData.
// Cloned readers inherit the current reconstruction state, then advance in parallel.
impl<T> Clone for Consumer<T> {
fn clone(&self) -> Self {
Self {
track: self.track.clone(),
group: self.group.clone(),
current: self.current.clone(),
frames_read: self.frames_read,
_marker: PhantomData,
}
}
}

impl<T: DeserializeOwned> Consumer<T> {
/// Create a consumer reading from the given track subscriber.
pub fn new(track: moq_net::TrackConsumer) -> Self {
Expand Down Expand Up @@ -462,4 +476,36 @@ mod test {
}
assert_eq!(last.unwrap(), json!({ "a": 3 }));
}

#[test]
fn cloned_consumer_reconstructs_independently() {
// Deltas share one group, so a clone taken mid-group carries in-progress reconstruction state.
let config = Config {
delta_ratio: Some(100.0),
};
let (mut producer, track) = producer(config);
let mut consumer = Consumer::<Value>::new(track);
let waiter = kio::Waiter::noop();

producer.update(&json!({ "a": 1, "b": 1 })).unwrap(); // snapshot, group 0
match consumer.poll_next(&waiter) {
Poll::Ready(Ok(Some(value))) => assert_eq!(value, json!({ "a": 1, "b": 1 })),
other => panic!("expected snapshot, got {other:?}"),
}

// Clone after the snapshot: the copy inherits `current`/`frames_read` and an independent cursor.
let mut clone = consumer.clone();

producer.update(&json!({ "a": 1, "b": 2 })).unwrap(); // delta in group 0
producer.finish().unwrap();

// Each consumer applies the delta on top of its own reconstruction state.
let expected = json!({ "a": 1, "b": 2 });
for consumer in [&mut consumer, &mut clone] {
match consumer.poll_next(&waiter) {
Poll::Ready(Ok(Some(value))) => assert_eq!(value, expected),
other => panic!("expected delta, got {other:?}"),
}
}
}
}
2 changes: 1 addition & 1 deletion rs/moq-mux/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@ authors = ["Luke Curley <kixelated@gmail.com>"]
repository = "https://github.com/moq-dev/moq"
license = "MIT OR Apache-2.0"

version = "0.5.3"
version = "0.5.4"
edition = "2024"
rust-version.workspace = true

Expand Down
1 change: 1 addition & 0 deletions rs/moq-mux/src/catalog/hang/consumer.rs
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ use crate::Result;
///
/// This wraps a [`moq_json::Consumer`], reconstructing the JSON catalog from the latest
/// group's snapshot (plus any future deltas) to discover available audio and video tracks.
#[derive(Clone)]
pub struct Consumer {
inner: moq_json::Consumer<Catalog>,
}
Expand Down
18 changes: 10 additions & 8 deletions rs/moq-mux/src/import.rs
Original file line number Diff line number Diff line change
Expand Up @@ -28,10 +28,6 @@ pub enum FramedFormat {
Hev1,
/// AV1 with inline sequence headers
Av01,
/// VP8 (one frame per buffer; not self-delimiting).
Vp8,
/// VP9 (one frame per buffer; not self-delimiting).
Vp9,
/// Raw AAC frames (not ADTS).
Aac,
/// Raw Opus frames (not Ogg).
Expand All @@ -40,6 +36,12 @@ pub enum FramedFormat {
Mkv,
/// MPEG-TS (transport stream) container.
Ts,
// New variants go at the end: this enum has no repr, so inserting in the
// middle would shift the implicit discriminants of everything after it.
/// VP8 (one frame per buffer; not self-delimiting).
Vp8,
/// VP9 (one frame per buffer; not self-delimiting).
Vp9,
}

impl FromStr for FramedFormat {
Expand All @@ -52,12 +54,12 @@ impl FromStr for FramedFormat {
"hev1" => Ok(FramedFormat::Hev1),
"fmp4" | "cmaf" => Ok(FramedFormat::Fmp4),
"av01" | "av1" | "av1c" | "av1C" => Ok(FramedFormat::Av01),
"vp8" | "vp08" => Ok(FramedFormat::Vp8),
"vp9" | "vp09" => Ok(FramedFormat::Vp9),
"aac" => Ok(FramedFormat::Aac),
"opus" => Ok(FramedFormat::Opus),
"mkv" | "webm" | "matroska" => Ok(FramedFormat::Mkv),
"ts" | "mpegts" | "mpeg2ts" | "m2ts" => Ok(FramedFormat::Ts),
"vp8" | "vp08" => Ok(FramedFormat::Vp8),
"vp9" | "vp09" => Ok(FramedFormat::Vp9),
_ => Err(Error::UnknownFormat(s.to_string())),
}
}
Expand All @@ -71,12 +73,12 @@ impl fmt::Display for FramedFormat {
FramedFormat::Fmp4 => write!(f, "fmp4"),
FramedFormat::Hev1 => write!(f, "hev1"),
FramedFormat::Av01 => write!(f, "av01"),
FramedFormat::Vp8 => write!(f, "vp8"),
FramedFormat::Vp9 => write!(f, "vp9"),
FramedFormat::Aac => write!(f, "aac"),
FramedFormat::Opus => write!(f, "opus"),
FramedFormat::Mkv => write!(f, "mkv"),
FramedFormat::Ts => write!(f, "ts"),
FramedFormat::Vp8 => write!(f, "vp8"),
FramedFormat::Vp9 => write!(f, "vp9"),
}
}
}
Expand Down
Loading