Karp API improvements (#220)

moq-karp/src/catalog/broadcast.rs

@@ -130,7 +130,6 @@ mod test {
 					width: 1280,
 					height: 720,
 				},
-				layers: Default::default(),
 				bitrate: Some(6_000_000),
 			}],
 			audio: vec![Audio {

moq-karp/src/catalog/video.rs

@@ -27,20 +27,4 @@ pub struct Video {
 
 	#[serde(default)]
 	pub bitrate: Option<u32>,
-
-	// Additional enhancement layers
-	#[serde(default, skip_serializing_if = "Vec::is_empty")]
-	pub layers: Vec<VideoLayer>,
-}
-
-#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
-pub struct VideoLayer {
-	/// The name of the track, relative to the broadcast path.
-	pub name: String,
-
-	/// The priority of the track, relative to other tracks in the same broadcast.
-	pub priority: i8,
-
-	/// The resolution of this layer.
-	pub resolution: Dimensions,
 }

moq-karp/src/cmaf/import.rs

@@ -1,10 +1,14 @@
-use bytes::BytesMut;
+use bytes::{Bytes, BytesMut};
 use mp4_atom::{Any, AsyncReadFrom, Atom, DecodeMaybe, Esds, Moof, Moov, Trak};
 use std::collections::HashMap;
 use tokio::io::AsyncRead;
 
 use super::{util, Error, Result};
-use crate::{catalog, media::Timestamp, produce};
+use crate::{
+	catalog,
+	media::{self, Timestamp},
+	produce,
+};
 
 /// Converts fMP4 -> Karp
 pub struct Import {
@@ -15,8 +19,7 @@ pub struct Import {
 	broadcast: produce::Broadcast,
 
 	// A lookup to tracks in the broadcast
-	audio: HashMap<u32, produce::Audio>,
-	video: HashMap<u32, produce::Video>,
+	tracks: HashMap<u32, produce::Track>,
 
 	// The moov atom at the start of the file.
 	moov: Option<Moov>,
@@ -30,8 +33,7 @@ impl Import {
 		Self {
 			buffer: BytesMut::new(),
 			broadcast,
-			audio: HashMap::default(),
-			video: HashMap::default(),
+			tracks: HashMap::default(),
 			moov: None,
 			moof: None,
 		}
@@ -68,20 +70,20 @@ impl Import {
 			let track_id = trak.tkhd.track_id;
 			let handler = &trak.mdia.hdlr.handler;
 
-			match handler.as_ref() {
+			let track = match handler.as_ref() {
 				b"vide" => {
 					let track = Self::init_video(trak)?;
-					let video = self.broadcast.create_video(track)?;
-					self.video.insert(track_id, video);
+					self.broadcast.create_video(track)?
 				}
 				b"soun" => {
 					let track = Self::init_audio(trak)?;
-					let audio = self.broadcast.create_audio(track)?;
-					self.audio.insert(track_id, audio);
+					self.broadcast.create_audio(track)?
 				}
 				b"sbtl" => return Err(Error::UnsupportedTrack("subtitle")),
 				_ => return Err(Error::UnsupportedTrack("unknown")),
 			};
+
+			self.tracks.insert(track_id, track);
 		}
 
 		self.moov = Some(moov);
@@ -112,7 +114,6 @@ impl Import {
 				}
 				.into(),
 				description: description.freeze(),
-				layers: vec![],
 				bitrate: None,
 			}
 		} else if let Some(hev1) = &stsd.hev1 {
@@ -159,7 +160,6 @@ impl Import {
 					width: vp09.width,
 					height: vp09.height,
 				},
-				layers: vec![],
 				bitrate: None,
 			}
 		} else {
@@ -230,30 +230,6 @@ impl Import {
 				self.init(moov)?;
 			}
 			Any::Moof(moof) => {
-				let track_id = util::frame_track_id(&moof)?;
-				let keyframe = util::frame_is_key(&moof);
-
-				if keyframe {
-					let moov = self.moov.as_ref().ok_or(Error::MissingBox(Moov::KIND))?;
-					let trak = moov
-						.trak
-						.iter()
-						.find(|trak| trak.tkhd.track_id == track_id)
-						.ok_or(Error::UnknownTrack)?;
-
-					// If this is a video track, start a new group for the keyframe.
-					if trak.mdia.hdlr.handler == b"vide".into() {
-						// Start a new group for the keyframe.
-						for track in self.video.values_mut() {
-							track.keyframe();
-						}
-
-						for track in self.audio.values_mut() {
-							track.segment();
-						}
-					}
-				}
-
 				if self.moof.is_some() {
 					// Two moof boxes in a row.
 					return Err(Error::DuplicateBox(Moof::KIND));
@@ -269,15 +245,16 @@ impl Import {
 				let timestamp = util::frame_timestamp(&moof)?;
 				let timescale = util::frame_timescale(moov, &moof)?;
 
-				let timestamp = Timestamp::from_scale(timestamp, timescale as _);
+				let timestamp = Timestamp::from_units(timestamp, timescale as _);
 
-				if let Some(video) = self.video.get_mut(&track_id) {
-					video.write(timestamp, mdat.data.into());
-				} else if let Some(audio) = self.audio.get_mut(&track_id) {
-					audio.write(timestamp, mdat.data.into());
-				} else {
-					return Err(Error::UnknownTrack);
-				}
+				let frame = media::Frame {
+					timestamp,
+					keyframe: util::frame_is_key(&moof),
+					payload: Bytes::from(mdat.data),
+				};
+
+				let track = self.tracks.get_mut(&track_id).ok_or(Error::UnknownTrack)?;
+				track.write(frame);
 			}
 
 			_ => {

moq-karp/src/consume/broadcast.rs

@@ -3,7 +3,7 @@ use crate::{catalog, Error};
 use moq_transfork::{Path, Session};
 use tokio::sync::mpsc;
 
-use super::{Audio, Video};
+use super::Track;
 
 #[derive(Clone)]
 pub struct Broadcast {
@@ -41,7 +41,7 @@ impl Broadcast {
 	}
 
 	// This API could be improved
-	pub fn video(&self, name: &str) -> Result<Video, Error> {
+	pub fn video(&self, name: &str) -> Result<Track, Error> {
 		let info = self.find_video(name)?;
 
 		let track = moq_transfork::Track {
@@ -51,10 +51,10 @@ impl Broadcast {
 		};
 		let track = self.session.subscribe(track);
 
-		Ok(Video::new(track))
+		Ok(Track::new(track))
 	}
 
-	pub fn audio(&self, name: &str) -> Result<Audio, Error> {
+	pub fn audio(&self, name: &str) -> Result<Track, Error> {
 		let info = self.find_audio(name)?;
 
 		let track = moq_transfork::Track {
@@ -64,7 +64,7 @@ impl Broadcast {
 		};
 		let track = self.session.subscribe(track);
 
-		Ok(Audio::new(track))
+		Ok(Track::new(track))
 	}
 
 	fn find_audio(&self, name: &str) -> Result<&catalog::Audio, Error> {

moq-karp/src/consume/mod.rs

@@ -1,9 +1,7 @@
-mod audio;
 mod broadcast;
 mod resume;
-mod video;
+mod track;
 
-pub use audio::*;
 pub use broadcast::*;
 pub use resume::*;
-pub use video::*;
+pub use track::*;

moq-karp/src/consume/audio.rs → moq-karp/src/consume/track.rs

@@ -1,16 +1,16 @@
+use crate::media::Timestamp;
 use crate::Error;
 
-use crate::media::Timestamp;
 use crate::{media::Frame, util::FuturesExt};
 
 use moq_transfork::coding::*;
 
-pub struct Audio {
+pub struct Track {
 	track: moq_transfork::TrackConsumer,
 	group: Option<moq_transfork::GroupConsumer>,
 }
 
-impl Audio {
+impl Track {
 	pub(super) fn new(track: moq_transfork::TrackConsumer) -> Self {
 		Self { track, group: None }
 	}

moq-karp/src/lib.rs

@@ -8,3 +8,6 @@ mod error;
 pub use error::*;
 
 pub(crate) mod util;
+
+// export the moq-transfork version in use
+pub use moq_transfork;

moq-karp/src/main.rs

@@ -6,7 +6,6 @@ use url::Url;
 
 use moq_karp::{cmaf, produce};
 use moq_native::quic;
-use moq_transfork::*;
 
 #[derive(Parser, Clone)]
 struct Cli {
@@ -53,7 +52,7 @@ async fn main() -> anyhow::Result<()> {
 	let session = quic.client.connect(&cli.url).await?;
 	let session = moq_transfork::Session::connect(session).await?;
 
-	let path = Path::new(cli.path);
+	let path = moq_transfork::Path::new(cli.path);
 
 	match cli.command {
 		Command::Publish => publish(session, path).await,
@@ -62,7 +61,7 @@ async fn main() -> anyhow::Result<()> {
 }
 
 #[tracing::instrument("publish", skip_all, err, fields(?path))]
-async fn publish(mut session: moq_transfork::Session, path: Path) -> anyhow::Result<()> {
+async fn publish(mut session: moq_transfork::Session, path: moq_transfork::Path) -> anyhow::Result<()> {
 	let broadcast = produce::Resumable::new(path).broadcast();
 
 	let mut input = tokio::io::stdin();

moq-karp/src/media/timestamp.rs

@@ -1,6 +1,6 @@
-#[derive(Debug, Clone, Copy)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
 pub struct Timestamp {
-	pub micros: u64,
+	micros: u64,
 }
 
 impl Timestamp {
@@ -18,9 +18,9 @@ impl Timestamp {
 		}
 	}
 
-	pub fn from_scale(base: u64, scale: u64) -> Self {
+	pub fn from_units(value: u64, base: u64) -> Self {
 		Self {
-			micros: base * 1_000_000 / scale,
+			micros: (value * 1_000_000) / base,
 		}
 	}
 
@@ -36,7 +36,7 @@ impl Timestamp {
 		self.micros / 1_000_000
 	}
 
-	pub fn to_scale(&self, scale: u64) -> u64 {
-		self.micros * scale / 1_000_000
+	pub fn as_units(&self, base: u64) -> u64 {
+		(self.micros * base) / 1_000_000
 	}
 }