use std::marker::PhantomData;
use std::time::Duration;
use super::Estimate;
use super::Producer;
use super::hang::{Catalog, CatalogExt};
pub trait RenditionConfig<E: CatalogExt>: Clone + Send + 'static {
fn detects() -> bool {
false
}
fn insert(self, catalog: &mut Catalog<E>, name: &str);
fn get_mut<'a>(catalog: &'a mut Catalog<E>, name: &str) -> Option<&'a mut Self>;
fn remove(catalog: &mut Catalog<E>, name: &str);
fn estimate(&self) -> Estimate {
Estimate::default()
}
fn set_estimate(&mut self, _estimate: Estimate) {}
}
impl<E: CatalogExt> RenditionConfig<E> for hang::catalog::JsonConfig {
fn insert(self, catalog: &mut Catalog<E>, name: &str) {
catalog.json.tracks.insert(name.to_string(), self);
}
fn get_mut<'a>(catalog: &'a mut Catalog<E>, name: &str) -> Option<&'a mut Self> {
catalog.json.tracks.get_mut(name)
}
fn remove(catalog: &mut Catalog<E>, name: &str) {
catalog.json.tracks.remove(name);
}
fn detects() -> bool {
true
}
fn estimate(&self) -> Estimate {
Estimate::default()
.with_jitter(self.jitter)
.with_bitrate(self.bitrate)
.with_delay(self.delay)
}
fn set_estimate(&mut self, estimate: Estimate) {
self.jitter = estimate.jitter;
self.bitrate = estimate.bitrate;
self.delay = estimate.delay;
}
}
impl<E: CatalogExt> RenditionConfig<E> for hang::catalog::BinaryConfig {
fn insert(self, catalog: &mut Catalog<E>, name: &str) {
catalog.binary.tracks.insert(name.to_string(), self);
}
fn get_mut<'a>(catalog: &'a mut Catalog<E>, name: &str) -> Option<&'a mut Self> {
catalog.binary.tracks.get_mut(name)
}
fn remove(catalog: &mut Catalog<E>, name: &str) {
catalog.binary.tracks.remove(name);
}
fn detects() -> bool {
true
}
fn estimate(&self) -> Estimate {
Estimate::default()
.with_jitter(self.jitter)
.with_bitrate(self.bitrate)
.with_delay(self.delay)
}
fn set_estimate(&mut self, estimate: Estimate) {
self.jitter = estimate.jitter;
self.bitrate = estimate.bitrate;
self.delay = estimate.delay;
}
}
#[derive(Clone, Default, Debug, PartialEq)]
#[non_exhaustive]
pub struct VideoHint {
pub(crate) label: Option<String>,
pub codec: Option<hang::catalog::VideoCodec>,
pub coded_width: Option<u32>,
pub coded_height: Option<u32>,
pub display_aspect_width: Option<u32>,
pub display_aspect_height: Option<u32>,
pub bitrate: Option<u64>,
pub framerate: Option<f64>,
pub optimize_for_latency: Option<bool>,
pub jitter: Option<Duration>,
pub delay: Option<Duration>,
pub container: hang::catalog::Container,
}
fn fill<T>(slot: &mut Option<T>, value: Option<T>) {
if slot.is_none() {
*slot = value;
}
}
impl From<hang::catalog::VideoConfig> for VideoHint {
fn from(config: hang::catalog::VideoConfig) -> Self {
Self {
label: config.label,
codec: Some(config.codec),
coded_width: config.coded_width,
coded_height: config.coded_height,
display_aspect_width: config.display_aspect_width,
display_aspect_height: config.display_aspect_height,
bitrate: config.bitrate,
framerate: config.framerate,
optimize_for_latency: config.optimize_for_latency,
jitter: config.jitter,
delay: config.delay,
container: config.container,
}
}
}
impl VideoHint {
pub fn apply(&self, config: &mut hang::catalog::VideoConfig) {
fill(&mut config.label, self.label.clone());
fill(&mut config.coded_width, self.coded_width);
fill(&mut config.coded_height, self.coded_height);
fill(&mut config.display_aspect_width, self.display_aspect_width);
fill(&mut config.display_aspect_height, self.display_aspect_height);
fill(&mut config.bitrate, self.bitrate);
fill(&mut config.framerate, self.framerate);
fill(&mut config.optimize_for_latency, self.optimize_for_latency);
fill(&mut config.jitter, self.jitter);
fill(&mut config.delay, self.delay);
config.container = self.container.clone();
}
pub fn to_config(&self) -> Option<hang::catalog::VideoConfig> {
let codec = self.codec.clone()?;
let mut config = hang::catalog::VideoConfig::new(codec);
self.apply(&mut config);
Some(config)
}
}
impl<E: CatalogExt> RenditionConfig<E> for hang::catalog::VideoConfig {
fn detects() -> bool {
true
}
fn insert(self, catalog: &mut Catalog<E>, name: &str) {
catalog.video.renditions.insert(name.to_string(), self);
}
fn get_mut<'a>(catalog: &'a mut Catalog<E>, name: &str) -> Option<&'a mut Self> {
catalog.video.renditions.get_mut(name)
}
fn remove(catalog: &mut Catalog<E>, name: &str) {
catalog.video.renditions.remove(name);
}
fn estimate(&self) -> Estimate {
Estimate::default()
.with_jitter(self.jitter)
.with_bitrate(self.bitrate)
.with_delay(self.delay)
.with_framerate(self.framerate)
}
fn set_estimate(&mut self, estimate: Estimate) {
self.jitter = estimate.jitter;
self.bitrate = estimate.bitrate;
self.delay = estimate.delay;
self.framerate = estimate.framerate;
}
}
impl<E: CatalogExt> RenditionConfig<E> for hang::catalog::AudioConfig {
fn detects() -> bool {
true
}
fn insert(self, catalog: &mut Catalog<E>, name: &str) {
catalog.audio.renditions.insert(name.to_string(), self);
}
fn get_mut<'a>(catalog: &'a mut Catalog<E>, name: &str) -> Option<&'a mut Self> {
catalog.audio.renditions.get_mut(name)
}
fn remove(catalog: &mut Catalog<E>, name: &str) {
catalog.audio.renditions.remove(name);
}
fn estimate(&self) -> Estimate {
Estimate::default()
.with_jitter(self.jitter)
.with_bitrate(self.bitrate)
.with_delay(self.delay)
}
fn set_estimate(&mut self, estimate: Estimate) {
self.jitter = estimate.jitter;
self.bitrate = estimate.bitrate;
self.delay = estimate.delay;
}
}
impl<E: CatalogExt> RenditionConfig<E> for hang::catalog::TextConfig {
fn insert(self, catalog: &mut Catalog<E>, name: &str) {
catalog.text.renditions.insert(name.to_string(), self);
}
fn get_mut<'a>(catalog: &'a mut Catalog<E>, name: &str) -> Option<&'a mut Self> {
catalog.text.renditions.get_mut(name)
}
fn remove(catalog: &mut Catalog<E>, name: &str) {
catalog.text.renditions.remove(name);
}
fn estimate(&self) -> Estimate {
Estimate::default().with_jitter(self.jitter).with_delay(self.delay)
}
fn set_estimate(&mut self, estimate: Estimate) {
self.jitter = estimate.jitter;
self.delay = estimate.delay;
}
}
pub struct Reserved<E: CatalogExt = ()> {
catalog: Producer<E>,
}
impl<E: CatalogExt> Reserved<E> {
pub(super) fn new(catalog: Producer<E>) -> Self {
catalog.add_reserver();
Self { catalog }
}
pub fn track_info(&self, priority: u8) -> moq_net::track::Info {
self.catalog.track_info(priority)
}
pub(crate) fn init<C: RenditionConfig<E>>(&self, name: impl Into<String>) -> crate::Result<Rendition<E, C>> {
Rendition::new(self.clone(), name.into())
}
pub fn video<C: crate::container::Container>(
&self,
track: moq_net::track::Producer,
container: C,
config: impl Into<Option<hang::catalog::VideoConfig>>,
) -> crate::Result<crate::container::Producer<C, hang::catalog::VideoConfig>>
where
crate::Error: From<C::Error>,
{
self.track(track, container, config)
}
pub fn audio<C: crate::container::Container>(
&self,
track: moq_net::track::Producer,
container: C,
config: impl Into<Option<hang::catalog::AudioConfig>>,
) -> crate::Result<crate::container::Producer<C, hang::catalog::AudioConfig>>
where
crate::Error: From<C::Error>,
{
self.track(track, container, config)
}
pub fn text<C: crate::container::Container>(
&self,
track: moq_net::track::Producer,
container: C,
config: impl Into<Option<hang::catalog::TextConfig>>,
) -> crate::Result<crate::container::Producer<C, hang::catalog::TextConfig>>
where
crate::Error: From<C::Error>,
{
self.track(track, container, config)
}
pub fn track<C, R>(
&self,
track: moq_net::track::Producer,
container: C,
config: impl Into<Option<R>>,
) -> crate::Result<crate::container::Producer<C, R>>
where
C: crate::container::Container,
R: RenditionConfig<E>,
crate::Error: From<C::Error>,
{
let rendition = self.init(track.name())?;
self.catalog.media(track, container, rendition, config.into())
}
pub fn timestamp(&self, hint: Option<moq_net::Timestamp>) -> crate::Result<moq_net::Timestamp> {
self.catalog.timestamp(hint)
}
pub(crate) fn producer(&self) -> Producer<E> {
self.catalog.clone()
}
}
impl<E: CatalogExt> Clone for Reserved<E> {
fn clone(&self) -> Self {
self.catalog.add_reserver();
Self {
catalog: self.catalog.clone(),
}
}
}
impl<E: CatalogExt> Drop for Reserved<E> {
fn drop(&mut self) {
self.catalog.release_reserver();
}
}
pub(crate) struct Rendition<E: CatalogExt, C: RenditionConfig<E>> {
catalog: Producer<E>,
name: String,
gate: Option<Reserved<E>>,
present: bool,
supplied: Estimate,
detected: Estimate,
published: Option<Estimate>,
_config: PhantomData<fn() -> C>,
}
pub(crate) type VideoTrack<E = ()> = Rendition<E, hang::catalog::VideoConfig>;
pub(crate) type AudioTrack<E = ()> = Rendition<E, hang::catalog::AudioConfig>;
impl<E: CatalogExt, C: RenditionConfig<E>> Rendition<E, C> {
fn new(reserved: Reserved<E>, name: String) -> crate::Result<Self> {
Self::owned(reserved.catalog.clone(), Some(reserved), name)
}
pub(super) fn live(catalog: Producer<E>, name: String) -> crate::Result<Self> {
Self::owned(catalog, None, name)
}
fn owned(catalog: Producer<E>, gate: Option<Reserved<E>>, name: String) -> crate::Result<Self> {
catalog.acquire::<C>(&name)?;
Ok(Self {
catalog,
gate,
name,
present: false,
supplied: Estimate::default(),
detected: Estimate::default(),
published: None,
_config: PhantomData,
})
}
pub fn name(&self) -> &str {
&self.name
}
pub(crate) fn estimator(&self) -> super::Estimator {
self.catalog.estimator()
}
pub(crate) fn clock(&self) -> crate::Clock {
self.catalog.clock()
}
pub fn timestamp(&self, hint: Option<moq_net::Timestamp>) -> crate::Result<moq_net::Timestamp> {
self.catalog.timestamp(hint)
}
pub(crate) fn set(&mut self, mut config: C) -> crate::Result<()> {
let supplied = config.estimate();
let resolved = Self::resolved(&supplied, &self.detected);
self.check_decrease(&resolved)?;
config.set_estimate(resolved.clone());
{
let mut guard = self.catalog.modify()?;
let mut next = (*guard).clone();
config.insert(&mut next, &self.name);
serde_json::to_writer(std::io::sink(), &next).map_err(moq_json::Error::from)?;
self.present = true;
*guard = next;
guard.commit()?;
}
self.supplied = supplied;
self.published = Some(resolved);
self.gate = None;
Ok(())
}
fn resolved(supplied: &Estimate, detected: &Estimate) -> Estimate {
let mut estimate = supplied.clone();
if estimate.jitter.is_none() {
estimate.jitter = detected.jitter;
}
if estimate.bitrate.is_none() {
estimate.bitrate = detected.bitrate;
}
if estimate.delay.is_none() {
estimate.delay = detected.delay;
}
if estimate.framerate.is_none() {
estimate.framerate = detected.framerate;
}
estimate
}
pub(crate) fn estimate(&mut self, estimate: Estimate) -> crate::Result<()> {
if !C::detects() {
return Ok(());
}
if !self.present {
self.detected = estimate;
return Ok(());
}
let mut resolved = Self::resolved(&self.supplied, &estimate);
let published = self.config()?.estimate();
resolved.jitter = resolved.jitter.max(published.jitter);
resolved.delay = resolved.delay.max(published.delay);
self.detected = estimate;
if self.published.as_ref() != Some(&resolved) {
let mut config = self.config()?;
config.set_estimate(resolved.clone());
self.replace(config)?;
self.published = Some(resolved);
}
Ok(())
}
fn check_decrease(&self, next: &Estimate) -> crate::Result<()> {
if !self.present {
return Ok(());
}
let previous = self.config()?.estimate();
if next.jitter < previous.jitter {
return Err(crate::Error::JitterDecreased);
}
if next.delay < previous.delay {
return Err(crate::Error::DelayDecreased);
}
Ok(())
}
pub(crate) fn config(&self) -> crate::Result<C> {
if !self.present {
return Err(crate::Error::NotPublished);
}
let mut catalog = self.catalog.snapshot();
C::get_mut(&mut catalog, &self.name)
.cloned()
.ok_or(crate::Error::NotPublished)
}
pub(crate) fn replace(&mut self, config: C) -> crate::Result<()> {
if !self.present {
return Err(crate::Error::NotPublished);
}
self.check_decrease(&config.estimate())?;
let mut guard = self.catalog.modify()?;
let mut next = (*guard).clone();
config.insert(&mut next, &self.name);
serde_json::to_writer(std::io::sink(), &next).map_err(moq_json::Error::from)?;
*guard = next;
guard.commit()
}
}
impl<E: CatalogExt, C: RenditionConfig<E>> std::fmt::Debug for Rendition<E, C> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Rendition")
.field("name", &self.name)
.field("published", &self.present)
.finish_non_exhaustive()
}
}
impl<E: CatalogExt, C: RenditionConfig<E>> Drop for Rendition<E, C> {
fn drop(&mut self) {
if let Ok(mut catalog) = self.catalog.modify() {
catalog.release::<C>(&self.name, self.present);
}
}
}
#[cfg(test)]
mod tests {
use super::*;
fn video_track() -> (moq_net::broadcast::Producer, super::super::Producer, VideoTrack) {
let mut broadcast = moq_net::broadcast::Info::new().produce();
let catalog = super::super::Producer::new(&mut broadcast, super::super::Config::default()).unwrap();
let reserved = catalog.reserve();
let rendition = reserved.init::<hang::catalog::VideoConfig>("v").unwrap();
drop(reserved);
(broadcast, catalog, rendition)
}
fn config(bitrate: Option<u64>, jitter: Option<Duration>) -> hang::catalog::VideoConfig {
let mut config = hang::catalog::VideoConfig::new(hang::catalog::VideoCodec::VP8);
config.bitrate = bitrate;
config.jitter = jitter;
config
}
#[test]
fn rejected_jitter_is_not_retained() {
let (_broadcast, catalog, mut rendition) = video_track();
for jitter in [Duration::ZERO, Duration::MAX] {
assert!(rendition.set(config(None, Some(jitter))).is_err());
assert!(catalog.snapshot().video.renditions.is_empty());
}
rendition.set(config(None, Some(Duration::from_millis(100)))).unwrap();
let mut invalid = rendition.config().unwrap();
invalid.jitter = Some(Duration::ZERO);
assert!(rendition.replace(invalid).is_err());
assert_eq!(
catalog.snapshot().video.renditions.values().next().unwrap().jitter,
Some(Duration::from_millis(100))
);
}
#[test]
fn published_jitter_never_decreases() {
let (_broadcast, catalog, mut rendition) = video_track();
rendition.set(config(None, Some(Duration::from_millis(100)))).unwrap();
let smaller = config(None, Some(Duration::from_millis(50)));
assert!(matches!(
rendition.set(smaller.clone()),
Err(crate::Error::JitterDecreased)
));
assert!(matches!(rendition.replace(smaller), Err(crate::Error::JitterDecreased)));
assert_eq!(
catalog.snapshot().video.renditions["v"].jitter,
Some(Duration::from_millis(100))
);
let (_broadcast, catalog, mut detected) = video_track();
detected.set(config(None, None)).unwrap();
detected
.estimate(Estimate::default().with_jitter(Duration::from_millis(100)))
.unwrap();
detected
.estimate(Estimate::default().with_jitter(Duration::from_millis(50)))
.unwrap();
assert_eq!(
catalog.snapshot().video.renditions["v"].jitter,
Some(Duration::from_millis(100))
);
}
#[test]
fn published_delay_never_decreases() {
let (_broadcast, catalog, mut rendition) = video_track();
let delayed = |delay| {
let mut config = config(None, None);
config.delay = delay;
config
};
rendition.set(delayed(Some(Duration::from_millis(200)))).unwrap();
for smaller in [Some(Duration::from_millis(100)), None] {
assert!(matches!(
rendition.set(delayed(smaller)),
Err(crate::Error::DelayDecreased)
));
assert!(matches!(
rendition.replace(delayed(smaller)),
Err(crate::Error::DelayDecreased)
));
}
assert_eq!(
catalog.snapshot().video.renditions["v"].delay,
Some(Duration::from_millis(200))
);
let (_broadcast, catalog, mut detected) = video_track();
detected.set(config(None, None)).unwrap();
detected
.estimate(Estimate::default().with_delay(Duration::from_millis(200)))
.unwrap();
detected
.estimate(Estimate::default().with_delay(Duration::from_millis(100)))
.unwrap();
assert_eq!(
catalog.snapshot().video.renditions["v"].delay,
Some(Duration::from_millis(200))
);
}
#[test]
fn measurement_keeps_a_jitter_raised_by_modify() {
let (_broadcast, catalog, mut rendition) = video_track();
rendition.set(config(None, None)).unwrap();
rendition
.estimate(Estimate::default().with_jitter(Duration::from_millis(60)))
.unwrap();
let mut raised = rendition.config().unwrap();
raised.jitter = Some(Duration::from_millis(200));
rendition.replace(raised).unwrap();
rendition
.estimate(Estimate::default().with_jitter(Duration::from_millis(80)))
.unwrap();
assert_eq!(
catalog.snapshot().video.renditions["v"].jitter,
Some(Duration::from_millis(200))
);
rendition
.estimate(Estimate::default().with_jitter(Duration::from_millis(300)))
.unwrap();
assert_eq!(
catalog.snapshot().video.renditions["v"].jitter,
Some(Duration::from_millis(300))
);
}
#[test]
fn published_text_jitter_never_decreases() {
let mut broadcast = moq_net::broadcast::Info::new().produce();
let catalog = super::super::Producer::new(&mut broadcast, super::super::Config::default()).unwrap();
let reserved = catalog.reserve();
let mut rendition = reserved.init::<hang::catalog::TextConfig>("t").unwrap();
drop(reserved);
let text = |jitter| {
let mut config = hang::catalog::TextConfig::new(hang::catalog::TextFormat::Utf8);
config.jitter = jitter;
config
};
rendition.set(text(Some(Duration::from_millis(100)))).unwrap();
for smaller in [Some(Duration::from_millis(50)), None] {
assert!(matches!(
rendition.set(text(smaller)),
Err(crate::Error::JitterDecreased)
));
}
assert_eq!(
catalog.snapshot().text.renditions["t"].jitter,
Some(Duration::from_millis(100))
);
}
#[test]
fn importer_returns_rejected_jitter() {
let mut broadcast = moq_net::broadcast::Info::new().produce();
let catalog = super::super::Producer::new(&mut broadcast, super::super::Config::default()).unwrap();
let reserved = catalog.reserve();
let track = broadcast
.create_track("audio", hang::container::track_info(hang::catalog::PRIORITY.audio))
.unwrap();
let mut config: hang::catalog::AudioConfig = crate::codec::aac::Config {
profile: 2,
sample_rate: 48_000,
channel_count: 2,
}
.into();
config.jitter = Some(Duration::ZERO);
assert!(crate::codec::aac::Import::new(track, reserved, config).is_err());
assert!(catalog.snapshot().audio.renditions.is_empty());
}
fn ts(micros: u64) -> moq_net::Timestamp {
moq_net::Timestamp::from_micros(micros).unwrap()
}
#[test]
fn a_video_hint_applies_its_container_to_every_config() {
let mut config = hang::catalog::VideoConfig::new(hang::catalog::VideoCodec::VP8);
VideoHint::default().apply(&mut config);
assert_eq!(config.container, hang::catalog::Container::Legacy);
let hint = VideoHint {
container: hang::catalog::Container::Loc,
..Default::default()
};
hint.apply(&mut config);
assert_eq!(config.container, hang::catalog::Container::Loc);
}
fn feed<E: CatalogExt, C: RenditionConfig<E>>(rendition: &mut Rendition<E, C>) {
let mut estimator = super::super::Estimator::new();
estimator.burst(Duration::from_millis(40));
for i in 0..60u64 {
let t = ts(i * 40_000);
estimator.cut(Some(t));
estimator.write(t, 100_000);
rendition.estimate(estimator.estimate()).unwrap();
}
}
#[test]
fn detects_absent_jitter_and_bitrate() {
let (_broadcast, catalog, mut rendition) = video_track();
rendition.set(config(None, None)).unwrap();
feed(&mut rendition);
let snapshot = catalog.snapshot();
let config = snapshot.video.renditions.get("v").unwrap();
assert!(config.jitter.is_some(), "absent jitter should be auto-detected");
assert!(config.bitrate.is_some(), "absent bitrate should be auto-detected");
assert_eq!(config.framerate, Some(25.0), "absent framerate should be auto-detected");
}
#[test]
fn keeps_provided_jitter_and_bitrate() {
let (_broadcast, catalog, mut rendition) = video_track();
rendition
.set(config(Some(123), Some(Duration::from_millis(50))))
.unwrap();
feed(&mut rendition);
let snapshot = catalog.snapshot();
let config = snapshot.video.renditions.get("v").unwrap();
assert_eq!(config.bitrate, Some(123), "a provided bitrate must not be overwritten");
assert_eq!(
config.jitter,
Some(Duration::from_millis(50)),
"a provided jitter must not be overwritten"
);
}
#[test]
fn a_config_converts_into_hints_without_losing_the_label() {
let (_broadcast, catalog, mut rendition) = video_track();
let mut source = config(Some(456), None);
source.label = Some("Main camera".to_string());
source.coded_width = Some(1920);
source.coded_height = Some(1080);
let hint = VideoHint::from(source);
let mut first = config(None, None);
hint.apply(&mut first);
rendition.set(first).unwrap();
feed(&mut rendition);
assert_eq!(
catalog.snapshot().video.renditions.get("v").unwrap().label.as_deref(),
Some("Main camera"),
"the label must survive the config the importer publishes up front"
);
let mut resolved = config(None, None);
resolved.coded_width = Some(1280);
resolved.coded_height = Some(720);
hint.apply(&mut resolved);
rendition.set(resolved).unwrap();
feed(&mut rendition);
let snapshot = catalog.snapshot();
let published = snapshot.video.renditions.get("v").unwrap();
assert_eq!(
published.label.as_deref(),
Some("Main camera"),
"the label must survive a config the stream resolved later"
);
assert_eq!(published.bitrate, Some(456), "the rest of the config converts too");
}
#[test]
fn hinted_fields_count_as_supplied() {
let (_broadcast, catalog, mut rendition) = video_track();
let hint = VideoHint {
bitrate: Some(456),
framerate: Some(30.0),
..Default::default()
};
let mut config = config(None, None);
hint.apply(&mut config);
rendition.set(config).unwrap();
feed(&mut rendition);
let snapshot = catalog.snapshot();
let config = snapshot.video.renditions.get("v").unwrap();
assert_eq!(config.bitrate, Some(456), "a hinted bitrate must not be overwritten");
assert_eq!(
config.framerate,
Some(30.0),
"a hinted framerate must not be overwritten"
);
assert!(config.jitter.is_some(), "the unhinted jitter should still be detected");
}
#[test]
fn measurements_before_set_seed_the_config() {
let (_broadcast, catalog, mut rendition) = video_track();
feed(&mut rendition);
rendition.set(config(None, None)).unwrap();
let snapshot = catalog.snapshot();
let config = snapshot.video.renditions.get("v").unwrap();
assert!(
config.jitter.is_some(),
"the jitter measured before `set` should carry over"
);
assert!(
config.bitrate.is_some(),
"the bitrate measured before `set` should carry over"
);
}
#[test]
fn resetting_recaptures_supplied() {
let (_broadcast, catalog, mut rendition) = video_track();
rendition.set(config(None, None)).unwrap();
feed(&mut rendition);
assert!(catalog.snapshot().video.renditions.get("v").unwrap().bitrate.is_some());
rendition.set(config(Some(789), None)).unwrap();
feed(&mut rendition);
let snapshot = catalog.snapshot();
let config = snapshot.video.renditions.get("v").unwrap();
assert_eq!(config.bitrate, Some(789), "the re-set bitrate is now authoritative");
}
#[test]
fn an_unresolved_rendition_still_owns_its_name() {
let (_broadcast, catalog, mut rendition) = video_track();
assert!(
catalog.snapshot().video.renditions.is_empty(),
"nothing is published until the config resolves"
);
let err = catalog
.reserve()
.init::<hang::catalog::VideoConfig>("v")
.expect_err("an unresolved rendition still owns its name");
assert!(matches!(err, crate::Error::Hang(hang::Error::Duplicate(name)) if name == "v"));
rendition.set(config(Some(123), None)).unwrap();
assert_eq!(
catalog.snapshot().video.renditions.get("v").unwrap().bitrate,
Some(123),
"the importer's config owns the rendition"
);
}
#[test]
fn a_resolved_rendition_still_owns_its_name() {
let (_broadcast, catalog, mut rendition) = video_track();
rendition.set(config(Some(789), None)).unwrap();
assert!(
catalog.reserve().init::<hang::catalog::VideoConfig>("v").is_err(),
"owned by the live rendition"
);
assert_eq!(
catalog.snapshot().video.renditions.get("v").unwrap().bitrate,
Some(789),
"the refused reservation left the rendition intact"
);
}
#[test]
fn dropping_a_rendition_frees_its_name() {
let (_broadcast, catalog, mut rendition) = video_track();
rendition.set(config(None, None)).unwrap();
drop(rendition);
assert!(
catalog.snapshot().video.renditions.is_empty(),
"the entry is retired with its owner"
);
let mut replacement = catalog
.reserve()
.init::<hang::catalog::VideoConfig>("v")
.expect("the name is free once the rendition drops");
replacement.set(config(Some(456), None)).unwrap();
assert_eq!(catalog.snapshot().video.renditions.get("v").unwrap().bitrate, Some(456));
}
#[test]
fn a_name_is_owned_per_section() {
let (_broadcast, catalog, _video) = video_track();
catalog
.reserve()
.init::<hang::catalog::AudioConfig>("v")
.expect("an audio rendition is a different section from a video one");
}
#[test]
fn an_existing_entry_owns_its_name() {
let mut broadcast = moq_net::broadcast::Info::new().produce();
let mut catalog = super::super::Producer::new(&mut broadcast, super::super::Config::default()).unwrap();
catalog.modify().unwrap().video.insert("v", config(None, None)).unwrap();
assert!(
catalog.reserve().init::<hang::catalog::VideoConfig>("v").is_err(),
"the catalog already carries it"
);
}
#[test]
fn renditions_share_the_broadcast_timeline() {
let mut broadcast = moq_net::broadcast::Info::new().produce();
let mut catalog = super::super::Producer::new(&mut broadcast, super::super::Config::default()).unwrap();
let _recorder = catalog.enroll("video0").unwrap();
let timeline = catalog.timeline();
assert_eq!(timeline.section().track, hang::timeline::DEFAULT_NAME);
assert_eq!(
catalog.snapshot().archive,
Some(timeline.section()),
"the one timeline is advertised at the catalog root"
);
}
mod custom {
use std::collections::BTreeMap;
use serde::{Deserialize, Serialize};
use super::*;
#[derive(Serialize, Deserialize, Clone, Default, Debug, PartialEq)]
struct TelemetryExt {
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
telemetry: BTreeMap<String, Telemetry>,
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
exploding: BTreeMap<String, Exploding>,
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
stubborn: BTreeMap<String, Stubborn>,
}
impl CatalogExt for TelemetryExt {}
#[derive(Serialize, Deserialize, Clone, Default, Debug, PartialEq)]
struct Telemetry {
schema: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
bitrate: Option<u64>,
}
impl RenditionConfig<TelemetryExt> for Telemetry {
fn detects() -> bool {
true
}
fn insert(self, catalog: &mut Catalog<TelemetryExt>, name: &str) {
catalog.ext.telemetry.insert(name.to_string(), self);
}
fn get_mut<'a>(catalog: &'a mut Catalog<TelemetryExt>, name: &str) -> Option<&'a mut Self> {
catalog.ext.telemetry.get_mut(name)
}
fn remove(catalog: &mut Catalog<TelemetryExt>, name: &str) {
catalog.ext.telemetry.remove(name);
}
fn estimate(&self) -> Estimate {
Estimate::default().with_bitrate(self.bitrate)
}
fn set_estimate(&mut self, estimate: Estimate) {
self.bitrate = estimate.bitrate;
}
}
fn telemetry(bitrate: Option<u64>) -> Telemetry {
Telemetry {
schema: "gps/v1".to_string(),
bitrate,
}
}
fn produce() -> (moq_net::broadcast::Producer, crate::catalog::Producer<TelemetryExt>) {
let mut broadcast = moq_net::broadcast::Info::new().produce();
let config = crate::catalog::Config::default().with_catalog(Catalog::<TelemetryExt>::default());
let catalog = crate::catalog::Producer::new(&mut broadcast, config).unwrap();
(broadcast, catalog)
}
#[derive(Serialize, Deserialize, Clone, Default, Debug, PartialEq)]
struct Exploding {
wrote: bool,
}
impl RenditionConfig<TelemetryExt> for Exploding {
fn insert(self, catalog: &mut Catalog<TelemetryExt>, name: &str) {
if self.wrote {
catalog.ext.exploding.insert(name.to_string(), self);
}
panic!("insert exploded");
}
fn get_mut<'a>(catalog: &'a mut Catalog<TelemetryExt>, name: &str) -> Option<&'a mut Self> {
catalog.ext.exploding.get_mut(name)
}
fn remove(catalog: &mut Catalog<TelemetryExt>, name: &str) {
catalog.ext.exploding.remove(name);
}
}
#[test]
fn a_poisoned_lock_does_not_take_the_producer_down() {
let (_broadcast, catalog) = produce();
let reserved = catalog.reserve();
let err = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
let mut boom = reserved.init::<Exploding>("gps").unwrap();
boom.set(Exploding { wrote: false }).unwrap();
}))
.expect_err("the config's insert panics");
assert!(
err.downcast_ref::<&str>().is_some_and(|msg| *msg == "insert exploded"),
"the panic is the one the config raised, not a lock failure"
);
assert!(catalog.snapshot().ext.telemetry.is_empty());
reserved
.init::<Exploding>("gps")
.expect("the unwound rendition released its name");
}
#[derive(Serialize, Deserialize, Clone, Default, Debug, PartialEq)]
struct Stubborn;
impl RenditionConfig<TelemetryExt> for Stubborn {
fn insert(self, catalog: &mut Catalog<TelemetryExt>, name: &str) {
catalog.ext.stubborn.insert(name.to_string(), self);
}
fn get_mut<'a>(catalog: &'a mut Catalog<TelemetryExt>, name: &str) -> Option<&'a mut Self> {
catalog.ext.stubborn.get_mut(name)
}
fn remove(catalog: &mut Catalog<TelemetryExt>, name: &str) {
catalog.ext.stubborn.remove(name);
panic!("remove exploded");
}
}
#[test]
fn a_panicking_remove_still_frees_the_name() {
let (_broadcast, catalog) = produce();
let reserved = catalog.reserve();
let mut rendition = reserved.init::<Stubborn>("gps").unwrap();
rendition.set(Stubborn).unwrap();
std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| drop(rendition)))
.expect_err("the config's remove panics");
reserved
.init::<Stubborn>("gps")
.expect("a name held by a blown-up cleanup hook would be lost forever");
}
#[test]
fn an_unwinding_rendition_retires_a_partially_written_entry() {
let (_broadcast, catalog) = produce();
let reserved = catalog.reserve();
std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
let mut boom = reserved.init::<Exploding>("gps").unwrap();
boom.set(Exploding { wrote: true }).unwrap();
}))
.expect_err("the config's insert panics after writing");
assert!(
catalog.snapshot().ext.exploding.is_empty(),
"the entry the panicking insert wrote is retired with its owner"
);
reserved
.init::<Exploding>("gps")
.expect("a stranded entry would refuse this forever");
}
#[test]
fn detects_and_advertises() {
let (_broadcast, catalog) = produce();
let reserved = catalog.reserve();
let mut rendition = reserved.init::<Telemetry>("gps").unwrap();
drop(reserved);
rendition.set(telemetry(None)).unwrap();
feed(&mut rendition);
let snapshot = catalog.snapshot();
let config = snapshot.ext.telemetry.get("gps").unwrap();
assert!(config.bitrate.is_some(), "absent bitrate should be auto-detected");
drop(rendition);
assert!(
!catalog.snapshot().ext.telemetry.contains_key("gps"),
"the rendition should be removed on drop"
);
}
#[test]
fn measures_a_custom_track() {
let (broadcast, catalog) = produce();
let reserved = catalog.reserve();
let net = broadcast.create_track("gps", None).unwrap();
let mut track = reserved
.track(
net,
crate::catalog::hang::Container::Legacy(crate::container::Kind::Data),
None,
)
.unwrap();
drop(reserved);
assert!(matches!(track.modify(), Err(crate::Error::NotPublished)));
track.set(telemetry(None)).unwrap();
for i in 0..60u64 {
track
.write(crate::container::Frame {
timestamp: ts(i * 40_000),
duration: None,
payload: bytes::Bytes::from(vec![0u8; 5_000]),
keyframe: true,
})
.unwrap();
}
let snapshot = catalog.snapshot();
let config = snapshot.ext.telemetry.get("gps").unwrap();
assert_eq!(config.bitrate, Some(1_000_000));
}
#[test]
fn keeps_supplied_bitrate() {
let (_broadcast, catalog) = produce();
let reserved = catalog.reserve();
let mut rendition = reserved.init::<Telemetry>("gps").unwrap();
drop(reserved);
rendition.set(telemetry(Some(4_200))).unwrap();
feed(&mut rendition);
let snapshot = catalog.snapshot();
assert_eq!(snapshot.ext.telemetry.get("gps").unwrap().bitrate, Some(4_200));
}
#[test]
fn gates_with_media_tracks() {
let (_broadcast, catalog) = produce();
let mut consumer = catalog.consume().unwrap();
let reserved = catalog.reserve();
let mut video = reserved.init::<hang::catalog::VideoConfig>("v").unwrap();
let mut gps = reserved.init::<Telemetry>("gps").unwrap();
drop(reserved);
let waiter = kio::Waiter::noop();
video.set(config(None, None)).unwrap();
assert!(
matches!(consumer.poll_next(&waiter), std::task::Poll::Pending),
"the catalog stays withheld while the telemetry rendition is unresolved"
);
gps.set(telemetry(None)).unwrap();
let mut latest = None;
while let std::task::Poll::Ready(Ok(Some(catalog))) = consumer.poll_next(&waiter) {
latest = Some(catalog);
}
let published = latest.expect("catalog published");
assert!(published.video.renditions.contains_key("v"));
assert!(published.ext.telemetry.contains_key("gps"));
}
}
}