diff --git a/.github/homebrew/Formula/moq-cli.rb.tmpl b/.github/homebrew/Formula/moq-cli.rb.tmpl index 2f92dcf94..ac9e3ad7f 100644 --- a/.github/homebrew/Formula/moq-cli.rb.tmpl +++ b/.github/homebrew/Formula/moq-cli.rb.tmpl @@ -27,10 +27,10 @@ class MoqCli < Formula end def install - bin.install "bin/moq-cli" + bin.install "bin/moq" end test do - system bin/"moq-cli", "--help" + system bin/"moq", "--help" end end diff --git a/.github/workflows/moq-cli.yml b/.github/workflows/moq-cli.yml index 4b7fe8637..e98d27b19 100644 --- a/.github/workflows/moq-cli.yml +++ b/.github/workflows/moq-cli.yml @@ -54,21 +54,10 @@ jobs: # and AlmaLinux/RHEL/Rocky 9+ (.rpm). run: cargo zigbuild --release --target ${{ matrix.target }}.2.34 -p moq-cli - - name: Rename binary - shell: bash - env: - TARGET: ${{ matrix.target }} - run: | - # The crate is named moq-cli but the .deb / .rpm and Homebrew - # formula present the binary as `moq`. Cargo produces - # target/release/moq-cli by default (no [[bin]] override), - # so make a sibling copy that the downstream packaging steps - # can point at without each one needing to know the rename. - cp "target/${TARGET}/release/moq-cli" "target/${TARGET}/release/moq" - - name: Package bare binary shell: bash run: | + # The crate is named moq-cli but its `[[bin]]` ships as `moq`. version="${{ steps.parse.outputs.version }}" target="${{ matrix.target }}" name="moq-cli-v${version}-${target}" @@ -83,7 +72,7 @@ jobs: run: | name="moq-cli-${VERSION}-${TARGET}" mkdir -p "dist/${name}/bin" - cp "target/${TARGET}/release/moq-cli" "dist/${name}/bin/moq-cli" + cp "target/${TARGET}/release/moq" "dist/${name}/bin/moq" cp LICENSE-MIT LICENSE-APACHE "dist/${name}/" if [[ -f rs/moq-cli/README.md ]]; then cp rs/moq-cli/README.md "dist/${name}/" @@ -166,6 +155,7 @@ jobs: run: | ./rs/scripts/package-binary.sh \ --crate moq-cli \ + --bin moq \ --target ${{ matrix.target }} \ --version ${{ steps.parse.outputs.version }} \ --output dist @@ -213,6 +203,7 @@ jobs: run: | ./rs/scripts/package-windows.sh \ --crate moq-cli \ + --bin moq \ --target ${{ matrix.target }} \ --version ${{ steps.parse.outputs.version }} \ --output dist diff --git a/Cargo.lock b/Cargo.lock index b4197f32e..4b529279b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2982,12 +2982,6 @@ dependencies = [ "time", ] -[[package]] -name = "http-range-header" -version = "0.4.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9171a2ea8a68358193d15dd5d70c1c10a2afc3e7e4c5bc92bc9f025cebd7359c" - [[package]] name = "http-serde" version = "2.1.1" @@ -3937,16 +3931,6 @@ version = "0.3.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" -[[package]] -name = "mime_guess" -version = "2.0.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f7c44f8e672c00fe5308fa235f821cb4198414e1c77935c1ab6948d3fd78550e" -dependencies = [ - "mime", - "unicase", -] - [[package]] name = "minimal-lexical" version = "0.2.1" @@ -4057,7 +4041,7 @@ dependencies = [ [[package]] name = "moq-cli" -version = "0.7.35" +version = "0.8.0" dependencies = [ "anyhow", "axum", @@ -4069,6 +4053,9 @@ dependencies = [ "moq-hls", "moq-mux", "moq-native", + "moq-rtc", + "moq-rtmp", + "moq-srt", "rustls", "sd-notify", "tokio", @@ -8013,19 +8000,9 @@ checksum = "b11f75e912b0c2be01b63d8cf8057b8c3f97cf34abb3d431a3a4c8675498e233" dependencies = [ "bitflags", "bytes", - "futures-core", - "futures-util", "http", - "http-body", - "http-body-util", - "http-range-header", - "httpdate", - "mime", - "mime_guess", "percent-encoding", "pin-project-lite", - "tokio", - "tokio-util", "tower-layer", "tower-service", ] @@ -8197,12 +8174,6 @@ version = "0.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "eaea85b334db583fe3274d12b4cd1880032beab409c0d774be044d4480ab9a94" -[[package]] -name = "unicase" -version = "2.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dbc4bc3a9f746d862c45cb89d705aa10f187bb96c76001afab07a0d35ce60142" - [[package]] name = "unicode-ident" version = "1.0.24" diff --git a/Cargo.toml b/Cargo.toml index 02478e5d3..eae6a17c1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -69,6 +69,9 @@ moq-msf = { version = "0.3", path = "rs/moq-msf" } moq-mux = { version = "0.7", path = "rs/moq-mux" } moq-native = { version = "0.17", path = "rs/moq-native", default-features = false } moq-net = { version = "0.1", path = "rs/moq-net" } +moq-rtc = { version = "0.0.1", path = "rs/moq-rtc" } +moq-rtmp = { version = "0.0.1", path = "rs/moq-rtmp" } +moq-srt = { version = "0.0.1", path = "rs/moq-srt" } moq-token = { version = "0.6", path = "rs/moq-token" } moq-video = { version = "0.0.6", path = "rs/moq-video" } qmux = { version = "0.2", default-features = false } diff --git a/Dockerfile b/Dockerfile index 230958d10..c77df07b7 100644 --- a/Dockerfile +++ b/Dockerfile @@ -28,9 +28,13 @@ RUN --mount=type=cache,target=/root/.cache --mount=type=cache,target=/nix,from=n # Default to `/bin/sh` for the entrypoint if no package is specified ARG package="sh" -# Create entry.sh script that knows which binary to run -RUN echo '#!/bin/sh' > /output/entry.sh && \ - echo "exec /bin/${package} \"\$@\"" >> /output/entry.sh && \ +# Create entry.sh script that knows which binary to run. Derive it from the +# single binary the package produced (so a crate whose `[[bin]]` name differs +# from its package name, e.g. `moq-cli` shipping as `moq`, just works); fall +# back to the package name when there's no binary (the `sh` default). +RUN binary="$(ls /output/result/bin 2>/dev/null | head -n1)"; \ + [ -n "$binary" ] || binary="${package}"; \ + printf '#!/bin/sh\nexec /bin/%s "$@"\n' "${binary}" > /output/entry.sh && \ chmod +x /output/entry.sh # Final image (when no specific package is selected, defaults to sh) diff --git a/demo/pub/justfile b/demo/pub/justfile index 0cc6aed29..d61e49e10 100644 --- a/demo/pub/justfile +++ b/demo/pub/justfile @@ -52,25 +52,25 @@ tos url='http://localhost:4443' *args: # Publish a video using ffmpeg (CMAF / fMP4) to a relay server. cmaf name url='http://localhost:4443' *args: - cargo build --bin moq-cli + cargo build --bin moq just ffmpeg-cmaf "media/{{ name }}.mp4" |\ - cargo run --bin moq-cli -- \ - {{ args }} publish --url "{{ url }}" --name "{{ name }}.hang" fmp4 + cargo run --bin moq -- \ + {{ args }} --client-connect "{{ url }}" --broadcast "{{ name }}.hang" import fmp4 # Publish a video using ffmpeg (MPEG-TS) to a relay server. ts name url='http://localhost:4443' *args: - cargo build --bin moq-cli + cargo build --bin moq just ffmpeg-ts "media/{{ name }}.mp4" |\ - cargo run --bin moq-cli -- \ - {{ args }} publish --url "{{ url }}" --name "{{ name }}.hang" ts + cargo run --bin moq -- \ + {{ args }} --client-connect "{{ url }}" --broadcast "{{ name }}.hang" import ts # Publish a video via iroh. iroh name url prefix="": just download "{{ name }}" - cargo build --bin moq-cli + cargo build --bin moq just ffmpeg-cmaf "media/{{ name }}.mp4" |\ - cargo run --bin moq-cli -- \ - --iroh-enabled publish --url "{{ url }}" --name "{{ prefix }}{{ name }}.hang" fmp4 + cargo run --bin moq -- \ + --iroh-enabled --client-connect "{{ url }}" --broadcast "{{ prefix }}{{ name }}.hang" import fmp4 # Generate and ingest an HLS stream from a video file. hls name relay="http://localhost:4443": @@ -142,7 +142,7 @@ hls name relay="http://localhost:4443": trap cleanup SIGINT SIGTERM EXIT echo ">>> Importing HLS via moq-cli" - cargo run --bin moq-cli -- hls --url "{{ relay }}" import --broadcast "{{ name }}.hang" --playlist "$OUT_DIR/master.m3u8" + cargo run --bin moq -- --client-connect "{{ relay }}" --broadcast "{{ name }}.hang" import hls "$OUT_DIR/master.m3u8" EXIT_CODE=$? cleanup @@ -151,7 +151,7 @@ hls name relay="http://localhost:4443": # Publish a video using H.264 Annex B format. h264 name url="http://localhost:4443" *args: just download "{{ name }}" - cargo build --bin moq-cli + cargo build --bin moq ffmpeg -hide_banner -v quiet \ -stream_loop -1 -re \ @@ -159,16 +159,16 @@ h264 name url="http://localhost:4443" *args: -c:v copy -an \ -bsf:v h264_mp4toannexb \ -f h264 \ - - | cargo run --bin moq-cli -- publish --url "{{ url }}" --name "{{ name }}.hang" avc3 {{ args }} + - | cargo run --bin moq -- {{ args }} --client-connect "{{ url }}" --broadcast "{{ name }}.hang" import avc3 # Publish a video using ffmpeg directly from moq to the localhost. serve name *args: just download "{{ name }}" - cargo build --bin moq-cli + cargo build --bin moq just ffmpeg-cmaf "media/{{ name }}.mp4" |\ - cargo run --bin moq-cli -- \ - {{ args }} serve --listen "[::]:4443" --tls-generate "localhost" \ - --name "{{ name }}.hang" fmp4 + cargo run --bin moq -- \ + {{ args }} --server-bind "[::]:4443" --tls-generate "localhost" \ + --broadcast "{{ name }}.hang" import fmp4 # Generate and serve an HLS stream from a video for testing. serve-hls name port="8000": diff --git a/doc/bin/cli.md b/doc/bin/cli.md index 91ce37a6d..89367cf4a 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -5,7 +5,9 @@ description: Command-line tools for MoQ media # FFmpeg / moq-cli -`moq-cli` is a command-line tool for publishing media to MoQ relays. It works with FFmpeg for encoding. +`moq-cli` is a media router: it wires one endpoint onto a shared MoQ Origin. It +moves media into MoQ from a source, or out of MoQ to a sink, bridging stdin/stdout +(via FFmpeg), HLS, RTMP, SRT, and WebRTC. ## Installation @@ -39,7 +41,7 @@ docker pull moqdev/moq-cli # moq-cli reads media from stdin, so pipe an MPEG-TS stream into the container. # `-i` forwards stdin to the container process. ffmpeg -i video.mp4 -c copy -f mpegts - | \ - docker run -i moqdev/moq-cli publish --url https://relay.example.com/anon --broadcast my-stream ts + docker run -i moqdev/moq-cli --client-connect https://relay.example.com/anon --broadcast my-stream.hang import ts ``` Multi-arch images (`linux/amd64` and `linux/arm64`) are published to [Docker Hub](https://hub.docker.com/r/moqdev/moq-cli). @@ -54,11 +56,36 @@ cargo build --release --bin moq-cli The binary will be in `target/release/moq-cli`. +## The grammar + +``` +moq [endpoint options] +``` + +- **MoQ side** attaches the Origin to the network, and comes before the verb. At + least one of: + - `--client-connect ` dials a relay. The URL path is the relay auth path + (e.g. `/anon`), `?jwt=` supplies a token, and `--broadcast` names the + broadcast. + - `--server-bind ` hosts MoQ sessions directly (with `--tls-generate` / + `--tls-cert` + `--tls-key`). + + Both may be given at once (dial a relay *and* accept incoming sessions). +- **`import`** routes media INTO MoQ (a source fills the Origin); **`export`** + routes it OUT (a sink drains the Origin). The verb fixes the data direction. +- **endpoint** is one subcommand: a container format (`avc3`, `fmp4`, `ts`, `flv` + read from stdin on import; `fmp4`, `mkv`, `ts`, `flv` written to stdout on + export), or a gateway (`hls`, `rtmp`, `srt`, `rtc`). For the bidirectional + gateways, `--connect` dials out and `--listen` binds a socket; the parent verb + decides whether that pushes or pulls. + +Run `moq import --help` / `moq export --help` to see the endpoints, and +`moq import rtmp --help` for a specific one. + ## Basic Usage -`moq-cli publish` reads media from stdin and selects the input container with a -subcommand (`ts`, `fmp4`, `flv`, `avc3`, `hls`). The destination is set with -`--url` (the server) and `--broadcast` (the broadcast name), not a path on the URL. +`moq import ` reads a container from stdin; +`moq export ` writes one to stdout. ### Publish a Video File @@ -66,15 +93,7 @@ Remux a file to MPEG-TS and pipe it in (`-c copy` avoids re-encoding): ```bash ffmpeg -i video.mp4 -c copy -f mpegts - | \ - moq-cli publish --url https://relay.example.com/anon --broadcast my-stream ts -``` - -### Publish from FFmpeg - -Pipe FFmpeg output directly to moq-cli: - -```bash -ffmpeg -i input.mp4 -f mpegts - | moq-cli publish --url https://relay.example.com/anon --broadcast my-stream ts + moq --client-connect https://relay.example.com/anon --broadcast my-stream.hang import ts ``` ### Capture a Webcam @@ -83,34 +102,24 @@ Pipe an external FFmpeg process as MPEG-TS: ```bash # macOS -ffmpeg -f avfoundation -i "0:0" -f mpegts - | moq-cli publish --url https://relay.example.com/anon --broadcast webcam ts +ffmpeg -f avfoundation -i "0:0" -f mpegts - | \ + moq --client-connect https://relay.example.com/anon --broadcast webcam.hang import ts # Linux -ffmpeg -f v4l2 -i /dev/video0 -f mpegts - | moq-cli publish --url https://relay.example.com/anon --broadcast webcam ts +ffmpeg -f v4l2 -i /dev/video0 -f mpegts - | \ + moq --client-connect https://relay.example.com/anon --broadcast webcam.hang import ts ``` -### Publish Screen +### Play a Broadcast -```bash -# macOS -ffmpeg -f avfoundation -i "1:" -f mpegts - | moq-cli publish --url https://relay.example.com/anon --broadcast screen ts +Pull a broadcast back out and play it: -# Linux (X11) -ffmpeg -f x11grab -i :0.0 -f mpegts - | moq-cli publish --url https://relay.example.com/anon --broadcast screen ts +```bash +moq --client-connect https://relay.example.com/anon --broadcast my-stream.hang export fmp4 | ffplay - ``` ## Encoding Options -### Custom Video Settings - -```bash -ffmpeg -i input.mp4 \ - -c:v libx264 -preset ultrafast -tune zerolatency \ - -b:v 2500k -maxrate 2500k -bufsize 5000k \ - -c:a aac -b:a 128k \ - -f mpegts - | moq-cli publish --url https://relay.example.com/anon --broadcast my-stream ts -``` - ### Low Latency Settings ```bash @@ -118,39 +127,29 @@ ffmpeg -i input.mp4 \ -c:v libx264 -preset ultrafast -tune zerolatency \ -g 30 -keyint_min 30 \ -c:a aac \ - -f mpegts - | moq-cli publish --url https://relay.example.com/anon --broadcast my-stream ts -``` - -### H.265/HEVC - -```bash -ffmpeg -i input.mp4 \ - -c:v libx265 -preset ultrafast \ - -c:a aac \ - -f mpegts - | moq-cli publish --url https://relay.example.com/anon --broadcast my-stream ts + -f mpegts - | moq --client-connect https://relay.example.com/anon --broadcast my-stream.hang import ts ``` ## Container Formats -`publish` selects its input container with a subcommand; `subscribe` selects its -output container with `--format`. +The container format is the endpoint subcommand: `import ` reads it from +stdin, `export ` writes it to stdout. -Publish (read from stdin unless noted): +Import formats: - `avc3` - raw H.264 Annex-B - `fmp4` - fragmented MP4 / CMAF - `ts` - MPEG-TS (H.264 / H.265 video; AAC, MP2, AC-3, or E-AC-3 audio) - `flv` - FLV / RTMP (H.264 video, AAC audio) -- `capture` - capture local devices directly (camera H.264 + microphone Opus; requires the `capture` build feature; does not read stdin) -Subscribe (`--format`): +Export formats: - `fmp4` - fragmented MP4 / CMAF - `mkv` - Matroska / WebM - `ts` - MPEG-TS - `flv` - FLV / RTMP (H.264 video, AAC audio) -`subscribe` also takes `--catalog` to pick which catalog track to read for track +`export` also takes `--catalog-format` to pick which catalog track to read for track discovery. When omitted, it's auto-detected from the broadcast name suffix (`.hang` -> `hang`, `.msf` -> `msf`), falling back to `hang`: @@ -158,33 +157,17 @@ discovery. When omitted, it's auto-detected from the broadcast name suffix - `hangz` - the DEFLATE-compressed `catalog.json.z` catalog (opt-in; shares the `.hang` suffix and is never auto-detected) - `msf` - the MSF `catalog` track -### HLS / LL-HLS - -Import an HLS master/media playlist into a MoQ broadcast: - -```bash -moq-cli hls --url https://relay.example.com/anon import \ - --broadcast my-stream.hang \ - --playlist https://example.com/live/master.m3u8 -``` - -Serve MoQ broadcasts as HLS / LL-HLS over HTTP: - -```bash -moq-cli hls --url https://relay.example.com/anon export --listen '[::]:8089' -``` - ### MPEG-TS Ingest an MPEG-TS stream from FFmpeg and play one back out: ```bash -# Publish: remux a file to MPEG-TS and pipe it in +# Import: remux a file to MPEG-TS and pipe it in ffmpeg -i input.mp4 -c copy -f mpegts - | \ - moq-cli publish --url https://relay.example.com --broadcast my-stream ts + moq --client-connect https://relay.example.com --broadcast my-stream.hang import ts -# Subscribe: pull MPEG-TS back out and play it -moq-cli subscribe --url https://relay.example.com --broadcast my-stream --format ts | ffplay - +# Export: pull MPEG-TS back out and play it +moq --client-connect https://relay.example.com --broadcast my-stream.hang export ts | ffplay - ``` TS export carries H.264 / H.265 as Annex-B and AAC as ADTS. Both in-band @@ -194,35 +177,102 @@ sources work: the parameter sets are read from the bitstream or the catalog Broadcast audio (MP2, AC-3, E-AC-3) is carried verbatim: complete, well-formed frames pass through byte-exact, never transcoded; malformed input is rejected -rather than mis-described. The catalog describes the codec honestly so a -subscriber that can decode it (typically TS gear) picks it up; browsers cannot -play these codecs and should skip the rendition. - -Elementary streams the CLI does not decode (SCTE-35 cues, teletext, DVB -subtitles, private data, ...) are carried verbatim too, one MoQ track per PID, -described in the catalog `mpegts` section. They survive `publish ts | relay | -subscribe --format ts` end-to-end with their original PIDs, PMT descriptors, and -PES stream\_ids, so a contribution feed keeps its ancillary streams. The relay -forwards them transparently and never parses the payload. +rather than mis-described. Elementary streams the CLI does not decode (SCTE-35 +cues, teletext, DVB subtitles, ...) are carried verbatim too, one MoQ track per +PID, described in the catalog `mpegts` section, and survive `import ts` / +`export ts` end-to-end. ### FLV -Ingest an FLV stream from FFmpeg and play one back out: - ```bash -# Publish: remux a file to FLV and pipe it in +# Import: remux a file to FLV and pipe it in ffmpeg -i input.mp4 -c copy -f flv - | \ - moq-cli publish --url https://relay.example.com --broadcast my-stream flv + moq --client-connect https://relay.example.com --broadcast my-stream.hang import flv -# Subscribe: pull FLV back out and play it -moq-cli subscribe --url https://relay.example.com --broadcast my-stream --format flv | ffplay - +# Export: pull FLV back out and play it +moq --client-connect https://relay.example.com --broadcast my-stream.hang export flv | ffplay - ``` -FLV is the classic RTMP container: H.264 video carried as length-prefixed NALU -with an out-of-band avcC, and AAC audio carried raw with an out-of-band -AudioSpecificConfig. Both pass straight through to the catalog `description`. The -enhanced E-RTMP FourCC payloads (HEVC, AV1, Opus) and the older codecs (VP6, MP3) -are not supported. +FLV is the classic RTMP container: H.264 video and AAC audio, each with an +out-of-band header. The enhanced E-RTMP FourCC payloads (HEVC, AV1, Opus) and the +older codecs (VP6, MP3) are not supported on the stdin/stdout container path. + +## HLS / LL-HLS + +Import a remote HLS master/media playlist into a MoQ broadcast: + +```bash +moq --client-connect https://relay.example.com/anon --broadcast my-stream.hang \ + import hls https://example.com/live/master.m3u8 +``` + +Serve one MoQ broadcast as HLS / LL-HLS over HTTP (reached at +`http://host:8089//master.m3u8`): + +```bash +moq --client-connect https://relay.example.com/anon --broadcast my-stream.hang \ + export hls --listen '[::]:8089' +``` + +## Network Gateways (RTMP / SRT / WebRTC) + +The `rtmp`, `srt`, and `rtc` endpoints bridge other live protocols. Each takes +either `--connect ` (dial out) or `--listen ` (bind a socket), and the +parent verb decides the role: + +- **import `--listen`** accepts pushes only (an RTMP/SRT publish, a WHIP publish). +- **export `--listen`** serves plays only (an RTMP/SRT play, a WHEP play). + +A listener is directional: an import listener rejects plays, and an export +listener rejects publishes. The operator declares the direction; the connecting +peer can't choose. + +Every gateway is scoped to the single `--broadcast` (required for a `--listen`): +a listener bridges only that broadcast, ignoring the RTMP app/key and SRT stream +id. (Multi-broadcast routing by app/key belongs behind a relay, via the gateway +libraries' auth-aware API.) + +### RTMP ingest to a relay + +Accept OBS / FFmpeg RTMP pushes and forward one broadcast to a relay: + +```bash +moq --client-connect https://relay.example.com/anon --broadcast my-stream.hang \ + import rtmp --listen '[::]:1935' +``` + +### Restream MoQ to Twitch (RTMP) + +Pull a broadcast from a relay and push it to a remote RTMP server: + +```bash +moq --client-connect https://relay.example.com/anon --broadcast my-stream.hang \ + export rtmp --connect 'rtmp://live.twitch.tv/app/' +``` + +### SRT + +```bash +# Accept incoming SRT publishes as one broadcast and forward to a relay +moq --client-connect https://relay.example.com/anon --broadcast my-stream.hang import srt --listen '[::]:9000' + +# Serve a broadcast to SRT players +moq --client-connect https://relay.example.com/anon --broadcast my-stream.hang export srt --listen '[::]:9000' +``` + +### WebRTC (WHIP / WHEP) + +Direction picks the HTTP role: import `--listen` is a WHIP server, export +`--listen` is a WHEP server. Peers reach the broadcast at +`http://host:8080/`. + +```bash +# WHIP ingest: browsers publish one broadcast to us, we forward to a relay +moq --client-connect https://relay.example.com/anon --broadcast my-stream.hang import rtc --listen '[::]:8080' + +# WHEP playback: serve a broadcast to browsers +moq --client-connect https://relay.example.com/anon --broadcast my-stream.hang export rtc --listen '[::]:8080' +``` ## Authentication @@ -230,7 +280,7 @@ Pass a JWT token via the URL's `?jwt=` query parameter: ```bash ffmpeg -i video.mp4 -c copy -f mpegts - | \ - moq-cli publish --url "https://relay.example.com/?jwt=" --broadcast my-stream ts + moq --client-connect "https://relay.example.com/?jwt=" --broadcast my-stream.hang import ts ``` See [Authentication](/bin/relay/auth) for token generation. @@ -247,25 +297,13 @@ just pub bbb https://relay.example.com/anon just pub tos https://relay.example.com/anon ``` -## Clock Synchronization - -Publish and subscribe to clock broadcasts for testing: - -```bash -# Publish a clock -just pub clock publish https://relay.example.com/anon - -# Subscribe to a clock -just pub clock subscribe https://relay.example.com/anon -``` - ## Debugging ### Verbose Output ```bash ffmpeg -i video.mp4 -c copy -f mpegts - | \ - RUST_LOG=debug moq-cli publish --url https://relay.example.com/anon --broadcast my-stream ts + RUST_LOG=debug moq --client-connect https://relay.example.com/anon --broadcast my-stream.hang import ts ``` ### Check Connection diff --git a/doc/bin/index.md b/doc/bin/index.md index 233b9c151..faf7663fa 100644 --- a/doc/bin/index.md +++ b/doc/bin/index.md @@ -26,7 +26,7 @@ Another tool does the encoding (ex. ffmpeg), making it easy to pipe any media in ```bash # Publish your webcam -ffmpeg -f avfoundation -i "0" -f mpegts - | moq-cli publish --url https://relay.example.com/anon --broadcast my-stream ts +ffmpeg -f avfoundation -i "0" -f mpegts - | moq --client-connect https://relay.example.com/anon --broadcast my-stream import ts ``` ## [OBS Plugin](/bin/obs) diff --git a/doc/bin/relay/auth.md b/doc/bin/relay/auth.md index ecd26223b..eb109a67a 100644 --- a/doc/bin/relay/auth.md +++ b/doc/bin/relay/auth.md @@ -278,7 +278,7 @@ logs a warning when the address is not loopback but does not refuse to start, so firewalling the port is your responsibility. ```bash -moq publish "tcp://127.0.0.1:4444/my-broadcast?jwt=$TOKEN" < video.mp4 +moq --client-connect "tcp://127.0.0.1:4444/my-broadcast.hang?jwt=$TOKEN" import fmp4 < video.mp4 ``` ### Unix socket (with a uid/gid/pid allowlist) @@ -307,7 +307,7 @@ read. A pid requirement rejects peers whose PID the platform doesn't report of the JWT, not a replacement for it. ```bash -moq publish "unix:///run/moq/internal.sock/my-broadcast?jwt=$TOKEN" < video.mp4 +moq --client-connect "unix:///run/moq/internal.sock/?jwt=$TOKEN" --broadcast my-broadcast.hang import fmp4 < video.mp4 ``` ### Notes diff --git a/doc/concept/layer/hang.md b/doc/concept/layer/hang.md index fde480738..f7a3b08b1 100644 --- a/doc/concept/layer/hang.md +++ b/doc/concept/layer/hang.md @@ -52,7 +52,7 @@ The catalog is published on two tracks with identical content: `catalog.json` (p A publisher always serves both; a consumer reads whichever it prefers and defaults to the uncompressed `catalog.json`. The compression is the group-scoped `deflate-raw` ([RFC 1951](https://www.rfc-editor.org/rfc/rfc1951.html)) stream used by `@moq/json` / `moq-json`, interoperable between the browser and native. -To read the compressed track, opt in explicitly: pass `--catalog hangz` to `moq-cli subscribe`, `CatalogFormat::HangZ` in Rust, or `catalogFormat: "hangz"` to `@moq/watch`. +To read the compressed track, opt in explicitly: pass `--catalog-format hangz` to `moq export`, `CatalogFormat::HangZ` in Rust, or `catalogFormat: "hangz"` to `@moq/watch`. The `.hang` broadcast suffix is unchanged: the compressed track is an extra track on the same broadcast, not a different broadcast name. ### Audio diff --git a/doc/concept/standard/interop.md b/doc/concept/standard/interop.md index fa82e51d2..4c4efefa8 100644 --- a/doc/concept/standard/interop.md +++ b/doc/concept/standard/interop.md @@ -27,13 +27,13 @@ A test pattern plus tone, so you don't need a media file: ffmpeg -re -f lavfi -i testsrc=size=1280x720:rate=30 -f lavfi -i sine=frequency=440 \ -c:v libx264 -preset ultrafast -tune zerolatency -g 60 -c:a aac \ -f mp4 -movflags cmaf+frag_keyframe+empty_moov+default_base_moof - \ -| moq-cli publish --url https://your-relay.example.com --broadcast bbb.hang fmp4 +| moq --client-connect https://your-relay.example.com --broadcast bbb.hang import fmp4 ``` ## Subscribe ```bash -moq-cli subscribe --url https://your-relay.example.com --broadcast bbb.hang --format fmp4 | ffplay - +moq --client-connect https://your-relay.example.com --broadcast bbb.hang export fmp4 | ffplay - ``` If it plays, you interop. That's the whole test. diff --git a/doc/lib/rs/crate/hang.md b/doc/lib/rs/crate/hang.md index b783a4047..c7fc30829 100644 --- a/doc/lib/rs/crate/hang.md +++ b/doc/lib/rs/crate/hang.md @@ -134,14 +134,14 @@ cargo install moq-cli # Publish a video file (remux to MPEG-TS and pipe it in) ffmpeg -i input.mp4 -c copy -f mpegts - | \ - moq-cli publish --url https://relay.example.com/anon --broadcast my-stream ts + moq --client-connect https://relay.example.com/anon --broadcast my-stream import ts # Publish from FFmpeg ffmpeg -i input.mp4 -f mpegts - | \ - moq-cli publish --url https://relay.example.com/anon --broadcast my-stream ts + moq --client-connect https://relay.example.com/anon --broadcast my-stream import ts ``` -See `moq-cli --help` for all options, or [FFmpeg documentation](/bin/cli). +See `moq --help` for all options, or [FFmpeg documentation](/bin/cli). ## API Reference diff --git a/doc/lib/rs/crate/moq-mux.md b/doc/lib/rs/crate/moq-mux.md index 07f3669fa..5e6bd6511 100644 --- a/doc/lib/rs/crate/moq-mux.md +++ b/doc/lib/rs/crate/moq-mux.md @@ -87,11 +87,11 @@ cargo install moq-cli # Publish a video file (remux to MPEG-TS and pipe it in) ffmpeg -i input.mp4 -c copy -f mpegts - | \ - moq-cli publish --url https://relay.example.com/anon --broadcast my-stream ts + moq --client-connect https://relay.example.com/anon --broadcast my-stream import ts # Publish from FFmpeg ffmpeg -i input.mp4 -f mpegts - | \ - moq-cli publish --url https://relay.example.com/anon --broadcast my-stream ts + moq --client-connect https://relay.example.com/anon --broadcast my-stream import ts ``` ## Next Steps diff --git a/doc/lib/rs/index.md b/doc/lib/rs/index.md index 8a0db2393..a2e726787 100644 --- a/doc/lib/rs/index.md +++ b/doc/lib/rs/index.md @@ -124,11 +124,11 @@ cargo install moq-cli ```bash # Publish a video file (remux to MPEG-TS and pipe it in) ffmpeg -i input.mp4 -c copy -f mpegts - | \ - moq-cli publish --url https://relay.example.com/anon --broadcast my-stream ts + moq --client-connect https://relay.example.com/anon --broadcast my-stream import ts # Publish from FFmpeg ffmpeg -i input.mp4 -f mpegts - | \ - moq-cli publish --url https://relay.example.com/anon --broadcast my-stream ts + moq --client-connect https://relay.example.com/anon --broadcast my-stream import ts ``` [Learn more](/bin/cli) diff --git a/nix/overlay.nix b/nix/overlay.nix index 8f8e34371..8c514b0a5 100644 --- a/nix/overlay.nix +++ b/nix/overlay.nix @@ -57,6 +57,8 @@ let moqCliArgs = crateInfo ../rs/moq-cli/Cargo.toml // { src = craneLib.cleanCargoSource ../.; cargoExtraArgs = "-p moq-cli"; + # The crate is `moq-cli`, but its `[[bin]]` ships as `moq`. + meta.mainProgram = "moq"; }; moqTokenCliArgs = crateInfo ../rs/moq-token-cli/Cargo.toml // { diff --git a/rs/moq-cli/Cargo.toml b/rs/moq-cli/Cargo.toml index dd0154b3f..b6634b967 100644 --- a/rs/moq-cli/Cargo.toml +++ b/rs/moq-cli/Cargo.toml @@ -5,13 +5,18 @@ authors = ["Luke Curley "] repository = "https://github.com/moq-dev/moq" license = "MIT OR Apache-2.0" -version = "0.7.35" +version = "0.8.0" edition = "2024" rust-version.workspace = true keywords = ["quic", "http3", "webtransport", "media", "live"] categories = ["multimedia", "network-programming", "web-programming"] +# The package is `moq-cli`, but the binary it ships is `moq`. +[[bin]] +name = "moq" +path = "src/main.rs" + [features] default = ["iroh", "quinn", "websocket"] iroh = ["moq-native/iroh"] @@ -30,9 +35,12 @@ humantime = "2.3" moq-hls = { workspace = true, features = ["server"] } moq-mux = { workspace = true } moq-native = { workspace = true, default-features = false, features = ["aws-lc-rs"] } +moq-rtc = { workspace = true } +moq-rtmp = { workspace = true } +moq-srt = { workspace = true } rustls = { version = "0.23", features = ["aws-lc-rs"], default-features = false } tokio = { workspace = true, features = ["full"] } -tower-http = { version = "0.7", features = ["cors", "fs"] } +tower-http = { version = "0.7", features = ["cors"] } tracing = "0.1" url = "2" diff --git a/rs/moq-cli/README.md b/rs/moq-cli/README.md index 22b6f037d..6e413e172 100644 --- a/rs/moq-cli/README.md +++ b/rs/moq-cli/README.md @@ -19,40 +19,40 @@ Multi-arch images (`linux/amd64` and `linux/arm64`) are published to [Docker Hub ## Usage -`moq-cli` reads media from stdin (or writes media to stdout) and exchanges it with a MoQ relay. Pick a subcommand based on whether you want to publish or subscribe, and whether your relay is local or remote. +`moq-cli` routes one endpoint onto a shared MoQ Origin: `moq `. The MoQ side (before the verb) is either `--client-connect ` (dial a relay) or `--server-bind ` (self-host). `import` moves media into MoQ, `export` moves it out. The endpoint is a container format (`fmp4`, `ts`, `flv`, ... read from stdin / written to stdout), or a gateway (`hls`, `rtmp`, `srt`, `rtc`). ### Publish to a remote relay ```bash ffmpeg -i input.mp4 -f mp4 -movflags cmaf - | \ - moq-cli publish --url https://relay.example.com --broadcast my-stream fmp4 + moq --client-connect https://relay.example.com --broadcast my-stream.hang import fmp4 ``` ### Subscribe from a remote relay ```bash -moq-cli subscribe --url https://relay.example.com --broadcast my-stream --format fmp4 | \ +moq --client-connect https://relay.example.com --broadcast my-stream.hang export fmp4 | \ ffplay - ``` -### Self-host: publish into a local relay (`serve`) +### Self-host: publish into a local relay -Runs a relay and publishes a single broadcast read from stdin into it. Useful for local testing without a separate relay process. +Hosts a MoQ server and publishes a single broadcast read from stdin into it. Useful for local testing without a separate relay process. ```bash ffmpeg -i input.mp4 -f mp4 -movflags cmaf - | \ - moq-cli serve --broadcast my-stream --tls-generate "localhost" fmp4 + moq --server-bind '[::]:4443' --tls-generate localhost --broadcast my-stream.hang import fmp4 ``` -### Self-host: subscribe to an inbound broadcast (`accept`) +### Self-host: subscribe to an inbound broadcast -Runs a relay and writes the first incoming broadcast's media to stdout. The inverse of `serve`. +Hosts a MoQ server and writes an incoming broadcast's media to stdout. The inverse of the above. ```bash -moq-cli accept --broadcast my-stream --tls-generate "localhost" --format fmp4 | ffplay - +moq --server-bind '[::]:4443' --tls-generate localhost --broadcast my-stream.hang export fmp4 | ffplay - ``` -### Input formats (`publish` / `serve`) +### Import formats - `avc3` raw H.264 Annex-B from stdin - `fmp4` fragmented MP4 from stdin diff --git a/rs/moq-cli/src/args.rs b/rs/moq-cli/src/args.rs new file mode 100644 index 000000000..835a0f75f --- /dev/null +++ b/rs/moq-cli/src/args.rs @@ -0,0 +1,197 @@ +//! The unified moq-cli argument surface. +//! +//! Grammar: `moq [endpoint opts]`. +//! +//! - The MoQ side (`--client-connect` / `--server-bind`, both optional, at least +//! one) attaches the shared Origin to the MoQ network, and comes before the +//! verb. Both may be given: dial a relay *and* accept incoming sessions. +//! - `import` routes media INTO MoQ from one source; `export` routes it OUT to +//! one sink. The verb fixes the data direction (and thus, for the +//! bidirectional gateways, whether `--connect`/`--listen` push or pull). +//! - The endpoint is one subcommand: a container format (`ts`, `fmp4`, ... read +//! from stdin on import, written to stdout on export) or a gateway (`hls`, +//! `rtmp`, `srt`, `rtc`). Exactly one per invocation, so "which endpoint" is +//! unambiguous and there's no silently-ignored flag. + +use std::time::Duration; + +use clap::{ArgGroup, Args, Parser, Subcommand}; +use url::Url; + +use crate::publish::PublishFormat; +use crate::subscribe::{CatalogFormatArg, SubscribeFormat}; + +/// moq-cli: a media router that wires one endpoint onto a shared MoQ Origin. +#[derive(Parser, Clone)] +#[command(name = "moq", version = env!("VERSION"))] +pub struct Cli { + /// Logging configuration. + #[command(flatten)] + pub log: moq_native::Log, + + /// The MoQ attachment, shared by both directions. + #[command(flatten)] + pub moq: MoqSide, + + /// The routing direction and endpoint. + #[command(subcommand)] + pub direction: Direction, +} + +/// The MoQ attachment. At least one of `--client-connect` / `--server-bind`; +/// both may be given at once. +#[derive(Args, Clone)] +#[command(group = ArgGroup::new("moq").required(true).multiple(true).args(["client-connect", "server-bind"]))] +pub struct MoqSide { + /// Dial a MoQ relay/server over WebTransport. + /// + /// The URL path is the relay auth path (e.g. `/anon` for a public relay); the + /// broadcast rides on top of it (via `--broadcast` or the endpoint). `?jwt=` + /// supplies a token. `http://` first fetches `/certificate.sha256` for the + /// (insecure) self-signed fingerprint; `https://` connects directly. + #[arg( + id = "client-connect", + long = "client-connect", + env = "MOQ_CLIENT_CONNECT", + help_heading = "MoQ" + )] + pub client_connect: Option, + + /// The broadcast name. Optional for the point endpoints (stdin/stdout, HLS + /// import, and the `--connect` dials), which default to the root broadcast at + /// the connection path; required by the `--listen` endpoints and `hls export`, + /// which bridge one named broadcast. + #[arg(long, alias = "name", help_heading = "MoQ")] + pub broadcast: Option, + + /// MoQ client transport config (`--client-bind`, `--client-tls-*`, ...). + #[command(flatten)] + pub client: moq_native::ClientConfig, + + /// MoQ server transport config (`--server-bind`, `--server-tls-*`, `--tls-*`). + #[command(flatten)] + pub server: moq_native::ServerConfig, + + /// Iroh transport config (`--iroh-*`), used by both the client and server. + #[cfg(feature = "iroh")] + #[command(flatten)] + pub iroh: moq_native::iroh::EndpointConfig, +} + +/// The data direction: the pivot between the MoQ side and the endpoint. +#[derive(Subcommand, Clone)] +pub enum Direction { + /// Route media INTO MoQ from one source. + Import(Import), + /// Route media OUT OF MoQ to one sink. + Export(Export), +} + +// ------------------------------------------------------------------ import + +/// import = one source -> MoQ. +#[derive(Args, Clone)] +pub struct Import { + /// The single source feeding the Origin. + #[command(subcommand)] + pub source: ImportSource, +} + +/// The single source feeding the Origin on an import. The container formats read +/// from stdin; the gateways bridge another protocol. +#[derive(Subcommand, Clone)] +pub enum ImportSource { + /// Raw H.264 Annex-B from stdin. + Avc3, + /// Fragmented MP4 / CMAF from stdin. + Fmp4, + /// MPEG-TS from stdin. + Ts, + /// FLV / RTMP container from stdin. + Flv, + /// Pull a remote HLS / LL-HLS playlist (http/https URL or local file) into MoQ. + Hls(crate::hls::ImportArgs), + /// RTMP: pull a remote play (`--connect`) or accept incoming publishes (`--listen`). + Rtmp(crate::rtmp::Args), + /// SRT: pull a remote stream (`--connect`) or accept incoming publishes (`--listen`). + Srt(crate::srt::Args), + /// WebRTC: WHEP client pulling a remote (`--connect`) or WHIP server accepting publishes (`--listen`). + Rtc(crate::rtc::Args), +} + +impl ImportSource { + /// The stdin container format, when this source is one of the container formats. + pub fn stdin_format(&self) -> Option { + Some(match self { + Self::Avc3 => PublishFormat::Avc3, + Self::Fmp4 => PublishFormat::Fmp4, + Self::Ts => PublishFormat::Ts, + Self::Flv => PublishFormat::Flv, + _ => return None, + }) + } +} + +// ------------------------------------------------------------------ export + +/// export = MoQ -> one sink. +#[derive(Args, Clone)] +pub struct Export { + /// Maximum latency before skipping groups (e.g. `500ms`, `1s`), for the stdout + /// container formats. The gateways (`hls`, `srt`, ...) have their own latency + /// controls. + #[arg(long = "latency-max", default_value = "500ms", value_parser = humantime::parse_duration)] + pub latency_max: Duration, + + /// Catalog format to read for track discovery (default: detect from the broadcast suffix). + #[arg(long = "catalog-format")] + pub catalog_format: Option, + + /// The single sink draining the Origin. + #[command(subcommand)] + pub sink: ExportSink, +} + +/// The single sink draining the Origin on an export. The container formats write +/// to stdout; the gateways bridge another protocol. +#[derive(Subcommand, Clone)] +pub enum ExportSink { + /// Fragmented MP4 / CMAF to stdout. + Fmp4(Fragmented), + /// Matroska / WebM to stdout. + Mkv(Fragmented), + /// MPEG-TS to stdout. + Ts, + /// FLV / RTMP container to stdout. + Flv, + /// Serve HLS / LL-HLS over HTTP. + Hls(crate::hls::ExportArgs), + /// RTMP: push to a remote (`--connect`) or serve plays (`--listen`). + Rtmp(crate::rtmp::Args), + /// SRT: push to a remote (`--connect`) or serve requests (`--listen`). + Srt(crate::srt::Args), + /// WebRTC: WHIP client pushing to a remote (`--connect`) or WHEP server serving plays (`--listen`). + Rtc(crate::rtc::Args), +} + +impl ExportSink { + /// The stdout container format and its fragment cap, when this sink writes to + /// stdout (the container formats). The fragment cap is fmp4/mkv-only. + pub fn stdout(&self) -> Option<(SubscribeFormat, Option)> { + Some(match self { + Self::Fmp4(args) => (SubscribeFormat::Fmp4, args.fragment_duration), + Self::Mkv(args) => (SubscribeFormat::Mkv, args.fragment_duration), + Self::Ts => (SubscribeFormat::Ts, None), + Self::Flv => (SubscribeFormat::Flv, None), + _ => return None, + }) + } +} + +/// Fragmenting option shared by the fmp4 / mkv stdout containers. +#[derive(Args, Clone)] +pub struct Fragmented { + /// Cap the output fragment/cluster duration (e.g. `2s`). Default: one GOP. + #[arg(long, value_parser = humantime::parse_duration)] + pub fragment_duration: Option, +} diff --git a/rs/moq-cli/src/client.rs b/rs/moq-cli/src/client.rs deleted file mode 100644 index 64a179dff..000000000 --- a/rs/moq-cli/src/client.rs +++ /dev/null @@ -1,24 +0,0 @@ -use crate::Publish; - -use hang::moq_net; -use url::Url; - -pub async fn run_client(client: moq_native::Client, url: Url, name: String, publish: Publish) -> anyhow::Result<()> { - // Create an origin producer to publish to the broadcast. - let origin = moq_net::Origin::random().produce(); - origin.publish_broadcast(&name, publish.consume()); - - tracing::info!(%url, %name, "connecting"); - - let reconnect = client.with_publish(origin.consume()).reconnect(url); - - #[cfg(unix)] - // Notify systemd that we're ready. - let _ = sd_notify::notify(&[sd_notify::NotifyState::Ready]); - - tokio::select! { - res = publish.run() => res, - res = reconnect.closed() => Ok(res?), - _ = tokio::signal::ctrl_c() => Ok(()), - } -} diff --git a/rs/moq-cli/src/hls.rs b/rs/moq-cli/src/hls.rs new file mode 100644 index 000000000..0d58e454d --- /dev/null +++ b/rs/moq-cli/src/hls.rs @@ -0,0 +1,89 @@ +//! HLS / LL-HLS endpoints: pull a remote playlist into MoQ (import), or serve +//! HLS over HTTP from MoQ broadcasts (export). + +use std::net::SocketAddr; +use std::time::Duration; + +use anyhow::Context; +use hang::moq_net; +use hang::moq_net::AsPath; +use tower_http::cors::{Any, CorsLayer}; + +use crate::moq::notify_ready; + +/// HLS import (pull a remote playlist) args. +#[derive(clap::Args, Clone)] +pub struct ImportArgs { + /// Playlist URL (http/https) or local file path. + pub playlist: String, +} + +/// HLS export (serve over HTTP) args. +#[derive(clap::Args, Clone)] +pub struct ExportArgs { + /// HTTP listener for the HLS endpoints. + #[arg(long, default_value = "[::]:8089")] + pub listen: SocketAddr, + + /// TLS certificates, keys, self-signed generation, and optional mTLS roots. + #[command(flatten)] + pub tls: moq_native::tls::Server, + + /// LL-HLS part target duration. + #[arg(long, default_value = "500ms", value_parser = humantime::parse_duration)] + pub part_target: Duration, + + /// Minimum media kept in each rendition's sliding window. + #[arg(long, default_value = "16s", value_parser = humantime::parse_duration)] + pub window: Duration, +} + +/// Pull a remote HLS/LL-HLS playlist (URL or file path) into the Origin under `name`. +pub async fn import(origin: &moq_net::OriginProducer, name: String, playlist: String) -> anyhow::Result<()> { + let mut producer = moq_net::Broadcast::new().produce(); + anyhow::ensure!( + origin.publish_broadcast(&name, producer.consume()), + "failed to publish broadcast" + ); + + let catalog = moq_mux::catalog::Producer::new(&mut producer)?; + let mut importer = moq_hls::import::Import::new(producer, catalog, moq_hls::import::Config::new(playlist))?; + + tracing::info!(%name, "importing HLS"); + + importer.init().await?; + notify_ready(); + Ok(importer.run().await?) +} + +/// Serve HLS/LL-HLS over HTTP for the single broadcast `name` (reached at +/// `//master.m3u8`); other broadcasts in the Origin are not served. +pub async fn export(origin: moq_net::OriginConsumer, args: ExportArgs, name: String) -> anyhow::Result<()> { + let scoped = origin + .scope(&[name.as_path()]) + .with_context(|| format!("failed to scope origin to broadcast `{name}`"))?; + + let config = moq_hls::export::Config { + part_target: args.part_target, + window: args.window, + ..Default::default() + }; + let server = moq_hls::Server::new(scoped, config); + let app = server + .router() + .layer(CorsLayer::new().allow_origin(Any).allow_methods(Any).allow_headers(Any)); + + let tls = if args.tls.cert.is_empty() && args.tls.generate.is_empty() { + None + } else { + let alpn = vec![b"h2".to_vec(), b"http/1.1".to_vec()]; + Some(args.tls.server_config(alpn)?) + }; + + let listener = moq_native::bind::tcp(args.listen)?; + + tracing::info!(listen = %args.listen, "serving HLS"); + notify_ready(); + + crate::web::serve(listener, app, tls).await +} diff --git a/rs/moq-cli/src/main.rs b/rs/moq-cli/src/main.rs index 152330d65..4edce1446 100644 --- a/rs/moq-cli/src/main.rs +++ b/rs/moq-cli/src/main.rs @@ -1,163 +1,49 @@ -mod client; +//! moq-cli: a media router that wires one endpoint onto a shared MoQ Origin. +//! +//! The binary is `moq`. See [`args`] for the `import`/`export` command grammar; +//! this module orchestrates the shared Origin and spawns the MoQ side plus the +//! selected endpoint. + +mod args; +mod hls; +mod moq; mod publish; -mod server; +mod rtc; +mod rtmp; +mod srt; mod subscribe; mod web; -use client::*; +use args::{Cli, Direction, Export, ExportSink, Import, ImportSource, MoqSide}; use hang::moq_net; -use publish::*; -use server::*; -use subscribe::*; -use web::*; +use publish::Publish; +use subscribe::{Subscribe, SubscribeArgs}; -use clap::{Parser, Subcommand}; -use std::net::SocketAddr; -use std::path::PathBuf; -use std::sync::Arc; -use std::time::Duration; -use tower_http::cors::{Any, CorsLayer}; -use url::Url; +use clap::Parser; +use tokio::task::JoinSet; -#[derive(Parser, Clone)] -#[command(version = env!("VERSION"))] -pub struct Cli { - #[command(flatten)] - log: moq_native::Log, - - /// Iroh configuration - #[command(flatten)] +/// Everything needed to build MoQ clients/servers, encapsulating the optional +/// iroh endpoint so the rest of the code is feature-agnostic. +#[derive(Clone)] +struct Net { #[cfg(feature = "iroh")] - iroh: moq_native::iroh::EndpointConfig, - - #[command(subcommand)] - command: Command, -} - -#[derive(Subcommand, Clone)] -pub enum Command { - /// Run a relay and publish a single broadcast read from stdin into it. - Serve { - #[command(flatten)] - config: moq_native::ServerConfig, - - /// The name of the broadcast to serve. - #[arg(long, alias = "name")] - broadcast: String, - - /// Optionally serve static files from the given directory. - #[arg(long)] - dir: Option, - - /// The format of the input media. - #[command(subcommand)] - format: PublishFormat, - }, - /// Run a relay and write the first incoming broadcast's media to stdout. - Accept { - #[command(flatten)] - config: moq_native::ServerConfig, - - /// The name of the broadcast to accept. - #[arg(long, alias = "name")] - broadcast: String, - - /// Optionally serve static files from the given directory. - #[arg(long)] - dir: Option, - - #[command(flatten)] - args: SubscribeArgs, - }, - /// Publish a broadcast read from stdin to a remote relay. - Publish { - /// The MoQ client configuration. - #[command(flatten)] - config: moq_native::ClientConfig, - - /// The URL of the MoQ server. - /// - /// The URL must start with `https://` or `http://`. - /// - If `http` is used, a HTTP fetch to "/certificate.sha256" is first made to get the TLS certificiate fingerprint (insecure). - /// - If `https` is used, then A WebTransport connection is made via QUIC to the provided host/port. - /// - /// The `?jwt=` query parameter is used to provide a JWT token from moq-token-cli. - /// Otherwise, the public path (if any) is used instead. - /// - /// The path currently must be `/` or you'll get an error on connect. - #[arg(long)] - url: Url, - - /// The name of the broadcast to publish. - #[arg(long, alias = "name")] - broadcast: String, - - /// The format of the input media. - #[command(subcommand)] - format: PublishFormat, - }, - /// Subscribe to a broadcast on a remote relay and write the media to stdout. - Subscribe { - /// The MoQ client configuration. - #[command(flatten)] - config: moq_native::ClientConfig, - - /// The URL of the MoQ server. - #[arg(long)] - url: Url, - - /// The name of the broadcast to subscribe to. - #[arg(long, alias = "name")] - broadcast: String, - - #[command(flatten)] - args: SubscribeArgs, - }, - /// Import or export HLS / LL-HLS via a MoQ relay. - Hls { - /// The MoQ client configuration. - #[command(flatten)] - config: moq_native::ClientConfig, - - /// The URL of the MoQ server. - #[arg(long, alias = "relay", env = "MOQ_HLS_RELAY")] - url: Url, - - #[command(subcommand)] - command: HlsCommand, - }, + iroh: Option, } -#[derive(Subcommand, Clone)] -pub enum HlsCommand { - /// Serve HLS / LL-HLS over HTTP from MoQ broadcasts. - Export { - /// HTTP listener for the HLS endpoints. - #[arg(long, env = "MOQ_HLS_LISTEN", default_value = "[::]:8089")] - listen: SocketAddr, - - /// TLS certificates, keys, self-signed generation, and optional mTLS roots. - #[command(flatten)] - tls: moq_native::tls::Server, - - /// LL-HLS part target duration. - #[arg(long, env = "MOQ_HLS_PART_TARGET", default_value = "500ms", value_parser = humantime::parse_duration)] - part_target: Duration, - - /// Minimum duration of media kept in each rendition's sliding window. - #[arg(long, env = "MOQ_HLS_WINDOW", default_value = "16s", value_parser = humantime::parse_duration)] - window: Duration, - }, - /// Pull a remote HLS master/media playlist and publish it into MoQ. - Import { - /// Broadcast name to publish on the relay. - #[arg(long, alias = "name", env = "MOQ_HLS_BROADCAST")] - broadcast: String, +impl Net { + fn client(&self, config: moq_native::ClientConfig) -> anyhow::Result { + let client = config.init()?; + #[cfg(feature = "iroh")] + let client = client.with_iroh(self.iroh.clone()); + Ok(client) + } - /// Remote HLS playlist URL (http/https) or local file path. - #[arg(long, env = "MOQ_HLS_PLAYLIST")] - playlist: String, - }, + fn server(&self, config: moq_native::ServerConfig) -> anyhow::Result { + let server = config.init()?; + #[cfg(feature = "iroh")] + let server = server.with_iroh(self.iroh.clone()); + Ok(server) + } } #[tokio::main] @@ -171,268 +57,212 @@ async fn main() -> anyhow::Result<()> { let cli = Cli::parse(); cli.log.init()?; - #[cfg(feature = "iroh")] - let iroh = cli.iroh.bind().await?; - - match cli.command { - Command::Serve { - config, - dir, - broadcast, - format, - } => { - warn_if_missing_format(&broadcast); - let publish = Publish::new(&format)?; - let web_bind = config.bind.clone().unwrap_or_else(|| "[::]:443".to_string()); + let net = Net { + #[cfg(feature = "iroh")] + iroh: cli.moq.iroh.clone().bind().await?, + }; - let server = config.init()?; - #[cfg(feature = "iroh")] - let server = server.with_iroh(iroh); + match cli.direction { + Direction::Import(import) => run_import(cli.moq, import, net).await, + Direction::Export(export) => run_export(cli.moq, export, net).await, + } +} - let web_tls = server.tls_info(); +/// Route one source INTO the shared Origin, exposing it to the MoQ network. +async fn run_import(moq: MoqSide, import: Import, net: Net) -> anyhow::Result<()> { + let origin = moq_net::Origin::random().produce(); + // The broadcast defaults to "": MoQ names each broadcast by the connection + // path plus any explicit `--broadcast`, so an unset name is the root broadcast. + let name = moq.broadcast.clone().unwrap_or_default(); + let mut tasks: JoinSet> = JoinSet::new(); + + // MoQ side: publish the Origin outward. + if let Some(url) = moq.client_connect.clone() { + let client = net.client(moq.client.clone())?; + let origin = origin.clone(); + tasks.spawn(async move { moq::client_import(client, url, &origin).await }); + } + if let Some(web_bind) = moq.server.bind.clone() { + let server = net.server(moq.server.clone())?; + let tls_info = server.tls_info(); + tasks.spawn(moq::server_import(server, origin.clone())); + tasks.spawn(async move { web::run_web(&web_bind, tls_info).await }); + } - tokio::select! { - res = run_server(server, broadcast, publish.consume()) => res, - res = run_web(&web_bind, web_tls, dir) => res, - res = publish.run() => res, + // Foreign side: the single source. + if let Some(format) = import.source.stdin_format() { + warn_if_missing_format(&name); + let publish = Publish::new(&format)?; + anyhow::ensure!( + origin.publish_broadcast(&name, publish.consume()), + "failed to publish broadcast" + ); + tasks.spawn(async move { publish.run().await }); + } else { + match import.source { + ImportSource::Hls(hls) => { + warn_if_missing_format(&name); + let origin = origin.clone(); + tasks.spawn(async move { hls::import(&origin, name, hls.playlist).await }); } - } - Command::Accept { - config, - broadcast, - dir, - args, - } => { - let web_bind = config.bind.clone().unwrap_or_else(|| "[::]:443".to_string()); - - let server = config.init()?; - #[cfg(feature = "iroh")] - let server = server.with_iroh(iroh); - - let web_tls = server.tls_info(); - - let origin = moq_net::Origin::random().produce(); - let consumer = origin.consume(); - - tokio::select! { - res = run_accept(server, origin) => res, - res = run_web(&web_bind, web_tls, dir) => res, - res = run_announced_subscribe(consumer, broadcast, args) => res, - _ = tokio::signal::ctrl_c() => Ok(()), + ImportSource::Rtmp(rtmp) => { + if let Some(addr) = rtmp.listen { + let name = require_broadcast(name, "import rtmp --listen")?; + tasks.spawn(rtmp::listen_import(origin.clone(), addr, name)); + } else if let Some(url) = rtmp.connect { + tasks.spawn(rtmp::connect_import(origin.clone(), url, name)); + } } - } - Command::Publish { - config, - url, - broadcast, - format, - } => { - warn_if_missing_format(&broadcast); - let publish = Publish::new(&format)?; - let client = config.init()?; - - #[cfg(feature = "iroh")] - let client = client.with_iroh(iroh); - - run_client(client, url, broadcast, publish).await - } - Command::Subscribe { - config, - url, - broadcast, - args, - } => { - let client = config.init()?; - - #[cfg(feature = "iroh")] - let client = client.with_iroh(iroh); - - run_subscribe(client, url, broadcast, args).await - } - Command::Hls { config, url, command } => { - let client = config.init()?; - - #[cfg(feature = "iroh")] - let client = client.with_iroh(iroh); - - match command { - HlsCommand::Export { - listen, - tls, - part_target, - window, - } => run_hls_export(client, url, listen, tls, part_target, window).await, - HlsCommand::Import { broadcast, playlist } => run_hls_import(client, url, broadcast, playlist).await, + ImportSource::Srt(srt) => { + if let Some(addr) = srt.listen { + let name = require_broadcast(name, "import srt --listen")?; + tasks.spawn(srt::listen_import(origin.clone(), addr, name, srt.latency)); + } else if let Some(url) = srt.connect { + tasks.spawn(srt::connect_import(origin.clone(), url, name, srt.latency)); + } } + ImportSource::Rtc(rtc) => { + if let Some(addr) = rtc.listen { + let name = require_broadcast(name, "import rtc --listen")?; + tasks.spawn(rtc::listen_import( + origin.clone(), + addr, + rtc.udp_bind, + rtc.public_addr, + name, + )); + } else if let Some(url) = rtc.connect { + tasks.spawn(rtc::connect_import(origin.clone(), url, name)); + } + } + _ => unreachable!("container formats are handled by stdin_format above"), } } -} -fn warn_if_missing_format(name: &str) { - if moq_mux::catalog::CatalogFormat::detect(name).is_none() { - tracing::warn!( - name, - "You should append .hang to your broadcast name to make the catalog format explicit." - ); - } + drive(tasks).await } -async fn run_subscribe( - client: moq_native::Client, - url: Url, - broadcast: String, - args: SubscribeArgs, -) -> anyhow::Result<()> { +/// Route the shared Origin OUT to one sink, filling it from the MoQ network. +async fn run_export(moq: MoqSide, export: Export, net: Net) -> anyhow::Result<()> { let origin = moq_net::Origin::random().produce(); - let consumer = origin.consume(); - - tracing::info!(%url, %broadcast, "connecting"); - - let reconnect = client.with_consume(origin).reconnect(url); - - #[cfg(unix)] - let _ = sd_notify::notify(&[sd_notify::NotifyState::Ready]); - - tokio::select! { - res = run_announced_subscribe(consumer, broadcast, args) => res, - res = reconnect.closed() => Ok(res?), - _ = tokio::signal::ctrl_c() => Ok(()), + // The broadcast defaults to "": MoQ names each broadcast by the connection + // path plus any explicit `--broadcast`, so an unset name is the root broadcast. + let name = moq.broadcast.clone().unwrap_or_default(); + let mut tasks: JoinSet> = JoinSet::new(); + + // MoQ side: fill the Origin. + if let Some(url) = moq.client_connect.clone() { + let client = net.client(moq.client.clone())?; + let origin = origin.clone(); + tasks.spawn(async move { moq::client_export(client, url, origin).await }); + } + if let Some(web_bind) = moq.server.bind.clone() { + let server = net.server(moq.server.clone())?; + let tls_info = server.tls_info(); + tasks.spawn(moq::server_export(server, origin.clone())); + tasks.spawn(async move { web::run_web(&web_bind, tls_info).await }); } -} - -async fn run_announced_subscribe( - consumer: moq_net::OriginConsumer, - broadcast: String, - args: SubscribeArgs, -) -> anyhow::Result<()> { - let catalog = args.catalog_format(&broadcast); - - let consumer = consumer - .announced_broadcast(&broadcast) - .await - .ok_or_else(|| anyhow::anyhow!("origin closed before broadcast was announced"))?; - - Subscribe::new(consumer, catalog, args).run().await -} - -async fn run_hls_export( - client: moq_native::Client, - url: Url, - listen: SocketAddr, - tls: moq_native::tls::Server, - part_target: Duration, - window: Duration, -) -> anyhow::Result<()> { - let subscriber = moq_net::Origin::random().produce(); - let consumer = subscriber.consume(); - let reconnect = client.with_consume(subscriber).reconnect(url.clone()); - - let config = moq_hls::export::Config { - part_target, - window, - ..Default::default() - }; - let server = moq_hls::Server::new(consumer, config); - let app = server - .router() - .layer(CorsLayer::new().allow_origin(Any).allow_methods(Any).allow_headers(Any)); - let tls = if tls.cert.is_empty() && tls.generate.is_empty() { - None + // Foreign side: the single sink. + if let Some((format, fragment_duration)) = export.sink.stdout() { + let args = SubscribeArgs { + format, + max_latency: export.latency_max, + fragment_duration, + catalog: export.catalog_format, + }; + let consumer = origin.consume(); + tasks.spawn(async move { run_stdout(consumer, name, args).await }); } else { - let alpn = vec![b"h2".to_vec(), b"http/1.1".to_vec()]; - Some(tls.server_config(alpn)?) - }; - - let listener = moq_native::bind::tcp(listen)?; - - tracing::info!(%url, %listen, "serving HLS"); - - #[cfg(unix)] - let _ = sd_notify::notify(&[sd_notify::NotifyState::Ready]); - - tokio::select! { - res = serve_hls(listener, app, tls) => res, - res = reconnect.closed() => res.map_err(Into::into), - _ = shutdown_signal() => Ok(()), + match export.sink { + ExportSink::Hls(args) => { + let name = require_broadcast(name, "export hls")?; + tasks.spawn(hls::export(origin.consume(), args, name)); + } + ExportSink::Rtmp(rtmp) => { + if let Some(addr) = rtmp.listen { + let name = require_broadcast(name, "export rtmp --listen")?; + tasks.spawn(rtmp::listen_export(origin.consume(), addr, name)); + } else if let Some(url) = rtmp.connect { + tasks.spawn(rtmp::connect_export(origin.consume(), url, name)); + } + } + ExportSink::Srt(srt) => { + if let Some(addr) = srt.listen { + let name = require_broadcast(name, "export srt --listen")?; + tasks.spawn(srt::listen_export(origin.consume(), addr, name, srt.latency)); + } else if let Some(url) = srt.connect { + tasks.spawn(srt::connect_export(origin.consume(), url, name, srt.latency)); + } + } + ExportSink::Rtc(rtc) => { + if let Some(addr) = rtc.listen { + let name = require_broadcast(name, "export rtc --listen")?; + tasks.spawn(rtc::listen_export( + origin.consume(), + addr, + rtc.udp_bind, + rtc.public_addr, + name, + )); + } else if let Some(url) = rtc.connect { + tasks.spawn(rtc::connect_export(origin.consume(), url, name)); + } + } + _ => unreachable!("container formats are handled by stdout_format above"), + } } -} -async fn run_hls_import( - client: moq_native::Client, - url: Url, - broadcast: String, - playlist: String, -) -> anyhow::Result<()> { - warn_if_missing_format(&broadcast); - - let publisher = moq_net::Origin::random().produce(); - let reconnect = client.with_publish(publisher.consume()).reconnect(url.clone()); - - let mut producer = moq_net::Broadcast::new().produce(); - let consumer = producer.consume(); - anyhow::ensure!( - publisher.publish_broadcast(&broadcast, consumer), - "failed to publish broadcast" - ); - - let catalog = moq_mux::catalog::Producer::new(&mut producer)?; - let mut importer = moq_hls::import::Import::new(producer, catalog, moq_hls::import::Config::new(playlist))?; - - tracing::info!(%url, %broadcast, "importing HLS"); - - tokio::select! { - res = async { - importer.init().await?; + drive(tasks).await +} - #[cfg(unix)] - let _ = sd_notify::notify(&[sd_notify::NotifyState::Ready]); +/// Subscribe to `name` from the Origin and write it to stdout. +async fn run_stdout(consumer: moq_net::OriginConsumer, name: String, args: SubscribeArgs) -> anyhow::Result<()> { + let catalog = args.catalog_format(&name); + let broadcast = consumer + .announced_broadcast(&name) + .await + .ok_or_else(|| anyhow::anyhow!("origin closed before broadcast `{name}` was announced"))?; - importer.run().await - } => res.map_err(Into::into), - res = reconnect.closed() => res.map_err(Into::into), - _ = shutdown_signal() => Ok(()), - } + Subscribe::new(broadcast, catalog, args).run().await } -async fn serve_hls( - listener: std::net::TcpListener, - app: axum::Router, - tls: Option>, -) -> anyhow::Result<()> { - let service = app.into_make_service(); - match tls { - Some(config) => { - let config = axum_server::tls_rustls::RustlsConfig::from_config(config); - axum_server::from_tcp_rustls(listener, config)?.serve(service).await?; - } - None => { - axum_server::from_tcp(listener)?.serve(service).await?; +/// Run every endpoint until the first finishes (stdin EOF, Ctrl-C, or an error), +/// then drop the rest. +async fn drive(mut tasks: JoinSet>) -> anyhow::Result<()> { + tasks.spawn(async { + let _ = tokio::signal::ctrl_c().await; + Ok(()) + }); + + while let Some(res) = tasks.join_next().await { + match res { + Ok(Ok(())) => return Ok(()), + Ok(Err(err)) => return Err(err), + Err(err) if err.is_cancelled() => continue, + Err(err) => return Err(err.into()), } } + Ok(()) } -async fn shutdown_signal() { - #[cfg(unix)] - { - use tokio::signal::unix::{SignalKind, signal}; - - let mut term = match signal(SignalKind::terminate()) { - Ok(term) => term, - Err(_) => { - let _ = tokio::signal::ctrl_c().await; - return; - } - }; - tokio::select! { - _ = tokio::signal::ctrl_c() => {} - _ = term.recv() => {} - } - } +/// The listener / HTTP-serving endpoints bridge one named broadcast, so an +/// empty `--broadcast` is rejected rather than silently defaulting to the root. +fn require_broadcast(name: String, endpoint: &str) -> anyhow::Result { + anyhow::ensure!( + !name.is_empty(), + "`{endpoint}` requires a broadcast: pass --broadcast " + ); + Ok(name) +} - #[cfg(not(unix))] - { - let _ = tokio::signal::ctrl_c().await; +fn warn_if_missing_format(name: &str) { + // The empty (root) broadcast has no name to suffix, so there's nothing to warn about. + if !name.is_empty() && moq_mux::catalog::CatalogFormat::detect(name).is_none() { + tracing::warn!( + name, + "You should append .hang to your broadcast name to make the catalog format explicit." + ); } } diff --git a/rs/moq-cli/src/moq.rs b/rs/moq-cli/src/moq.rs new file mode 100644 index 000000000..1a410d0fa --- /dev/null +++ b/rs/moq-cli/src/moq.rs @@ -0,0 +1,83 @@ +//! The MoQ side of the router: attaching the shared Origin to the network via a +//! dialed client (`--client-connect`) and/or a hosted server (`--server-bind`). +//! +//! Direction determines data flow off the one shared Origin: +//! - import: MoQ publishes the Origin outward (client publishes to the relay; +//! each accepted server session serves subscribers). +//! - export: MoQ fills the Origin (client subscribes from the relay; each +//! accepted server session ingests a remote publish). + +use hang::moq_net; +use url::Url; + +/// Dial a relay and publish the Origin's broadcasts to it (import). +pub async fn client_import( + client: moq_native::Client, + url: Url, + origin: &moq_net::OriginProducer, +) -> anyhow::Result<()> { + let reconnect = client.with_publish(origin.consume()).reconnect(url); + notify_ready(); + Ok(reconnect.closed().await?) +} + +/// Dial a relay and subscribe its broadcasts into the Origin (export). +pub async fn client_export( + client: moq_native::Client, + url: Url, + origin: moq_net::OriginProducer, +) -> anyhow::Result<()> { + let reconnect = client.with_consume(origin).reconnect(url); + notify_ready(); + Ok(reconnect.closed().await?) +} + +/// Host a MoQ server; each accepted session serves the Origin to subscribers (import). +pub async fn server_import(mut server: moq_native::Server, origin: moq_net::OriginProducer) -> anyhow::Result<()> { + notify_ready(); + tracing::info!(addr = ?server.local_addr(), "listening"); + + while let Some(session) = server.accept().await { + let origin = origin.clone(); + tokio::spawn(async move { + if let Err(err) = serve_session(session, origin).await { + tracing::warn!(%err, "session ended with error"); + } + }); + } + + Ok(()) +} + +async fn serve_session(session: moq_native::Request, origin: moq_net::OriginProducer) -> anyhow::Result<()> { + let session = session.with_publish(origin.consume()).ok().await?; + Ok(session.closed().await?) +} + +/// Host a MoQ server; each accepted session ingests a remote publish into the Origin (export). +pub async fn server_export(mut server: moq_native::Server, origin: moq_net::OriginProducer) -> anyhow::Result<()> { + notify_ready(); + tracing::info!(addr = ?server.local_addr(), "listening"); + + while let Some(session) = server.accept().await { + let origin = origin.clone(); + tokio::spawn(async move { + if let Err(err) = accept_session(session, origin).await { + tracing::warn!(%err, "session ended with error"); + } + }); + } + + Ok(()) +} + +async fn accept_session(session: moq_native::Request, origin: moq_net::OriginProducer) -> anyhow::Result<()> { + let session = session.with_consume(origin).ok().await?; + Ok(session.closed().await?) +} + +/// Notify systemd (if any) that the endpoint is up. +pub fn notify_ready() { + #[cfg(unix)] + let _ = sd_notify::notify(&[sd_notify::NotifyState::Ready]); +} diff --git a/rs/moq-cli/src/publish.rs b/rs/moq-cli/src/publish.rs index 853774309..8db1e332d 100644 --- a/rs/moq-cli/src/publish.rs +++ b/rs/moq-cli/src/publish.rs @@ -1,14 +1,16 @@ -use clap::Subcommand; use hang::moq_net; use moq_mux::container::{flv, fmp4, ts}; -#[derive(Subcommand, Clone)] +/// Container format read from stdin on the import (source) side. +#[derive(Clone, Copy)] pub enum PublishFormat { + /// Raw AVC (H.264) Annex B elementary stream. Avc3, + /// Fragmented MP4 (CMAF). Fmp4, - /// MPEG-TS (transport stream) read from stdin. + /// MPEG-TS (transport stream). Ts, - /// FLV (Flash Video / RTMP) read from stdin. + /// FLV (Flash Video / RTMP). Flv, } diff --git a/rs/moq-cli/src/rtc.rs b/rs/moq-cli/src/rtc.rs new file mode 100644 index 000000000..df52dee7d --- /dev/null +++ b/rs/moq-cli/src/rtc.rs @@ -0,0 +1,125 @@ +//! WebRTC (WHIP/WHEP) endpoints. Direction picks the HTTP role: +//! - import `--listen` = WHIP server (accept publishes); `--connect` = WHEP client (pull). +//! - export `--listen` = WHEP server (serve plays); `--connect` = WHIP client (push). + +use std::net::SocketAddr; + +use anyhow::Context; +use hang::moq_net; +use hang::moq_net::AsPath; +use tower_http::cors::{Any, CorsLayer}; +use url::Url; + +use crate::moq::notify_ready; + +/// WebRTC endpoint args: exactly one of `--connect` (WHIP/WHEP client) / +/// `--listen` (WHIP/WHEP server). The parent direction picks WHIP vs WHEP. +#[derive(clap::Args, Clone)] +#[command(group = clap::ArgGroup::new("rtc-mode").required(true).multiple(false).args(["rtc-connect", "rtc-listen"]))] +pub struct Args { + /// Dial a remote WHIP/WHEP endpoint URL. + #[arg(id = "rtc-connect", long = "connect", value_name = "URL")] + pub connect: Option, + + /// Bind an HTTP listener for WHIP/WHEP, scoped to the single `--broadcast` + /// (peers reach it at `http://host:port/`). + #[arg(id = "rtc-listen", long = "listen", value_name = "ADDR")] + pub listen: Option, + + /// Shared UDP socket for ICE/media (one port for all sessions). + #[arg(long, requires = "rtc-listen", default_value = "[::]:0")] + pub udp_bind: SocketAddr, + + /// Public UDP address(es) advertised as ICE host candidates (repeatable). + #[arg(long, requires = "rtc-listen")] + pub public_addr: Vec, +} + +/// WHIP server: accept incoming WebRTC publishes into the Origin as `name` (import). +pub async fn listen_import( + origin: moq_net::OriginProducer, + listen: SocketAddr, + udp_bind: SocketAddr, + public_addr: Vec, + name: String, +) -> anyhow::Result<()> { + let publisher = scope_producer(&origin, &name)?; + let server = server(publisher, origin.consume(), udp_bind, public_addr); + serve(server.publish_router(), listen, "WHIP").await +} + +/// WHEP server: serve WebRTC plays of `name` from the Origin (export). +pub async fn listen_export( + origin: moq_net::OriginConsumer, + listen: SocketAddr, + udp_bind: SocketAddr, + public_addr: Vec, + name: String, +) -> anyhow::Result<()> { + let subscriber = origin + .scope(&[name.as_path()]) + .with_context(|| format!("failed to scope origin to broadcast `{name}`"))?; + // A WHEP server only reads; it still needs a publisher handle for the shared + // glue, so hand it an unused, empty Origin producer. + let publisher = moq_net::Origin::random().produce(); + let server = server(publisher, subscriber, udp_bind, public_addr); + serve(server.subscribe_router(), listen, "WHEP").await +} + +/// Restrict a producer to the single broadcast `name` so a WHIP peer can only publish it. +fn scope_producer(origin: &moq_net::OriginProducer, name: &str) -> anyhow::Result { + origin + .scope(&[name.as_path()]) + .with_context(|| format!("failed to scope origin to broadcast `{name}`")) +} + +/// WHEP client: pull a remote broadcast into the Origin under `name` (import). +pub async fn connect_import(origin: moq_net::OriginProducer, url: Url, name: String) -> anyhow::Result<()> { + let producer = moq_net::Broadcast::new().produce(); + anyhow::ensure!( + origin.publish_broadcast(&name, producer.consume()), + "failed to publish broadcast" + ); + + tracing::info!(%url, %name, "WHEP client pulling"); + notify_ready(); + + let client = moq_rtc::Client::new(moq_rtc::client::Config::default()); + Ok(client.subscribe(url, producer).await?) +} + +/// WHIP client: push a broadcast from the Origin to a remote (export). +pub async fn connect_export(origin: moq_net::OriginConsumer, url: Url, name: String) -> anyhow::Result<()> { + let broadcast = origin + .announced_broadcast(&name) + .await + .with_context(|| format!("origin closed before broadcast `{name}` was announced"))?; + + tracing::info!(%url, %name, "WHIP client pushing"); + notify_ready(); + + let client = moq_rtc::Client::new(moq_rtc::client::Config::default()); + Ok(client.publish(url, broadcast).await?) +} + +fn server( + publisher: moq_net::OriginProducer, + subscriber: moq_net::OriginConsumer, + udp_bind: SocketAddr, + public_addr: Vec, +) -> moq_rtc::Server { + let mut config = moq_rtc::server::Config::default(); + config.udp_bind = udp_bind; + config.ice_candidates = public_addr; + moq_rtc::Server::new(config, publisher, subscriber) +} + +async fn serve(router: axum::Router, listen: SocketAddr, role: &str) -> anyhow::Result<()> { + let app = router.layer(CorsLayer::new().allow_origin(Any).allow_methods(Any).allow_headers(Any)); + let listener = moq_native::bind::tcp(listen)?; + + tracing::info!(%listen, role, "serving WebRTC"); + notify_ready(); + + crate::web::serve(listener, app, None).await +} diff --git a/rs/moq-cli/src/rtmp.rs b/rs/moq-cli/src/rtmp.rs new file mode 100644 index 000000000..04d9692d6 --- /dev/null +++ b/rs/moq-cli/src/rtmp.rs @@ -0,0 +1,170 @@ +//! RTMP endpoints. Listeners are directional: an import listener accepts +//! publishes only (rejecting plays), an export listener serves plays only +//! (rejecting publishes). The operator declares direction; the peer can't choose. + +use std::net::SocketAddr; + +use anyhow::Context; +use hang::moq_net; +use moq_rtmp::{Client, Request, Server}; +use url::Url; + +use crate::moq::notify_ready; + +/// RTMP endpoint args: exactly one of `--connect` (dial) / `--listen` (bind). +/// The parent direction fixes whether that dial/bind pushes or pulls. +#[derive(clap::Args, Clone)] +#[command(group = clap::ArgGroup::new("rtmp-mode").required(true).multiple(false).args(["rtmp-connect", "rtmp-listen"]))] +pub struct Args { + /// Dial `rtmp://host[:1935]//`. + #[arg(id = "rtmp-connect", long = "connect", value_name = "URL")] + pub connect: Option, + + /// Bind an RTMP listener, bridging the single `--broadcast` (the RTMP app/key + /// is accepted but not used for routing). + #[arg(id = "rtmp-listen", long = "listen", value_name = "ADDR")] + pub listen: Option, +} + +/// Accept incoming RTMP publishes into the Origin as `name`; reject plays (import). +pub async fn listen_import(origin: moq_net::OriginProducer, addr: SocketAddr, name: String) -> anyhow::Result<()> { + let mut server = Server::bind(addr).await?; + tracing::info!(%addr, %name, "RTMP listening (import)"); + notify_ready(); + + while let Some(request) = server.accept().await { + match request { + Request::Publish(publish) => { + let origin = origin.clone(); + let name = name.clone(); + tokio::spawn(async move { + if let Err(err) = publish.accept(&origin, &name).await { + tracing::warn!(%name, %err, "RTMP ingest ended with error"); + } + }); + } + Request::Play(play) => { + tokio::spawn(async move { + let _ = play.reject("this is an import listener; it does not serve plays").await; + }); + } + _ => {} + } + } + + Ok(()) +} + +/// Serve RTMP plays of `name` from the Origin; reject publishes (export). +pub async fn listen_export(origin: moq_net::OriginConsumer, addr: SocketAddr, name: String) -> anyhow::Result<()> { + let mut server = Server::bind(addr).await?; + tracing::info!(%addr, %name, "RTMP listening (export)"); + notify_ready(); + + while let Some(request) = server.accept().await { + match request { + Request::Play(play) => { + let origin = origin.clone(); + let name = name.clone(); + tokio::spawn(async move { + if let Err(err) = play.accept(&origin, &name).await { + tracing::warn!(%name, %err, "RTMP play ended with error"); + } + }); + } + Request::Publish(publish) => { + tokio::spawn(async move { + let _ = publish + .reject("this is an export listener; it does not accept publishes") + .await; + }); + } + _ => {} + } + } + + Ok(()) +} + +/// Dial a remote RTMP server and pull its play into the Origin under `name` (import). +pub async fn connect_import(origin: moq_net::OriginProducer, url: Url, name: String) -> anyhow::Result<()> { + let (addr, app, key) = parse_url(&url).await?; + tracing::info!(%url, %name, "RTMP client pulling"); + notify_ready(); + + let client = Client::connect(addr, &app).await?; + Ok(client.pull(&key, &origin, &name).await?) +} + +/// Push a broadcast from the Origin to a remote RTMP server (export). +pub async fn connect_export(origin: moq_net::OriginConsumer, url: Url, name: String) -> anyhow::Result<()> { + let (addr, app, key) = parse_url(&url).await?; + let broadcast = origin + .announced_broadcast(&name) + .await + .with_context(|| format!("origin closed before broadcast `{name}` was announced"))?; + + tracing::info!(%url, %name, "RTMP client pushing"); + notify_ready(); + + let client = Client::connect(addr, &app).await?; + Ok(client.publish(&key, broadcast).await?) +} + +/// Parse `rtmp://host[:1935]//` into a resolved address, app, and stream key. +async fn parse_url(url: &Url) -> anyhow::Result<(SocketAddr, String, String)> { + anyhow::ensure!(url.scheme() == "rtmp", "rtmp url must use the rtmp scheme: {url}"); + + let host = url + .host_str() + .with_context(|| format!("rtmp url missing host: {url}"))?; + let port = url.port().unwrap_or(1935); + let addr = tokio::net::lookup_host((host, port)) + .await? + .next() + .with_context(|| format!("could not resolve {host}:{port}"))?; + + let mut segments = url.path().trim_matches('/').splitn(2, '/'); + let app = segments.next().unwrap_or_default().to_string(); + let key = segments.next().unwrap_or_default().to_string(); + anyhow::ensure!( + !app.is_empty() && !key.is_empty(), + "rtmp url must include an app and stream key: rtmp://host//" + ); + + Ok((addr, app, key)) +} + +#[cfg(test)] +mod tests { + use super::*; + + // Numeric hosts resolve without touching DNS, so these stay offline. + async fn parse(url: &str) -> anyhow::Result<(SocketAddr, String, String)> { + parse_url(&Url::parse(url).unwrap()).await + } + + #[tokio::test] + async fn ok_default_port() { + let (addr, app, key) = parse("rtmp://127.0.0.1/live/cam0").await.unwrap(); + assert_eq!(addr.port(), 1935); + assert_eq!((app.as_str(), key.as_str()), ("live", "cam0")); + } + + #[tokio::test] + async fn ok_explicit_port() { + let (addr, _, _) = parse("rtmp://127.0.0.1:1936/live/cam0").await.unwrap(); + assert_eq!(addr.port(), 1936); + } + + #[tokio::test] + async fn rejects_non_rtmp_scheme() { + assert!(parse("http://127.0.0.1/live/cam0").await.is_err()); + } + + #[tokio::test] + async fn requires_app_and_key() { + assert!(parse("rtmp://127.0.0.1/live").await.is_err()); + assert!(parse("rtmp://127.0.0.1/").await.is_err()); + } +} diff --git a/rs/moq-cli/src/server.rs b/rs/moq-cli/src/server.rs deleted file mode 100644 index 5c38594bc..000000000 --- a/rs/moq-cli/src/server.rs +++ /dev/null @@ -1,88 +0,0 @@ -use hang::moq_net; - -pub async fn run_server( - mut server: moq_native::Server, - name: String, - consumer: moq_net::BroadcastConsumer, -) -> anyhow::Result<()> { - #[cfg(unix)] - // Notify systemd that we're ready. - let _ = sd_notify::notify(&[sd_notify::NotifyState::Ready]); - - let mut conn_id = 0; - - tracing::info!(addr = ?server.local_addr(), "listening"); - - while let Some(session) = server.accept().await { - let id = conn_id; - conn_id += 1; - - let name = name.clone(); - - let consumer = consumer.clone(); - // Handle the connection in a new task. - tokio::spawn(async move { - if let Err(err) = run_serve_session(id, session, name, consumer).await { - tracing::warn!(%err, "failed to accept session"); - } - }); - } - - Ok(()) -} - -#[tracing::instrument("session", skip_all, fields(id))] -async fn run_serve_session( - id: u64, - session: moq_native::Request, - name: String, - consumer: moq_net::BroadcastConsumer, -) -> anyhow::Result<()> { - // Create an origin producer to publish to the broadcast. - let origin = moq_net::Origin::random().produce(); - origin.publish_broadcast(&name, consumer); - - // Blindly accept the session (WebTransport or QUIC), regardless of the URL. - let session = session.with_publish(origin.consume()).ok().await?; - - tracing::info!(id, "accepted session"); - - session.closed().await.map_err(Into::into) -} - -pub async fn run_accept(mut server: moq_native::Server, origin: moq_net::OriginProducer) -> anyhow::Result<()> { - #[cfg(unix)] - // Notify systemd that we're ready. - let _ = sd_notify::notify(&[sd_notify::NotifyState::Ready]); - - let mut conn_id = 0; - - tracing::info!(addr = ?server.local_addr(), "listening"); - - while let Some(session) = server.accept().await { - let id = conn_id; - conn_id += 1; - - let origin = origin.clone(); - tokio::spawn(async move { - if let Err(err) = run_accept_session(id, session, origin).await { - tracing::warn!(%err, "failed to accept session"); - } - }); - } - - Ok(()) -} - -#[tracing::instrument("session", skip_all, fields(id))] -async fn run_accept_session( - id: u64, - session: moq_native::Request, - origin: moq_net::OriginProducer, -) -> anyhow::Result<()> { - let session = session.with_consume(origin).ok().await?; - - tracing::info!(id, "accepted session"); - - session.closed().await.map_err(Into::into) -} diff --git a/rs/moq-cli/src/srt.rs b/rs/moq-cli/src/srt.rs new file mode 100644 index 000000000..90a55807f --- /dev/null +++ b/rs/moq-cli/src/srt.rs @@ -0,0 +1,182 @@ +//! SRT endpoints. Like RTMP, listeners are directional: an import listener +//! accepts publishes only, an export listener serves requests only. + +use std::net::SocketAddr; +use std::time::Duration; + +use anyhow::Context; +use hang::moq_net; +use moq_srt::{Request, Server}; +use url::Url; + +use crate::moq::notify_ready; + +/// SRT endpoint args: exactly one of `--connect` (dial) / `--listen` (bind). +#[derive(clap::Args, Clone)] +#[command(group = clap::ArgGroup::new("srt-mode").required(true).multiple(false).args(["srt-connect", "srt-listen"]))] +pub struct Args { + /// Dial `srt://host:port?streamid=...`. + #[arg(id = "srt-connect", long = "connect", value_name = "URL")] + pub connect: Option, + + /// Bind an SRT listener, bridging the single `--broadcast` (the SRT stream id + /// is accepted but not used for routing). + #[arg(id = "srt-listen", long = "listen", value_name = "ADDR")] + pub listen: Option, + + /// SRT receive latency: the buffering delay traded for loss-recovery headroom. + #[arg(long, default_value = "200ms", value_parser = humantime::parse_duration)] + pub latency: Duration, +} + +/// Accept incoming SRT publishes into the Origin as `name`; reject requests (import). +pub async fn listen_import( + origin: moq_net::OriginProducer, + addr: SocketAddr, + name: String, + latency: Duration, +) -> anyhow::Result<()> { + let mut server = Server::bind(addr, latency).await?; + tracing::info!(%addr, %name, "SRT listening (import)"); + notify_ready(); + + while let Some(request) = server.accept().await { + match request { + Request::Publish(publish) => { + let origin = origin.clone(); + let name = name.clone(); + tokio::spawn(async move { + if let Err(err) = publish.accept(&origin, &name).await { + tracing::warn!(%name, %err, "SRT ingest ended with error"); + } + }); + } + Request::Subscribe(subscribe) => { + tokio::spawn(async move { + let _ = subscribe.reject().await; + }); + } + _ => {} + } + } + + Ok(()) +} + +/// Serve SRT requests for `name` from the Origin; reject publishes (export). +pub async fn listen_export( + origin: moq_net::OriginConsumer, + addr: SocketAddr, + name: String, + latency: Duration, +) -> anyhow::Result<()> { + let mut server = Server::bind(addr, latency).await?; + tracing::info!(%addr, %name, "SRT listening (export)"); + notify_ready(); + + while let Some(request) = server.accept().await { + match request { + Request::Subscribe(subscribe) => { + let origin = origin.clone(); + let name = name.clone(); + tokio::spawn(async move { + if let Err(err) = subscribe.accept(&origin, &name).await { + tracing::warn!(%name, %err, "SRT request ended with error"); + } + }); + } + Request::Publish(publish) => { + tokio::spawn(async move { + let _ = publish.reject().await; + }); + } + _ => {} + } + } + + Ok(()) +} + +/// Dial a remote SRT server and pull its stream into the Origin under `name` (import). +pub async fn connect_import( + origin: moq_net::OriginProducer, + url: Url, + name: String, + latency: Duration, +) -> anyhow::Result<()> { + let (addr, resource) = parse_url(&url).await?; + tracing::info!(%url, %name, "SRT client pulling"); + notify_ready(); + + Ok(moq_srt::dial::pull(addr, &resource, latency, &origin, &name).await?) +} + +/// Push a broadcast from the Origin to a remote SRT server (export). +pub async fn connect_export( + origin: moq_net::OriginConsumer, + url: Url, + name: String, + latency: Duration, +) -> anyhow::Result<()> { + let (addr, resource) = parse_url(&url).await?; + tracing::info!(%url, %name, "SRT client pushing"); + notify_ready(); + + Ok(moq_srt::dial::publish(addr, &resource, latency, &origin, &name).await?) +} + +/// Parse `srt://host:port?streamid=` into a resolved address and resource. +/// The resource falls back to the URL path when `streamid` is absent. +async fn parse_url(url: &Url) -> anyhow::Result<(SocketAddr, String)> { + anyhow::ensure!(url.scheme() == "srt", "srt url must use the srt scheme: {url}"); + + let host = url.host_str().with_context(|| format!("srt url missing host: {url}"))?; + let port = url.port().context("srt url must include a port: srt://host:port")?; + let addr = tokio::net::lookup_host((host, port)) + .await? + .next() + .with_context(|| format!("could not resolve {host}:{port}"))?; + + let resource = url + .query_pairs() + .find(|(key, _)| key == "streamid") + .map(|(_, value)| value.into_owned()) + .unwrap_or_else(|| url.path().trim_matches('/').to_string()); + anyhow::ensure!(!resource.is_empty(), "srt url must include a streamid or path"); + + Ok((addr, resource)) +} + +#[cfg(test)] +mod tests { + use super::*; + + // Numeric hosts resolve without touching DNS, so these stay offline. + async fn parse(url: &str) -> anyhow::Result<(SocketAddr, String)> { + parse_url(&Url::parse(url).unwrap()).await + } + + #[tokio::test] + async fn resource_from_streamid() { + let (addr, resource) = parse("srt://127.0.0.1:9000?streamid=live/cam").await.unwrap(); + assert_eq!(addr.port(), 9000); + assert_eq!(resource, "live/cam"); + } + + #[tokio::test] + async fn resource_from_path() { + let (_, resource) = parse("srt://127.0.0.1:9000/live/cam").await.unwrap(); + assert_eq!(resource, "live/cam"); + } + + #[tokio::test] + async fn rejects_non_srt_scheme() { + assert!(parse("udp://127.0.0.1:9000").await.is_err()); + } + + #[tokio::test] + async fn requires_port_and_resource() { + assert!(parse("srt://127.0.0.1").await.is_err()); + assert!(parse("srt://127.0.0.1:9000").await.is_err()); + } +} diff --git a/rs/moq-cli/src/subscribe.rs b/rs/moq-cli/src/subscribe.rs index 36a6e2dd3..cf47fcf6b 100644 --- a/rs/moq-cli/src/subscribe.rs +++ b/rs/moq-cli/src/subscribe.rs @@ -5,7 +5,8 @@ use hang::moq_net; use moq_mux::catalog::CatalogFormat; use tokio::io::AsyncWriteExt; -#[derive(ValueEnum, Clone, Copy)] +/// Container format written to stdout on the export (sink) side. +#[derive(Clone, Copy)] pub enum SubscribeFormat { Fmp4, Mkv, @@ -33,30 +34,19 @@ impl From for CatalogFormat { } } -#[derive(clap::Args, Clone)] +/// The resolved stdout export settings (built from the `export` flags + format). +#[derive(Clone)] pub struct SubscribeArgs { /// The format to write to stdout. - #[arg(long)] pub format: SubscribeFormat, - /// Maximum latency before skipping groups (e.g. `500ms`, `1s`). - #[arg(long, default_value = "500ms", value_parser = humantime::parse_duration)] + /// Maximum latency before skipping groups. pub max_latency: Duration, - /// Cap the output fragment duration (e.g. `2s`, `500ms`). - /// - /// By default a fragment covers one GOP (rolled over on video keyframes). - /// Setting this caps each fragment to roughly the given duration. - /// The cap applies in addition to GOP rollover. - #[arg(long, value_parser = humantime::parse_duration)] + /// Cap the output fragment duration (default: one GOP). Applies to fmp4 / mkv. pub fragment_duration: Option, - /// Catalog format to subscribe to for track discovery. - /// - /// When omitted, the format is auto-detected from the broadcast name suffix - /// (`.hang` -> hang, `.msf` -> msf), falling back to hang. Pass `hangz` to read - /// the DEFLATE-compressed `catalog.json.z` track instead (same `.hang` broadcast). - #[arg(long)] + /// Catalog format for track discovery (default: detect from the broadcast suffix). pub catalog: Option, } diff --git a/rs/moq-cli/src/web.rs b/rs/moq-cli/src/web.rs index 43e2ec962..652c60073 100644 --- a/rs/moq-cli/src/web.rs +++ b/rs/moq-cli/src/web.rs @@ -3,17 +3,32 @@ use axum::handler::HandlerWithoutStateExt; use axum::http::StatusCode; use axum::response::IntoResponse; use axum::{Router, http::Method, routing::get}; -use std::path::PathBuf; use std::sync::{Arc, RwLock}; use tower_http::cors::{Any, CorsLayer}; -use tower_http::services::ServeDir; -// Initialize the HTTP server (but don't serve yet). -pub async fn run_web( - bind: &str, - tls_info: Arc>, - public: Option, +/// Serve an axum router over TCP, optionally terminating TLS. Used by the HLS +/// and WebRTC (WHIP/WHEP) HTTP endpoints. +pub async fn serve( + listener: std::net::TcpListener, + app: Router, + tls: Option>, ) -> anyhow::Result<()> { + let service = app.into_make_service(); + match tls { + Some(config) => { + let config = axum_server::tls_rustls::RustlsConfig::from_config(config); + axum_server::from_tcp_rustls(listener, config)?.serve(service).await?; + } + None => { + axum_server::from_tcp(listener)?.serve(service).await?; + } + } + Ok(()) +} + +/// Serve the `/certificate.sha256` self-signed fingerprint over HTTP, so an +/// `http://` client can pin a `--server-bind` server's generated cert. +pub async fn run_web(bind: &str, tls_info: Arc>) -> anyhow::Result<()> { let listen = tokio::net::lookup_host(bind) .await .context("invalid listen address")? @@ -36,20 +51,10 @@ pub async fn run_web( .clone() }; - let mut app = Router::new() + let app = Router::new() .route("/certificate.sha256", get(fingerprint_handler)) - .layer(CorsLayer::new().allow_origin(Any).allow_methods([Method::GET])); - - // If a public directory is provided, serve it. - // We use this for local development to serve the index.html file and friends. - if let Some(public) = public.as_ref() { - tracing::info!(public = %public.display(), "serving directory"); - - let public = ServeDir::new(public).not_found_service(handle_404.into_service()); - app = app.fallback_service(public); - } else { - app = app.fallback_service(handle_404.into_service()); - } + .layer(CorsLayer::new().allow_origin(Any).allow_methods([Method::GET])) + .fallback_service(handle_404.into_service()); // Dual-stack so the cert endpoint answers over IPv4 too, even on Windows // where `[::]` is IPv6-only by default. diff --git a/rs/moq-rtmp/README.md b/rs/moq-rtmp/README.md index f6e244872..c427b4b3f 100644 --- a/rs/moq-rtmp/README.md +++ b/rs/moq-rtmp/README.md @@ -7,7 +7,7 @@ both directions with [`moq-mux`](../moq-mux): on **publish** it re-wraps a client's messages as FLV tags, demuxes them, and publishes the result into a MoQ origin as ordinary broadcasts; on **play** it subscribes to a broadcast from the origin, muxes it back to FLV, and streams the tags down to the player. It's the -sibling of `moq-srt`, `moq-cli hls` import/export, and `moq-rtc`'s WHIP/WHEP. Both +sibling of `moq-srt`, `moq-cli` HLS import/export, and `moq-rtc`'s WHIP/WHEP. Both legacy RTMP (H.264 + AAC) and enhanced RTMP (E-RTMP: HEVC, AV1, VP9, Opus, AC-3) work in each direction, since the codec handling lives in the `moq-mux` FLV demuxer/muxer. Pure Rust: the protocol is provided by `rml_rtmp`, with no librtmp diff --git a/rs/moq-rtmp/src/lib.rs b/rs/moq-rtmp/src/lib.rs index 031038d4e..8429cb2a8 100644 --- a/rs/moq-rtmp/src/lib.rs +++ b/rs/moq-rtmp/src/lib.rs @@ -8,11 +8,11 @@ //! its audio/video messages as FLV tags, demux them with [`moq_mux`], and //! publish the result into a [`moq_net::OriginProducer`] as ordinary MoQ //! broadcasts. This is the contribution-ingest analogue of `moq-srt`, -//! `moq-cli hls import`, and `moq-rtc`'s WHIP. +//! `moq-cli import ... hls`, and `moq-rtc`'s WHIP. //! - **Play (egress)**: a client (VLC, ffplay, mpv) pulls //! `rtmp://host//`; we subscribe to that broadcast from a //! [`moq_net::OriginConsumer`], mux it back to FLV with [`moq_mux`], and stream -//! the tags down as RTMP. The counterpart to `moq-cli hls export`. +//! the tags down as RTMP. The counterpart to `moq-cli export ... hls`. //! //! Both legacy RTMP (H.264 + AAC, plus MP3) and enhanced RTMP (E-RTMP: the HEVC, //! AV1, VP9, Opus, AC-3, and MP3 FourCC payloads) are supported in each direction, diff --git a/rs/moq-srt/README.md b/rs/moq-srt/README.md index 97f067633..aa4de6db4 100644 --- a/rs/moq-srt/README.md +++ b/rs/moq-srt/README.md @@ -7,7 +7,7 @@ by its stream id `m=` mode: - `m=publish` (the default): ingest. Demux the connection's transport stream with [`moq-mux`](../moq-mux) and publish it into a MoQ origin as an ordinary - broadcast. The contribution-ingest analogue of `moq-cli hls import` and + broadcast. The contribution-ingest analogue of `moq-cli` HLS import and `moq-rtc`'s WHIP. - `m=request`: egress. Re-mux a broadcast from the origin back to MPEG-TS and stream it to the caller, so `vlc srt://...` and `ffmpeg -i srt://...` can play diff --git a/rs/moq-srt/src/lib.rs b/rs/moq-srt/src/lib.rs index 60fd08b3a..b07213cc1 100644 --- a/rs/moq-srt/src/lib.rs +++ b/rs/moq-srt/src/lib.rs @@ -6,7 +6,7 @@ //! //! - `m=publish` (the default): demux the MPEG-TS the connection carries with //! [`moq_mux`] and publish it into the origin as an ordinary broadcast. The -//! contribution-ingest analogue of `moq-cli hls import` and `moq-rtc`'s WHIP. +//! contribution-ingest analogue of `moq-cli import ... hls` and `moq-rtc`'s WHIP. //! - `m=request`: re-mux a broadcast from the origin back to MPEG-TS and stream //! it to the caller, so a plain SRT player (VLC, ffmpeg) can watch it. //! diff --git a/rs/moq-srt/src/ts.rs b/rs/moq-srt/src/ts.rs index 5df51b58e..63bda19c6 100644 --- a/rs/moq-srt/src/ts.rs +++ b/rs/moq-srt/src/ts.rs @@ -17,7 +17,7 @@ use crate::Result; /// /// Each chunk is handed straight to the TS importer, which consumes whole /// transport packets and retains any partial trailing packet internally for the -/// next call (the same pattern `moq-cli publish ts` uses against stdin). +/// next call (the same pattern `moq-cli import ... stdin ts` uses against stdin). /// Dropping the publisher ends the broadcast: the importer's producer clone /// closes, which unannounces it from the origin. pub struct Publisher { diff --git a/rs/scripts/package-binary.sh b/rs/scripts/package-binary.sh index 672260968..94a0cce31 100755 --- a/rs/scripts/package-binary.sh +++ b/rs/scripts/package-binary.sh @@ -2,7 +2,10 @@ set -euo pipefail # Build and package a workspace binary for release. -# Usage: ./package-binary.sh --crate CRATE [--target TARGET] [--version VERSION] [--output DIR] +# Usage: ./package-binary.sh --crate CRATE [--bin NAME] [--target TARGET] [--version VERSION] [--output DIR] +# +# --bin overrides the binary/command name when it differs from the crate (e.g. +# the `moq-cli` crate ships its binary as `moq`); it defaults to the crate name. # # Builds via `nix build .#` against the flake-pinned toolchain so # artifacts are reproducible across hosts. The resulting tarball matches @@ -14,19 +17,21 @@ RS_DIR="$(cd "$SCRIPT_DIR/.." && pwd)" WORKSPACE_DIR="$(cd "$RS_DIR/.." && pwd)" CRATE="" +BIN="" TARGET="" VERSION="" OUTPUT_DIR="dist" while [[ $# -gt 0 ]]; do case $1 in - --crate | --target | --version | --output) + --crate | --bin | --target | --version | --output) if [[ $# -lt 2 ]]; then echo "Error: $1 requires a value" >&2 exit 1 fi case $1 in --crate) CRATE="$2" ;; + --bin) BIN="$2" ;; --target) TARGET="$2" ;; --version) VERSION="$2" ;; --output) OUTPUT_DIR="$2" ;; @@ -34,7 +39,7 @@ while [[ $# -gt 0 ]]; do shift 2 ;; -h | --help) - echo "Usage: $0 --crate CRATE [--target TARGET] [--version VERSION] [--output DIR]" + echo "Usage: $0 --crate CRATE [--bin NAME] [--target TARGET] [--version VERSION] [--output DIR]" exit 0 ;; *) @@ -49,6 +54,9 @@ if [[ -z "$CRATE" ]]; then exit 1 fi +# The binary/command name defaults to the crate name. +BIN="${BIN:-$CRATE}" + CRATE_DIR="$RS_DIR/$CRATE" if [[ ! -f "$CRATE_DIR/Cargo.toml" ]]; then echo "Error: no Cargo.toml found at $CRATE_DIR" >&2 @@ -90,9 +98,9 @@ trap 'rm -rf "$BUILD_TMP"' EXIT RESULT_LINK="$BUILD_TMP/result" nix build "$WORKSPACE_DIR#$NIX_ATTR" --out-link "$RESULT_LINK" -# Locate the built binary. Crane installs to result/bin/. -# By convention each of our binaries shares its crate name. -BIN_FILE="$RESULT_LINK/bin/$CRATE" +# Locate the built binary. Crane installs to result/bin/. The binary +# name is usually the crate name; the `moq-cli` crate ships as `moq`. +BIN_FILE="$RESULT_LINK/bin/$BIN" if [[ ! -f "$BIN_FILE" ]]; then echo "Error: no binary found at $BIN_FILE" >&2 echo "Contents of $RESULT_LINK/bin:" >&2 @@ -109,8 +117,8 @@ mkdir -p "$PACKAGE_DIR/bin" # Dereference the nix-store symlink and drop perms so the file is writable # enough to archive cleanly. -cp -L "$BIN_FILE" "$PACKAGE_DIR/bin/$CRATE" -chmod 0755 "$PACKAGE_DIR/bin/$CRATE" +cp -L "$BIN_FILE" "$PACKAGE_DIR/bin/$BIN" +chmod 0755 "$PACKAGE_DIR/bin/$BIN" # On macOS the nix toolchain links the nix-store copy of libiconv, whose # absolute path doesn't exist on a user's Mac, so dyld aborts at startup @@ -119,7 +127,7 @@ chmod 0755 "$PACKAGE_DIR/bin/$CRATE" # /usr/lib), like a plain `cargo build` does. scrub-macho.sh also asserts no # /nix path survives, so this can't silently ship again. if [[ "$(uname)" == "Darwin" ]]; then - bin="$PACKAGE_DIR/bin/$CRATE" + bin="$PACKAGE_DIR/bin/$BIN" "$SCRIPT_DIR/scrub-macho.sh" "$bin" # Prove dyld actually loads the scrubbed binary. --help triggers dyld's diff --git a/rs/scripts/package-windows.sh b/rs/scripts/package-windows.sh index d1c6a6b18..11b595923 100755 --- a/rs/scripts/package-windows.sh +++ b/rs/scripts/package-windows.sh @@ -2,7 +2,10 @@ set -euo pipefail # Build and package a workspace binary as a Windows .zip for release. -# Usage: ./package-windows.sh --crate CRATE [--version VERSION] [--target TARGET] [--output DIR] +# Usage: ./package-windows.sh --crate CRATE [--bin NAME] [--version VERSION] [--target TARGET] [--output DIR] +# +# --bin overrides the binary/command name when it differs from the crate (e.g. +# the `moq-cli` crate ships its binary as `moq`); it defaults to the crate name. # # Runs under Git Bash on a windows-latest runner. Unlike package-binary.sh # (nix, macOS + Linux), this is a plain cargo build against the MSVC target. @@ -16,19 +19,21 @@ RS_DIR="$(cd "$SCRIPT_DIR/.." && pwd)" WORKSPACE_DIR="$(cd "$RS_DIR/.." && pwd)" CRATE="" +BIN="" VERSION="" TARGET="x86_64-pc-windows-msvc" OUTPUT_DIR="dist" while [[ $# -gt 0 ]]; do case $1 in - --crate | --version | --target | --output) + --crate | --bin | --version | --target | --output) if [[ $# -lt 2 ]]; then echo "Error: $1 requires a value" >&2 exit 1 fi case $1 in --crate) CRATE="$2" ;; + --bin) BIN="$2" ;; --version) VERSION="$2" ;; --target) TARGET="$2" ;; --output) OUTPUT_DIR="$2" ;; @@ -36,7 +41,7 @@ while [[ $# -gt 0 ]]; do shift 2 ;; -h | --help) - echo "Usage: $0 --crate CRATE [--version VERSION] [--target TARGET] [--output DIR]" + echo "Usage: $0 --crate CRATE [--bin NAME] [--version VERSION] [--target TARGET] [--output DIR]" exit 0 ;; *) @@ -51,6 +56,9 @@ if [[ -z "$CRATE" ]]; then exit 1 fi +# The binary/command name defaults to the crate name. +BIN="${BIN:-$CRATE}" + CRATE_DIR="$RS_DIR/$CRATE" if [[ ! -f "$CRATE_DIR/Cargo.toml" ]]; then echo "Error: no Cargo.toml found at $CRATE_DIR" >&2 @@ -69,9 +77,9 @@ fi echo "Building $CRATE for $TARGET..." cargo build --release --target "$TARGET" -p "$CRATE" -# The command name matches the crate, like `cargo install` and the Homebrew -# formula (only the .deb/.rpm rename moq-cli to moq). -BIN_FILE="$WORKSPACE_DIR/target/$TARGET/release/$CRATE.exe" +# The command name is the crate's `[[bin]]` name (usually the crate name; the +# `moq-cli` crate ships as `moq`), matching `cargo install` and the other packages. +BIN_FILE="$WORKSPACE_DIR/target/$TARGET/release/$BIN.exe" if [[ ! -f "$BIN_FILE" ]]; then echo "Error: no binary found at $BIN_FILE" >&2 ls "$WORKSPACE_DIR/target/$TARGET/release" >&2 || true @@ -85,7 +93,7 @@ NAME="$CRATE-$VERSION-$TARGET" STAGING="$(mktemp -d)" trap 'rm -rf "$STAGING"' EXIT -cp "$BIN_FILE" "$STAGING/$CRATE.exe" +cp "$BIN_FILE" "$STAGING/$BIN.exe" cp "$WORKSPACE_DIR/LICENSE-MIT" "$STAGING/" cp "$WORKSPACE_DIR/LICENSE-APACHE" "$STAGING/" if [[ -f "$CRATE_DIR/README.md" ]]; then diff --git a/test/smoke/smoke.sh b/test/smoke/smoke.sh index 4a3674ff8..f316df8bf 100755 --- a/test/smoke/smoke.sh +++ b/test/smoke/smoke.sh @@ -175,9 +175,8 @@ build_relay_cli() { exit 1 } [[ -n "$RELAY" ]] || RELAY="$TARGET_BASE/$PROFILE/moq-relay" - # `cargo build -p moq-cli` names the binary after the crate (moq-cli); the - # apt/rpm packages rename it to `moq`, but from source it's moq-cli. - [[ -n "$MOQ" ]] || MOQ="$TARGET_BASE/$PROFILE/moq-cli" + # The `moq-cli` crate ships its binary as `moq` (a `[[bin]]` override). + [[ -n "$MOQ" ]] || MOQ="$TARGET_BASE/$PROFILE/moq" } # Editable-install the workspace Python build (maturin builds rs/moq-ffi, then @@ -351,7 +350,7 @@ start_publisher() { local lang="$1" broadcast="$2" log="$TMP/pub-$1.log" case "$lang" in rust) - (ffmpeg_h264 | "$MOQ" publish --url "$URL" --broadcast "$broadcast" avc3) >"$log" 2>&1 & + (ffmpeg_h264 | "$MOQ" --client-connect "$URL" --broadcast "$broadcast" import avc3) >"$log" 2>&1 & ;; python) (ffmpeg_h264 | "$PY" "$CLIENTS/python/smoke.py" \ @@ -390,8 +389,8 @@ run_subscriber() { # moq-cli only handles SIGINT, so -k forces SIGKILL if it ignores the # SIGTERM that fires when no data arrives within the timeout. local n - n=$(timeout -k 3 "$TIMEOUT" "$MOQ" subscribe --url "$URL" --broadcast "$broadcast" \ - --format fmp4 2>/dev/null | head -c 1 | wc -c | tr -d ' ' || true) + n=$(timeout -k 3 "$TIMEOUT" "$MOQ" --client-connect "$URL" --broadcast "$broadcast" \ + export fmp4 2>/dev/null | head -c 1 | wc -c | tr -d ' ' || true) [[ "${n:-0}" -ge 1 ]] ;; python)