use std::collections::HashMap;
use std::net::{IpAddr, Ipv6Addr, SocketAddr};
use std::sync::{Arc, Mutex as StdMutex};
use anyhow::{bail, Context, Result};
use futures::{stream::FuturesUnordered, StreamExt};
use moq_transport::coding::TrackNamespace;
use moq_transport::serve::{self, ServeError, TracksReader};
use moq_transport::session::{Announced, Publisher, Subscriber};
type LocalRegistry = Arc<StdMutex<HashMap<TrackNamespace, TracksReader>>>;
struct RegistryEntry {
namespace: TrackNamespace,
registry: LocalRegistry,
}
impl Drop for RegistryEntry {
fn drop(&mut self) {
self.registry.lock().unwrap().remove(&self.namespace);
}
}
pub async fn run(port: u16) -> Result<()> {
let identity = wtransport::Identity::self_signed(["localhost", "127.0.0.1"])
.context("self-sign Draft 14 MoQ relay")?;
let certificate = &identity.certificate_chain().as_slice()[0];
let certificate_hash = certificate
.hash()
.fmt(wtransport::tls::Sha256DigestFmt::BytesArray);
let tls_directory = std::env::temp_dir().join(format!(
"openrtc-moq-draft14-{}-{}",
std::process::id(),
uuid::Uuid::new_v4()
));
std::fs::create_dir_all(&tls_directory).context("create local MoQ TLS directory")?;
let certificate_path = tls_directory.join("certificate.pem");
let private_key_path = tls_directory.join("private-key.pem");
std::fs::write(&certificate_path, certificate.to_pem())
.context("write local MoQ certificate")?;
std::fs::write(&private_key_path, identity.private_key().to_secret_pem())
.context("write local MoQ private key")?;
let tls = moq_native_ietf::tls::Args {
cert: vec![certificate_path],
key: vec![private_key_path],
..Default::default()
}
.load()
.context("load local MoQ TLS identity")?;
let _ = std::fs::remove_dir_all(&tls_directory);
let endpoint = moq_native_ietf::quic::Endpoint::new(
moq_native_ietf::quic::Config::new(
SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), port),
None,
tls,
)
.context("configure local Draft 14 MoQ endpoint")?,
)
.context("start local Draft 14 MoQ endpoint")?;
let mut server = endpoint
.server
.context("local MoQ endpoint is not a server")?;
let registry: LocalRegistry = Arc::new(StdMutex::new(HashMap::new()));
println!(
"[test-moq-relay] Listening on https://localhost:{port}/moq certificate_hash={certificate_hash} draft=14"
);
while let Some((webtransport, _connection_id, transport)) = server.accept().await {
println!("[test-moq-relay] accepted Draft 14 WebTransport session");
let registry = registry.clone();
tokio::spawn(async move {
let result = async {
let (session, publisher, subscriber) =
moq_transport::session::Session::accept(webtransport, None, transport)
.await
.context("accept Draft 14 MoQ session")?;
println!("[test-moq-relay] completed Draft 14 SETUP exchange");
let publisher = publisher.context("Draft 14 client did not enable publishing")?;
let subscriber =
subscriber.context("Draft 14 client did not enable subscribing")?;
tokio::select! {
result = session.run() => result.context("run Draft 14 MoQ session")?,
result = serve_subscriptions(publisher, registry.clone()) => result?,
result = serve_announcements(subscriber, registry) => result?,
}
Ok::<(), anyhow::Error>(())
}
.await;
if let Err(error) = result {
eprintln!("[test-moq-relay] Draft 14 session ended: {error:#}");
}
});
}
bail!("local Draft 14 MoQ endpoint stopped accepting sessions")
}
async fn serve_subscriptions(mut publisher: Publisher, registry: LocalRegistry) -> Result<()> {
let mut subscriptions = FuturesUnordered::new();
loop {
tokio::select! {
subscribed = publisher.subscribed() => {
let Some(subscribed) = subscribed else {
return Ok(());
};
let tracks = registry
.lock()
.unwrap()
.get(&subscribed.info.track_namespace)
.cloned();
subscriptions.push(async move {
let Some(mut tracks) = tracks else {
let _ = subscribed.close(ServeError::NotFound);
return;
};
let Some(track) = tracks.subscribe(
subscribed.info.track_namespace.clone(),
&subscribed.info.track_name,
) else {
let _ = subscribed.close(ServeError::NotFound);
return;
};
if let Err(error) = subscribed.serve(track).await {
eprintln!("[test-moq-relay] failed serving Draft 14 subscription: {error}");
}
});
}
_ = subscriptions.next(), if !subscriptions.is_empty() => {}
}
}
}
async fn serve_announcements(mut subscriber: Subscriber, registry: LocalRegistry) -> Result<()> {
let mut announcements = FuturesUnordered::new();
loop {
tokio::select! {
announced = subscriber.announced() => {
let Some(announced) = announced else {
return Ok(());
};
let registry = registry.clone();
let subscriber = subscriber.clone();
announcements.push(async move {
if let Err(error) = serve_announcement(announced, subscriber, registry).await {
eprintln!("[test-moq-relay] failed serving Draft 14 announcement: {error:#}");
}
});
}
_ = announcements.next(), if !announcements.is_empty() => {}
}
}
}
async fn serve_announcement(
mut announced: Announced,
subscriber: Subscriber,
registry: LocalRegistry,
) -> Result<()> {
let namespace = announced.info.namespace.clone();
let (_writer, mut requests, reader) = serve::Tracks::new(namespace.clone()).produce();
{
let mut entries = registry.lock().unwrap();
if entries.contains_key(&namespace) {
announced.close(ServeError::Duplicate)?;
bail!("duplicate Draft 14 namespace announcement");
}
entries.insert(namespace.clone(), reader);
}
let _registration = RegistryEntry {
namespace,
registry,
};
announced.ok()?;
let mut forwarded_tracks = FuturesUnordered::new();
loop {
tokio::select! {
result = announced.closed() => {
result?;
return Ok(());
}
requested = requests.next() => {
let Some(track) = requested else {
return Ok(());
};
let mut subscriber = subscriber.clone();
forwarded_tracks.push(async move {
if let Err(error) = subscriber.subscribe(track).await {
eprintln!("[test-moq-relay] failed forwarding Draft 14 track: {error}");
}
});
}
_ = forwarded_tracks.next(), if !forwarded_tracks.is_empty() => {}
}
}
}