diff --git a/.github/workflows/alert.yml b/.github/workflows/alert.yml index 5d53c749184..b822dfebae0 100644 --- a/.github/workflows/alert.yml +++ b/.github/workflows/alert.yml @@ -44,6 +44,7 @@ on: - Platform - Quest - release-brew + - Release C++ - Release Dart - Release Dart FFI - Release Go diff --git a/.github/workflows/cpp.yml b/.github/workflows/cpp.yml new file mode 100644 index 00000000000..7f8cb9d4096 --- /dev/null +++ b/.github/workflows/cpp.yml @@ -0,0 +1,91 @@ +name: C++ + +# Builds and tests the moq C++ package (cpp/moq) with clang and MSVC, outside +# Nix. `just check` already runs `just cpp check` with the dev shell's gcc; the +# dev shell has no clang++, and adding one would change which compiler every +# other recipe gets as `cc`. +# +# Outside Nix the generator comes from `cargo install` at the tag flake.nix +# pins; see the comment there for every other place that names it. + +permissions: + contents: read + +on: + pull_request: + # `closed` is here only so merging/closing a PR cancels its in-flight run + # via the concurrency group below; the job itself is skipped on close. + types: [opened, synchronize, reopened, closed] + paths: + - "cpp/moq/**" + - "cpp/justfile" + - "rs/moq-ffi/**" + - "sh/cpp/**" + - "rs/moq-c/native-libs/**" + - "doc/lib/cpp/**" + - "doc/lib/samples.sh" + - "Cargo.lock" + - ".github/workflows/cpp.yml" + +concurrency: + group: cpp-${{ github.ref }} + cancel-in-progress: true + +jobs: + clang: + name: C++ (clang) + if: github.event.action != 'closed' + runs-on: ubuntu-24.04 + timeout-minutes: 60 + + steps: + - name: Checkout + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + + - name: Install Rust + uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8 # stable + + - uses: ./.github/actions/just + + - name: Install uniffi-bindgen-cpp + run: cargo install uniffi-bindgen-cpp --locked --git https://github.com/kixelated/uniffi-bindgen-cpp --tag v0.11.0-kixelated.4+v0.32.2 + + - name: Check + run: just cpp check + env: + CXX: clang++ + + # MSVC, with exceptions and RTTI off, plus the package's tests. + msvc: + name: C++ (MSVC) + if: github.event.action != 'closed' + runs-on: windows-latest + timeout-minutes: 60 + + steps: + - name: Checkout + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + + - name: Install Rust + uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8 # stable + + # aws-lc-rs assembles its x86_64 crypto with NASM on Windows. + - name: Install NASM + shell: pwsh + run: | + choco install nasm -y --no-progress + "C:\Program Files\NASM" | Out-File -FilePath $env:GITHUB_PATH -Encoding utf8 -Append + + - uses: ./.github/actions/just + + - name: Install uniffi-bindgen-cpp + shell: bash + run: cargo install uniffi-bindgen-cpp --locked --git https://github.com/kixelated/uniffi-bindgen-cpp --tag v0.11.0-kixelated.4+v0.32.2 + + - name: Check + shell: bash + run: just cpp check diff --git a/.github/workflows/interop.yml b/.github/workflows/interop.yml index 2c4f8affffe..dd624345899 100644 --- a/.github/workflows/interop.yml +++ b/.github/workflows/interop.yml @@ -27,6 +27,7 @@ on: - "package.json" # The Go client builds against the modules this script stages. - "sh/go/**" + - "cpp/moq/**" - ".github/workflows/interop.yml" concurrency: diff --git a/.github/workflows/moq-c.yml b/.github/workflows/moq-c.yml index 329f81b605e..e95b9de48d5 100644 --- a/.github/workflows/moq-c.yml +++ b/.github/workflows/moq-c.yml @@ -116,109 +116,3 @@ jobs: RELEASE_PREV_TAG: ${{ steps.prev_tag.outputs.tag }} run: just gh release create artifacts - # The OBS plugin (cpp/obs) statically links moq-c, so it ships with every - # moq-c release: rebuild it against the release we just published (no - # second cargo build, guaranteed-matching ABI) and cut a parallel - # obs-moq-v release. Runs after `release` so the moq-c archives - # the plugin's CMake fetches are already live. - obs-build: - name: OBS plugin (${{ matrix.target }}) - needs: release - runs-on: ${{ matrix.os }} - - # macOS + Windows only. These link the ffmpeg the obs-deps bundle ships - # (the same one OBS uses), so the binaries are portable. A Linux build - # would link nix/distro ffmpeg and not be portable, so Linux is - # build-from-source (see doc/bin/obs.md) and ships no prebuilt binary. - strategy: - fail-fast: false - matrix: - include: - - target: aarch64-apple-darwin - os: macos-latest - - target: x86_64-pc-windows-msvc - # The plugin preset uses the Visual Studio 17 2022 generator. - os: windows-2022 - - steps: - # Shallow: this job only compiles, it never reads git history or tags. - - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 - with: - persist-credentials: false - - - uses: ./.github/actions/just - - - name: Parse version - id: parse - shell: bash - run: just gh release parse-version moq-c - - # Native Xcode / Visual Studio; libobs/Qt6/ffmpeg come from the obs-deps - # bundle and moq-c from the release we just published (--moq-c-release). - - name: Build and package - shell: bash - env: - TARGET: ${{ matrix.target }} - VERSION: ${{ steps.parse.outputs.version }} - run: | - just obs package --target "$TARGET" --moq-c-release "$VERSION" --output dist - - - name: Upload artifact - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 - with: - name: obs-moq-${{ matrix.target }} - path: dist/* - - obs-release: - name: OBS plugin release - needs: obs-build - runs-on: ubuntu-latest - permissions: - contents: write - - steps: - - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 - with: - fetch-depth: 0 - persist-credentials: false - - - uses: ./.github/actions/just - - - name: Parse version - id: parse - run: just gh release parse-version moq-c - - # Only the obs-moq archives: the moq-c-* artifacts from the build job - # belong to the moq-c release, not this one. - - name: Download artifacts - uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8 - with: - path: artifacts - pattern: obs-moq-* - merge-multiple: true - - # release.sh prev-tag keys off the current tag, but the trigger here is a - # moq-c-v* tag, so it can't find the previous obs-moq tag. Take the - # newest existing obs-moq tag (excluding this version, for safe re-runs) - # so the release notes span only since the last obs-moq release. - - name: Find previous obs-moq tag - id: prev_tag - env: - VERSION: ${{ steps.parse.outputs.version }} - run: | - cur="obs-moq-v${VERSION}" - # awk (not grep -vx | head) so an empty result is exit 0, not 1: - # under `set -eo pipefail` the inaugural release has no prior tag and - # grep's exit 1 would fail the step. Empty tag is fine downstream. - prev=$(git tag --list 'obs-moq-v*' --sort=-v:refname | awk -v cur="$cur" '$0 != cur { print; exit }') - echo "tag=${prev}" >>"$GITHUB_OUTPUT" - - # Cut a parallel obs-moq-v release at this commit (gh creates - # the tag). - - name: Create or update release - env: - GH_TOKEN: ${{ github.token }} - RELEASE_TAG: "obs-moq-v${{ steps.parse.outputs.version }}" - RELEASE_TITLE: "obs-moq v${{ steps.parse.outputs.version }}" - RELEASE_PREV_TAG: ${{ steps.prev_tag.outputs.tag }} - run: just gh release create artifacts diff --git a/.github/workflows/nightly.yml b/.github/workflows/nightly.yml index 1736442f8a7..d0c080c1fbb 100644 --- a/.github/workflows/nightly.yml +++ b/.github/workflows/nightly.yml @@ -9,8 +9,8 @@ name: Nightly # no artifacts with the default one, which is several minutes of duplicate work # that a PR gate can't justify for how rarely it catches something. `obs` is # diff-independent because its PR-time counterpart can't be: obs.yml has to -# name the paths that can break an external link to libmoq.a, and no list of -# paths covers every one of them (see the comment there). +# name the paths that can break an external link to a Rust staticlib, and no +# list of paths covers every one of them (see the comment there). # # The `release-*` jobs are diff-independent because the release workflows only # ever run on a tag: a change that compiles under plain cargo can still break a @@ -181,9 +181,9 @@ jobs: if: ${{ !cancelled() }} run: nix develop --command just rs audit - # The OBS plugin and moq-c's C fixtures are the only things that link - # libmoq.a from outside cargo, so a stale `rs/moq-c/native-libs/*.txt` - # breaks them and nothing else. obs.yml lists the inputs that can do that, + # The OBS plugin (over moq-ffi) and moq-c's C fixtures link a Rust + # staticlib from outside cargo, so a stale `rs/moq-c/native-libs/*.txt` + # breaks them and no cargo build. obs.yml lists the inputs that can do that, # but a list of paths can't be complete: a build script emitting # `cargo::rustc-link-lib` needs no manifest change, and widening that # filter to `rs/**` would run a link on nearly every PR. Being @@ -384,6 +384,12 @@ jobs: permissions: contents: write + release-cpp: + name: Release C++ + uses: ./.github/workflows/release-cpp.yml + permissions: + contents: write + release-moq-relay: name: moq-relay uses: ./.github/workflows/moq-relay.yml diff --git a/.github/workflows/obs.yml b/.github/workflows/obs.yml index 29e0e9dc889..cc6524cad9f 100644 --- a/.github/workflows/obs.yml +++ b/.github/workflows/obs.yml @@ -5,35 +5,35 @@ name: OBS # multi-hundred-MB obs-deps bundle on macOS and Windows. # # Linux is the platform that needs no bundle at all -- nixpkgs has obs-studio, -# Qt6 and ffmpeg, so the `.#obs` shell is the whole dependency set. The plugin is -# platform-independent C++ over moq-c's C ABI, so a Linux compile catches the -# breakage a macOS developer would otherwise ship uncompiled. The macOS and -# Windows release builds cover the rest: each platform's link line and the -# obs-deps bundle. +# Qt6 and ffmpeg, so the `.#obs` shell is the whole dependency set. The plugin +# is platform-independent C++ over the generated moq-ffi bindings (cpp/moq), so +# a Linux compile catches the breakage a macOS developer would otherwise ship +# uncompiled. The macOS and Windows release builds cover the rest: each +# platform's link line and the obs-deps bundle. # -# Narrow trigger, like swift.yml: the plugin's own sources, the moq-c header -# they call through, and the flake that supplies the toolchain. +# Narrow trigger, like swift.yml: the plugin's own sources, the C++ package and +# the moq-ffi API it is generated from, and the flake that supplies the +# toolchain, including uniffi-bindgen-cpp. # # moq-c's own C fixtures (`just rs c-tests`) run here too. They link libmoq.a -# from outside cargo the same way, which is too slow for nextest's hang budget -# and needs the debug staticlib this job has just built, so this is the one -# PR gate that already has everything they need. They also compile the C doc -# samples, hence `doc/lib/c/**` in the trigger. +# from outside cargo the same way the plugin links moq-ffi, which is too slow +# for nextest's hang budget, so this job, which already does that kind of link, +# is where they gate PRs. They also compile the C doc samples, hence +# `doc/lib/c/**` in the trigger. # # The manifests and Cargo.lock are in there for the one way a crate the plugin -# never names can break it: linking libmoq.a from outside cargo needs the native -# libraries in `rs/moq-c/native-libs/*.txt`, which are maintained by hand. A -# dependency that starts pulling in a new one leaves that list stale, and every -# Rust gate stays green, because cargo passes the flag itself and only this job -# links without it. +# never names can break it: linking a Rust staticlib from outside cargo needs +# the native libraries in `rs/moq-c/native-libs/*.txt`, which moq-ffi and +# moq-c share and which are maintained by hand. A dependency that starts +# pulling in a new one leaves that list stale, and every Rust gate stays green, +# because cargo passes the flag itself and only this job links without it. # # Manifests and build scripts rather than Cargo.lock alone, because neither a -# feature toggle nor a build script has to touch the lockfile. moq-c takes -# moq-native with default features, so flipping one of those defaults on -# (`jemalloc` gates `dep:tikv-jemallocator`) changes what libmoq.a needs at link -# time while Cargo.lock stays byte-identical, since moq-relay already put that -# package in the graph; a build.rs emitting `cargo::rustc-link-lib` needs no -# manifest change at all. +# feature toggle nor a build script has to touch the lockfile. Flipping a +# default feature on (`jemalloc` gates `dep:tikv-jemallocator`) changes what +# the staticlib needs at link time while Cargo.lock stays byte-identical, since +# moq-relay already put that package in the graph; a build.rs emitting +# `cargo::rustc-link-lib` needs no manifest change at all. # # A list of paths still can't be complete here -- a vendored C source or a # .cargo/config.toml could do it too -- and `rs/**` would run this on nearly @@ -51,6 +51,7 @@ on: paths: - "rs/**" - "cpp/obs/**" + - "cpp/moq/**" - "doc/lib/c/**" - "doc/lib/samples.sh" - ".cargo/**" @@ -66,7 +67,9 @@ on: types: [opened, synchronize, reopened, closed] paths: - "cpp/obs/**" + - "cpp/moq/**" - "sh/obs/**" + - "rs/moq-ffi/**" - "rs/moq-c/**" - "doc/lib/c/**" - "doc/lib/samples.sh" @@ -115,7 +118,7 @@ jobs: # cache.yml warms the default-branch store on `main`. Pull requests restore # it; the action refuses to publish from a pull request. It holds - # dev-profile artifacts, which is why `just obs ci` builds moq-c in debug + # dev-profile artifacts, which is why `just obs ci` builds moq-ffi in debug # rather than the preset's release. - name: Rust cache uses: ./.github/actions/rust-cache @@ -125,15 +128,13 @@ jobs: - name: Build run: nix develop .#obs --command just obs ci - # The plugin build above already produced the debug libmoq.a, so the - # cargo step inside is a no-op and this is just a cc and a run. - name: moq-c C fixtures run: nix develop .#obs --command just rs c-tests - # The release build of the OBS plugin, minus the published moq-c: build.sh - # without `--moq-c-release` links the in-tree rs/moq-c instead, so a PR is - # checked against its own C ABI. The obs-deps bundle (libobs, Qt6, ffmpeg) - # comes from buildspec.json at configure time, same as a tag build. + # The release build of the OBS plugin, minus the published C++ package: + # build.sh without `--moq-release` builds the in-tree cpp/moq instead, so a PR + # is checked against its own moq-ffi API. The obs-deps bundle (libobs, Qt6, + # ffmpeg) comes from buildspec.json at configure time, same as a tag build. release: name: OBS (${{ matrix.name }}) if: github.event.action != 'closed' @@ -162,6 +163,19 @@ jobs: - uses: ./.github/actions/just + # aws-lc-rs assembles its x86_64 crypto with NASM on Windows. + - name: Install NASM + if: runner.os == 'Windows' + shell: pwsh + run: | + choco install nasm -y --no-progress + "C:\Program Files\NASM" | Out-File -FilePath $env:GITHUB_PATH -Encoding utf8 -Append + + # The tag flake.nix pins; see the comment there. + - name: Install uniffi-bindgen-cpp + shell: bash + run: cargo install uniffi-bindgen-cpp --locked --git https://github.com/kixelated/uniffi-bindgen-cpp --tag v0.11.0-kixelated.4+v0.32.2 + - name: Build shell: bash env: diff --git a/.github/workflows/release-cpp.yml b/.github/workflows/release-cpp.yml new file mode 100644 index 00000000000..4d34c99a10f --- /dev/null +++ b/.github/workflows/release-cpp.yml @@ -0,0 +1,244 @@ +name: Release C++ + +# Builds the moq C++ package (cpp/moq) for every target and publishes one archive per +# target to a cpp-v GitHub release. The vcpkg and Conan packages fetch these. + +on: + push: + tags: + - "cpp-v*" + # Dry-run: build everything, publish nothing. Nightly calls every release + # workflow this way so a broken build surfaces before the tag is cut. + workflow_call: + workflow_dispatch: + +permissions: + contents: read + +jobs: + build: + name: Build (${{ matrix.target }}) + runs-on: ${{ matrix.os }} + + strategy: + fail-fast: false + matrix: + include: + - target: x86_64-unknown-linux-gnu + os: ubuntu-latest + - target: aarch64-unknown-linux-gnu + os: ubuntu-24.04-arm + - target: aarch64-apple-darwin + os: macos-latest + - target: x86_64-pc-windows-msvc + os: windows-latest + + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + + - uses: ./.github/actions/just + + - name: Install Rust + uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8 # stable + + # aws-lc-rs assembles its x86_64 crypto with NASM on Windows. + - name: Install NASM + if: runner.os == 'Windows' + shell: pwsh + run: | + choco install nasm -y --no-progress + "C:\Program Files\NASM" | Out-File -FilePath $env:GITHUB_PATH -Encoding utf8 -Append + + # The tag flake.nix pins; see the comment there. + - name: Install uniffi-bindgen-cpp + shell: bash + run: cargo install uniffi-bindgen-cpp --locked --git https://github.com/kixelated/uniffi-bindgen-cpp --tag v0.11.0-kixelated.4+v0.32.2 + + # The package version lives in cpp/moq/VERSION; a tag that disagrees is a mistake. + - name: Check version + if: github.ref_type == 'tag' + shell: bash + env: + TAG: ${{ github.ref_name }} + run: | + version=$(tr -d '[:space:]' < cpp/moq/VERSION) + if [[ "$TAG" != "cpp-v$version" ]]; then + echo "::error::tag $TAG does not match cpp/moq/VERSION ($version)" + exit 1 + fi + + - name: Build and package + shell: bash + env: + TARGET: ${{ matrix.target }} + run: just cpp package --target "$TARGET" --output dist + + - name: Upload artifact + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 + with: + name: moq-cpp-${{ matrix.target }} + path: dist/* + + release: + name: Release + needs: build + if: github.ref_type == 'tag' + runs-on: ubuntu-latest + permissions: + contents: write + + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + fetch-depth: 0 + persist-credentials: false + + - uses: ./.github/actions/just + + - name: Parse version + id: parse + run: just gh release parse-version cpp + + - name: Find previous tag + id: prev_tag + run: just gh release prev-tag cpp + + - name: Download artifacts + uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8 + with: + path: artifacts + merge-multiple: true + + - name: Create or update release + env: + GH_TOKEN: ${{ github.token }} + RELEASE_TAG: ${{ github.ref_name }} + RELEASE_TITLE: "moq C++ v${{ steps.parse.outputs.version }}" + RELEASE_PREV_TAG: ${{ steps.prev_tag.outputs.tag }} + run: just gh release create artifacts + + # The OBS plugin (cpp/obs) statically links the C++ package, so it ships with + # every release: rebuild it against the release just published (no second + # cargo build, guaranteed-matching bindings) and cut a parallel + # obs-moq-v release. The plugin's version is its own, + # cpp/obs/VERSION. Runs after `release` so the archives + # the plugin's CMake fetches are already live. + obs-build: + name: OBS plugin (${{ matrix.target }}) + needs: release + runs-on: ${{ matrix.os }} + + # macOS + Windows only. These link the ffmpeg the obs-deps bundle ships + # (the same one OBS uses), so the binaries are portable. A Linux build + # would link nix/distro ffmpeg and not be portable, so Linux is + # build-from-source (see doc/bin/obs.md) and ships no prebuilt binary. + strategy: + fail-fast: false + matrix: + include: + - target: aarch64-apple-darwin + os: macos-latest + - target: x86_64-pc-windows-msvc + # The plugin preset uses the Visual Studio 17 2022 generator. + os: windows-2022 + + steps: + # Shallow: this job only compiles, it never reads git history or tags. + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + + - uses: ./.github/actions/just + + # The build job already checked the tag against this file. + - name: Read C++ version + id: parse + shell: bash + run: echo "version=$(tr -d '[:space:]' < cpp/moq/VERSION)" >>"$GITHUB_OUTPUT" + + # Native Xcode / Visual Studio; libobs/Qt6/ffmpeg come from the obs-deps + # bundle and the C++ package from the release just published (--moq-release). + # build.sh versions the plugin from cpp/obs/VERSION. + - name: Build and package + shell: bash + env: + TARGET: ${{ matrix.target }} + MOQ_RELEASE: ${{ steps.parse.outputs.version }} + run: | + just obs package --target "$TARGET" --moq-release "$MOQ_RELEASE" --output dist + + - name: Upload artifact + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 + with: + name: obs-moq-${{ matrix.target }} + path: dist/* + + obs-release: + name: OBS plugin release + needs: obs-build + runs-on: ubuntu-latest + permissions: + contents: write + + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + fetch-depth: 0 + persist-credentials: false + + - uses: ./.github/actions/just + + - name: Read plugin version + id: parse + run: echo "version=$(tr -d '[:space:]' < cpp/obs/VERSION)" >>"$GITHUB_OUTPUT" + + # An obs-moq tag names one build. A later C++ release without a plugin + # version bump would overwrite that release's archives, so refuse it; a + # re-run of this same release finds its tag on this commit and carries on. + - name: Refuse a reused plugin version + env: + VERSION: ${{ steps.parse.outputs.version }} + run: | + tag="obs-moq-v${VERSION}" + if git rev-parse -q --verify "refs/tags/$tag" >/dev/null && + [ "$(git rev-parse "refs/tags/$tag^{commit}")" != "$(git rev-parse HEAD)" ]; then + echo "::error::$tag already names another commit; bump cpp/obs/VERSION" + exit 1 + fi + + # Only the obs-moq archives: the moq-cpp-* artifacts from the build job + # belong to the C++ release, not this one. + - name: Download artifacts + uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8 + with: + path: artifacts + pattern: obs-moq-* + merge-multiple: true + + # `just gh release prev-tag` keys off the current tag, but the trigger here is a + # cpp-v* tag, so it can't find the previous obs-moq tag. Take the newest + # existing obs-moq tag (excluding this version, for safe re-runs) so the + # release notes span only since the last obs-moq release. + - name: Find previous obs-moq tag + id: prev_tag + env: + VERSION: ${{ steps.parse.outputs.version }} + run: | + cur="obs-moq-v${VERSION}" + # awk (not grep -vx | head) so an empty result is exit 0, not 1: + # under `set -eo pipefail` the inaugural release has no prior tag and + # grep's exit 1 would fail the step. Empty tag is fine downstream. + prev=$(git tag --list 'obs-moq-v*' --sort=-v:refname | awk -v cur="$cur" '$0 != cur { print; exit }') + echo "tag=${prev}" >>"$GITHUB_OUTPUT" + + # Cut a parallel obs-moq-v release at this commit (gh creates + # the tag). + - name: Create or update release + env: + GH_TOKEN: ${{ github.token }} + RELEASE_TAG: "obs-moq-v${{ steps.parse.outputs.version }}" + RELEASE_TITLE: "obs-moq v${{ steps.parse.outputs.version }}" + RELEASE_PREV_TAG: ${{ steps.prev_tag.outputs.tag }} + run: just gh release create artifacts diff --git a/AGENTS.md b/AGENTS.md index f275e74c683..471b2095f01 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -98,7 +98,7 @@ The `quest` binary comes from the kixelated/quest flake input and serves the que | Change in | Also update | |---|---| -| `rs/moq-ffi` | `rs/moq-c`, `{py,swift,kt,dart}/`, `go/wrapper/moq/*.go` (the `go/ffi` and `dart/moq_ffi` bindings regenerate automatically, but a new method needs a hand-written wrapper too, like `py/moq-rs` or `dart/moq`), `doc/lib/{py,swift,kt,go,dart,c}` | +| `rs/moq-ffi` | `rs/moq-c`, `{py,swift,kt,dart}/`, `go/wrapper/moq/*.go` (the `go/ffi` and `dart/moq_ffi` bindings regenerate automatically, but a new method needs a hand-written wrapper too, like `py/moq-rs` or `dart/moq`), `cpp/moq`, `cpp/obs/src`, `doc/lib/{py,swift,kt,go,dart,c,cpp}`, `doc/bin/obs.md` | | `rs/moq-net` wire/API | `js/net`, `doc/concept`, `drafts/draft-lcurley-moq-lite.md` (if the wire spec changes) | | `rs/hang` catalog/container | `js/hang`, `doc/concept`, `drafts/draft-lcurley-moq-hang.md` (if the format spec changes) | | `rs/moq-auth` | `js/auth`, `doc/bin/relay/auth.md` | @@ -106,7 +106,7 @@ The `quest` binary comes from the kixelated/quest flake input and serves the que | `rs/moq-relay` config/behavior | `doc/bin/relay/` | | `rs/moq-cli` | `doc/bin/cli.md` | | `rs/moq-gst` | `doc/bin/gstreamer.md` | -| `rs/moq-c` C ABI (`moq.h`) | `cpp/obs/src`, `doc/bin/obs.md` | +| `rs/moq-c` C ABI (`moq.h`) | `doc/lib/c` | | `js/{watch,publish}` UI/API | `demo/web` if it consumes the API | | a kramdown-rfc construct new to `drafts/` | `doc/.vitepress/drafts.ts`, which translates the drafts into `/draft/` site pages | diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index f88011bf64e..00834b72505 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -97,4 +97,5 @@ Releases are cut separately; bump only when asked. Each package's version lives - **Python**: `py/moq-rs/pyproject.toml`, plus the matching `moq-rs` entry in the root `uv.lock`; `py/moq-ffi` follows the `moq-ffi-v*` tag and Rust crate. - **Swift**: `swift/VERSION`. **Kotlin**: `moq.version` in `kt/gradle.properties`. **Dart**: `version` in `dart/moq/pubspec.yaml`. Their FFI counterparts track the Rust crate. - **Go**: `go/wrapper/VERSION` holds a human-owned `MAJOR.MINOR` line; CI derives the patch, so only edit it for a breaking API. Leave the placeholder FFI version in `go.mod` alone. -- **OBS**: release builds take their version from the `moq-c-v*` tag that release-plz cuts for the `moq-c` crate (`cpp/obs/build.sh --moq-c-release`), not from any manifest, so there is nothing to bump. +- **C++**: `cpp/moq/VERSION`, human-owned; a `cpp-v` tag releases it. +- **OBS**: `cpp/obs/VERSION`, human-owned; each C++ release also cuts `obs-moq-v`, and refuses one whose version already names an earlier release. diff --git a/cpp/justfile b/cpp/justfile new file mode 100644 index 00000000000..963443c8fbe --- /dev/null +++ b/cpp/justfile @@ -0,0 +1,26 @@ +#!/usr/bin/env just --justfile +# +# The moq C++ package over rs/moq-ffi. Invoked from the repo root as `just cpp ` via +# the `mod cpp` import. The OBS plugin has its own module, `just obs`. + +set working-directory := '.' + +default: + just check + +# Build the package and run the probe and doc samples against its install. +check: + ../sh/cpp/check.sh + +# Runs in the caller's directory, so a relative `--output` means what it would +# to the script itself. + +# Build and package the C++ package for release. +[no-cd] +[positional-arguments] +package *args: + {{ quote(justfile_directory() / "cpp/moq/build.sh") }} "$@" + +# Remove the package builds. +clean: + rm -rf moq/build diff --git a/cpp/moq/.gitignore b/cpp/moq/.gitignore new file mode 100644 index 00000000000..8a468423f13 --- /dev/null +++ b/cpp/moq/.gitignore @@ -0,0 +1,2 @@ +# `just cpp check` builds, installs, and tests here. +/build/ diff --git a/cpp/moq/CMakeLists.txt b/cpp/moq/CMakeLists.txt new file mode 100644 index 00000000000..529fbf372b1 --- /dev/null +++ b/cpp/moq/CMakeLists.txt @@ -0,0 +1,128 @@ +cmake_minimum_required(VERSION 3.21) + +# The package version is human-owned, like go/wrapper/VERSION; release-cpp.yml tags cpp-v. +file(STRINGS "${CMAKE_CURRENT_LIST_DIR}/VERSION" _version LIMIT_COUNT 1) +if(NOT _version MATCHES "^[0-9]+\\.[0-9]+\\.[0-9]+$") + message(FATAL_ERROR "Unsupported version '${_version}' in cpp/moq/VERSION (want x.y.z)") +endif() + +project(moq-cpp VERSION ${_version} LANGUAGES CXX) + +# Builds moq-ffi as a staticlib with cargo and renders its bindings with uniffi-bindgen-cpp, +# the same no-Corrosion custom target rs/moq-c/CMakeLists.txt uses. The cargo profile follows +# CMAKE_BUILD_TYPE (release unless Debug), even on a multi-config generator (Visual Studio), +# whose --config never changes it: the staticlib links the release CRT either way. +set(_workspace "${CMAKE_CURRENT_SOURCE_DIR}/../..") +if(CMAKE_BUILD_TYPE STREQUAL "Debug") + set(_profile debug) + set(_cargo_flags "") +else() + set(_profile release) + set(_cargo_flags --release) +endif() +if(DEFINED ENV{CARGO_TARGET_DIR}) + set(_cargo_target "$ENV{CARGO_TARGET_DIR}") +else() + set(_cargo_target "${_workspace}/target") +endif() +if(WIN32) + set(MOQ_FFI_LIB "${_cargo_target}/${_profile}/moq_ffi.lib") +else() + set(MOQ_FFI_LIB "${_cargo_target}/${_profile}/libmoq_ffi.a") +endif() + +find_program(UNIFFI_BINDGEN_CPP uniffi-bindgen-cpp REQUIRED) + +# Rendered to a scratch directory and copied only when changed, so a no-op cargo build does +# not recompile the bindings. +set(_generated "${CMAKE_CURRENT_BINARY_DIR}/include/moq/ffi") +set(_generated_files moq.hpp moq.cpp moq_scaffolding.hpp uniffi_expected.hpp) +list(TRANSFORM _generated_files PREPEND "${_generated}/" OUTPUT_VARIABLE _generated_paths) +set(_render "${CMAKE_CURRENT_BINARY_DIR}/render") +list(TRANSFORM _generated_files PREPEND "${_render}/" OUTPUT_VARIABLE _render_paths) +add_custom_target(moq_ffi ALL + COMMAND cargo build --locked --package moq-ffi ${_cargo_flags} + COMMAND ${CMAKE_COMMAND} -E rm -rf "${_render}" + COMMAND "${UNIFFI_BINDGEN_CPP}" --library "${MOQ_FFI_LIB}" + --config "${CMAKE_CURRENT_SOURCE_DIR}/uniffi.toml" --out-dir "${_render}" + COMMAND ${CMAKE_COMMAND} -E make_directory "${_generated}" + COMMAND ${CMAKE_COMMAND} -E copy_if_different ${_render_paths} "${_generated}" + WORKING_DIRECTORY "${_workspace}" + BYPRODUCTS "${MOQ_FFI_LIB}" ${_generated_paths} + COMMENT "Building moq-ffi and rendering its C++ bindings" + VERBATIM +) +# System libraries the staticlib needs from an external linker. moq-ffi and moq-c share one +# dependency tree, so they share rs/moq-c/native-libs/, which says how to regenerate it. +if(APPLE) + set(_platform apple) +elseif(WIN32) + set(_platform windows) +else() + set(_platform linux) +endif() +file(STRINGS "${_workspace}/rs/moq-c/native-libs/${_platform}.txt" _lines) +set(MOQ_NATIVE_LIBS "") +foreach(_line IN LISTS _lines) + string(STRIP "${_line}" _line) + if(_line STREQUAL "" OR _line MATCHES "^#") + continue() + elseif(_line MATCHES "^framework:(.+)$") + list(APPEND MOQ_NATIVE_LIBS "-framework ${CMAKE_MATCH_1}") + else() + list(APPEND MOQ_NATIVE_LIBS "${_line}") + endif() +endforeach() + +# The bindings compile here at the including project's CMAKE_CXX_STANDARD (else the compiler's +# default, at least C++17) and with its flags; see src/moq.cpp. A standard raised only on a +# consumer's own target does not reach this library. +# Named moq-cpp so an in-tree build can sit beside moq-c's `moq` target. +add_library(moq-cpp STATIC src/moq.cpp) +add_library(moq::cpp ALIAS moq-cpp) +add_dependencies(moq-cpp moq_ffi) +target_compile_features(moq-cpp PUBLIC cxx_std_17) +target_include_directories(moq-cpp PUBLIC + "$" + "$" +) +find_package(Threads REQUIRED) +target_link_libraries(moq-cpp PUBLIC "${MOQ_FFI_LIB}" ${MOQ_NATIVE_LIBS} Threads::Threads) + +if(PROJECT_IS_TOP_LEVEL) + include(GNUInstallDirs) + include(CMakePackageConfigHelpers) + + install(DIRECTORY include/moq DESTINATION ${CMAKE_INSTALL_INCLUDEDIR}) + install(FILES ${_generated_paths} DESTINATION ${CMAKE_INSTALL_INCLUDEDIR}/moq/ffi) + install(FILES src/moq.cpp DESTINATION ${CMAKE_INSTALL_DATADIR}/moq-cpp) + install(FILES "${MOQ_FFI_LIB}" DESTINATION ${CMAKE_INSTALL_LIBDIR}) + + # The config takes MOQ_NATIVE_LIBS as a list; pkg-config wants linker flags. + get_filename_component(MOQ_FFI_LIB_NAME "${MOQ_FFI_LIB}" NAME) + set(MOQ_NATIVE_LIBS_PC "") + foreach(_lib IN LISTS MOQ_NATIVE_LIBS) + if(_lib MATCHES "^-framework ") + string(APPEND MOQ_NATIVE_LIBS_PC " ${_lib}") + else() + string(APPEND MOQ_NATIVE_LIBS_PC " -l${_lib}") + endif() + endforeach() + + configure_package_config_file(cmake/moq-cpp-config.cmake.in + "${CMAKE_CURRENT_BINARY_DIR}/moq-cpp-config.cmake" + INSTALL_DESTINATION ${CMAKE_INSTALL_LIBDIR}/cmake/moq-cpp + PATH_VARS CMAKE_INSTALL_INCLUDEDIR CMAKE_INSTALL_LIBDIR CMAKE_INSTALL_DATADIR + ) + write_basic_package_version_file("${CMAKE_CURRENT_BINARY_DIR}/moq-cpp-config-version.cmake" + COMPATIBILITY SameMinorVersion + ) + install(FILES + "${CMAKE_CURRENT_BINARY_DIR}/moq-cpp-config.cmake" + "${CMAKE_CURRENT_BINARY_DIR}/moq-cpp-config-version.cmake" + DESTINATION ${CMAKE_INSTALL_LIBDIR}/cmake/moq-cpp + ) + + configure_file(cmake/moq-cpp.pc.in "${CMAKE_CURRENT_BINARY_DIR}/moq-cpp.pc" @ONLY) + install(FILES "${CMAKE_CURRENT_BINARY_DIR}/moq-cpp.pc" DESTINATION ${CMAKE_INSTALL_LIBDIR}/pkgconfig) +endif() diff --git a/cpp/moq/README.md b/cpp/moq/README.md new file mode 100644 index 00000000000..6035d1274f0 --- /dev/null +++ b/cpp/moq/README.md @@ -0,0 +1,46 @@ +# moq C++ package + +C++17 bindings for [rs/moq-ffi](../../rs/moq-ffi): the sources `uniffi-bindgen-cpp` generates, plus `include/moq/moq.hpp`, a thin layer that renames them (`moq::MoqClient` is also `moq::Client`) and adds the executor, shutdown, and `co_await` glue. Nothing here re-implements a method; a shape problem is fixed in moq-ffi, where every binding benefits. User docs: [doc/lib/cpp](../../doc/lib/cpp/index.md). + +## Layout + +- `CMakeLists.txt` builds moq-ffi as a staticlib with cargo, renders the bindings into the build tree, and compiles them as the `moq::cpp` target. Use it with `add_subdirectory`, or install it for `find_package(moq-cpp)` and `moq-cpp.pc`. +- `src/moq.cpp` compiles the generated `moq/ffi/moq.cpp`. It ships as source so it builds with the consumer's standard and flags. +- `cmake/` holds the installed CMake package and pkg-config templates. +- `build.sh` builds the release archive `release-cpp.yml` publishes. +- `test/probe.cpp` exercises the installed package over a real QUIC session. + +## Checking + +`just cpp check` builds and installs the package, builds the probe against it with `find_package` at C++17 and C++23 and, outside Windows, with pkg-config at C++17, runs each, compiles the doc samples (also outside Windows), and fails if a generated type lacks its short alias. The probe builds with exceptions and RTTI disabled. + +The generator is a fork, [kixelated/uniffi-bindgen-cpp](https://github.com/kixelated/uniffi-bindgen-cpp). It carries LiveKit's async support ported to uniffi 0.32, plus the `error_style = "expected"` option `uniffi.toml` turns on. The dev shell provides it. Without Nix: + +```bash +cargo install uniffi-bindgen-cpp --locked \ + --git https://github.com/kixelated/uniffi-bindgen-cpp \ + --tag v0.11.0-kixelated.4+v0.32.2 +``` + +`flake.nix` pins the same tag and lists every other place that names it. + +## Shape + +Every generated type lives in `namespace moq`. Objects are `std::shared_ptr`; records and enums are values. + +- A fallible call returns `moq::expected`, which is `std::expected` on C++23 and a bundled `tl::expected` below it. The generated sources and every includer must agree on which, so `moq/moq.hpp` references a symbol named for the one it sees and `src/moq.cpp` defines the one it was built with: a mismatch fails to link. +- An async call returns `moq::Future`. Block on it with `get()` or `wait_for()`, attach a continuation with `std::move(future).then(executor, callback)`, which returns a `moq::Continuation`, or `co_await` it on C++20. +- `moq::Error` is a value holding a `std::variant` of its cases, and `to_string()` gives Rust's `Display` message. Nothing throws. A Rust panic aborts with a message on stderr, and so does reading an invalid future (see [Cancellation](#cancellation)). +- Futures are polled and continuations run on one process-wide executor thread unless the application installs its own with `moq::set_executor` before the first async call. `moq::shutdown()` stops the moq-ffi runtime thread (`moq_ffi_shutdown`), then the executor, before unloading the code a continuation could call into. + +## C++ standard + +The installed package compiles `moq.cpp` inside the consumer's project, at its `CMAKE_CXX_STANDARD` (else the compiler's default, at least C++17). To use C++23, set `CMAKE_CXX_STANDARD 23` before `find_package(moq-cpp)`, or pass `-DCMAKE_CXX_STANDARD=23`. `target_compile_features(app PRIVATE cxx_std_23)` alone does not reach the separate `moq-cpp` library, so `` and the bindings pick different `expected` types and the link fails on `moq_abi_std_expected` or `moq_abi_tl_expected`. `just cpp check` builds the probe at C++23 both ways. + +## Cancellation + +Cancelling a future (`cancel()`, destroying it, destroying the `moq::Continuation` returned by `then`, or destroying a coroutine suspended on it) drops the Rust future. Native moq-ffi runs each async call as a spawned task that holds an `AbortOnDrop` on it (`rs/moq-ffi/src/ffi.rs`), so dropping the future aborts the work at its next await point instead of letting it finish unobserved. The continuation of a cancelled future never runs. + +Like `std::future`, `valid()` is false once `get()`, `then()`, `cancel()`, or a move took the future's state. Reading an invalid future aborts with "it was consumed or cancelled", so check `valid()` first when a future may have been cancelled, consumed, or moved from. + +The abort discards the operation, not the handle it ran on, so the next call on that object works. Where a method keeps partial progress, its doc comment says so. For example, a cancelled `read_frame` leaves the current group for the next read. Every write in moq-ffi (`write_frame`, `finish`, `abort`) is synchronous, so cancelling a future can never tear a write. Only the async operations (reads, subscribes, connects, accepts, and `reject`) can be cut short. diff --git a/cpp/moq/VERSION b/cpp/moq/VERSION new file mode 100644 index 00000000000..6e8bf73aa55 --- /dev/null +++ b/cpp/moq/VERSION @@ -0,0 +1 @@ +0.1.0 diff --git a/cpp/moq/build.sh b/cpp/moq/build.sh new file mode 100755 index 00000000000..e9a3c14e342 --- /dev/null +++ b/cpp/moq/build.sh @@ -0,0 +1,86 @@ +#!/usr/bin/env bash +set -euo pipefail + +# Build the moq C++ package for the host and archive it for release. +# Usage: ./build.sh [--target TARGET] [--output DIR] +# +# The archive holds what `cmake --install` lays out: the moq-ffi staticlib, the wrapper and +# generated headers, the generated source, the CMake package, and moq-cpp.pc. Needs cargo, cmake, +# and uniffi-bindgen-cpp on PATH. The version is cpp/moq/VERSION. + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" + +# Shrink the release staticlib the way rs/moq-c/build.sh does: thin LTO with one codegen +# unit dead-strips the monomorphizations Rust bakes into a staticlib. +export CARGO_PROFILE_RELEASE_LTO="${CARGO_PROFILE_RELEASE_LTO:-thin}" +export CARGO_PROFILE_RELEASE_CODEGEN_UNITS="${CARGO_PROFILE_RELEASE_CODEGEN_UNITS:-1}" + +TARGET="" +OUTPUT_DIR="dist" + +while [[ $# -gt 0 ]]; do + case $1 in + --target) + TARGET="$2" + shift 2 + ;; + --output) + OUTPUT_DIR="$2" + shift 2 + ;; + -h | --help) + echo "Usage: $0 [--target TARGET] [--output DIR]" + exit 0 + ;; + *) + echo "Unknown option: $1" >&2 + exit 1 + ;; + esac +done + +VERSION=$(tr -d '[:space:]' <"$SCRIPT_DIR/VERSION") +HOST_TARGET=$(rustc -vV | sed -n 's/^host: //p') +TARGET="${TARGET:-$HOST_TARGET}" + +# CMake builds for the host. A mismatch would silently mislabel the archive. +if [[ "$TARGET" != "$HOST_TARGET" ]]; then + echo "Error: unsupported cross ($HOST_TARGET -> $TARGET); refusing to mislabel the archive." >&2 + exit 1 +fi + +NAME="moq-cpp-${VERSION}-${TARGET}" +mkdir -p "$OUTPUT_DIR" +OUTPUT_DIR="$(cd "$OUTPUT_DIR" && pwd)" +PACKAGE_DIR="$OUTPUT_DIR/$NAME" +BUILD_DIR="$SCRIPT_DIR/build/release" + +echo "Packaging $NAME..." +rm -rf "$PACKAGE_DIR" +# lib/, not the lib64/ GNUInstallDirs picks on some hosts, so every archive has one layout. +cmake -S "$SCRIPT_DIR" -B "$BUILD_DIR" -DCMAKE_BUILD_TYPE=Release -DCMAKE_INSTALL_LIBDIR=lib +cmake --build "$BUILD_DIR" --config Release +cmake --install "$BUILD_DIR" --config Release --prefix "$PACKAGE_DIR" + +# A placeholder the configure step missed would reach a consumer's linker as a literal. +# Comments are skipped: configure_package_config_file names @PACKAGE_INIT@ in one. +if grep -rnE '^[^#]*@[A-Z_]+@' "$PACKAGE_DIR/lib/cmake" "$PACKAGE_DIR/lib/pkgconfig"; then + echo "Error: unsubstituted placeholder in the package config (see above)" >&2 + exit 1 +fi + +cd "$OUTPUT_DIR" +if [[ "$TARGET" == *"-windows-"* ]]; then + ARCHIVE="$NAME.zip" + if command -v 7z &>/dev/null; then + 7z a "$ARCHIVE" "$NAME" + else + zip -r "$ARCHIVE" "$NAME" + fi +else + ARCHIVE="$NAME.tar.gz" + tar -czf "$ARCHIVE" "$NAME" +fi +rm -rf "$NAME" + +echo "Created: $OUTPUT_DIR/$ARCHIVE" diff --git a/cpp/moq/cmake/moq-cpp-config.cmake.in b/cpp/moq/cmake/moq-cpp-config.cmake.in new file mode 100644 index 00000000000..e49d90440af --- /dev/null +++ b/cpp/moq/cmake/moq-cpp-config.cmake.in @@ -0,0 +1,28 @@ +# find_package(moq-cpp): the C++ bindings for moq-ffi as the moq::cpp target. Every moq +# package exports into the one moq:: namespace, and the package names keep installs apart. + +@PACKAGE_INIT@ + +if(NOT TARGET moq::cpp) + find_package(Threads REQUIRED) + + # The prebuilt Rust staticlib, and the system libraries an external linker must add for + # it, baked in for the platform this package was built for. + add_library(moq::ffi STATIC IMPORTED) + set_target_properties(moq::ffi PROPERTIES + IMPORTED_LOCATION "@PACKAGE_CMAKE_INSTALL_LIBDIR@/@MOQ_FFI_LIB_NAME@" + INTERFACE_LINK_LIBRARIES "@MOQ_NATIVE_LIBS@;Threads::Threads" + ) + + # The generated bindings compile here, in the consumer's project, at its CMAKE_CXX_STANDARD + # (else the compiler's default, at least C++17) and with its flags. A standard raised only + # on the consumer's own target does not reach this library, so and the + # bindings disagree on uniffi::expected and the link fails: set CMAKE_CXX_STANDARD instead. + add_library(moq-cpp STATIC "@PACKAGE_CMAKE_INSTALL_DATADIR@/moq-cpp/moq.cpp") + target_compile_features(moq-cpp PUBLIC cxx_std_17) + target_include_directories(moq-cpp PUBLIC "@PACKAGE_CMAKE_INSTALL_INCLUDEDIR@") + target_link_libraries(moq-cpp PUBLIC moq::ffi) + add_library(moq::cpp ALIAS moq-cpp) +endif() + +check_required_components(moq-cpp) diff --git a/cpp/moq/cmake/moq-cpp.pc.in b/cpp/moq/cmake/moq-cpp.pc.in new file mode 100644 index 00000000000..4c59f60249f --- /dev/null +++ b/cpp/moq/cmake/moq-cpp.pc.in @@ -0,0 +1,12 @@ +# The generated bindings ship as source: compile `pkg-config --variable=sources moq-cpp` with the +# same standard and flags as the code that includes . +prefix=${pcfiledir}/../.. +libdir=${prefix}/@CMAKE_INSTALL_LIBDIR@ +includedir=${prefix}/@CMAKE_INSTALL_INCLUDEDIR@ +sources=${prefix}/@CMAKE_INSTALL_DATADIR@/moq-cpp/moq.cpp + +Name: moq-cpp +Description: Media over QUIC for C++ +Version: @PROJECT_VERSION@ +Cflags: -I${includedir} +Libs: -L${libdir} -lmoq_ffi@MOQ_NATIVE_LIBS_PC@ diff --git a/cpp/moq/include/moq/moq.hpp b/cpp/moq/include/moq/moq.hpp new file mode 100644 index 00000000000..f9fe75c22ee --- /dev/null +++ b/cpp/moq/include/moq/moq.hpp @@ -0,0 +1,286 @@ +// The moq C++ package: the generated moq-ffi bindings under short names, plus the +// executor, shutdown, and coroutine glue they need. Every type and method is generated; +// this header only renames and wires them. +#pragma once + +#include + +#include +#include +#include + +#if defined(__cpp_impl_coroutine) && __has_include() +#include +#include +#include +#include +#include +#define MOQ_COROUTINES 1 +#endif + +// Not a uniffi export, so the generated bindings do not declare it; see moq::shutdown. +extern "C" void moq_ffi_shutdown(void); + +// uniffi::expected is std::expected or tl::expected depending on the standard, and the +// generated sources (src/moq.cpp) and every includer must agree. The bindings define the +// symbol for the type they saw and every includer references the one it sees, so a mismatch +// fails to link instead of corrupting memory. +#if defined(__cpp_lib_expected) && __cpp_lib_expected >= 202211L +#define MOQ_ABI moq_abi_std_expected +#define MOQ_ABI_NAME "moq_abi_std_expected" +#else +#define MOQ_ABI moq_abi_tl_expected +#define MOQ_ABI_NAME "moq_abi_tl_expected" +#endif +extern "C" const int MOQ_ABI; +#if defined(MOQ_IMPLEMENTATION) +extern "C" const int MOQ_ABI = 1; +#elif defined(_MSC_VER) +#pragma comment(linker, "/include:" MOQ_ABI_NAME) +#else +namespace moq::detail { +[[maybe_unused]] __attribute__((used)) static const int *const abi = &MOQ_ABI; +} // namespace moq::detail +#endif + +namespace moq { + +// Every generated `moq::MoqFoo` is also `moq::Foo`. `just cpp check` fails when one is missing. +using AnnounceConsumer = MoqAnnounceConsumer; +using AnnouncedBroadcast = MoqAnnouncedBroadcast; +using AudioCodec = MoqAudioCodec; +using AudioConsumer = MoqAudioConsumer; +using AudioProducer = MoqAudioProducer; +using Bandwidth = MoqBandwidth; +using BroadcastConsumer = MoqBroadcastConsumer; +using BroadcastDynamic = MoqBroadcastDynamic; +using BroadcastProducer = MoqBroadcastProducer; +using BroadcastRequest = MoqBroadcastRequest; +using CatalogConsumer = MoqCatalogConsumer; +using Client = MoqClient; +using ContainerProducer = MoqContainerProducer; +using ContainerStreamProducer = MoqContainerStreamProducer; +using FlateSnapshotProducer = MoqFlateSnapshotProducer; +using FlateStreamProducer = MoqFlateStreamProducer; +using GroupConsumer = MoqGroupConsumer; +using GroupProducer = MoqGroupProducer; +using GroupRequest = MoqGroupRequest; +using JsonSnapshotConsumer = MoqJsonSnapshotConsumer; +using JsonSnapshotProducer = MoqJsonSnapshotProducer; +using JsonStreamConsumer = MoqJsonStreamConsumer; +using JsonStreamProducer = MoqJsonStreamProducer; +using MediaConsumer = MoqMediaConsumer; +using MediaGroupConsumer = MoqMediaGroupConsumer; +using MediaProducer = MoqMediaProducer; +using MediaStreamProducer = MoqMediaStreamProducer; +using OriginConsumer = MoqOriginConsumer; +using OriginDynamic = MoqOriginDynamic; +using OriginProducer = MoqOriginProducer; +using Request = MoqRequest; +using Reservation = MoqReservation; +using Server = MoqServer; +using Session = MoqSession; +using TrackConsumer = MoqTrackConsumer; +using TrackDemand = MoqTrackDemand; +using TrackDynamic = MoqTrackDynamic; +using TrackProducer = MoqTrackProducer; +using TrackRequest = MoqTrackRequest; +using VideoConsumer = MoqVideoConsumer; +using VideoProducer = MoqVideoProducer; +using Announce = MoqAnnounce; +using AnnounceConfig = MoqAnnounceConfig; +using Audio = MoqAudio; +using AudioDecoderOutput = MoqAudioDecoderOutput; +using AudioEncoderInput = MoqAudioEncoderInput; +using AudioEncoderOutput = MoqAudioEncoderOutput; +using AudioFrame = MoqAudioFrame; +using AudioInit = MoqAudioInit; +using Backoff = MoqBackoff; +using Catalog = MoqCatalog; +using ConnectionStats = MoqConnectionStats; +using ContainerInit = MoqContainerInit; +using Datagram = MoqDatagram; +using Dimensions = MoqDimensions; +using FetchGroupOptions = MoqFetchGroupOptions; +using FlateConfig = MoqFlateConfig; +using Frame = MoqFrame; +using JsonSnapshotConfig = MoqJsonSnapshotConfig; +using JsonStreamConfig = MoqJsonStreamConfig; +using MediaFrame = MoqMediaFrame; +using OriginConfig = MoqOriginConfig; +using ProtocolError = MoqProtocolError; +using Route = MoqRoute; +using Subscription = MoqSubscription; +using TrackInfo = MoqTrackInfo; +using Video = MoqVideo; +using VideoDecodedFrame = MoqVideoDecodedFrame; +using VideoDecoderOutput = MoqVideoDecoderOutput; +using VideoEncoderInput = MoqVideoEncoderInput; +using VideoEncoderOutput = MoqVideoEncoderOutput; +using VideoFrame = MoqVideoFrame; +using VideoHint = MoqVideoHint; +using VideoInit = MoqVideoInit; +using VideoProperties = MoqVideoProperties; +using AnnounceEvent = MoqAnnounceEvent; +using AudioFormat = MoqAudioFormat; +using AudioSampleFormat = MoqAudioSampleFormat; +using ConnectionStatus = MoqConnectionStatus; +using Container = MoqContainer; +using ContainerFormat = MoqContainerFormat; +using Error = MoqError; +using ErrorScope = MoqErrorScope; +using ProtocolKind = MoqProtocolKind; +using Transport = MoqTransport; +using VideoCodec = MoqVideoCodec; +using VideoEncoderKind = MoqVideoEncoderKind; +using VideoFormat = MoqVideoFormat; +using VideoPixelFormat = MoqVideoPixelFormat; +using VideoSurface = MoqVideoSurface; + +// The value of a fallible call, or the Error that stopped it. std::expected on C++23. +template +using expected = ::uniffi::expected; + +// Wraps an Error so it converts to any moq::expected. +using ::uniffi::unexpected; + +// A pending async call: block with get() or wait_for(), or attach a continuation with then(). +template +using Future = ::uniffi::Future; + +// Owns a continuation attached with then(); destroying it cancels the call. +using Continuation = ::uniffi::FutureContinuation; + +// One unit of work handed to an Executor. +using Task = ::uniffi::AsyncTask; + +// Runs a Task somewhere and returns true, or returns false to refuse it. +using Executor = ::uniffi::AsyncDispatcher; + +// Runs a continuation on the thread that completed the future, for work that never blocks. +inline bool inline_executor(Task task) { + task(); + return true; +} + +// Replaces the default executor that polls futures, before the first async call. +// `shutdown` stops accepting tasks and returns once none can still run; moq::shutdown calls it. +inline void set_executor(Executor executor, std::function shutdown) noexcept { + ::uniffi::set_async_dispatcher(std::move(executor), std::move(shutdown)); +} + +// Stops the moq-ffi runtime thread, then the executor, before the process or module goes away. +// Pending calls resolve Cancelled. Call it once from a thread that is not running a continuation. +inline void shutdown() noexcept { + moq_ffi_shutdown(); + ::uniffi::shutdown_async_dispatcher(); +} + +// Sets the log level: "error", "warn", "info", "debug", "trace", or "". Errors if called twice. +inline expected log_level(const std::string &level) { + return moq_log_level(level); +} + +#ifdef MOQ_COROUTINES +namespace detail { + +// Resumes a coroutine with a future's result. Completion and cancellation share +// `state_`. The continuation resumes a claimed handle before it drops the mutex, +// so the destructor cannot free the frame between the claim and the resume. +template +class Awaiter { +public: + using Output = typename Future::Output; + + explicit Awaiter(Future future) noexcept: + future_(std::move(future)), state_(std::make_shared()) {} + + Awaiter(const Awaiter &) = delete; + Awaiter &operator=(const Awaiter &) = delete; + + // Destroying a suspended coroutine cancels the call, and it never resumes. + ~Awaiter() { + std::optional continuation; + { + // Recursive: resume() calls this on the completing thread while that + // thread still holds the mutex inside the continuation. + std::lock_guard guard(state_->mutex); + if (state_->resuming && state_->resume_on == std::this_thread::get_id()) { + return; + } + state_->destroying = true; + state_->handle = nullptr; + continuation = std::move(state_->continuation); + } + } + + bool await_ready() const noexcept { + return false; + } + + bool await_suspend(std::coroutine_handle<> handle) noexcept { + auto continuation = std::move(future_).then(inline_executor, [state = state_](Output output) { + std::lock_guard guard(state->mutex); + state->output.emplace(std::move(output)); + if (state->destroying || !state->handle) { + return; + } + auto resume = std::exchange(state->handle, nullptr); + state->resuming = true; + state->resume_on = std::this_thread::get_id(); + struct Clear { + State &state; + ~Clear() { + state.resuming = false; + } + } clear{*state}; + resume.resume(); + }); + + std::lock_guard guard(state_->mutex); + state_->continuation.emplace(std::move(continuation)); + if (state_->output) { + // Completed before suspending: carry on without a round trip. + return false; + } + state_->handle = handle; + return true; + } + + Output await_resume() noexcept { + std::lock_guard guard(state_->mutex); + return std::move(*state_->output); + } + +private: + struct State { + std::recursive_mutex mutex; + std::coroutine_handle<> handle; + std::optional output; + std::optional continuation; + bool destroying = false; + bool resuming = false; + std::thread::id resume_on{}; + }; + + Future future_; + std::shared_ptr state_; +}; + +} // namespace detail +#endif + +} // namespace moq + +#ifdef MOQ_COROUTINES +// Found by argument-dependent lookup on uniffi::Future. +namespace uniffi { + +// `co_await future` resumes with the future's result on the thread that completed it. +template +moq::detail::Awaiter operator co_await(Future &&future) noexcept { + return moq::detail::Awaiter(std::move(future)); +} + +} // namespace uniffi +#endif diff --git a/cpp/moq/src/moq.cpp b/cpp/moq/src/moq.cpp new file mode 100644 index 00000000000..7502a9f3ee4 --- /dev/null +++ b/cpp/moq/src/moq.cpp @@ -0,0 +1,6 @@ +// Compiles the generated bindings. Build it with the same C++ standard as the code that +// includes ; the MOQ_ABI symbol below turns a mismatch into a link error. +#define MOQ_IMPLEMENTATION +#include + +#include diff --git a/cpp/moq/test/CMakeLists.txt b/cpp/moq/test/CMakeLists.txt new file mode 100644 index 00000000000..440d6e93b0b --- /dev/null +++ b/cpp/moq/test/CMakeLists.txt @@ -0,0 +1,32 @@ +# A consumer of the installed package: `just cpp check` installs cpp/moq to a prefix, then +# builds this with find_package(moq-cpp) at C++17 and C++23, the standard given as +# -DCMAKE_CXX_STANDARD or as -DPROBE_CXX_STANDARD, which sets it in the project and also +# requires it on the app's own target, the way a consumer following doc/lib/cpp would. +cmake_minimum_required(VERSION 3.21) +project(moq-probe LANGUAGES CXX) + +if(PROBE_CXX_STANDARD) + set(CMAKE_CXX_STANDARD ${PROBE_CXX_STANDARD}) +endif() +if(NOT CMAKE_CXX_STANDARD) + set(CMAKE_CXX_STANDARD 17) +endif() +set(CMAKE_CXX_STANDARD_REQUIRED ON) +set(CMAKE_CXX_EXTENSIONS OFF) + +# No exceptions and no RTTI, the way Unreal builds. Set before find_package so the bindings, +# which compile in this project, get the same flags. +if(MSVC) + add_compile_options(/permissive- /EHs-c- /GR-) + add_compile_definitions(_HAS_EXCEPTIONS=0) +else() + add_compile_options(-Wall -Wextra -Werror -pedantic-errors -fno-exceptions -fno-rtti) +endif() + +find_package(moq-cpp REQUIRED) + +add_executable(probe probe.cpp) +target_link_libraries(probe PRIVATE moq::cpp) +if(PROBE_CXX_STANDARD) + target_compile_features(probe PRIVATE cxx_std_${PROBE_CXX_STANDARD}) +endif() diff --git a/cpp/moq/test/doc-samples.hpp b/cpp/moq/test/doc-samples.hpp new file mode 100644 index 00000000000..4e691e91ef4 --- /dev/null +++ b/cpp/moq/test/doc-samples.hpp @@ -0,0 +1,34 @@ +// Inputs the samples in doc/lib/cpp/index.md leave undefined, and the ok() helper its +// Example section shows. `just cpp check` compiles every sample against the installed +// package with this included first. +#pragma once + +#include + +#include +#include +#include +#include +#include + +template +T ok(moq::expected result) { + if (!result) { + std::fprintf(stderr, "moq: %s\n", result.error().to_string().c_str()); + std::abort(); + } + return std::move(*result); +} + +inline void ok(moq::expected result) { + if (!result) { + std::fprintf(stderr, "moq: %s\n", result.error().to_string().c_str()); + std::abort(); + } +} + +inline std::shared_ptr session; +inline std::shared_ptr media; +inline std::vector opus_init; +inline std::vector packet; +inline std::vector rgba; diff --git a/cpp/moq/test/probe.cpp b/cpp/moq/test/probe.cpp new file mode 100644 index 00000000000..012274c03e4 --- /dev/null +++ b/cpp/moq/test/probe.cpp @@ -0,0 +1,229 @@ +// Exercises the installed package end to end over a real QUIC session: connect, subscribe, +// read a frame through a future, a continuation, and (on C++20) a coroutine, cancel pending +// reads, and observe errors as returned values. Built with exceptions and RTTI disabled. + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace std::chrono_literals; + +// Unlike assert, this survives a release build. +#define CHECK(expr) \ + do { \ + if (!(expr)) { \ + std::fprintf(stderr, "%s:%d: CHECK failed: %s\n", __FILE__, __LINE__, #expr); \ + std::abort(); \ + } \ + } while (0) + +namespace { + +[[noreturn]] void fail(const char *what, const moq::Error &error) { + std::fprintf(stderr, "%s failed: %s\n", what, error.to_string().c_str()); + std::abort(); +} + +// Unwraps a result the probe expects to succeed. +template +T ok(moq::expected result, const char *what) { + if (!result) { + fail(what, result.error()); + } + return std::move(*result); +} + +void ok(moq::expected result, const char *what) { + if (!result) { + fail(what, result.error()); + } +} + +std::vector bytes(const std::string &text) { + return std::vector(text.begin(), text.end()); +} + +// Blocks main until a continuation or coroutine on the executor thread reports back. +class Latch { +public: + // Notifies under the lock: the waiter may destroy the latch as soon as it sees done_. + void set() { + std::lock_guard guard(mutex_); + done_ = true; + ready_.notify_all(); + } + + bool wait_for(std::chrono::milliseconds timeout) { + std::unique_lock lock(mutex_); + return ready_.wait_for(lock, timeout, [this] { return done_; }); + } + +private: + std::mutex mutex_; + std::condition_variable ready_; + bool done_ = false; +}; + +#ifdef MOQ_COROUTINES +// Just enough of a coroutine type to start one from main. It frees itself when it finishes, +// and main destroys one that is still suspended. +struct Coroutine { + struct promise_type { + Coroutine get_return_object() noexcept { + return Coroutine{std::coroutine_handle::from_promise(*this)}; + } + std::suspend_never initial_suspend() noexcept { return {}; } + std::suspend_never final_suspend() noexcept { return {}; } + void return_void() noexcept {} + void unhandled_exception() noexcept { std::abort(); } + }; + + std::coroutine_handle handle; +}; + +Coroutine read_one(std::shared_ptr consumer, std::optional &out, Latch &done) { + auto frame = co_await consumer->read_frame(); + out = ok(std::move(frame), "co_await read_frame"); + done.set(); +} +#endif + +} // namespace + +int main() { + // A synchronous error is a returned value, not an exception. + auto unbound = moq::Server::init(); + auto fingerprints = unbound->cert_fingerprints(); + CHECK(!fingerprints); + CHECK(std::holds_alternative(fingerprints.error().get_variant())); + // It prints the message Rust's Display gives. + CHECK(fingerprints.error().to_string() == "bind: not listening; call listen() first"); + + // Publisher: a server whose origin serves one broadcast with one track. + auto origin = moq::OriginProducer::init({}); + auto broadcast = ok(origin->create_broadcast("probe"), "create_broadcast"); + auto track = ok(broadcast->publish_track("data", std::nullopt), "publish_track"); + ok(broadcast->announce({}), "announce"); + + auto server = moq::Server::init(); + ok(server->set_bind("127.0.0.1:0"), "set_bind"); + ok(server->set_tls_generate({"localhost"}), "set_tls_generate"); + ok(server->set_publish(origin), "set_publish"); + auto addr = ok(server->listen().get(), "listen"); + + // Both halves of the handshake are futures, so they run concurrently while this + // thread blocks on one at a time. + auto client = moq::Client::init(); + ok(client->set_tls_verify(false), "set_tls_verify"); + auto accepting = server->accept(); + auto connecting = client->connect("https://" + addr); + auto request = ok(accepting.get(), "accept"); + CHECK(request != nullptr); + auto served = ok(request->accept().get(), "request accept"); + auto session = ok(connecting.get(), "connect"); + + // Subscriber: resolve the announced broadcast and subscribe to its track. + auto announced = ok(session->consume()->announced_broadcast("probe"), "announced_broadcast"); + auto remote = ok(announced->available().get(), "available"); + auto consumer = ok(remote->subscribe_track("data", std::nullopt).get(), "subscribe_track"); + + // Cancel a read that has nothing to deliver yet. + auto pending = consumer->read_frame(); + CHECK(pending.wait_for(50ms) == std::future_status::timeout); + CHECK(pending.valid()); + pending.cancel(); + CHECK(!pending.valid()); + + // A move takes the state, so the source is invalid and the destination is not. + { + auto source = consumer->read_frame(); + CHECK(source.valid()); + auto destination = std::move(source); + CHECK(!source.valid()); + CHECK(destination.valid()); + } + + // The consumer survives the cancelled read and delivers the next frame. + auto reading = consumer->read_frame(); + ok(track->write_frame({bytes("hello"), 1000}), "write_frame"); + auto frame = ok(reading.get(), "read_frame"); + // get() consumed the state. + CHECK(!reading.valid()); + CHECK(frame.has_value()); + CHECK(frame->payload == bytes("hello")); + CHECK(frame->timestamp_us == 1000); + + // A continuation receives the next frame on the executor. + { + Latch done; + std::optional received; + auto reading_next = consumer->read_frame(); + auto continuation = std::move(reading_next).then(moq::inline_executor, [&](moq::expected> result) { + received = ok(std::move(result), "then read_frame"); + done.set(); + }); + // then() took the future's state. + CHECK(!reading_next.valid()); + ok(track->write_frame({bytes("then"), 2000}), "write_frame"); + CHECK(done.wait_for(5s)); + CHECK(received && received->payload == bytes("then")); + } + +#ifdef MOQ_COROUTINES + // Destroying a coroutine suspended on a read cancels the read, and it never resumes. + { + Latch done; + std::optional received; + auto abandoned = read_one(consumer, received, done); + CHECK(!done.wait_for(50ms)); + abandoned.handle.destroy(); + } + + // A coroutine resumes with the next frame. + { + Latch done; + std::optional received; + auto reader = read_one(consumer, received, done); + ok(track->write_frame({bytes("co_await"), 3000}), "write_frame"); + CHECK(done.wait_for(5s)); + CHECK(received && received->payload == bytes("co_await")); + (void)reader; + } +#endif + + // An async error arrives through the future as a returned value too: this track + // finished, so reading past its end reports it. + ok(track->finish(), "track finish"); + auto ended = consumer->read_frame().get(); + CHECK(ended && !ended->has_value()); + + auto closed = moq::Client::init(); + closed->cancel(); + auto refused = closed->connect("https://" + addr).get(); + CHECK(!refused); + CHECK(std::holds_alternative(refused.error().get_variant())); + + ok(broadcast->close(), "broadcast close"); + session->cancel(0); + served->cancel(0); + server->cancel(); + + // Stop the runtime and the executor before the process tears down. The handles above + // stay safe to drop afterwards. + moq::shutdown(); + +#ifdef MOQ_COROUTINES + std::printf("probe: ok (with coroutines)\n"); +#else + std::printf("probe: ok\n"); +#endif + return 0; +} diff --git a/cpp/moq/uniffi.toml b/cpp/moq/uniffi.toml new file mode 100644 index 00000000000..3dccc7caa84 --- /dev/null +++ b/cpp/moq/uniffi.toml @@ -0,0 +1,4 @@ +# uniffi-bindgen-cpp config for rs/moq-ffi. Errors are returned, never thrown, so the +# bindings build with exceptions disabled (Unreal's default). +[bindings.cpp] +error_style = "expected" diff --git a/cpp/obs/CMakeLists.txt b/cpp/obs/CMakeLists.txt index b24656608b2..0fefebd895c 100644 --- a/cpp/obs/CMakeLists.txt +++ b/cpp/obs/CMakeLists.txt @@ -1,35 +1,35 @@ cmake_minimum_required(VERSION 3.28) -# Build the in-tree moq-c by default. Release-only builds clear MOQ_LOCAL and -# must select the published archive explicitly, so this never silently drifts -# to an old static version. -set(MOQ_LOCAL "${CMAKE_CURRENT_SOURCE_DIR}/../.." CACHE PATH "Path to the moq repo (builds rs/moq-c in-tree)") -set(MOQ_VERSION "" CACHE STRING "Published moq-c version (required without MOQ_LOCAL)") +# Build the in-tree C++ package (cpp/moq) by default. Release-only builds clear +# MOQ_LOCAL and must select the published archive explicitly, so this never +# silently drifts to an old static version. +set(MOQ_LOCAL "${CMAKE_CURRENT_SOURCE_DIR}/../.." CACHE PATH "Path to the moq repo (builds cpp/moq in-tree)") +set(MOQ_VERSION "" CACHE STRING "Published moq-cpp version (required without MOQ_LOCAL)") # One predicate for both the version derivation here and the link/fetch choice # below, so the two can't drift apart. set(_moq_local_build FALSE) -if(MOQ_LOCAL AND EXISTS "${MOQ_LOCAL}/rs/moq-c/CMakeLists.txt") +if(MOQ_LOCAL AND EXISTS "${MOQ_LOCAL}/cpp/moq/CMakeLists.txt") set(_moq_local_build TRUE) endif() if(_moq_local_build) - set(_moq_manifest "${MOQ_LOCAL}/rs/moq-c/Cargo.toml") - file(STRINGS "${_moq_manifest}" _moq_version_lines REGEX "^version = \"[^\"]+\"$") - list(LENGTH _moq_version_lines _moq_version_count) - if(NOT _moq_version_count EQUAL 1) - message(FATAL_ERROR "Expected one package version in ${_moq_manifest}, found ${_moq_version_count}") + file(STRINGS "${MOQ_LOCAL}/cpp/moq/VERSION" _moq_link_version LIMIT_COUNT 1) + if(NOT _moq_link_version MATCHES "^[0-9]+\\.[0-9]+\\.[0-9]+$") + message(FATAL_ERROR "Unsupported version '${_moq_link_version}' in ${MOQ_LOCAL}/cpp/moq/VERSION") endif() - list(GET _moq_version_lines 0 _moq_version_line) - string(REGEX REPLACE "^version = \"([^\"]+)\"$" "\\1" _moq_link_version "${_moq_version_line}") else() if(NOT MOQ_VERSION) - message(FATAL_ERROR "MOQ_VERSION is required when MOQ_LOCAL does not contain rs/moq-c") + message(FATAL_ERROR "MOQ_VERSION is required when MOQ_LOCAL does not contain cpp/moq") endif() set(_moq_link_version "${MOQ_VERSION}") endif() include("${CMAKE_CURRENT_SOURCE_DIR}/cmake/common/bootstrap.cmake" NO_POLICY_SCOPE) +# The template turns this on for every directory, which puts cpp/obs and cpp/moq on the +# include path; on case-insensitive filesystems their VERSION files then answer +# `#include `. Every plugin include is a same-directory quoted one, so nothing needs it. +set(CMAKE_INCLUDE_CURRENT_DIR FALSE) project(${_name} VERSION ${_version}) @@ -67,36 +67,20 @@ endif() target_link_libraries(obs-moq PRIVATE OBS::libobs) if(_moq_local_build) - add_subdirectory("${MOQ_LOCAL}/rs/moq-c" moq) - target_link_libraries(obs-moq PRIVATE moq::c) + add_subdirectory("${MOQ_LOCAL}/cpp/moq" moq-cpp) else() include(FetchContent) FetchContent_Declare( - moq + moq-cpp URL - https://github.com/moq-dev/moq/releases/download/moq-c-v${_moq_link_version}/moq-c-${_moq_link_version}-${MOQ_TARGET}.${MOQ_ARCHIVE} + https://github.com/moq-dev/moq/releases/download/cpp-v${_moq_link_version}/moq-cpp-${_moq_link_version}-${MOQ_TARGET}.${MOQ_ARCHIVE} ) - FetchContent_MakeAvailable(moq) + FetchContent_MakeAvailable(moq-cpp) - find_package(moq-c REQUIRED PATHS ${moq_SOURCE_DIR} NO_DEFAULT_PATH) - target_link_libraries(obs-moq PRIVATE moq::c) -endif() - -# moq-c is a Rust static library; its std runtime pulls in symbols from Windows -# system libraries (e.g. NtCreateNamedPipeFile from ntdll) that MSVC does not -# auto-resolve when the archive is linked into this module. These are the OS -# import libraries reported by `rustc --print native-static-libs` that aren't -# linked by default; they resolve by name from the Windows SDK (no hard-coded -# paths). The moq-c package config also declares these, but linking them here -# keeps the plugin buildable against any moq-c (local dev tree or prebuilt -# release) whose config may predate that declaration. Duplicate links are -# harmless. -if(WIN32) - target_link_libraries( - obs-moq - PRIVATE ntdll userenv ws2_32 dbghelp bcrypt - ) + find_package(moq-cpp REQUIRED PATHS ${moq-cpp_SOURCE_DIR} NO_DEFAULT_PATH) endif() +# The package carries the system libraries its Rust staticlib needs, on every platform. +target_link_libraries(obs-moq PRIVATE moq::cpp) # The obs-deps Qt6 build references the AGL framework transitively (via # WrapOpenGL), but recent macOS SDKs ship no linkable AGL binary -- it exists @@ -150,6 +134,7 @@ target_sources( obs-moq PRIVATE src/obs-moq.cpp + src/moq-worker.h src/moq-output.h src/moq-service.h src/moq-output.cpp diff --git a/cpp/obs/README.md b/cpp/obs/README.md index e5aff1fd1b7..0af672f572e 100644 --- a/cpp/obs/README.md +++ b/cpp/obs/README.md @@ -3,7 +3,8 @@ An OBS Studio plugin for publishing to and subscribing from MoQ relays. It loads into a stock OBS Studio install (no OBS source build required) and links -`moq-c`, built from the in-tree [`rs/moq-c`](../../rs/moq-c) crate. +the moq C++ package, built from the in-tree [`cpp/moq`](../moq) over +[`rs/moq-ffi`](../../rs/moq-ffi). Build instructions for each platform live in [`doc/bin/obs.md`](../../doc/bin/obs.md). In short, from the repo root: diff --git a/cpp/obs/VERSION b/cpp/obs/VERSION new file mode 100644 index 00000000000..faef31a4357 --- /dev/null +++ b/cpp/obs/VERSION @@ -0,0 +1 @@ +0.7.0 diff --git a/cpp/obs/build.sh b/cpp/obs/build.sh index 64a5f0baa85..9c00bff2b9d 100755 --- a/cpp/obs/build.sh +++ b/cpp/obs/build.sh @@ -36,16 +36,16 @@ while [[ $# -gt 0 ]]; do OUTPUT_DIR="$2" shift 2 ;; - --moq-c-release) - # Link a published moq-c release of this version instead of - # building rs/moq-c from source. CMake fetches the matching - # moq-c-- archive from the GitHub release and the - # plugin is versioned to match. Used by CI on a moq-c-v* tag. + --moq-release) + # Link a published moq-cpp release of this version instead of + # building cpp/moq from source. CMake fetches the matching + # moq-cpp-- archive from the GitHub release. The + # plugin keeps its own version. Used by CI on a cpp-v* tag. MOQ_RELEASE="$2" shift 2 ;; -h | --help) - echo "Usage: $0 [--target TARGET] [--version VERSION] [--output DIR] [--moq-c-release VERSION]" + echo "Usage: $0 [--target TARGET] [--version VERSION] [--output DIR] [--moq-release VERSION]" exit 0 ;; *) @@ -55,20 +55,14 @@ while [[ $# -gt 0 ]]; do esac done -# In moq-c-release mode the plugin version tracks the moq-c version. -if [[ -n "$MOQ_RELEASE" ]]; then - VERSION="$MOQ_RELEASE" -fi - if [[ -z "$TARGET" ]]; then TARGET=$(cc -dumpmachine 2>/dev/null || echo unknown) echo "Detected target: $TARGET" fi -# Default the version from buildspec.json's top-level "version" (the nested -# dependency entries also have "version" keys, hence the leading-indent anchor). +# Default to the plugin's own version. if [[ -z "$VERSION" ]]; then - VERSION=$(grep -E '^[[:space:]]{4}"version"' "$SCRIPT_DIR/buildspec.json" | head -1 | sed 's/.*: *"\([^"]*\)".*/\1/') + VERSION=$(tr -d '[:space:]' <"$SCRIPT_DIR/VERSION") echo "Detected version: $VERSION" fi @@ -112,7 +106,7 @@ if [[ -n "$MOQ_RELEASE" ]]; then # Empty MOQ_LOCAL forces CMake's release-download branch; MOQ_VERSION and # MOQ_TARGET steer it at this target's archive. MOQ_ARCHIVE is correct per # preset already. - echo "Linking moq-c release v$MOQ_RELEASE ($TARGET)" + echo "Linking moq-cpp release v$MOQ_RELEASE ($TARGET)" CONFIGURE_ARGS+=(-DMOQ_LOCAL= "-DMOQ_VERSION=$MOQ_RELEASE" "-DMOQ_TARGET=$TARGET") fi cmake --preset "$PRESET" ${CONFIGURE_ARGS[@]+"${CONFIGURE_ARGS[@]}"} diff --git a/cpp/obs/buildspec.json b/cpp/obs/buildspec.json index 35dd8f23326..136d3eb975b 100644 --- a/cpp/obs/buildspec.json +++ b/cpp/obs/buildspec.json @@ -38,7 +38,6 @@ }, "name": "obs-moq", "displayName": "MOQ for OBS", - "version": "0.0.1", "author": "Montevideo Tech", "website": "https://montevideotech.dev", "email": "emil@qualabs.com" diff --git a/cpp/obs/cmake/common/bootstrap.cmake b/cpp/obs/cmake/common/bootstrap.cmake index 4420dbbb8e4..48312f052f4 100644 --- a/cpp/obs/cmake/common/bootstrap.cmake +++ b/cpp/obs/cmake/common/bootstrap.cmake @@ -49,12 +49,15 @@ string(JSON _name GET ${buildspec} name) string(JSON _website GET ${buildspec} website) string(JSON _author GET ${buildspec} author) string(JSON _email GET ${buildspec} email) -string(JSON _version GET ${buildspec} version) +# The plugin's own version, human-owned like cpp/moq/VERSION. +file(STRINGS "${CMAKE_CURRENT_SOURCE_DIR}/VERSION" _version LIMIT_COUNT 1) +if(NOT _version MATCHES "^[0-9]+\\.[0-9]+\\.[0-9]+$") + message(FATAL_ERROR "Unsupported version '${_version}' in ${CMAKE_CURRENT_SOURCE_DIR}/VERSION (want x.y.z)") +endif() string(JSON _bundleId GET ${buildspec} platformConfig macos bundleId) -# Release CI builds the plugin against a specific moq-c release and versions it -# to match (see build.sh --moq-c-release), so allow overriding the buildspec -# version. Everything below derives PLUGIN_VERSION_* from _version. +# build.sh --version stamps a different version than VERSION's, so allow +# overriding it. Everything below derives PLUGIN_VERSION_* from _version. if(DEFINED PLUGIN_VERSION_OVERRIDE AND NOT PLUGIN_VERSION_OVERRIDE STREQUAL "") set(_version "${PLUGIN_VERSION_OVERRIDE}") endif() diff --git a/cpp/obs/justfile b/cpp/obs/justfile index 5fc4ff39161..199067ec1c5 100644 --- a/cpp/obs/justfile +++ b/cpp/obs/justfile @@ -10,8 +10,9 @@ # libobs/Qt6/ffmpeg via buildspec.json (obs-deps). # Windows: needs Visual Studio 2022; same buildspec download (run from Git Bash). # -# moq-c is built from the in-tree crate (rs/moq-c) via cargo. MOQ_LOCAL points -# CMake at the repo root by default, so there's no prebuilt release to download. +# The C++ package (cpp/moq) is built in-tree: cargo builds rs/moq-ffi and +# uniffi-bindgen-cpp renders its bindings. MOQ_LOCAL points CMake at the repo root +# by default, so there's no prebuilt release to download. set quiet @@ -20,7 +21,7 @@ default: just --list # Configure via CMake presets. MOQ_LOCAL defaults to the repo root inside -# CMakeLists.txt (so moq-c builds from rs/moq-c); pass `path=/some/moq` to +# CMakeLists.txt (so the C++ package builds from cpp/moq); pass `path=/some/moq` to # override it. setup preset="" path="": cmake --preset "$(../../sh/obs/preset.sh '{{ preset }}')" {{ if path == "" { "" } else { "-DMOQ_LOCAL=" + quote(path) } }} @@ -38,15 +39,16 @@ compile: # Compile and link the plugin the way CI does, against the `.#obs` shell's own # libobs/Qt6/ffmpeg, then run the unit tests. Linux-only in practice: it's the # one platform where nixpkgs has obs-studio. The Debug build type reuses the -# dev-profile Rust artifacts CI restores, where a release moq-c would rebuild +# dev-profile Rust artifacts CI restores, where a release moq-ffi would rebuild # its whole dependency tree. ci preset="": cmake --preset "$(../../sh/obs/preset.sh '{{ preset }}' ci)" -DCMAKE_BUILD_TYPE=Debug cmake --build --preset "$(../../sh/obs/preset.sh '{{ preset }}' ci)" ../../sh/obs/unit.sh -# Unit-test the plugin sources against stubbed libobs/moq-c/ffmpeg, under -# ThreadSanitizer. Manual; `just obs ci` runs them without the sanitizer. +# Unit-test the plugin sources against stubbed libobs and ffmpeg and the real +# moq-ffi, under ThreadSanitizer. Manual; `just obs ci` runs them without the +# sanitizer. test: ../../sh/obs/test.sh diff --git a/cpp/obs/src/moq-dock.cpp b/cpp/obs/src/moq-dock.cpp index ff6e85e1b97..d527bd78a1b 100644 --- a/cpp/obs/src/moq-dock.cpp +++ b/cpp/obs/src/moq-dock.cpp @@ -435,7 +435,7 @@ MoQDock::MoQDock(QWidget *parent) : QWidget(parent) advancedButton = new QPushButton("Advanced…", streamPage); advancedButton->setCursor(Qt::PointingHandCursor); - advancedButton->setToolTip("Protocol pin, TLS, reconnect, QUIC and WebSocket settings. " + advancedButton->setToolTip("TLS, reconnect, QUIC and WebSocket settings. " "Opens them in a window. Changes apply on the next Go Live."); connect(advancedButton, &QPushButton::clicked, this, &MoQDock::OpenAdvanced); @@ -565,12 +565,12 @@ MoQDock::MoQDock(QWidget *parent) : QWidget(parent) QString("%1").arg(QString::fromUtf8(PLUGIN_VERSION_STRING))); pluginVer->setToolTip("OBS plugin docs on doc.moq.dev"); - auto *libmoqVer = new QLabel(aboutPage); - libmoqVer->setOpenExternalLinks(true); - libmoqVer->setTextInteractionFlags(Qt::TextBrowserInteraction); - libmoqVer->setText( - QString("%1").arg(QString::fromUtf8(MOQ_VERSION_STRING))); - libmoqVer->setToolTip("libmoq C API docs on doc.moq.dev"); + auto *moqVer = new QLabel(aboutPage); + moqVer->setOpenExternalLinks(true); + moqVer->setTextInteractionFlags(Qt::TextBrowserInteraction); + moqVer->setText( + QString("%1").arg(QString::fromUtf8(MOQ_VERSION_STRING))); + moqVer->setToolTip("moq C++ API docs on doc.moq.dev"); auto *moqDevLink = new QLabel(aboutPage); moqDevLink->setOpenExternalLinks(true); @@ -587,7 +587,7 @@ MoQDock::MoQDock(QWidget *parent) : QWidget(parent) detectedLabel->setTextInteractionFlags(Qt::TextSelectableByMouse); verForm->addRow("Plugin", pluginVer); - verForm->addRow("libmoq", libmoqVer); + verForm->addRow("moq-cpp", moqVer); verForm->addRow("moq.dev", moqDevLink); verForm->addRow("moq.pro", moqProLink); verForm->addRow("Available video encoders", detectedLabel); diff --git a/cpp/obs/src/moq-error.h b/cpp/obs/src/moq-error.h index 76005ea7b83..fb2ae1973e5 100644 --- a/cpp/obs/src/moq-error.h +++ b/cpp/obs/src/moq-error.h @@ -7,6 +7,9 @@ namespace MoQError { enum class Kind { Unauthorized, Forbidden, Certificate, Timeout, Network, Connect, Offline, Other }; +// What the output recorded a failure as, beside its text. None means no failure. +enum Code : int { None = 0, Unauthorized, Forbidden, Connect, Offline, Other }; + inline Kind Classify(int code, std::string reason) { for (char &c : reason) { @@ -16,9 +19,9 @@ inline Kind Classify(int code, std::string reason) auto has = [&](const char *text) { return reason.find(text) != std::string::npos; }; - if (has("unauthorized") || code == -34) + if (has("unauthorized") || code == Code::Unauthorized) return Kind::Unauthorized; - if (has("forbidden") || code == -35) + if (has("forbidden") || code == Code::Forbidden) return Kind::Forbidden; // Fetching the pin happens before TLS verification and can fail on an outage. const bool fetching = has("failed to fetch fingerprint") || has("fingerprint request failed") || @@ -29,9 +32,9 @@ inline Kind Classify(int code, std::string reason) return Kind::Timeout; if (fetching || has("failed to connect") || has("dns")) return Kind::Network; - if (code == -5 || has("connect error")) + if (code == Code::Connect || has("connect error")) return Kind::Connect; - if (code == -17 || reason == "offline") + if (code == Code::Offline || reason == "offline") return Kind::Offline; return Kind::Other; } diff --git a/cpp/obs/src/moq-output.cpp b/cpp/obs/src/moq-output.cpp index 49de54d3ffc..855b825038c 100644 --- a/cpp/obs/src/moq-output.cpp +++ b/cpp/obs/src/moq-output.cpp @@ -2,15 +2,17 @@ #include #include "moq-output.h" +#include "moq-error.h" #include "moq-settings.h" #include "moq-url.h" #include "logger.h" #include "util/util_uint64.h" +#include #include #include - -#include "moq.h" +#include +#include namespace { @@ -31,20 +33,8 @@ bool LooksGenericOffline(const std::string &reason) return false; } -bool IsAuthFailure(int code, const std::string &reason) -{ - if (code == -34 || code == -35) - return true; - std::string lower = reason; - for (char &c : lower) { - if (c >= 'A' && c <= 'Z') - c = static_cast(c - 'A' + 'a'); - } - return lower.find("unauthorized") != std::string::npos || lower.find("forbidden") != std::string::npos; -} - -// Dial URL scheme only. https races WebTransport vs WebSocket in moq-native; -// libmoq does not yet expose which won, so do not invent a transport name here. +// Dial URL scheme only. https races WebTransport vs WebSocket, and the session +// does not report which won, so do not invent a transport name here. std::string DialSchemeLabel(const std::string &url) { const auto colon = url.find(':'); @@ -64,47 +54,61 @@ std::string DialSchemeLabel(const std::string &url) return scheme; } -} // namespace - -MoQOutput::MoQOutput(obs_data_t *, obs_output_t *output) - : output(output), - state(std::make_shared(output)), - path(), - total_bytes_sent(0), - origin(moq_origin_create()), - broadcast(0) +// The failure code the dock classifies a moq::Error by, beside its text. +MoQError::Code FailureCode(const moq::Error &error) { + const auto &variant = error.get_variant(); + if (std::holds_alternative(variant)) + return MoQError::Code::Unauthorized; + if (std::holds_alternative(variant)) + return MoQError::Code::Forbidden; + if (std::holds_alternative(variant)) + return MoQError::Code::Connect; + if (const auto *protocol = std::get_if(&variant)) { + if (protocol->details.kind == moq::ProtocolKind::kUnauthorized) + return MoQError::Code::Unauthorized; + } + return MoQError::Code::Other; } +} // namespace + +MoQOutput::MoQOutput(obs_data_t *, obs_output_t *output) : output(output), path(), total_bytes_sent(0) {} + MoQOutput::~MoQOutput() { - // Retires the attempt, so a terminal callback that is still in flight won't - // signal a stop on an output that is going away. + // Retires the attempt, so a continuation that is already queued won't signal a + // stop on an output that is going away. Stop(); - // Reset() closed the session and finished the tracks and the broadcast, so - // the origin has no children left. - moq_origin_close(origin); - - // Give up the frontend. A terminal callback still in flight holds its own - // reference to the shared state, so nothing here waits on the libmoq runtime: - // whenever it arrives it finds the state detached and reports nothing. - state->Detach(); -} + // Let the sessions finish draining, so the relay sees the tracks end rather than a + // cancel. moq-ffi bounds each drain to about a second. + std::vector> drains; + { + std::lock_guard signal_lock(signal_mutex); + drains.swap(draining); + } + for (auto &drain : drains) + drain.wait(); -void MoQOutput::SessionState::Detach() -{ - // A callback that has already decided to report parks here, so this returns - // only once nothing is inside an OBS call and the output can be freed. - std::lock_guard signal_lock(signal_mutex); - output = nullptr; + // Drops whatever is still queued and waits out a continuation in flight, which + // may be parked on signal_mutex; so this must not hold it. Nothing touches this + // object once it returns. + worker.Stop(); } -void MoQOutput::SessionState::SignalStop(int code) +template +std::function MoQOutput::Current(const std::shared_ptr ¤t, Callback callback) { - std::lock_guard signal_lock(signal_mutex); - if (output) - obs_output_signal_stop(output, code); + return [this, weak = std::weak_ptr(current), callback = std::move(callback)](Output output) { + // Held across the check and whatever the callback reports, so a Stop() can't + // retire the attempt in between. + std::lock_guard signal_lock(signal_mutex); + auto current = weak.lock(); + if (!current || current != attempt) + return; + callback(current, std::move(output)); + }; } bool MoQOutput::Start() @@ -116,7 +120,8 @@ bool MoQOutput::Start() obs_service_t *service = obs_output_get_service(output); if (!service) { LOG_ERROR("Failed to get service from output"); - state->SignalStop(OBS_OUTPUT_ERROR); + std::lock_guard signal_lock(signal_mutex); + obs_output_signal_stop(output, OBS_OUTPUT_ERROR); return false; } @@ -131,10 +136,11 @@ bool MoQOutput::Start() } const char *server_value = obs_service_get_connect_info(service, OBS_SERVICE_CONNECT_INFO_SERVER_URL); - const std::string url = server_value ? server_value : ""; - if (url.empty()) { + const std::string server = server_value ? server_value : ""; + if (server.empty()) { LOG_ERROR("Server URL is empty"); - state->SignalStop(OBS_OUTPUT_BAD_PATH); + std::lock_guard signal_lock(signal_mutex); + obs_output_signal_stop(output, OBS_OUTPUT_BAD_PATH); return false; } @@ -155,155 +161,245 @@ bool MoQOutput::Start() return false; } + auto next = std::make_shared(); + next->client = moq::Client::init(); + next->url = server; + // Advanced settings live on the service alongside the URL and path. With the group - // switched off Pointer() is NULL, which dials with the library defaults. The config - // borrows its strings, so it has to outlive the connect below. + // switched off the client keeps the library defaults. OBSDataAutoRelease service_settings = obs_service_get_settings(service); - MoQSettings::Config client; - if (!MoQSettings::BuildConfig(service_settings, &client)) { - // BuildConfig logged why. Refusing to start beats connecting with a setting - // the user asked for quietly dropped. - obs_output_set_last_error(output, "Invalid advanced MoQ settings; see the log for details."); - state->SignalStop(OBS_OUTPUT_CONNECT_FAILED); + std::string invalid; + if (!MoQSettings::Configure(service_settings, *next->client, &invalid)) { + // Refusing to start beats connecting with a setting the user asked for + // quietly dropped. + LOG_ERROR("Invalid advanced MoQ settings: %s", invalid.c_str()); + std::lock_guard signal_lock(signal_mutex); + obs_output_set_last_error(output, ("Invalid advanced MoQ settings: " + invalid).c_str()); + obs_output_signal_stop(output, OBS_OUTPUT_CONNECT_FAILED); return false; } - LOG_INFO("Connecting to MoQ server: %s", MoQRedactUrl(url).c_str()); + // Create the broadcast on an origin of our own, then announce it so subscribers + // can discover it. Stop() finishes it, so each Start creates a fresh one. + LOG_INFO("Publishing broadcast: %s", path.c_str()); + auto next_origin = moq::OriginProducer::init(moq::OriginConfig{}); + auto next_broadcast = next_origin->create_broadcast(path); + if (!next_broadcast) { + LOG_ERROR("Failed to create broadcast: %s", next_broadcast.error().to_string().c_str()); + return false; + } + if (auto announced = (*next_broadcast)->announce(moq::Route{}); !announced) { + LOG_ERROR("Failed to announce broadcast: %s", announced.error().to_string().c_str()); + return false; + } + if (auto wired = next->client->set_publish(next_origin); !wired) { + LOG_ERROR("Failed to publish through the client: %s", wired.error().to_string().c_str()); + return false; + } - // Held from the connect through obs_output_begin_data_capture. The status - // callback takes the same lock before reporting anything, so this attempt's - // terminal cannot signal a failure against an output that is only half - // started: it waits until the output is committed, and OBS then handles the - // stop through its normal active-output path. - std::lock_guard signal_lock(state->signal_mutex); + LOG_INFO("Connecting to MoQ server: %s", MoQRedactUrl(server).c_str()); - uint64_t attempt; + // Held from the connect through obs_output_begin_data_capture. Every + // continuation takes the same lock before reporting anything, so this attempt's + // failure cannot signal against an output that is only half started: it waits + // until the output is committed, and OBS then handles the stop through its + // normal active-output path. + std::lock_guard signal_lock(signal_mutex); + + attempt = next; { - std::lock_guard lock(state->mutex); - attempt = state->attempt; - state->url = url; + std::lock_guard lock(mutex); + url = server; + } + { + std::lock_guard lock(media_mutex); + origin = next_origin; + broadcast = *next_broadcast; } - // Everything the callbacks need about this attempt, copied so they never read - // a member the next Start() may be rewriting, plus the reference that keeps - // the shared state alive. Freed by the terminal callback, so it outlives this - // scope. - auto ref = new SessionRef{state, attempt, url, std::chrono::steady_clock::now()}; - - // Start establishing a session with the MoQ server // NOTE: You could publish the same broadcasts to multiple sessions if you want (redundant ingest). - int handle = - moq_session_connect(url.data(), url.size(), client.Pointer(), origin, 0, MoQOutput::SessionStatus, ref); - - if (handle < 0) { - const char *reason = moq_error(); - LOG_ERROR("Failed to initialize MoQ server: %d: %s", handle, reason ? reason : "unknown error"); - obs_output_set_last_error(output, reason ? reason : "Failed to initialize MoQ connection"); - // No subscription was created, so no terminal will fire; drop the reference. - delete ref; - return false; + next->started = std::chrono::steady_clock::now(); + next->pending = next->client->connect(server).then( + worker.Executor(), Current>>( + next, [this](const std::shared_ptr ¤t, auto result) { + OnConnect(current, std::move(result)); + })); + + obs_output_begin_data_capture(output, 0); + + return true; +} + +void MoQOutput::OnConnect(const std::shared_ptr ¤t, moq::expected> result) +{ + if (!result) { + Fail(current, result.error()); + return; } - bool superseded; + auto elapsed = std::chrono::steady_clock::now() - current->started; + auto ms = static_cast(std::chrono::duration_cast(elapsed).count()); + const int connect_epoch = static_cast((*result)->epoch()); + + current->session = *result; + current->connected = true; + current->live = true; { - std::lock_guard lock(state->mutex); - // Holding signal_mutex keeps this attempt current: libmoq delivers the - // terminal on its runtime thread, which parks in SessionState::Closed until - // we commit. The check stands guard over that assumption rather than - // trusting libmoq's threading to stay as it is. - superseded = state->attempt != attempt; - if (!superseded) - state->session = handle; - } - - if (superseded) { - // The session died during connect without going through the callback's - // signal path. Refuse the start rather than capturing into a dead session; - // OBS surfaces the last error we recorded when info.start returns false. - LOG_ERROR("MoQ session failed before the output started: %s", MoQRedactUrl(url).c_str()); - return false; + std::lock_guard lock(mutex); + session = current->session; + connected = true; + live = true; + last_failure_code = MoQError::None; + last_failure_reason.clear(); } + // OBS and older dock paths treat connect_time_ms == 0 as "never connected". + // Clamp sub-millisecond connects to 1 so that sentinel stays honest. + connect_time_ms = ms > 0 ? ms : 1; - LOG_INFO("Publishing broadcast: %s", path.c_str()); + LOG_INFO("MoQ session connected (%d ms, epoch %d): %s", ms, connect_epoch, MoQRedactUrl(current->url).c_str()); - // Create the broadcast on the origin we created, then announce it so - // subscribers can discover it. Stop() finishes it, so each Start creates a fresh one. - broadcast = moq_origin_create_broadcast(origin, path.data(), path.size()); - if (broadcast < 0) { - LOG_ERROR("Failed to create broadcast: %d", broadcast); - broadcast = 0; - // The session connected above; close it so a retry on this same output - // doesn't reuse the stale handle. Its terminal callback releases the - // outstanding-session reference the destructor waits on. - Stop(false); - return false; + WatchStatus(current); +} + +void MoQOutput::WatchStatus(const std::shared_ptr ¤t) +{ + // Replacing `pending` from inside its own continuation is fine: that call has + // already completed, so dropping it cancels nothing. + current->pending = current->session->status().then( + worker.Executor(), Current>( + current, [this](const std::shared_ptr ¤t, auto result) { + OnStatus(current, std::move(result)); + })); +} + +void MoQOutput::OnStatus(const std::shared_ptr ¤t, moq::expected result) +{ + // The status stream ends in an error once reconnecting gives up for good. + if (!result) { + Fail(current, result.error()); + return; } - if (moq_publish_announce(broadcast, nullptr) < 0) { - LOG_ERROR("Failed to announce broadcast"); - Stop(false); - return false; + + switch (*result) { + case moq::ConnectionStatus::kConnected: { + const int connect_epoch = static_cast(current->session->epoch()); + current->live = true; + { + std::lock_guard lock(mutex); + live = true; + last_failure_code = MoQError::None; + last_failure_reason.clear(); + } + LOG_INFO("MoQ session reconnected (epoch %d): %s", connect_epoch, MoQRedactUrl(current->url).c_str()); + break; + } + case moq::ConnectionStatus::kDisconnected: { + current->live = false; + std::lock_guard lock(mutex); + live = false; + LOG_WARNING("MoQ session dropped, reconnecting: %s", MoQRedactUrl(current->url).c_str()); + break; + } + case moq::ConnectionStatus::kMigrating: + // The old session keeps serving while the replacement dials. + LOG_INFO("MoQ session migrating: %s", MoQRedactUrl(current->url).c_str()); + break; } - obs_output_begin_data_capture(output, 0); + WatchStatus(current); +} - return true; +void MoQOutput::Fail(const std::shared_ptr ¤t, const moq::Error &error) +{ + const std::string reason = error.to_string(); + const bool was_connected = current->connected; + + // Retire the attempt, which also limits the failure signal to one per Start(). + // The broadcast stays until Reset(), as OBS stops or restarts the output next. + attempt.reset(); + { + std::lock_guard lock(mutex); + session.reset(); + connected = false; + live = false; + last_failure_code = FailureCode(error); + last_failure_reason = reason; + } + connect_time_ms = 0; + + LOG_ERROR("MoQ session failed: %s: %s", MoQRedactUrl(current->url).c_str(), reason.c_str()); + + // Reconnection gave up, so nothing is reaching the server any more. Without + // this OBS keeps encoding and reporting the stream as live forever. + obs_output_set_last_error(output, reason.c_str()); + // CONNECT_FAILED is terminal for OBS. DISCONNECTED lets its own reconnect + // logic retry, which is only worth offering once we know the server works. + obs_output_signal_stop(output, was_connected ? OBS_OUTPUT_DISCONNECTED : OBS_OUTPUT_CONNECT_FAILED); } void MoQOutput::Stop(bool signal) { - std::lock_guard signal_lock(state->signal_mutex); + std::lock_guard signal_lock(signal_mutex); Reset(); - if (signal) { - state->SignalStop(OBS_OUTPUT_SUCCESS); - } + if (signal) + obs_output_signal_stop(output, OBS_OUTPUT_SUCCESS); } void MoQOutput::Reset() { - // Excludes the status callback's report: retiring the attempt and signalling - // a failure must not interleave, or a stop that already happened gets followed - // by a failure OBS turns into a reconnect. - std::lock_guard signal_lock(state->signal_mutex); + // Excludes a continuation's report: retiring the attempt and signalling a + // failure must not interleave, or a stop that already happened gets followed by + // a failure OBS turns into a reconnect. + std::lock_guard signal_lock(signal_mutex); + + // Dropping the attempt cancels its pending connect or status call. + std::shared_ptr retired = attempt ? attempt->session : nullptr; + attempt.reset(); - int stale; { - std::lock_guard lock(state->mutex); - // Retire the attempt so an in-flight terminal callback stays out of the way. - state->attempt++; - state->connected = false; - state->connect_time_ms = 0; - state->epoch = 0; - state->url.clear(); - state->last_failure_code = 0; - state->last_failure_reason.clear(); - stale = state->session; - state->session = 0; - } - - // Outside the lock: the status callback takes state->mutex, and libmoq is - // free to run it on its own thread while we're here. - if (stale > 0) - moq_session_close(stale); - - for (auto &[encoder, handle] : video_tracks) { - if (handle > 0) - moq_publish_media_finish(handle); + std::lock_guard lock(mutex); + session.reset(); + connected = false; + live = false; + url.clear(); + last_failure_code = MoQError::None; + last_failure_reason.clear(); + } + connect_time_ms = 0; + + std::lock_guard lock(media_mutex); + for (auto &[encoder, track] : video_tracks) { + if (track) + track->finish(); } video_tracks.clear(); - for (auto &[encoder, handle] : audio_tracks) { - if (handle > 0) - moq_publish_media_finish(handle); + for (auto &[encoder, track] : audio_tracks) { + if (track) + track->finish(); } audio_tracks.clear(); // Close the broadcast so the origin retracts it immediately; Start() // creates a fresh one on restart. - if (broadcast > 0) { - moq_publish_close(broadcast); - broadcast = 0; + if (broadcast) + broadcast->close(); + broadcast.reset(); + origin.reset(); + + // Drain once the tracks are finished, so the session delivers their tails instead + // of cutting them off. The future holds the session until the drain settles, and + // dropping it would abort the drain, so keep it past this attempt. + if (retired) { + draining.erase(std::remove_if(draining.begin(), draining.end(), + [](const moq::Future &drain) { + return drain.wait_for(std::chrono::seconds(0)) == + std::future_status::ready; + }), + draining.end()); + draining.push_back(retired->shutdown()); } } @@ -312,176 +408,81 @@ bool MoQOutput::TryGetConnectionStats(ConnectionStats *out) if (!out) return false; - uint32_t handle; - uint64_t attempt; - std::string dialUrl; + std::shared_ptr current; + std::string dial_url; { - std::lock_guard lock(state->mutex); - if (state->session == 0) + std::lock_guard lock(mutex); + if (!session) return false; - handle = static_cast(state->session); - attempt = state->attempt; - dialUrl = state->url; - } - - moq_connection_snapshot connection{}; - const int32_t rc = moq_session_snapshot(handle, &connection); - if (rc != 0) { - const char *message = moq_error(); - const std::string next = message && *message ? message : "offline"; - std::lock_guard lock(state->mutex); - if (state->attempt != attempt) + if (!live) { + // Between reconnects. Keep a more specific prior failure (unauthorized) + // over a generic offline blip. + if (last_failure_reason.empty() || LooksGenericOffline(last_failure_reason)) { + last_failure_code = MoQError::Offline; + last_failure_reason = "offline"; + } return false; - // Keep a more specific prior failure (unauthorized) over a generic offline blip. - if (state->last_failure_reason.empty() || LooksGenericOffline(state->last_failure_reason) || - !LooksGenericOffline(next) || IsAuthFailure(rc, next)) { - state->last_failure_code = rc; - state->last_failure_reason = next; } - return false; + current = session; + dial_url = url; } - const auto &raw = connection.stats; + const moq::ConnectionStats raw = current->stats(); + ConnectionStats snapshot; snapshot.reconnects = GetReconnectCount(); - snapshot.rtt_valid = raw.rtt_valid; - snapshot.rtt_ms = raw.rtt_valid ? static_cast(raw.rtt_us) / 1000.0 : 0; - snapshot.estimated_send_rate_valid = raw.estimated_send_rate_valid; + snapshot.rtt_valid = raw.rtt_us.has_value(); + snapshot.rtt_ms = raw.rtt_us ? static_cast(*raw.rtt_us) / 1000.0 : 0; + snapshot.estimated_send_rate_valid = raw.estimated_send_rate_bps.has_value(); snapshot.estimated_send_rate_bps = - raw.estimated_send_rate_valid ? static_cast(raw.estimated_send_rate_bps) : 0; - snapshot.estimated_recv_rate_valid = raw.estimated_recv_rate_valid; + raw.estimated_send_rate_bps ? static_cast(*raw.estimated_send_rate_bps) : 0; + snapshot.estimated_recv_rate_valid = raw.estimated_recv_rate_bps.has_value(); snapshot.estimated_recv_rate_bps = - raw.estimated_recv_rate_valid ? static_cast(raw.estimated_recv_rate_bps) : 0; - snapshot.bytes_sent_valid = raw.bytes_sent_valid; - snapshot.bytes_sent = raw.bytes_sent_valid ? raw.bytes_sent : 0; - if (raw.packets_sent_valid && raw.packets_lost_valid && raw.packets_sent > 0) { + raw.estimated_recv_rate_bps ? static_cast(*raw.estimated_recv_rate_bps) : 0; + snapshot.bytes_sent_valid = raw.bytes_sent.has_value(); + snapshot.bytes_sent = raw.bytes_sent.value_or(0); + if (raw.packets_sent && raw.packets_lost && *raw.packets_sent > 0) { snapshot.loss_valid = true; snapshot.loss_pct = - 100.0 * static_cast(raw.packets_lost) / static_cast(raw.packets_sent); + 100.0 * static_cast(*raw.packets_lost) / static_cast(*raw.packets_sent); } - snapshot.dial = DialSchemeLabel(dialUrl); + snapshot.dial = DialSchemeLabel(dial_url); - snapshot.protocol.assign(connection.protocol.data, connection.protocol.len); + // A Stop(), restart, or disconnect while stats() ran retired this session, so + // its numbers no longer describe the output. + std::lock_guard lock(mutex); + if (session != current || !live) + return false; + *out = std::move(snapshot); + return true; +} +int MoQOutput::GetReconnectCount() +{ + std::shared_ptr current; { - std::lock_guard lock(state->mutex); - if (state->attempt != attempt || static_cast(state->session) != handle) - return false; - *out = std::move(snapshot); + std::lock_guard lock(mutex); + current = session; } - - return true; + // Read live rather than tracked from status(), which coalesces a drop that + // reconnects before it is asked again. + const uint64_t connect_epoch = current ? current->epoch() : 0; + return connect_epoch > 1 ? static_cast(connect_epoch - 1) : 0; } bool MoQOutput::IsLiveSession() { - std::lock_guard lock(state->mutex); - return state->connected && state->session != 0; + std::lock_guard lock(mutex); + return connected && session != nullptr; } void MoQOutput::CopyLastFailure(int *code, std::string *reason) { - std::lock_guard lock(state->mutex); + std::lock_guard lock(mutex); if (code) - *code = state->last_failure_code; + *code = last_failure_code; if (reason) - *reason = state->last_failure_reason; -} - -// libmoq status codes (>= 0.3.0): > 0 = (re)connected, carrying the connection -// epoch; 0 = closed cleanly (terminal); < 0 = fatal, reconnection gave up (terminal). -void MoQOutput::SessionStatus(void *user_data, int code) -{ - auto ref = static_cast(user_data); - - if (code > 0) { - ref->state->Connected(*ref, code); - return; - } - - // Terminal: libmoq never touches user_data again, so we own the reference. - // Destroying it drops the last hold on the shared state when the output is - // already gone, so this is where a detached state is finally freed. - std::unique_ptr owned(ref); - owned->state->Closed(*owned, code); -} - -void MoQOutput::SessionState::Connected(const SessionRef &ref, int connect_epoch) -{ - auto elapsed = std::chrono::steady_clock::now() - ref.started; - auto ms = static_cast(std::chrono::duration_cast(elapsed).count()); - - { - std::lock_guard lock(mutex); - if (attempt != ref.attempt) - return; - connected = true; - // OBS and older dock paths treat connect_time_ms == 0 as "never connected". - // Clamp sub-millisecond connects to 1 so that sentinel stays honest. - connect_time_ms = ms > 0 ? ms : 1; - epoch = connect_epoch; - last_failure_code = 0; - last_failure_reason.clear(); - } - - LOG_INFO("MoQ session connected (%d ms, epoch %d): %s", ms, connect_epoch, MoQRedactUrl(ref.url).c_str()); -} - -void MoQOutput::SessionState::Closed(const SessionRef &ref, int code) -{ - // moq_error() only describes the most recent call on this thread, so copy the - // reason out before anything below can overwrite it. - std::string reason; - if (code < 0) { - const char *message = moq_error(); - reason = message ? message : "unknown error"; - } - - // Held across both the decision below and the signal itself. Taking it before - // mutex is the required lock order, and holding it through the signal is what - // stops a teardown from landing in between: without that, Stop() can retire - // the attempt after we decide to report, and OBS turns the late - // OBS_OUTPUT_DISCONNECTED into a reconnect of a stopped stream. It is also - // what Detach() waits on, so `output` cannot go away mid-report. - std::lock_guard signal_lock(signal_mutex); - - bool current; - bool was_connected = false; - { - std::lock_guard lock(mutex); - current = attempt == ref.attempt; - if (current) { - was_connected = connected; - // The session task has ended and dropped the handle, so retire it here - // rather than closing a handle libmoq no longer knows about. - session = 0; - attempt++; - connected = false; - connect_time_ms = 0; - epoch = 0; - if (code < 0) { - last_failure_code = code; - last_failure_reason = reason; - } - } - } - - if (code == 0) { - LOG_INFO("MoQ session closed: %s", MoQRedactUrl(ref.url).c_str()); - } else { - LOG_ERROR("MoQ session failed (%d): %s: %s", code, MoQRedactUrl(ref.url).c_str(), reason.c_str()); - } - - // Reconnection gave up, so nothing is reaching the server any more. Without - // this OBS keeps encoding and reporting the stream as live forever. Only the - // current attempt signals, which is what keeps it to one signal per Start(). - // A detached output has already been destroyed, so there is nobody to tell. - if (code < 0 && current && output) { - obs_output_set_last_error(output, reason.c_str()); - // CONNECT_FAILED is terminal for OBS. DISCONNECTED lets its own reconnect - // logic retry, which is only worth offering once we know the server works. - obs_output_signal_stop(output, was_connected ? OBS_OUTPUT_DISCONNECTED : OBS_OUTPUT_CONNECT_FAILED); - } + *reason = last_failure_reason; } void MoQOutput::Data(struct encoder_packet *packet) @@ -489,9 +490,9 @@ void MoQOutput::Data(struct encoder_packet *packet) if (!packet) { // One report for the pair, so a session failure can't slip between the // teardown and the encode error and report a second time. - std::lock_guard signal_lock(state->signal_mutex); + std::lock_guard signal_lock(signal_mutex); Stop(false); - state->SignalStop(OBS_OUTPUT_ENCODE_ERROR); + obs_output_signal_stop(output, OBS_OUTPUT_ENCODE_ERROR); return; } @@ -502,8 +503,26 @@ void MoQOutput::Data(struct encoder_packet *packet) } } +// The packet's timestamp on the broadcast clock, or nothing when it falls before zero. +static std::optional PacketTimestamp(const struct encoder_packet *packet) +{ + // Add ~1 second offset to handle negative PTS from audio priming frames, and + // the same to video for A/V sync. + // TODO: This is slightly wrong when den is not evenly divisible by num, but close enough. + int64_t pts = packet->pts + packet->timebase_den / packet->timebase_num; + if (pts < 0) + return std::nullopt; + return util_mul_div64(pts, 1000000ULL * packet->timebase_num, packet->timebase_den); +} + +static moq::Frame PacketFrame(const struct encoder_packet *packet, uint64_t timestamp_us) +{ + return moq::Frame{std::vector(packet->data, packet->data + packet->size), timestamp_us}; +} + void MoQOutput::AudioData(struct encoder_packet *packet) { + std::lock_guard lock(media_mutex); obs_encoder_t *encoder = packet->encoder; auto it = audio_tracks.find(encoder); @@ -511,25 +530,20 @@ void MoQOutput::AudioData(struct encoder_packet *packet) AudioInit(encoder); it = audio_tracks.find(encoder); } - if (it == audio_tracks.end() || it->second < 0) { + if (it == audio_tracks.end() || !it->second) { // We failed to initialize the audio track, so we can't write any data. return; } - int handle = it->second; + const auto &track = it->second; - // Add ~1 second offset to handle negative PTS from audio priming frames. - // TODO: This is slightly wrong when den is not evenly divisible by num, but close enough. - int64_t pts = packet->pts + packet->timebase_den / packet->timebase_num; - if (pts < 0) { + auto timestamp = PacketTimestamp(packet); + if (!timestamp) { LOG_WARNING("Dropping audio frame with negative PTS: %lld", (long long)packet->pts); return; } - auto pts_us = util_mul_div64(pts, 1000000ULL * packet->timebase_num, packet->timebase_den); - - auto result = moq_publish_media_frame(handle, packet->data, packet->size, pts_us); - if (result < 0) { - LOG_ERROR("Failed to write audio frame: %d", result); + if (auto result = track->write_frame(PacketFrame(packet, *timestamp)); !result) { + LOG_ERROR("Failed to write audio frame: %s", result.error().to_string().c_str()); return; } @@ -537,15 +551,13 @@ void MoQOutput::AudioData(struct encoder_packet *packet) // stream is one group. Cut per frame, which is one QUIC stream per packet forwarded without // waiting for the next, the right trade for live. Video groups at its own keyframes. // Cut before observing the flush so a failed observation never leaves the group open. - result = moq_publish_media_cut(handle); - if (result < 0) { - LOG_ERROR("Failed to cut audio group: %d", result); + if (auto result = track->cut(); !result) { + LOG_ERROR("Failed to cut audio group: %s", result.error().to_string().c_str()); return; } - result = moq_publish_media_flush(handle, pts_us); - if (result < 0) { - LOG_ERROR("Failed to observe audio encoder flush: %d", result); + if (auto result = track->flush(*timestamp); !result) { + LOG_ERROR("Failed to observe audio encoder flush: %s", result.error().to_string().c_str()); return; } @@ -554,6 +566,7 @@ void MoQOutput::AudioData(struct encoder_packet *packet) void MoQOutput::VideoData(struct encoder_packet *packet) { + std::lock_guard lock(media_mutex); obs_encoder_t *encoder = packet->encoder; auto it = video_tracks.find(encoder); @@ -561,41 +574,39 @@ void MoQOutput::VideoData(struct encoder_packet *packet) VideoInit(encoder); it = video_tracks.find(encoder); } - if (it == video_tracks.end() || it->second < 0) + if (it == video_tracks.end() || !it->second) return; - int handle = it->second; + const auto &track = it->second; - // Add ~1 second offset to match audio for A/V sync. - // TODO: This is slightly wrong when den is not evenly divisible by num, but close enough. - int64_t pts = packet->pts + packet->timebase_den / packet->timebase_num; - if (pts < 0) { + auto timestamp = PacketTimestamp(packet); + if (!timestamp) { LOG_WARNING("Dropping video frame with negative PTS: %lld", (long long)packet->pts); return; } - auto pts_us = util_mul_div64(pts, 1000000ULL * packet->timebase_num, packet->timebase_den); - - auto result = moq_publish_media_frame(handle, packet->data, packet->size, pts_us); - if (result < 0) { - LOG_ERROR("Failed to write video frame: %d", result); + if (auto result = track->write_frame(PacketFrame(packet, *timestamp)); !result) { + LOG_ERROR("Failed to write video frame: %s", result.error().to_string().c_str()); return; } - result = moq_publish_media_flush(handle, pts_us); - if (result < 0) { - LOG_ERROR("Failed to observe video encoder flush: %d", result); + if (auto result = track->flush(*timestamp); !result) { + LOG_ERROR("Failed to observe video encoder flush: %s", result.error().to_string().c_str()); return; } total_bytes_sent += packet->size; } +// NOTE: Caller must hold media_mutex. void MoQOutput::VideoInit(obs_encoder_t *encoder) { if (!encoder) { LOG_ERROR("Failed to get video encoder"); return; } + // Stopped, or not started yet: no broadcast to publish on. + if (!broadcast) + return; OBSDataAutoRelease settings = obs_encoder_get_settings(encoder); const auto video_width = obs_encoder_get_width(encoder); @@ -615,53 +626,54 @@ void MoQOutput::VideoInit(obs_encoder_t *encoder) // Map the OBS codec name onto a MoQ format. Both H.26x entries are the Annex-B framing // with inline parameter sets, which is what OBS hands us. - moq_video_init config{}; + moq::VideoInit config{}; if (strcmp(codec, "h264") == 0) { - config.format = MOQ_VIDEO_FORMAT_AVC3; + config.format = moq::VideoFormat::kAvc3; } else if (strcmp(codec, "hevc") == 0) { - config.format = MOQ_VIDEO_FORMAT_HEV1; + config.format = moq::VideoFormat::kHev1; } else if (strcmp(codec, "av1") == 0) { - config.format = MOQ_VIDEO_FORMAT_AV01; + config.format = moq::VideoFormat::kAv01; } else { LOG_ERROR("Unsupported video codec: %s", codec); + video_tracks[encoder] = nullptr; return; } - config.init = extra_data; - config.init_len = extra_size; + if (extra_data && extra_size > 0) + config.data.assign(extra_data, extra_data + extra_size); // Seed catalog fields a downstream moq-transcode needs before measured rates // arrive: coded size (also from SPS once parsed) and configured CBR bitrate so // same-height ladder rungs can undercut the mezzanine. - if (video_width > 0 && video_height > 0) { - config.hint.coded_width = video_width; - config.hint.coded_height = video_height; - config.hint.has_coded = true; - } + moq::VideoHint hint{}; + if (video_width > 0 && video_height > 0) + hint.coded = moq::Dimensions{video_width, video_height}; const std::string rate_control = settings ? obs_data_get_string(settings, "rate_control") : ""; - if (video_bitrate_kbps > 0 && (rate_control == "CBR" || rate_control == "cbr")) { - config.hint.bitrate = (uint64_t)video_bitrate_kbps * 1000ULL; - config.hint.has_bitrate = true; - } - config.hint.optimize_for_latency = true; - config.hint.has_optimize_for_latency = true; - - int handle = moq_publish_video(broadcast, &config); - video_tracks[encoder] = handle; - if (handle < 0) { - LOG_ERROR("Failed to initialize video track: %d", handle); + if (video_bitrate_kbps > 0 && (rate_control == "CBR" || rate_control == "cbr")) + hint.bitrate = (uint64_t)video_bitrate_kbps * 1000ULL; + hint.optimize_for_latency = true; + config.hint = hint; + + auto track = broadcast->publish_video(config); + if (!track) { + LOG_ERROR("Failed to initialize video track: %s", track.error().to_string().c_str()); + video_tracks[encoder] = nullptr; return; } + video_tracks[encoder] = *track; LOG_INFO("Video track initialized (%ux%u, %d kbps)", video_width, video_height, video_bitrate_kbps); } +// NOTE: Caller must hold media_mutex. void MoQOutput::AudioInit(obs_encoder_t *encoder) { if (!encoder) { LOG_ERROR("Failed to get audio encoder"); return; } + if (!broadcast) + return; // TODO Pass these along to the audio catalog somehow. /* @@ -685,28 +697,31 @@ void MoQOutput::AudioInit(obs_encoder_t *encoder) const char *codec = obs_encoder_get_codec(encoder); - // The codec string used to go straight through, so an unsupported one failed deep in the - // importer. Mapping it here means OBS says which codec it was. - moq_audio_init config{}; + // Mapping the codec here means OBS says which one it was, rather than the importer + // failing on it deep inside. + moq::AudioInit config{}; if (strcmp(codec, "opus") == 0) { - config.format = MOQ_AUDIO_FORMAT_OPUS; + config.format = moq::AudioFormat::kOpus; } else if (strcmp(codec, "aac") == 0) { - config.format = MOQ_AUDIO_FORMAT_AAC; + config.format = moq::AudioFormat::kAac; } else if (strcmp(codec, "flac") == 0) { - config.format = MOQ_AUDIO_FORMAT_FLAC; + config.format = moq::AudioFormat::kFlac; } else { LOG_ERROR("Unsupported audio codec: %s", codec); + audio_tracks[encoder] = nullptr; return; } - config.init = extra_data; - config.init_len = extra_size; - int handle = moq_publish_audio(broadcast, &config); - audio_tracks[encoder] = handle; - if (handle < 0) { - LOG_ERROR("Failed to initialize audio track: %d", handle); + if (extra_data && extra_size > 0) + config.data.assign(extra_data, extra_data + extra_size); + + auto track = broadcast->publish_audio(config); + if (!track) { + LOG_ERROR("Failed to initialize audio track: %s", track.error().to_string().c_str()); + audio_tracks[encoder] = nullptr; return; } + audio_tracks[encoder] = *track; LOG_INFO("Audio track initialized successfully"); } diff --git a/cpp/obs/src/moq-output.h b/cpp/obs/src/moq-output.h index bdff2a4cfd8..0509cc7902b 100644 --- a/cpp/obs/src/moq-output.h +++ b/cpp/obs/src/moq-output.h @@ -2,14 +2,19 @@ #pragma once #include +#include + #include #include #include #include #include #include +#include #include +#include #include "logger.h" +#include "moq-worker.h" class MoQOutput { public: @@ -22,10 +27,10 @@ class MoQOutput { inline size_t GetTotalBytes() { return total_bytes_sent; } - inline int GetConnectTime() { return state->connect_time_ms; } + inline int GetConnectTime() { return connect_time_ms; } // Point-in-time QUIC/WebTransport health for the live session. False when - // there is no session or libmoq is between reconnects (no live connection). + // there is no session or it is between reconnects (no live connection). // A failed read leaves the caller's snapshot unchanged. struct ConnectionStats { int reconnects = 0; @@ -47,11 +52,7 @@ class MoQOutput { bool TryGetConnectionStats(ConnectionStats *out); // Successful (re)connects after the first for this Start(); 0 until epoch >= 2. - inline int GetReconnectCount() - { - const int epoch = state->epoch.load(std::memory_order_relaxed); - return epoch > 1 ? epoch - 1 : 0; - } + int GetReconnectCount(); // True while the current Start() attempt has an open MoQ session. // Prefer this over obs_output_get_connect_time_ms, which stays 0 for sub-ms connects. @@ -62,107 +63,32 @@ class MoQOutput { void CopyLastFailure(int *code, std::string *reason); private: - struct SessionRef; - - // Every piece of state a libmoq status callback touches, held by shared_ptr - // so it outlives MoQOutput. libmoq delivers the terminal status callback - // (code <= 0) asynchronously on its runtime thread after moq_session_close, - // and OBS is free to destroy the output long before that runtime gets around - // to it. Keeping this separate is what makes a late callback harmless: it - // finds the state detached rather than freed, so teardown never has to - // outwait a stalled runtime. - // - // It is also the only state a callback may touch: everything a callback needs - // about its own attempt is copied into its SessionRef, since the members here - // belong to whichever attempt is current. - struct SessionState { - explicit SessionState(obs_output_t *output) : output(output) {} - - // Stop reporting to OBS. Returns only once no callback is inside an OBS - // call, so the output can be torn down behind it. The wait is bounded by - // an OBS call rather than by anything libmoq does. - void Detach(); - - // Tell OBS the output stopped, unless it has already been detached. Every - // obs_output_signal_stop goes through here or through an explicit - // signal_mutex hold. - void SignalStop(int code); - - void Connected(const SessionRef &ref, int connect_epoch); - void Closed(const SessionRef &ref, int code); - - // Serializes reporting the output's fate to OBS against tearing it down, and - // is held across the OBS calls themselves. Without it the status callback can - // decide to report a failure, lose the race to Stop(), and still signal: - // OBS_OUTPUT_DISCONNECTED then makes OBS reconnect a stream the user just - // stopped, since obs_output_signal_stop never checks whether the output is - // still active. Start() also holds it from the connect through - // obs_output_begin_data_capture, so a session that dies mid-startup cannot - // report against an output that isn't committed yet. - // - // Recursive because obs_output_signal_stop and obs_output_begin_data_capture - // run the frontend's handlers inline, and a frontend may call obs_output_stop - // straight back into Stop() on this thread. - // - // Lock order: signal_mutex first, then mutex. Never the reverse, and never - // take signal_mutex from inside a libmoq call. That keeps it ordered ahead - // of libmoq's runtime lock too, since the status callback runs with no - // libmoq locks held. - // - // Two costs this buys, both bounded and deliberate: - // - A frontend that re-enters this output from a *different* thread while we - // hold it would deadlock, which recursion does not help with. OBS reaches - // one such point, the pthread_join in end_data_capture_internal. Studio's - // frontend queues its handlers, so it does not arise there. - // - libmoq's runtime is single threaded, so a terminal callback parked here - // stalls every MoQ session in the process. The holds are short, and the - // expensive part of starting (obs_output_initialize_encoders) is - // deliberately outside the lock. - std::recursive_mutex signal_mutex; - - // The OBS output, or null once MoQOutput has been destroyed. Guarded by - // signal_mutex, which every call made through it is also held under. - obs_output_t *output; - - // Guards the group below, which the OBS thread and the libmoq runtime - // thread both touch. - std::mutex mutex; - // The live session handle, or 0 when there is none. libmoq drops the handle - // before firing the terminal callback, so it is retired there rather than - // closed later (the close would just fail with "session not found"). - int session = 0; - // Bumped whenever the publish state is torn down or restarted. A status - // callback stamped with an older value belongs to a superseded attempt and - // must leave both OBS and `session` alone, which is also what limits the - // failure signal to one per Start(). - uint64_t attempt = 0; - // Whether the current attempt ever reached the server, which picks between - // telling OBS the connection failed and telling it the stream dropped. - bool connected = false; - // Dial URL of the current attempt, for the stats snapshot's scheme label. - std::string url; - int last_failure_code = 0; - std::string last_failure_reason; - - // Written by the status callback (libmoq runtime thread), read by - // GetConnectTime() and GetReconnectCount() (OBS thread); atomic to avoid a - // data race. - std::atomic connect_time_ms{0}; - std::atomic epoch{0}; - }; - - // Handed to libmoq as the status callback's user_data. Carries everything a - // callback needs about its own Start() attempt, so it never has to read a - // member the OBS thread may already be rewriting for the next attempt, plus - // the reference that keeps that state alive. Freed by the terminal callback. - struct SessionRef { - std::shared_ptr state; - uint64_t attempt; + // One Start()'s connection: the client dialing, the session once it connects, + // and the pending call whose continuation reports on it. Replaced wholesale by + // the next Start() and dropped by Stop(), which cancels that call. + struct Attempt { + std::shared_ptr client; + std::shared_ptr session; + std::optional pending; std::string url; std::chrono::steady_clock::time_point started; + // Whether this attempt ever reached the server, which picks between telling + // OBS the connection failed and telling it the stream dropped. + bool connected = false; + // False while the session is between reconnects. + bool live = false; }; - static void SessionStatus(void *user_data, int code); + void OnConnect(const std::shared_ptr &attempt, moq::expected> result); + void OnStatus(const std::shared_ptr &attempt, moq::expected result); + void WatchStatus(const std::shared_ptr &attempt); + void Fail(const std::shared_ptr &attempt, const moq::Error &error); + + // Runs `callback` on the worker if `attempt` is still the current one once it + // holds signal_mutex. Cancelling the call on Stop() is not enough on its own: a + // result that already completed may be queued on the worker by then. + template + std::function Current(const std::shared_ptr &attempt, Callback callback); // Tear down the publish state without telling OBS. void Reset(); @@ -172,22 +98,59 @@ class MoQOutput { void AudioInit(obs_encoder_t *encoder); void AudioData(struct encoder_packet *packet); - // The OBS thread's view of the output, valid for this object's lifetime. - // Reporting a stop goes through state->SignalStop instead, which is the only - // path a callback thread may take. obs_output_t *output; - const std::shared_ptr state; + // Runs every continuation, so none runs once the destructor has stopped it. + MoQWorker worker; + + // Serializes reporting the output's fate to OBS against tearing it down, and is + // held across the OBS calls themselves. Without it a continuation can decide to + // report a failure, lose the race to Stop(), and still signal: + // OBS_OUTPUT_DISCONNECTED then makes OBS reconnect a stream the user just + // stopped, since obs_output_signal_stop never checks whether the output is still + // active. Start() also holds it from the connect through + // obs_output_begin_data_capture, so a session that fails mid-startup cannot + // report against an output that isn't committed yet. + // + // Recursive because obs_output_signal_stop and obs_output_begin_data_capture run + // the frontend's handlers inline, and a frontend may call obs_output_stop + // straight back into Stop() on this thread. + // + // Lock order: signal_mutex, then mutex, then media_mutex. + std::recursive_mutex signal_mutex; + + // The current attempt, or null once it was stopped or failed. Guarded by + // signal_mutex; the fields the dock polls are copied out under `mutex`. + std::shared_ptr attempt; + + // Guards the group below, which the dock reads from the UI thread. + std::mutex mutex; + std::shared_ptr session; + bool connected = false; + bool live = false; + std::string url; + int last_failure_code = 0; + std::string last_failure_reason; + + // Written on the worker, read by GetConnectTime(). + std::atomic connect_time_ms{0}; + + // Guards the publishing side, which the encoder threads write through Data(). + // Never held across an OBS call that can wait on an encoder thread. + std::mutex media_mutex; + std::shared_ptr origin; + std::shared_ptr broadcast; + // An encoder maps to null when its track failed to initialize, so it isn't retried. + std::map> video_tracks; + std::map> audio_tracks; + + // Retired sessions still draining their finished tracks, which the destructor + // waits out. Guarded by signal_mutex. + std::vector> draining; std::string path; size_t total_bytes_sent; - - int origin; - int broadcast; - - std::map video_tracks; - std::map audio_tracks; }; void register_moq_output(); diff --git a/cpp/obs/src/moq-settings.cpp b/cpp/obs/src/moq-settings.cpp index a1dcfec1924..495b90a4524 100644 --- a/cpp/obs/src/moq-settings.cpp +++ b/cpp/obs/src/moq-settings.cpp @@ -7,59 +7,43 @@ #include #include -#include "moq.h" - namespace MoQSettings { namespace { // Keys. These are stable: they land in scene collections and in the dock's settings // file, so renaming one silently drops whatever the user had configured. -constexpr const char *VERSION = "version"; constexpr const char *BIND = "bind"; -constexpr const char *CONNECT_TIMEOUT = "connect_timeout_ms"; -constexpr const char *FAILOVER_DELAY = "failover_delay_ms"; constexpr const char *TLS_DISABLE_VERIFY = "tls_disable_verify"; constexpr const char *TLS_FINGERPRINT = "tls_fingerprint"; constexpr const char *TLS_ROOT = "tls_root"; -constexpr const char *TLS_HOST_NAME = "tls_host_name"; constexpr const char *BACKOFF_INITIAL = "backoff_initial_ms"; constexpr const char *BACKOFF_MAX = "backoff_max_ms"; constexpr const char *BACKOFF_TIMEOUT = "backoff_timeout_ms"; -constexpr const char *QUIC_CONGESTION_CONTROL = "quic_congestion_control"; constexpr const char *QUIC_MAX_STREAMS = "quic_max_streams"; -constexpr const char *QUIC_IDLE_TIMEOUT = "quic_idle_timeout_ms"; -constexpr const char *QUIC_KEEP_ALIVE = "quic_keep_alive_ms"; -constexpr const char *QUIC_GSO = "quic_gso"; -constexpr const char *QUIC_MTU_DISCOVERY = "quic_mtu_discovery"; -constexpr const char *QUIC_QLOG = "quic_qlog"; constexpr const char *WEBSOCKET_ENABLED = "websocket_enabled"; constexpr const char *WEBSOCKET_DELAY = "websocket_delay_ms"; -// Read a string setting, returning nullptr when it's unset or empty. libmoq's optional -// string setters take exactly that: NULL means "leave at the default". +// moq-ffi documents these defaults on the MoqClient setters but does not report them, +// so they are repeated here. The backoff defaults come from moq::Backoff itself. +constexpr long long QUIC_MAX_STREAMS_DEFAULT = 1024; +constexpr bool WEBSOCKET_ENABLED_DEFAULT = true; +constexpr long long WEBSOCKET_DELAY_DEFAULT_MS = 200; + +// Read a string setting, returning nullptr when it's unset or empty, which leaves the +// knob at the library default. const char *OptionalString(obs_data_t *settings, const char *key) { const char *value = obs_data_get_string(settings, key); return (value && *value) ? value : nullptr; } -// Resolve a tri-state choice (AUTO / "on" / "off") into an override, if any. -bool TriState(obs_data_t *settings, const char *key, bool *out) -{ - const char *value = OptionalString(settings, key); - if (!value) - return false; - *out = strcmp(value, "on") == 0; - return true; -} - // Read an integer setting inside the range declared by its Field. Scene collections -// are editable JSON, so the widget bounds alone do not protect the unsigned C API. +// are editable JSON, so the widget bounds alone do not protect the unsigned API. uint64_t Amount(obs_data_t *settings, const char *key) { const long long value = obs_data_get_int(settings, key); @@ -72,41 +56,6 @@ uint64_t Amount(obs_data_t *settings, const char *key) return 0; } -// Enumerate a libmoq capability into menu options. The names live beside the caller's -// static field table because Option borrows their pointers. -std::vector