#[cfg(feature = "capture")]
use std::time::Instant;
use moq_mux::catalog::hang::CatalogExt;
#[cfg(any(feature = "capture", test))]
use moq_net::Timestamp;
use crate::Error;
#[cfg(any(feature = "capture", test))]
use crate::Frame;
#[cfg(feature = "capture")]
use crate::capture;
use super::Encoded;
#[cfg(feature = "capture")]
use super::Sink;
#[cfg(any(feature = "capture", test))]
use super::encoder;
#[cfg(feature = "capture")]
use super::encoder::Codec;
#[cfg(feature = "capture")]
use super::rate::{Control, Policy};
#[cfg(feature = "capture")]
const DEFAULT_FRAMERATE: u32 = 30;
fn rendition_hint(rendition: hang::catalog::VideoConfig) -> moq_mux::catalog::VideoHint {
let mut hint = moq_mux::catalog::VideoHint::default();
hint.codec = Some(rendition.codec);
hint.coded_width = rendition.coded_width;
hint.coded_height = rendition.coded_height;
hint.display_aspect_width = rendition.display_aspect_width;
hint.display_aspect_height = rendition.display_aspect_height;
hint.framerate = rendition.framerate;
hint.bitrate = rendition.bitrate;
hint.optimize_for_latency = rendition.optimize_for_latency;
hint
}
enum Codecs<E: CatalogExt> {
H264 {
split: moq_mux::codec::h264::Split,
import: moq_mux::codec::h264::Import<E>,
},
H265 {
split: moq_mux::codec::h265::Split,
import: moq_mux::codec::h265::Import<E>,
},
}
pub struct Producer<E: CatalogExt = ()> {
codecs: Codecs<E>,
}
impl<E: CatalogExt> Producer<E> {
pub fn new(
mut broadcast: moq_net::broadcast::Producer,
catalog: moq_mux::catalog::Producer<E>,
rendition: hang::catalog::VideoConfig,
) -> Result<Self, Error> {
let codecs = match &rendition.codec {
hang::catalog::VideoCodec::H264(_) => {
let track = broadcast.unique_track(".avc3", catalog.track_info())?;
Codecs::H264 {
split: moq_mux::codec::h264::Split::new(),
import: moq_mux::codec::h264::Import::new(track, catalog.reserve(), rendition_hint(rendition))?,
}
}
hang::catalog::VideoCodec::H265(_) => {
let track = broadcast.unique_track(".hev1", catalog.track_info())?;
Codecs::H265 {
split: moq_mux::codec::h265::Split::new(),
import: moq_mux::codec::h265::Import::new(track, catalog.reserve(), rendition_hint(rendition))?,
}
}
other => {
return Err(Error::Codec(anyhow::anyhow!(
"{other} is not a codec this producer can publish"
)));
}
};
Ok(Self { codecs })
}
pub fn demand(&self) -> moq_net::track::Demand {
match &self.codecs {
Codecs::H264 { import, .. } => import.demand(),
Codecs::H265 { import, .. } => import.demand(),
}
}
pub fn publish(&mut self, encoded: &[Encoded]) -> Result<(), Error> {
for frame in encoded {
let timestamp = Some(frame.timestamp);
match &mut self.codecs {
Codecs::H264 { split, import } => {
let mut frames = split.decode(&frame.payload, timestamp)?;
frames.extend(split.flush(timestamp)?);
import.decode(frames)?;
}
Codecs::H265 { split, import } => {
let mut frames = split.decode(&frame.payload, timestamp)?;
frames.extend(split.flush(timestamp)?);
import.decode(frames)?;
}
}
}
Ok(())
}
pub fn discontinuity(&mut self) -> Result<(), Error> {
match &mut self.codecs {
Codecs::H264 { import, .. } => import.discontinuity()?,
Codecs::H265 { import, .. } => import.discontinuity()?,
}
Ok(())
}
pub fn finish(mut self) -> Result<(), Error> {
match &mut self.codecs {
Codecs::H264 { import, .. } => import.finish()?,
Codecs::H265 { import, .. } => import.finish()?,
}
Ok(())
}
pub fn abort(self, err: moq_net::Error) {
match self.codecs {
Codecs::H264 { import, .. } => import.abort(err),
Codecs::H265 { import, .. } => import.abort(err),
}
}
}
#[derive(Clone, Default)]
#[non_exhaustive]
#[cfg(feature = "capture")]
pub struct Options {
pub bitrate: Option<u64>,
pub codec: Codec,
pub kind: encoder::Kind,
pub bandwidth: Option<moq_net::bandwidth::Consumer>,
}
#[cfg(feature = "capture")]
impl std::fmt::Debug for Options {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Options")
.field("bitrate", &self.bitrate)
.field("codec", &self.codec)
.field("kind", &self.kind)
.field("bandwidth", &self.bandwidth.is_some())
.finish()
}
}
#[cfg(feature = "capture")]
pub async fn publish_capture<E: CatalogExt>(
broadcast: moq_net::broadcast::Producer,
catalog: moq_mux::catalog::Producer<E>,
capture: capture::Config,
encode: Options,
clock: moq_mux::Clock,
) -> Result<(), Error> {
if capture.framerate == Some(0) {
return Err(Error::InvalidFramerate(0));
}
let rendition = {
let camera = capture::open(&capture).await?;
let mut probe_config = encoder::Config::new(
camera.width(),
camera.height(),
capture
.framerate
.or_else(|| camera.framerate())
.unwrap_or(DEFAULT_FRAMERATE),
);
probe_config.bitrate = encode.bitrate;
probe_config.codec = encode.codec;
probe_config.kind = encode.kind.clone();
probe_config.probe().await?
};
let mut producer = Producer::new(broadcast, catalog, rendition)?;
let demand = producer.demand();
let result = capture_loop(&mut producer, &demand, &capture, &encode, &clock).await;
match &result {
Ok(()) => {
if let Err(err) = producer.finish() {
tracing::debug!(error = %err, "video track finish after capture ended");
}
}
Err(err) => producer.abort(moq_net::Error::Transport(err.to_string())),
}
result
}
#[cfg(all(feature = "capture", not(target_os = "macos")))]
#[allow(dead_code)]
fn assert_publish_capture_send(
broadcast: moq_net::broadcast::Producer,
catalog: moq_mux::catalog::Producer,
capture: capture::Config,
encode: Options,
clock: moq_mux::Clock,
) {
fn is_send<T: Send>(_: &T) {}
is_send(&publish_capture(broadcast, catalog, capture, encode, clock));
}
#[cfg(feature = "capture")]
type Rate = Option<(moq_net::bandwidth::Consumer, Control)>;
#[cfg(feature = "capture")]
async fn next_estimate(rate: &mut Rate) -> Option<Option<u64>> {
match rate {
Some((bandwidth, _)) => bandwidth.changed().await.ok(),
None => std::future::pending().await,
}
}
#[cfg(feature = "capture")]
async fn apply_estimate(encoder: &mut Sink, rate: &mut Rate, estimate: Option<Option<u64>>) {
let Some((_, control)) = rate.as_mut() else { return };
let Some(estimate) = estimate else {
tracing::debug!("bandwidth estimate ended; holding the current encoder bitrate");
*rate = None;
return;
};
let Some(bitrate) = control.update(estimate, Instant::now()) else {
return;
};
match encoder.set_bitrate(bitrate).await {
Ok(()) => tracing::debug!(bitrate, estimate, "adjusted encoder bitrate"),
Err(Error::BitrateUnsupported(name)) => {
tracing::warn!(encoder = name, "encoder cannot follow the bandwidth estimate");
*rate = None;
}
Err(err) => tracing::warn!(error = %err, bitrate, "failed to adjust encoder bitrate"),
}
}
#[cfg(feature = "capture")]
fn log_track_ended(err: moq_net::Error) {
if matches!(err, moq_net::Error::Dropped | moq_net::Error::Closed) {
tracing::debug!("video track no longer announced; stopping capture");
} else {
tracing::warn!(error = %err, "video track aborted; stopping capture");
}
}
#[cfg(feature = "capture")]
async fn capture_loop<E: CatalogExt>(
producer: &mut Producer<E>,
demand: &moq_net::track::Demand,
capture: &capture::Config,
encode: &Options,
clock: &moq_mux::Clock,
) -> Result<(), Error> {
loop {
if let Err(err) = demand.used().await {
log_track_ended(err);
return Ok(());
}
let mut camera = capture::open(capture).await?;
let framerate = capture
.framerate
.or_else(|| camera.framerate())
.unwrap_or(DEFAULT_FRAMERATE);
let mut encoder_config = encoder::Config::new(camera.width(), camera.height(), framerate);
encoder_config.bitrate = encode.bitrate;
encoder_config.codec = encode.codec;
encoder_config.kind = encode.kind.clone();
let mut encoder = Sink::open(&encoder_config).await?;
let mut force_keyframe = true;
tracing::info!(encoder = encoder.name(), device = camera.device(), "capturing");
let mut rate = encode
.bandwidth
.clone()
.map(|bandwidth| (bandwidth, Control::new(Policy::new(encoder_config.resolved_bitrate()))));
loop {
let frame = tokio::select! {
biased;
res = demand.unused() => {
if let Err(err) = res {
log_track_ended(err);
return Ok(());
}
break; }
estimate = next_estimate(&mut rate) => {
apply_estimate(&mut encoder, &mut rate, estimate).await;
continue;
}
frame = camera.read() => frame,
};
let Some(surface) = frame else { break };
let frame = Frame::new(surface, Timestamp::from_micros(clock.micros())?);
if force_keyframe {
encoder.keyframe();
force_keyframe = false;
}
producer.publish(&encoder.encode(frame).await?)?;
}
drop(camera);
tracing::info!("no viewers: released camera");
}
}
#[cfg(test)]
mod tests {
use moq_mux::catalog::Stream as _;
use super::*;
use crate::encode::{Config, Encoder};
async fn roundtrip_rendition(codec: Codec, kind: encoder::Kind) -> (String, hang::catalog::VideoConfig) {
let mut broadcast = moq_net::broadcast::Info::new().produce();
let catalog = moq_mux::catalog::Producer::new(&mut broadcast).unwrap();
let mut config = Config::new(320, 240, 30);
config.codec = codec;
config.kind = kind;
let mut producer = Producer::new(broadcast, catalog.clone(), config.probe().await.unwrap()).unwrap();
let advertised = rendition(&catalog).expect("the rendition publishes before any frame").1;
let mut encoder = Encoder::new(&config).unwrap();
assert_eq!(encoder.codec(), codec);
let rgba = vec![0x80u8; 320 * 240 * 4];
for i in 0..10u64 {
let surface = crate::Surface::rgba(&rgba, crate::Size::new(320, 240)).unwrap();
let frame = Frame::new(surface, Timestamp::from_micros(i * 33_333).unwrap());
producer.publish(&encoder.encode(&frame).unwrap()).unwrap();
}
producer.publish(&encoder.finish().unwrap()).unwrap();
let (name, resolved) = rendition(&catalog).expect("the importer should have registered a video rendition");
let (mut before, mut after) = (advertised, resolved.clone());
before.jitter = None;
after.jitter = None;
assert_eq!(
before, after,
"the first keyframe should confirm the advertised rendition, not correct it"
);
(name, resolved)
}
fn rendition(catalog: &moq_mux::catalog::Producer) -> Option<(String, hang::catalog::VideoConfig)> {
let snapshot = catalog.snapshot();
let (name, config) = snapshot.video.renditions.iter().next()?;
Some((name.clone(), config.clone()))
}
#[tokio::test]
async fn the_rendition_reaches_the_wire_before_the_first_frame() {
let mut broadcast = moq_net::broadcast::Info::new().produce();
let consumer = broadcast.consume();
let catalog = moq_mux::catalog::Producer::new(&mut broadcast).unwrap();
let mut config = Config::new(1920, 1080, 30);
config.bitrate = Some(6_000_000);
config.kind = encoder::Kind::Software;
let _producer = Producer::new(broadcast, catalog, config.probe().await.unwrap()).unwrap();
let mut stream = moq_mux::catalog::Consumer::<()>::new(&consumer, moq_mux::catalog::CatalogFormat::Hang)
.await
.unwrap();
let snapshot = stream.next().await.unwrap().expect("a catalog before any frame");
let (name, rendition) = snapshot
.video
.renditions
.iter()
.next()
.expect("the track must be discoverable before it has encoded anything");
assert!(name.ends_with(".avc3"));
let hang::catalog::VideoCodec::H264(h264) = &rendition.codec else {
panic!("expected H.264, got {}", rendition.codec)
};
assert!(h264.inline, "an avc3 track carries its parameter sets in band");
assert_eq!(rendition.coded_width, Some(1920));
assert_eq!(rendition.coded_height, Some(1080));
assert_eq!(rendition.framerate, Some(30.0));
assert_eq!(rendition.bitrate, Some(6_000_000));
}
#[tokio::test]
async fn h264_roundtrip_publishes_avc3() {
let (name, config) = roundtrip_rendition(Codec::H264, encoder::Kind::Software).await;
assert!(name.ends_with(".avc3"));
assert_eq!(config.coded_width, Some(320));
assert_eq!(config.coded_height, Some(240));
}
#[cfg(target_os = "macos")]
#[tokio::test]
async fn h265_roundtrip_publishes_hev1() {
let (name, config) = roundtrip_rendition(Codec::H265, encoder::Kind::Hardware).await;
assert!(name.ends_with(".hev1"));
assert_eq!(config.coded_width, Some(320));
assert_eq!(config.coded_height, Some(240));
}
}