Update to hang 0.9.0 / moq-lite 0.10.1 / moq-native 0.10.1 (#27)

* Update to hang 0.9.0 / moq-lite 0.10.1 / moq-native 0.10.1

Update dependencies to latest versions and fix breaking API changes:

Sink (hangsink):
- hang::cmaf::Import → hang::import::Fmp4
- .parse() → .decode()
- Add BytesMut buffer to accumulate incoming data across GStreamer
  buffer boundaries, fixing "invalid size" panic when MP4 atoms span
  multiple buffers
- Add bytes crate dependency

Source (hangsrc):
- hang::Catalog::default_track() → hang::catalog::Catalog::default_track()
- TrackConsumer now requires explicit construction with max_latency
- .read() → .read_frame()
- frame.payload is now chunked, requires flattening to Vec<u8>
- Timestamp subtraction returns Timestamp, needs .into() for Duration

* Update to hang 0.9.0 / moq-lite 0.10.1 / moq-native 0.10.1

Update dependencies to latest versions and fix breaking API changes:

Sink (hangsink):
- hang::cmaf::Import → hang::import::Fmp4
- .parse() → .decode()
- Add BytesMut buffer to accumulate incoming data across GStreamer
  buffer boundaries, fixing "invalid size" panic when MP4 atoms span
  multiple buffers
- Add bytes crate dependency

Source (hangsrc):
- hang::Catalog::default_track() → hang::catalog::Catalog::default_track()
- TrackConsumer now requires explicit construction with max_latency
- .read() → .read_frame()
- frame.payload is now chunked, requires flattening to Vec<u8>
- Timestamp subtraction returns Timestamp, needs .into() for Duration

Author: davegullo

Cargo.lock

@@ -19,12 +19,13 @@ path = "src/lib.rs"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-# Using published versions from crates.io instead of workspace
-hang = "0.6"
-moq-lite = "0.8"
-moq-native = "0.8"
+# Using local path dependencies from moq-dev
+hang = "0.9.0"
+moq-lite = "0.10.1"
+moq-native = "0.10.1"
 
 anyhow = { version = "1", features = ["backtrace"] }
+bytes = "1"
 gst = { package = "gstreamer", version = "0.23" }
 gst-base = { package = "gstreamer-base", version = "0.23" }
 #gst-app = { package = "gstreamer-app", version = "0.23", features = ["v1_20"] }

Cargo.toml

@@ -1,4 +1,5 @@
 use anyhow::Context as _;
+use bytes::BytesMut;
 use gst::glib;
 use gst::prelude::*;
 use gst::subclass::prelude::*;
@@ -28,7 +29,8 @@ struct Settings {
 
 #[derive(Default)]
 struct State {
-	pub media: Option<hang::cmaf::Import>,
+	pub media: Option<hang::import::Fmp4>,
+	pub buffer: BytesMut,
 }
 
 #[derive(Default)]
@@ -140,12 +142,24 @@ impl BaseSinkImpl for HangSink {
 		let data = buffer.map_readable().map_err(|_| gst::FlowError::Error)?;
 
 		let mut state = self.state.lock().unwrap();
+
+		// Append incoming data to our buffer
+		state.buffer.extend_from_slice(data.as_slice());
+
+		// Take media out temporarily to avoid borrow conflict
 		let mut media = state.media.take().expect("not initialized");
 
-		// TODO avoid full media parsing? gst should be able to provide the necessary info
-		media.parse(data.as_slice()).expect("failed to parse");
+		// Try to decode what we have buffered
+		let result = media.decode(&mut state.buffer);
+
+		// Put media back
 		state.media = Some(media);
 
+		if let Err(e) = result {
+			gst::error!(gst::CAT_DEFAULT, "Failed to decode: {}", e);
+			return Err(gst::FlowError::Error);
+		}
+
 		Ok(gst::FlowSuccess::Ok)
 	}
 }
@@ -180,7 +194,7 @@ impl HangSink {
 				.await
 				.expect("failed to connect");
 
-			let media = hang::cmaf::Import::new(broadcast.producer);
+			let media = hang::import::Fmp4::new(broadcast.producer.into());
 
 			let mut state = self.state.lock().unwrap();
 			state.media = Some(media);

src/sink/imp.rs

@@ -185,7 +185,7 @@ impl HangSrc {
 			.consume_broadcast(&name)
 			.ok_or_else(|| anyhow::anyhow!("Broadcast '{}' not found", name))?;
 
-		let catalog = broadcast.subscribe_track(&hang::Catalog::default_track());
+		let catalog = broadcast.subscribe_track(&hang::catalog::Catalog::default_track());
 		let mut catalog = hang::catalog::CatalogConsumer::new(catalog);
 
 		// TODO handle catalog updates
@@ -194,7 +194,8 @@ impl HangSrc {
 		if let Some(video) = catalog.video {
 			for (track_name, config) in video.renditions {
 				let track_ref = hang::moq_lite::Track::new(&track_name);
-				let mut track: hang::TrackConsumer = broadcast.subscribe_track(&track_ref).into();
+				let mut track =
+					hang::TrackConsumer::new(broadcast.subscribe_track(&track_ref), std::time::Duration::from_secs(1));
 
 				let caps = match config.codec {
 					hang::catalog::VideoCodec::H264(_) => {
@@ -239,14 +240,15 @@ impl HangSrc {
 				let mut reference = None;
 				tokio::spawn(async move {
 					loop {
-						match track.read().await {
+						match track.read_frame().await {
 							Ok(Some(frame)) => {
-								let mut buffer = gst::Buffer::from_slice(frame.payload);
+								let payload: Vec<u8> = frame.payload.into_iter().flatten().collect();
+								let mut buffer = gst::Buffer::from_slice(payload);
 								let buffer_mut = buffer.get_mut().unwrap();
 
 								// Make timestamps relative to the first frame for proper playback
 								let pts = if let Some(reference_ts) = reference {
-									let timestamp: std::time::Duration = frame.timestamp - reference_ts;
+									let timestamp: std::time::Duration = (frame.timestamp - reference_ts).into();
 									gst::ClockTime::from_nseconds(timestamp.as_nanos() as _)
 								} else {
 									reference = Some(frame.timestamp);
@@ -287,7 +289,8 @@ impl HangSrc {
 		if let Some(audio) = catalog.audio {
 			for (track_name, config) in audio.renditions {
 				let track_ref = hang::moq_lite::Track::new(&track_name);
-				let mut track: hang::TrackConsumer = broadcast.subscribe_track(&track_ref).into();
+				let mut track =
+					hang::TrackConsumer::new(broadcast.subscribe_track(&track_ref), std::time::Duration::from_secs(1));
 
 				let caps = match &config.codec {
 					hang::catalog::AudioCodec::AAC(_aac) => {
@@ -346,14 +349,15 @@ impl HangSrc {
 				let mut reference = None;
 				tokio::spawn(async move {
 					loop {
-						match track.read().await {
+						match track.read_frame().await {
 							Ok(Some(frame)) => {
-								let mut buffer = gst::Buffer::from_slice(frame.payload);
+								let payload: Vec<u8> = frame.payload.into_iter().flatten().collect();
+								let mut buffer = gst::Buffer::from_slice(payload);
 								let buffer_mut = buffer.get_mut().unwrap();
 
 								// Make timestamps relative to the first frame for proper playback
 								let pts = if let Some(reference_ts) = reference {
-									let timestamp: std::time::Duration = frame.timestamp - reference_ts;
+									let timestamp: std::time::Duration = (frame.timestamp - reference_ts).into();
 									gst::ClockTime::from_nseconds(timestamp.as_nanos() as _)
 								} else {
 									reference = Some(frame.timestamp);