diff --git a/.github/workflows/libmoq.yml b/.github/workflows/libmoq.yml index 7eb12759de1..bf0e8456bfb 100644 --- a/.github/workflows/libmoq.yml +++ b/.github/workflows/libmoq.yml @@ -106,106 +106,3 @@ jobs: RELEASE_TITLE: "libmoq v${{ steps.parse.outputs.version }}" RELEASE_PREV_TAG: ${{ steps.prev_tag.outputs.tag }} run: .github/scripts/release.sh create artifacts - - # The OBS plugin (cpp/obs) statically links libmoq, so it ships with every - # libmoq 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 libmoq 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 - - - name: Parse version - id: parse - shell: bash - run: .github/scripts/release.sh parse-version libmoq - - # Native Xcode / Visual Studio; libobs/Qt6/ffmpeg come from the obs-deps - # bundle and libmoq from the release we just published (--libmoq-release). - - name: Build and package - shell: bash - env: - TARGET: ${{ matrix.target }} - VERSION: ${{ steps.parse.outputs.version }} - run: | - ./cpp/obs/build.sh --target "$TARGET" --libmoq-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 - - - name: Parse version - id: parse - run: .github/scripts/release.sh parse-version libmoq - - # Only the obs-moq archives: the libmoq-* artifacts from the build job - # belong to the libmoq 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 - # libmoq-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: .github/scripts/release.sh create artifacts diff --git a/.github/workflows/nightly.yml b/.github/workflows/nightly.yml index adbbea4d67e..ad77b178fe7 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). `windows` and `macos` +# 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). `windows` and `macos` # are diff-independent because what they catch is a whole target's worth of # `#[cfg(target_os = ...)]` code that no Linux job compiles at all, and those # runners are throttled too hard to gate every PR on. @@ -112,9 +112,9 @@ jobs: if: ${{ !cancelled() }} run: nix develop --command just rs audit - # The OBS plugin and libmoq's C fixtures are the only things that link - # libmoq.a from outside cargo, so a stale `rs/libmoq/native-libs/*.txt` - # breaks them and nothing else. obs.yml lists the inputs that can do that, + # The OBS plugin (over moq-ffi) and libmoq's C fixtures link a Rust + # staticlib from outside cargo, so a stale `rs/libmoq/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 diff --git a/.github/workflows/obs.yml b/.github/workflows/obs.yml index 8b4c03420ce..17a87e2f27f 100644 --- a/.github/workflows/obs.yml +++ b/.github/workflows/obs.yml @@ -6,32 +6,33 @@ name: OBS # # Linux is the platform that needs no bundle at all -- nixpkgs has obs-studio, # Qt6 and ffmpeg, so the dev shell is the whole dependency set. The plugin is -# platform-independent C++ over libmoq's C ABI, so a Linux compile catches the -# breakage a macOS developer would otherwise ship uncompiled. +# 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. # -# Narrow trigger, like swift.yml: the plugin's own sources, the libmoq 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. # # libmoq'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/libmoq/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/libmoq/native-libs/*.txt`, which moq-ffi and +# libmoq 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. libmoq 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 @@ -49,6 +50,8 @@ on: types: [opened, synchronize, reopened, closed] paths: - "cpp/obs/**" + - "cpp/moq/**" + - "rs/moq-ffi/**" - "rs/libmoq/**" - "doc/lib/c/**" - "doc/lib/samples.sh" @@ -94,7 +97,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 libmoq 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 @@ -104,7 +107,5 @@ jobs: - name: Build run: nix develop --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: libmoq C fixtures run: nix develop --command just rs c-tests diff --git a/.github/workflows/release-cpp.yml b/.github/workflows/release-cpp.yml index 3c79288f9a4..0a02075913e 100644 --- a/.github/workflows/release-cpp.yml +++ b/.github/workflows/release-cpp.yml @@ -109,3 +109,123 @@ jobs: RELEASE_TAG: ${{ github.ref_name }} RELEASE_PREV_TAG: ${{ steps.prev_tag.outputs.tag }} run: RELEASE_TITLE="moq C++ v${RELEASE_TAG#cpp-v}" .github/scripts/release.sh 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 + + # 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: | + ./cpp/obs/build.sh --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 + + - 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 + + # release.sh 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: .github/scripts/release.sh create artifacts diff --git a/AGENTS.md b/AGENTS.md index 4728f9bd687..8684747b1fc 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -98,7 +98,7 @@ A quest deleted on `dev` is done, even while `main` still lists it. | Change in | Also update | |---|---| -| `rs/moq-ffi` | `rs/libmoq`, `{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`, `doc/lib/{py,swift,kt,go,dart,c,cpp}` | +| `rs/moq-ffi` | `rs/libmoq`, `{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-token` | `js/token` | @@ -107,7 +107,7 @@ A quest deleted on `dev` is done, even while `main` still lists it. | `rs/moq-cli` | `doc/bin/cli.md` | | `rs/moq-token-cli` | `doc/bin/relay/auth.md`, `doc/lib/rs/moq-token.md`, `doc/lib/rs/index.md` | | `rs/moq-gst` | `doc/bin/gstreamer.md` | -| `rs/libmoq` C ABI (`moq.h`) | `cpp/obs/src`, `doc/bin/obs.md` | +| `rs/libmoq` 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 28673e0f47e..881d0ded887 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -67,4 +67,5 @@ Releases are cut separately; bump only when asked. Each package's version lives - **Python**: `py/moq-rs/pyproject.toml` only; `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**: the plugin takes the `libmoq` crate's version (`cpp/obs/CMakeLists.txt`), so release-plz owns it. +- **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/obs/CMakeLists.txt b/cpp/obs/CMakeLists.txt index 636b2ea81af..ad250c05f66 100644 --- a/cpp/obs/CMakeLists.txt +++ b/cpp/obs/CMakeLists.txt @@ -1,30 +1,26 @@ cmake_minimum_required(VERSION 3.28) -# Build the in-tree libmoq 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/libmoq in-tree)") -set(MOQ_VERSION "" CACHE STRING "Published libmoq 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/libmoq/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/libmoq/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/libmoq") + message(FATAL_ERROR "MOQ_VERSION is required when MOQ_LOCAL does not contain cpp/moq") endif() set(_moq_link_version "${MOQ_VERSION}") endif() @@ -67,36 +63,20 @@ endif() target_link_libraries(obs-moq PRIVATE OBS::libobs) if(_moq_local_build) - add_subdirectory("${MOQ_LOCAL}/rs/libmoq" moq) - target_link_libraries(obs-moq PRIVATE moq) + 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/libmoq-v${_moq_link_version}/moq-${_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 REQUIRED PATHS ${moq_SOURCE_DIR} NO_DEFAULT_PATH) - target_link_libraries(obs-moq PRIVATE moq::moq) -endif() - -# libmoq 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 libmoq package config also declares these, but linking them here -# keeps the plugin buildable against any libmoq (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 +130,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 2c16160fec6..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 -`libmoq`, built from the in-tree [`rs/libmoq`](../../rs/libmoq) 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 dfa153c6de9..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 ;; - --libmoq-release) - # Link a published libmoq release of this version instead of - # building rs/libmoq from source. CMake fetches the matching - # moq-- archive from the GitHub release and the - # plugin is versioned to match. Used by CI on a libmoq-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] [--libmoq-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 libmoq-release mode the plugin version tracks the libmoq 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 libmoq 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 c38b7d029a8..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 libmoq release and versions it -# to match (see build.sh --libmoq-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 1c1f59ed108..83401b44734 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). # -# libmoq is built from the in-tree crate (rs/libmoq) 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 libmoq builds from rs/libmoq); pass `path=/some/moq` to +# CMakeLists.txt (so the C++ package builds from cpp/moq); pass `path=/some/moq` to # override. We deliberately don't compute the path here: `justfile_directory()` # resolves to the root justfile when this runs as `just obs setup`, so the @@ -61,25 +62,23 @@ run: cp -a build_macos/RelWithDebInfo/obs-moq.plugin "$dest/" RUST_LOG=debug RUST_BACKTRACE=1 OBS_LOG_LEVEL=debug /Applications/OBS.app/Contents/MacOS/OBS -# Type-check every plugin source and unit test, without linking or an obs-deps -# download. +# Type-check every plugin source and unit test, without linking the plugin or an +# obs-deps download. # # This is the gate to run in a worktree. `just obs build` needs the -# multi-hundred-MB obs-deps bundle (macOS/Windows) and a full cargo build, per -# worktree; this needs neither, because the dev shell carries all three header -# sets: libobs pinned to the same OBS release buildspec.json downloads (see -# `obs-headers` in flake.nix), Qt6, and ffmpeg. CI links the real thing on Linux -# (.github/workflows/obs.yml). +# multi-hundred-MB obs-deps bundle (macOS/Windows); this doesn't, because the dev +# shell carries the OBS-side header sets: libobs pinned to the same OBS release +# buildspec.json downloads (see `obs-headers` in flake.nix), Qt6, and ffmpeg. The +# moq headers are generated, so it builds rs/moq-ffi to render them. CI links the +# real thing on Linux (.github/workflows/obs.yml). # # It compiles the Qt sources too, which the CMake build only does when # ENABLE_QT and ENABLE_FRONTEND_API are on, so the definitions they gate on are # set here to match. # -# test/ is in scope because each test file defines the libmoq entry points the -# plugin calls, so a signature that drifts from the generated moq.h is a -# conflicting C declaration. Catching that needs only headers, which is why it -# belongs here: `just obs ci` finds the same drift, but only where obs.yml's -# path filter reaches, and `just obs test` is manual. +# test/ is in scope because the tests drive the plugin through the same +# generated moq headers, so an API that drifts under them fails here rather than +# only where obs.yml's path filter reaches; `just obs test` is manual. compile: #!/usr/bin/env bash set -euo pipefail @@ -110,7 +109,7 @@ compile: ffmpeg=$(pkg-config --cflags libavcodec libavutil libswscale libswresample) # MOQ_VERSION_STRING only reaches a label in the dock, so any value - # type-checks the same; CMake stamps the real libmoq version. + # type-checks the same; CMake stamps the real moq-cpp version. status=0 for source in src/*.cpp test/*.cpp; do # shellcheck disable=SC2086 @@ -137,7 +136,7 @@ compile: # gates a merge. # # CMAKE_BUILD_TYPE overrides the preset's RelWithDebInfo because the Rust cache -# CI restores holds dev-profile artifacts. A release build of libmoq would +# CI restores holds dev-profile artifacts. A release build of moq-ffi would # rebuild its whole dependency tree on every run, and the gate is about whether # the plugin compiles, which the profile doesn't change. ci preset="": @@ -148,11 +147,8 @@ ci preset="": cmake --build --preset "$PRESET" just _unit -# Print the -I flags for libobs and libmoq, shared by `compile` and `_unit`. -# -# moq.h is generated by cbindgen from rs/libmoq/src, so regenerate rather than -# just locating it: a stale header is how a plugin call to a since-changed -# libmoq function type-checks anyway. +# Print the -I flags for libobs and the moq C++ package, shared by `compile` and +# `_unit`. [private] _includes: #!/usr/bin/env bash @@ -198,36 +194,45 @@ _includes: exit 1 fi - (cd ../.. && cargo check --locked --quiet -p libmoq) - - # rs/libmoq/build.rs writes the header into its OUT_DIR, so ask cargo where - # that is. Replays the check above from cache, so it costs a process rather - # than a compile. - out_dir=$(cd ../.. && cargo check --locked -p libmoq --message-format=json | - jq -r 'select(.reason == "build-script-executed") | select(.package_id | tostring | test("libmoq")) | .out_dir' | - tail -1) - if [ -z "$out_dir" ]; then - echo "cargo reported no build script output for libmoq" >&2 - exit 1 - fi - moq_include="$out_dir/include" - if [ ! -f "$moq_include/moq.h" ]; then - echo "$moq_include/moq.h missing after 'cargo check --locked -p libmoq'" >&2 - exit 1 - fi + moq_include="$(just _moq)/include" # One flag per line, path glued to the -I, so a caller can rebuild the # arguments without splitting on the spaces in a path. printf -- '-I%s\n' src "$obs_include" "$simde_include" "$moq_include" -# Unit-test the plugin sources against stubbed libobs/libmoq/ffmpeg, under -# ThreadSanitizer. +# Build the moq C++ package from the in-tree cpp/moq, install it to a prefix, +# and print the prefix. Rebuilt every time rather than trusted from a previous +# run: the headers are rendered from the moq-ffi build, so a stale copy would +# type-check the plugin against an API that has since changed. Cargo and the +# copy-if-different render keep a no-op rebuild cheap. +# +# Debug, for the same reason `ci` is: the Rust cache CI restores holds +# dev-profile artifacts. +[private] +_moq: + #!/usr/bin/env bash + set -euo pipefail + if ! command -v uniffi-bindgen-cpp >/dev/null 2>&1; then + echo "uniffi-bindgen-cpp not on PATH; run inside 'nix develop', which supplies it" >&2 + exit 1 + fi + build="$PWD/../moq/build/obs" + prefix="$build/prefix" + { + cmake -S ../moq -B "$build" -DCMAKE_BUILD_TYPE=Debug -DCMAKE_INSTALL_LIBDIR=lib + cmake --build "$build" --config Release + cmake --install "$build" --config Release --prefix "$prefix" + } >&2 + echo "$prefix" + +# Unit-test the plugin sources against stubbed libobs and ffmpeg and the real +# moq-ffi, under ThreadSanitizer. # # Manual, like `just rs macos` and `just rs windows`, because ThreadSanitizer # needs its own build. `just obs ci` runs the same tests without it, so a # regression these assertions catch still turns PR CI red; the sanitizer is what -# adds the races on top. Run this whenever you touch src/, especially the session -# callback plumbing, whose orderings the build can't check. +# adds the races on top. Run this whenever you touch src/, especially the +# continuation plumbing, whose orderings the build can't check. test: #!/usr/bin/env bash set -euo pipefail @@ -286,6 +291,13 @@ _unit mode="plain": ffmpeg=$(pkg-config --cflags libavcodec libavutil libswscale) + # The output and source tests run against the real moq-ffi, over a server in + # the same process. The bindings compile here, with the tests' flags. + PKG_CONFIG_PATH="$(just _moq)/lib/pkgconfig" + export PKG_CONFIG_PATH + moq_sources=$(pkg-config --variable=sources moq-cpp) + moq_libs=$(pkg-config --libs moq-cpp) + flags=(-std=c++17 -g -O0 -pthread) if [ "{{ mode }}" = "tsan" ]; then flags+=(-fsanitize=thread) @@ -294,21 +306,23 @@ _unit mode="plain": out=$(mktemp -d) trap 'rm -rf "$out"' EXIT - # One binary per source under test: each test file defines its own libobs and - # libmoq stubs, so two of them can't share a link. Header-only helpers - # (quality defaults) build as a third binary with no plugin source. - for name in moq-output moq-source; do + "$cxx" "${flags[@]}" "${includes[@]}" -c "$moq_sources" -o "$out/moq.o" + + # One binary per source under test: each test file defines its own libobs + # stubs, so two of them can't share a link. Header-only helpers (quality + # defaults) build as separate binaries with no plugin source. + for name in moq-output moq-source obs-moq; do # shellcheck disable=SC2086 - # $ffmpeg stays unquoted on purpose: pkg-config hands back one - # space-separated string, and splitting it is the intended reading. + # $ffmpeg and $moq_libs stay unquoted on purpose: pkg-config hands back + # one space-separated string, and splitting it is the intended reading. "$cxx" "${flags[@]}" "${includes[@]}" $ffmpeg \ - -o "$out/$name-test" "test/$name-test.cpp" "src/$name.cpp" - TSAN_OPTIONS="halt_on_error=1" "$out/$name-test" + -o "$out/$name-test" "test/$name-test.cpp" "src/$name.cpp" "$out/moq.o" $moq_libs + TSAN_OPTIONS="halt_on_error=1 suppressions=$PWD/test/tsan.supp" "$out/$name-test" done for name in moq-quality-defaults moq-dock-stop moq-error moq-encoder-latency moq-dial moq-spark; do "$cxx" "${flags[@]}" "${includes[@]}" \ -o "$out/$name-test" "test/$name-test.cpp" - TSAN_OPTIONS="halt_on_error=1" "$out/$name-test" + TSAN_OPTIONS="halt_on_error=1 suppressions=$PWD/test/tsan.supp" "$out/$name-test" done # Lint formatting. Skips a tool silently if it isn't on $PATH (matches repo convention). @@ -366,7 +380,7 @@ check: echo "CMake accepted a release build without MOQ_VERSION" >&2 exit 1 fi - if ! grep -Fq "MOQ_VERSION is required when MOQ_LOCAL does not contain rs/libmoq" "$build_dir/configure.log"; then + if ! grep -Fq "MOQ_VERSION is required when MOQ_LOCAL does not contain cpp/moq" "$build_dir/configure.log"; then cat "$build_dir/configure.log" >&2 exit 1 fi diff --git a/cpp/obs/src/moq-dock.cpp b/cpp/obs/src/moq-dock.cpp index ff6e85e1b97..0f7ea21fea8 100644 --- a/cpp/obs/src/moq-dock.cpp +++ b/cpp/obs/src/moq-dock.cpp @@ -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 7a5e071951f..63854e45b05 100644 --- a/cpp/obs/src/moq-output.cpp +++ b/cpp/obs/src/moq-output.cpp @@ -2,6 +2,7 @@ #include #include "moq-output.h" +#include "moq-error.h" #include "moq-settings.h" #include "moq-url.h" #include "logger.h" @@ -9,10 +10,8 @@ #include #include - -extern "C" { -#include "moq.h" -} +#include +#include namespace { @@ -33,20 +32,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(':'); @@ -66,47 +53,51 @@ 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(); + // 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::Detach() +template +std::function MoQOutput::Current(const std::shared_ptr ¤t, Callback callback) { - // 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; -} - -void MoQOutput::SessionState::SignalStop(int code) -{ - 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() @@ -118,7 +109,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; } @@ -133,10 +125,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; } @@ -157,156 +150,234 @@ 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. + if (attempt && attempt->session) + attempt->session->shutdown(); + 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(); } bool MoQOutput::TryGetConnectionStats(ConnectionStats *out) @@ -314,176 +385,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) @@ -491,9 +467,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; } @@ -504,8 +480,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); @@ -513,25 +507,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; } @@ -539,15 +528,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; } @@ -556,6 +543,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); @@ -563,41 +551,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); @@ -617,53 +603,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. /* @@ -687,28 +674,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..ed318dad80f 100644 --- a/cpp/obs/src/moq-output.h +++ b/cpp/obs/src/moq-output.h @@ -2,14 +2,18 @@ #pragma once #include +#include + #include #include #include #include #include #include +#include #include #include "logger.h" +#include "moq-worker.h" class MoQOutput { public: @@ -22,10 +26,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 +51,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 +62,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 +97,55 @@ 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; 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 8b3fc4b5a24..495b90a4524 100644 --- a/cpp/obs/src/moq-settings.cpp +++ b/cpp/obs/src/moq-settings.cpp @@ -7,61 +7,43 @@ #include #include -extern "C" { -#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); @@ -74,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