feat: hls import

justfile

@@ -71,6 +71,11 @@ leaf:
 pub name url='http://localhost:4443/anon' *args:
 	cd rs && just pub {{name}} {{url}} {{args}}
 
+# Ingest a live HLS media playlist and publish it via hang (full ladder).
+# Thin wrapper around the Rust justfile recipe.
+pub-hls url name='demo' relay='http://localhost:4443/anon':
+	cd rs && just pub-hls {{url}} {{name}} {{relay}}
+
 # Publish/subscribe using gstreamer - see https://github.com/moq-dev/gstreamer
 pub-gst name url='http://localhost:4443/anon':
 	@echo "GStreamer plugin has moved to: https://github.com/moq-dev/gstreamer"

rs/hang-cli/Cargo.toml

@@ -19,6 +19,7 @@ path = "src/main.rs"
 [dependencies]
 anyhow = { version = "1", features = ["backtrace"] }
 axum = { version = "0.8", features = ["tokio"] }
+bytes = "1.10"
 clap = { version = "4", features = ["derive"] }
 hang = { workspace = true }
 hyper-serve = { version = "0.6", features = ["tls-rustls"] }

rs/hang-cli/src/client.rs

@@ -1,49 +1,29 @@
-use crate::import::Import;
-use crate::ImportType;
-use anyhow::Context;
+use crate::Publish;
+
 use hang::moq_lite;
-use tokio::io::AsyncRead;
 use url::Url;
 
-pub async fn client<T: AsyncRead + Unpin>(
-	config: moq_native::ClientConfig,
-	url: Url,
-	name: String,
-	format: ImportType,
-	input: &mut T,
-) -> anyhow::Result<()> {
-	let broadcast = moq_lite::Broadcast::produce();
+pub async fn client(config: moq_native::ClientConfig, url: Url, name: String, publish: Publish) -> anyhow::Result<()> {
 	let client = config.init()?;
 
 	tracing::info!(%url, %name, "connecting");
 	let session = client.connect(url).await?;
 
 	// Create an origin producer to publish to the broadcast.
 	let origin = moq_lite::Origin::produce();
+	origin.producer.publish_broadcast(&name, publish.consume());
 
 	// Establish the connection, not providing a subscriber.
 	let session = moq_lite::Session::connect(session, origin.consumer, None).await?;
 
-	let mut import = Import::new(broadcast.producer.into(), format);
-	import
-		.init_from(input)
-		.await
-		.context("failed to initialize from media stream")?;
-
-	// Announce the broadcast as available once the catalog is ready.
-	origin.producer.publish_broadcast(&name, broadcast.consumer);
-
 	// Notify systemd that we're ready.
 	let _ = sd_notify::notify(true, &[sd_notify::NotifyState::Ready]);
 
 	tokio::select! {
-		res = import.read_from(input) => res,
+		res = publish.run() => res,
 		res = session.closed() => res.map_err(Into::into),
-
 		_ = tokio::signal::ctrl_c() => {
 			session.close(moq_lite::Error::Cancel);
-
-			// Give it a chance to close.
 			tokio::time::sleep(std::time::Duration::from_millis(100)).await;
 			Ok(())
 		},