Compare commits

..

No commits in common. "main" and "v0.1.0" have entirely different histories.
main ... v0.1.0

33 changed files with 348 additions and 1296 deletions

View file

@ -1,11 +1,11 @@
name: beta release name: beta release
# Publishes a beta-track build from a `vX.Y.Z-rc.N` tag. The leftover # Publishes a beta-track build on every push to `beta` — a frozen
# `beta` branch trigger is kept so an old branch push does not go silent; # stabilization branch cut from `dev` when ready to stabilize; only
# do not use that branch — cut beta from an RC tag. See CONTRIBUTING.md. # fix/<issue> branches merged into `beta` should land here afterward.
# See bread-ecosystem's docs/release-channels.md for the three-track policy.
on: on:
push: push:
tags: ['v*-rc.*']
branches: ['beta'] branches: ['beta']
jobs: jobs:
@ -16,7 +16,7 @@ jobs:
run: | run: |
set -euo pipefail set -euo pipefail
rm -rf src && mkdir src rm -rf src && mkdir src
git clone --branch "${GITHUB_REF_NAME}" --depth 1 \ git clone --branch beta --depth 1 \
"https://git.breadway.dev/${GITHUB_REPOSITORY}.git" src "https://git.breadway.dev/${GITHUB_REPOSITORY}.git" src
- name: build - name: build
@ -25,21 +25,19 @@ jobs:
- name: compute beta version - name: compute beta version
run: | run: |
set -euo pipefail set -euo pipefail
# An RC tag is already valid semver and is the beta version.
# The leftover `beta` branch path still synthesizes one from the
# latest stable tag so an old push does not publish as 0.0.0.
if [[ "${GITHUB_REF_NAME}" == v*-rc.* ]]; then
echo "VERSION=${GITHUB_REF_NAME#v}" >> "$GITHUB_ENV"
exit 0
fi
cd src cd src
# Base the beta version off the latest published stable tag,
# not Cargo.toml — Cargo.toml can go stale relative to the last
# real release (seen in practice: breadbox/breadpad/breadcrumbs/
# breadpaper), which would make a beta build sort as OLDER than
# what's already installed and bakery would correctly refuse it.
LATEST_TAG="$(git ls-remote --tags --refs \ LATEST_TAG="$(git ls-remote --tags --refs \
"https://git.breadway.dev/${GITHUB_REPOSITORY}.git" 'v*' \ "https://git.breadway.dev/${GITHUB_REPOSITORY}.git" 'v*' \
| awk -F/ '{print $NF}' | sed 's/^v//' | grep -v -- '-rc' | sort -V | tail -1)" | awk -F/ '{print $NF}' | sed 's/^v//' | sort -V | tail -1)"
if [ -n "${LATEST_TAG}" ]; then if [ -n "${LATEST_TAG}" ]; then
CUR="${LATEST_TAG}" CUR="${LATEST_TAG}"
else else
CUR="$(grep -m1 '^version' Cargo.toml | sed -E 's/.*"(.*)".*/\1/')" CUR="$(grep -m1 '^version' breadcast/Cargo.toml | sed -E 's/.*"(.*)".*/\1/')"
fi fi
IFS='.' read -r MA MI PA <<< "${CUR}" IFS='.' read -r MA MI PA <<< "${CUR}"
SHA="$(git rev-parse --short HEAD)" SHA="$(git rev-parse --short HEAD)"

View file

@ -4,7 +4,7 @@ name: check
# main and triggers a dev-track release build. # main and triggers a dev-track release build.
on: on:
push: push:
branches: ['main', 'feature/**', 'fix/**'] branches: ['feature/**', 'fix/**']
jobs: jobs:
check: check:

View file

@ -1,11 +1,12 @@
name: dev release name: dev release
# Publishes a dev-track build on every push to `main` (single-trunk). # Publishes a dev-track build on every push to `dev` —
# The leftover `dev` trigger is kept so an old branch push still works # separate from release.yml's tag-triggered stable releases. See
# until that branch is deleted. See CONTRIBUTING.md. # bread-ecosystem's docs/release-channels.md for the three-track policy
# this is part of.
on: on:
push: push:
branches: ['main', 'dev'] branches: ['dev']
jobs: jobs:
build: build:
@ -15,7 +16,7 @@ jobs:
run: | run: |
set -euo pipefail set -euo pipefail
rm -rf src && mkdir src rm -rf src && mkdir src
git clone --branch "${GITHUB_REF_NAME}" --depth 1 \ git clone --branch dev --depth 1 \
"https://git.breadway.dev/${GITHUB_REPOSITORY}.git" src "https://git.breadway.dev/${GITHUB_REPOSITORY}.git" src
- name: build - name: build
@ -36,7 +37,7 @@ jobs:
if [ -n "${LATEST_TAG}" ]; then if [ -n "${LATEST_TAG}" ]; then
CUR="${LATEST_TAG}" CUR="${LATEST_TAG}"
else else
CUR="$(grep -m1 '^version' Cargo.toml | sed -E 's/.*"(.*)".*/\1/')" CUR="$(grep -m1 '^version' breadcast/Cargo.toml | sed -E 's/.*"(.*)".*/\1/')"
fi fi
IFS='.' read -r MA MI PA <<< "${CUR}" IFS='.' read -r MA MI PA <<< "${CUR}"
SHA="$(git rev-parse --short HEAD)" SHA="$(git rev-parse --short HEAD)"

View file

@ -6,9 +6,6 @@ on:
jobs: jobs:
build: build:
# RC tags are `v*` too (`v0.1.1-rc.1`) — those belong on the beta
# track, see beta-release.yml. A stable cut is a plain `vX.Y.Z`.
if: ${{ !contains(github.ref_name, '-rc') }}
runs-on: [self-hosted, hestia] runs-on: [self-hosted, hestia]
steps: steps:
- name: checkout - name: checkout

3
.gitignore vendored
View file

@ -35,6 +35,3 @@ CLAUDE.md
# graphify knowledge-graph output (local tool cache, not for commit) # graphify knowledge-graph output (local tool cache, not for commit)
graphify-out/ graphify-out/
# .freebuff local tool state (not for commit)
.freebuff/

View file

@ -20,13 +20,15 @@ out of sync with `dev`/`beta` across most repos in this ecosystem.
- `github` — GitHub mirror. Push both when publishing. - `github` — GitHub mirror. Push both when publishing.
## CI ## CI
- `release.yml` triggers on `push: tags: ['v*']` but skips any tag whose - `release.yml` triggers on `push: tags: ['v*']` — tag a release to cut
name contains `-rc` — a plain `vX.Y.Z` cuts the signed stable build. the signed stable build.
- `beta-release.yml` publishes a beta-track build from a `vX.Y.Z-rc.N` - Leftover: `dev-release.yml` still triggers on `push: branches: ['dev']`
tag (and still on leftover `beta` so an old branch push does not go silent). and `beta-release.yml` still triggers on `push: branches: ['beta']`.
- `dev-release.yml` publishes a dev-track build on push to `main` (and Those files are leftover from the three-branch model. Do **not** rewrite
still on leftover `dev` so an old branch push does not go silent). them unless you are deliberately migrating CI; document and follow
- `check.yml` runs clippy + test on `main`, `feature/**`, and `fix/**`. single-trunk `main` + RC tags instead.
- `check.yml` runs clippy + test on `feature/**` and `fix/**`.
- No separate lint/PR-check pipeline on ordinary commits to `main`.
## Product cut ## Product cut
breadcast is a **bakery product**, not shipped on the BOS ISO, and not part breadcast is a **bakery product**, not shipped on the BOS ISO, and not part
@ -39,9 +41,9 @@ The daemon + GTK picker + Cast Streaming / DLNA pipelines are built and
validated against real hardware. `EVENTS.md` is the bus contract. validated against real hardware. `EVENTS.md` is the bus contract.
## Pins ## Pins
`bread-theme` pins `git.breadway.dev/Breadway/bread-ecosystem` tag `bread-theme` and `bread-utils` pin
`v0.7.4` (per-monitor palette). `bread-utils` pins the same repo at `git.breadway.dev/Breadway/bread-ecosystem` tag `v0.7.1`. Do not switch
`v0.7.2`. Do not switch those to github.com or `branch = "main"`. those to github.com or `branch = "main"`.
## Don't ## Don't
- Don't embed credentials in remote URLs — SSH or a credential helper only. - Don't embed credentials in remote URLs — SSH or a credential helper only.

228
Cargo.lock generated
View file

@ -4,9 +4,9 @@ version = 4
[[package]] [[package]]
name = "aho-corasick" name = "aho-corasick"
version = "1.1.5" version = "1.1.4"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c982642fa9e8606056828ee9a8505737230110bb1099153c79efe865c59d12ba" checksum = "ddd31a130427c27518df266943a5308ed92d4b226cc639f5a8f1002816174301"
dependencies = [ dependencies = [
"memchr", "memchr",
] ]
@ -63,9 +63,9 @@ dependencies = [
[[package]] [[package]]
name = "async-trait" name = "async-trait"
version = "0.1.92" version = "0.1.91"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "82f6aeea286b8eb4dd3431a1be1b59d290ace00f5bfd8e2a159bc2a05e2c1667" checksum = "ae36dc4177970ef04fde5178d3e2429882def40e57a451f919c098f72baa6cec"
dependencies = [ dependencies = [
"proc-macro2", "proc-macro2",
"quote", "quote",
@ -92,9 +92,9 @@ checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53"
[[package]] [[package]]
name = "aws-lc-rs" name = "aws-lc-rs"
version = "1.18.0" version = "1.17.3"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ce2b2dcc879c3bae0d371e77c99f2238400ef24ec001394befa67b6e543add9e" checksum = "00bdb5da18dac48ca2cc7cd4a98e533e8635a58e2361d13a1a4ee3888e0d72f1"
dependencies = [ dependencies = [
"aws-lc-sys", "aws-lc-sys",
"zeroize", "zeroize",
@ -102,9 +102,9 @@ dependencies = [
[[package]] [[package]]
name = "aws-lc-sys" name = "aws-lc-sys"
version = "0.44.0" version = "0.43.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f09fae7be8bb3174e05c6afdb34199e6dc0c7c04ba9fa237b1967adfbde27483" checksum = "43103168cc76fe62678a375e722fc9cb3a0146159ac5828bc4f0dfd755c2224c"
dependencies = [ dependencies = [
"cc", "cc",
"cmake", "cmake",
@ -132,8 +132,8 @@ dependencies = [
[[package]] [[package]]
name = "bread-theme" name = "bread-theme"
version = "0.7.4" version = "0.7.2"
source = "git+https://git.breadway.dev/Breadway/bread-ecosystem?tag=v0.7.4#fcba3760387e2523edb71350f8efea3bc851b21e" source = "git+https://git.breadway.dev/Breadway/bread-ecosystem?tag=v0.7.2#30517f161724132cdeb658c04cf5e490be07ee73"
dependencies = [ dependencies = [
"dirs", "dirs",
"gtk4", "gtk4",
@ -187,7 +187,6 @@ dependencies = [
"gstreamer", "gstreamer",
"gstreamer-app", "gstreamer-app",
"gstreamer-video", "gstreamer-video",
"if-addrs 0.15.0",
"mdns-sd", "mdns-sd",
"rupnp", "rupnp",
"rust_cast", "rust_cast",
@ -254,14 +253,14 @@ checksum = "f8b4985713047f5faee02b8db6a6ef32bbb50269ff53c1aee716d1d195b76d54"
dependencies = [ dependencies = [
"glib-sys", "glib-sys",
"libc", "libc",
"system-deps 7.0.8", "system-deps",
] ]
[[package]] [[package]]
name = "cc" name = "cc"
version = "1.4.3" version = "1.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "509591b7bcd67f4ef775afad7662703b4935daaa6ec0e5605cfb1090b32a2b6d" checksum = "5add81bb678e6cb321aff7fa0dc7689ad82b112dbc032cea19f91d6b8e3582b9"
dependencies = [ dependencies = [
"find-msvc-tools", "find-msvc-tools",
"jobserver", "jobserver",
@ -430,9 +429,9 @@ dependencies = [
[[package]] [[package]]
name = "find-msvc-tools" name = "find-msvc-tools"
version = "0.1.11" version = "0.1.9"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d45db016d36b838f563236e9193d0ee6ce38f3f68b6c94e914b4929c96bbb890" checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582"
[[package]] [[package]]
name = "flume" name = "flume"
@ -453,24 +452,24 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
[[package]] [[package]]
name = "futures-channel" name = "futures-channel"
version = "0.3.34" version = "0.3.33"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b1f9e3d69d39e4862ffed03ed071a76f9a13ba1d9109d355b0f0aa6b15e393c4" checksum = "262590f4fe6afeb0bc83be1daa64e52657fe185690a958af7f3ad0e92085c5ae"
dependencies = [ dependencies = [
"futures-core", "futures-core",
] ]
[[package]] [[package]]
name = "futures-core" name = "futures-core"
version = "0.3.34" version = "0.3.33"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" checksum = "2cd50c473c80f6d7c3670a752354b8e569b1a7cbfdc0419ec88e5edad85e0dc7"
[[package]] [[package]]
name = "futures-executor" name = "futures-executor"
version = "0.3.34" version = "0.3.33"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "031b47cf1a3c6cc8bc2fc76cd437f521619387907d469316e7c0bc278f1f5432" checksum = "6754879cc9f2c66f88c6e5c35344bb0bdb0708b0352b1201815667c7eabc7458"
dependencies = [ dependencies = [
"futures-core", "futures-core",
"futures-task", "futures-task",
@ -479,9 +478,9 @@ dependencies = [
[[package]] [[package]]
name = "futures-io" name = "futures-io"
version = "0.3.34" version = "0.3.33"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "53c0fa8157de1303bfffdaa1cc2a673bfffb60102f76b0ef4441659124373fed" checksum = "4577ecaa3c4f96589d473f679a71b596316f6641bc350038b962a5daf0085d7a"
[[package]] [[package]]
name = "futures-lite" name = "futures-lite"
@ -498,32 +497,32 @@ dependencies = [
[[package]] [[package]]
name = "futures-macro" name = "futures-macro"
version = "0.3.34" version = "0.3.33"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9fb9654ba8355388abeb8dcb4fc62f511300867002afc858860463bdd9fe0c44" checksum = "2d6d3cde68c518367be28956066ddfef33813991b77a55005a69dae04bf3b10b"
dependencies = [ dependencies = [
"proc-macro2", "proc-macro2",
"quote", "quote",
"syn 3.0.3", "syn 2.0.119",
] ]
[[package]] [[package]]
name = "futures-sink" name = "futures-sink"
version = "0.3.34" version = "0.3.33"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1944426bf7d03f1d14f708785e4b33efd750b36d48a157b836b3efc15ede8e1d" checksum = "e34418ac499d6305c2fb5ad0ed2f6ac998c5f8ca209b4510f7f94242c647e307"
[[package]] [[package]]
name = "futures-task" name = "futures-task"
version = "0.3.34" version = "0.3.33"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cd417de3d1d015fc3bfd2b1ea46dfc7bab72ef86f1cc7cc9c78e728b34a6d1fd" checksum = "b231ed28831efb4a61a08580c4bc233ec56bc009f4cd8f52da2c3cb97df0c109"
[[package]] [[package]]
name = "futures-util" name = "futures-util"
version = "0.3.34" version = "0.3.33"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" checksum = "a77a90a256fce34da66415271e30f94ee91c57b04b8a2c042d9cf3220179deaa"
dependencies = [ dependencies = [
"futures-core", "futures-core",
"futures-macro", "futures-macro",
@ -554,7 +553,7 @@ dependencies = [
"glib-sys", "glib-sys",
"gobject-sys", "gobject-sys",
"libc", "libc",
"system-deps 7.0.8", "system-deps",
] ]
[[package]] [[package]]
@ -587,7 +586,7 @@ dependencies = [
"libc", "libc",
"pango-sys", "pango-sys",
"pkg-config", "pkg-config",
"system-deps 7.0.8", "system-deps",
] ]
[[package]] [[package]]
@ -654,7 +653,7 @@ dependencies = [
"glib-sys", "glib-sys",
"gobject-sys", "gobject-sys",
"libc", "libc",
"system-deps 7.0.8", "system-deps",
"windows-sys 0.61.2", "windows-sys 0.61.2",
] ]
@ -718,7 +717,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "030967459f9f676851872c6304adea7825c6d462ec9b72554c733cf0c5952233" checksum = "030967459f9f676851872c6304adea7825c6d462ec9b72554c733cf0c5952233"
dependencies = [ dependencies = [
"libc", "libc",
"system-deps 7.0.8", "system-deps",
] ]
[[package]] [[package]]
@ -729,7 +728,7 @@ checksum = "22a861859b887a79cf461359c192c97a57d8fb0229dd291232e57aa11f6fa72c"
dependencies = [ dependencies = [
"glib-sys", "glib-sys",
"libc", "libc",
"system-deps 7.0.8", "system-deps",
] ]
[[package]] [[package]]
@ -750,7 +749,7 @@ checksum = "5c7ffdfde88f3570d3705e0d8a2433e036d387a1f2930bbf47eafcb5f569fd04"
dependencies = [ dependencies = [
"glib-sys", "glib-sys",
"libc", "libc",
"system-deps 7.0.8", "system-deps",
] ]
[[package]] [[package]]
@ -781,7 +780,7 @@ dependencies = [
"graphene-sys", "graphene-sys",
"libc", "libc",
"pango-sys", "pango-sys",
"system-deps 7.0.8", "system-deps",
] ]
[[package]] [[package]]
@ -806,7 +805,7 @@ dependencies = [
"pastey", "pastey",
"pin-project-lite", "pin-project-lite",
"smallvec", "smallvec",
"thiserror 2.0.20", "thiserror 2.0.19",
] ]
[[package]] [[package]]
@ -834,7 +833,7 @@ dependencies = [
"gstreamer-base-sys", "gstreamer-base-sys",
"gstreamer-sys", "gstreamer-sys",
"libc", "libc",
"system-deps 7.0.8", "system-deps",
] ]
[[package]] [[package]]
@ -861,7 +860,7 @@ dependencies = [
"gobject-sys", "gobject-sys",
"gstreamer-sys", "gstreamer-sys",
"libc", "libc",
"system-deps 7.0.8", "system-deps",
] ]
[[package]] [[package]]
@ -874,7 +873,7 @@ dependencies = [
"glib-sys", "glib-sys",
"gobject-sys", "gobject-sys",
"libc", "libc",
"system-deps 7.0.8", "system-deps",
] ]
[[package]] [[package]]
@ -890,7 +889,7 @@ dependencies = [
"gstreamer-base", "gstreamer-base",
"gstreamer-video-sys", "gstreamer-video-sys",
"libc", "libc",
"thiserror 2.0.20", "thiserror 2.0.19",
] ]
[[package]] [[package]]
@ -904,7 +903,7 @@ dependencies = [
"gstreamer-base-sys", "gstreamer-base-sys",
"gstreamer-sys", "gstreamer-sys",
"libc", "libc",
"system-deps 7.0.8", "system-deps",
] ]
[[package]] [[package]]
@ -930,9 +929,9 @@ dependencies = [
[[package]] [[package]]
name = "gtk4-layer-shell" name = "gtk4-layer-shell"
version = "0.8.1" version = "0.8.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "17c28ea0f4676fdaaae7ff2413a24d0d35c8657424f84856c1103c73454c9da4" checksum = "a4069987ff4793699511a251028cc336b438e46565b463f111250148d574752a"
dependencies = [ dependencies = [
"bitflags", "bitflags",
"gdk4", "gdk4",
@ -945,15 +944,15 @@ dependencies = [
[[package]] [[package]]
name = "gtk4-layer-shell-sys" name = "gtk4-layer-shell-sys"
version = "0.6.1" version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bcf19bb884ef0ef55b9e6b2b369c39b4fcc0c41e3a0c1cbc8c267720338b690b" checksum = "8f566a5ec5bcc454e7fcf2ab76930887ced5365afce12c1e5201bb296b95f1b9"
dependencies = [ dependencies = [
"gdk4-sys", "gdk4-sys",
"glib-sys", "glib-sys",
"gtk4-sys", "gtk4-sys",
"libc", "libc",
"system-deps 8.0.0", "system-deps",
] ]
[[package]] [[package]]
@ -984,7 +983,7 @@ dependencies = [
"gsk4-sys", "gsk4-sys",
"libc", "libc",
"pango-sys", "pango-sys",
"system-deps 7.0.8", "system-deps",
] ]
[[package]] [[package]]
@ -1036,9 +1035,9 @@ dependencies = [
[[package]] [[package]]
name = "http-body-util" name = "http-body-util"
version = "0.1.5" version = "0.1.4"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "23169fe34a5fbcdd3f3862e78fb9b6fccd5f02a6dc6f732547005d45631ce71c" checksum = "e9f41fd6a08e4d4ec69df65976da761afd5ad5e58a9d4acb46bd1c953a9e3ff2"
dependencies = [ dependencies = [
"bytes", "bytes",
"futures-core", "futures-core",
@ -1156,9 +1155,9 @@ dependencies = [
[[package]] [[package]]
name = "js-sys" name = "js-sys"
version = "0.3.104" version = "0.3.103"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0e0c1080212aad755ea003d18543e8768dd432c48819efd73a7bf1e39b7a5a3a" checksum = "53b44bfcdb3f8d5837a46dae1ca9660a837176eee74a28b229bc626816589102"
dependencies = [ dependencies = [
"cfg-if", "cfg-if",
"futures-util", "futures-util",
@ -1194,9 +1193,9 @@ checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2"
[[package]] [[package]]
name = "libredox" name = "libredox"
version = "0.1.20" version = "0.1.18"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "28d0a00925a9f930d679b6789b721e3a7f9ed110f41b86d2497caa780c3a070a" checksum = "c943259e342f1e06ff2da7a83eabdfe7f92ce10262688dbf1895ff0b3e6e4652"
dependencies = [ dependencies = [
"libc", "libc",
] ]
@ -1287,9 +1286,9 @@ dependencies = [
[[package]] [[package]]
name = "num-integer" name = "num-integer"
version = "0.1.47" version = "0.1.46"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7ce2d95d4b3734dc35aa2f45e1aa22cd416814592a4f9d9205e11affd5b8e10b" checksum = "7969661fd2958a5cb096e56c8e1ad0444ac2bbcd0061bd28660485a44879858f"
dependencies = [ dependencies = [
"num-traits", "num-traits",
] ]
@ -1370,7 +1369,7 @@ dependencies = [
"glib-sys", "glib-sys",
"gobject-sys", "gobject-sys",
"libc", "libc",
"system-deps 7.0.8", "system-deps",
] ]
[[package]] [[package]]
@ -1393,9 +1392,9 @@ checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"
[[package]] [[package]]
name = "pkg-config" name = "pkg-config"
version = "0.3.34" version = "0.3.33"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f6b464fbc74e149a392436b17d523f769e057cb6877f6a5c4618bc6f11800548" checksum = "19f132c84eca552bf34cab8ec81f1c1dcc229b811638f9d283dceabe58c5569e"
[[package]] [[package]]
name = "proc-macro-crate" name = "proc-macro-crate"
@ -1506,9 +1505,9 @@ dependencies = [
[[package]] [[package]]
name = "regex-automata" name = "regex-automata"
version = "0.4.18" version = "0.4.16"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ad8553b9b26413251cbf30e620595c7a41b3887f03da04579c0e6b0d6a06b4b2" checksum = "8fcfdb36bda0c880c5931cdc7a2bcdc8ba4556847b9d912bca70bc94708711ad"
dependencies = [ dependencies = [
"aho-corasick", "aho-corasick",
"memchr", "memchr",
@ -1573,7 +1572,7 @@ dependencies = [
"rustls-native-certs", "rustls-native-certs",
"serde", "serde",
"serde_json", "serde_json",
"thiserror 2.0.20", "thiserror 2.0.19",
] ]
[[package]] [[package]]
@ -1613,9 +1612,9 @@ dependencies = [
[[package]] [[package]]
name = "rustls" name = "rustls"
version = "0.23.43" version = "0.23.42"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0283386ce02abc0151e1761d08802dfe86c173b0b494af5cbc086574e453da06" checksum = "3c54fcab019b409d04215d3a17cb438fd7fbf192ee61461f20f4fe18704bc138"
dependencies = [ dependencies = [
"aws-lc-rs", "aws-lc-rs",
"log", "log",
@ -1649,9 +1648,9 @@ dependencies = [
[[package]] [[package]]
name = "rustls-webpki" name = "rustls-webpki"
version = "0.103.14" version = "0.103.13"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0527518605e68109d875e248ea259b6758801cf165e4b2c2733ae3b51f12535a" checksum = "61c429a8649f110dddef65e2a5ad240f747e85f7758a6bccc7e5777bd33f756e"
dependencies = [ dependencies = [
"aws-lc-rs", "aws-lc-rs",
"ring", "ring",
@ -1907,19 +1906,6 @@ dependencies = [
"version-compare", "version-compare",
] ]
[[package]]
name = "system-deps"
version = "8.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "83779a5c956bcb6ba627a4ecf0a9d7625db47d7537e0892d97f712ac995648a3"
dependencies = [
"cfg-expr",
"heck",
"pkg-config",
"toml 1.1.4+spec-1.1.0",
"version-compare",
]
[[package]] [[package]]
name = "target-lexicon" name = "target-lexicon"
version = "0.13.5" version = "0.13.5"
@ -1950,11 +1936,11 @@ dependencies = [
[[package]] [[package]]
name = "thiserror" name = "thiserror"
version = "2.0.20" version = "2.0.19"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ec86235f5fcc2a73650310756d2ac5b138a5780bbbdfae3eeccec992c435ba4f" checksum = "09a43598840e33d5b0331f38c5e30d13bb11c11210a4b58f0d9b18a5a5eefcd9"
dependencies = [ dependencies = [
"thiserror-impl 2.0.20", "thiserror-impl 2.0.19",
] ]
[[package]] [[package]]
@ -1970,9 +1956,9 @@ dependencies = [
[[package]] [[package]]
name = "thiserror-impl" name = "thiserror-impl"
version = "2.0.20" version = "2.0.19"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bc04cd3e1236dd4a98afca4569f2deb3f120e5422a4023be2cb683f8486292af" checksum = "43cbfe0cf76104d42a574802844187e84a305e531ed54455f11fbde0f10541cd"
dependencies = [ dependencies = [
"proc-macro2", "proc-macro2",
"quote", "quote",
@ -2019,13 +2005,13 @@ dependencies = [
[[package]] [[package]]
name = "tokio-macros" name = "tokio-macros"
version = "2.7.2" version = "2.7.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" checksum = "6328af13490e73a9b4694030fafd93f8c8c6a9dede33e821c3fc63eddf8042ba"
dependencies = [ dependencies = [
"proc-macro2", "proc-macro2",
"quote", "quote",
"syn 3.0.3", "syn 2.0.119",
] ]
[[package]] [[package]]
@ -2214,9 +2200,9 @@ checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1"
[[package]] [[package]]
name = "uuid" name = "uuid"
version = "1.24.1" version = "1.24.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2cefc03fd367c0c6d4305de1b312cf00248c4114f4a0418ce6a6af769e3b0bd9" checksum = "bf3923a6f5c4c6382e0b653c4117f48d631ea17f38ed86e2a828e6f7412f5239"
dependencies = [ dependencies = [
"js-sys", "js-sys",
"serde_core", "serde_core",
@ -2252,9 +2238,9 @@ checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b"
[[package]] [[package]]
name = "wasm-bindgen" name = "wasm-bindgen"
version = "0.2.127" version = "0.2.126"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1b70935747edd64d89de3efa29d73789b806c15798f8e7dca4d8ac356b50ce70" checksum = "4b067c0c11094aef6b7a801c1e34a26affafdf3d051dba08456b868789aaf9a4"
dependencies = [ dependencies = [
"cfg-if", "cfg-if",
"once_cell", "once_cell",
@ -2265,9 +2251,9 @@ dependencies = [
[[package]] [[package]]
name = "wasm-bindgen-macro" name = "wasm-bindgen-macro"
version = "0.2.127" version = "0.2.126"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "77775f8f3f7217702089053b94958f8f54061a3f663417df76e19cbdcca29bc1" checksum = "167ce5e579f6bcf889c4f7175a8a5a585de84e8ff93976ce393efa5f2837aab1"
dependencies = [ dependencies = [
"quote", "quote",
"wasm-bindgen-macro-support", "wasm-bindgen-macro-support",
@ -2275,9 +2261,9 @@ dependencies = [
[[package]] [[package]]
name = "wasm-bindgen-macro-support" name = "wasm-bindgen-macro-support"
version = "0.2.127" version = "0.2.126"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e11d33f857dc2fb11b8bc75aee111aa9cbeb12cd9f25efd3d4c2a3dd4e235284" checksum = "f3997c7839262f4ef12cf90b818d6340c18e80f263f1a94bf157d0ec4420380e"
dependencies = [ dependencies = [
"bumpalo", "bumpalo",
"proc-macro2", "proc-macro2",
@ -2288,9 +2274,9 @@ dependencies = [
[[package]] [[package]]
name = "wasm-bindgen-shared" name = "wasm-bindgen-shared"
version = "0.2.127" version = "0.2.126"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7ef64dbcc55df09c7e5a46182d181c2cfa3e925f3da937ea764728b4bbb9dcbf" checksum = "dc1b4cb0cc549fcf58d7dfc081778139b3d283a081644e833e84682ad71cea24"
dependencies = [ dependencies = [
"unicode-ident", "unicode-ident",
] ]
@ -2490,15 +2476,15 @@ dependencies = [
[[package]] [[package]]
name = "xml-rs" name = "xml-rs"
version = "0.8.29" version = "0.8.28"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e450f9b2ed1dff33c94c12589a87338689467b9c4f5d8a5710bd09a847d2c8a7" checksum = "3ae8337f8a065cfc972643663ea4279e04e7256de865aa66fe25cec5fb912d3f"
[[package]] [[package]]
name = "zbus" name = "zbus"
version = "5.19.0" version = "5.18.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5db4be7c075cb421e4b7ee645541604239bd243ba7c357511f4ff3a74b555907" checksum = "fe18fb60dc696039e738717b76eaea21e7a4489bbb1885020b43c94236d7e98a"
dependencies = [ dependencies = [
"async-broadcast", "async-broadcast",
"async-recursion", "async-recursion",
@ -2526,14 +2512,14 @@ dependencies = [
[[package]] [[package]]
name = "zbus_macros" name = "zbus_macros"
version = "5.19.0" version = "5.18.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2990635d09ade6df1868f72f8cac69a876a90981e8bd3c40b1be413f8dc88f40" checksum = "fe96480bed92df2b442a1a30df364e12d08eed03aeb061f2b8dc6afb2be91119"
dependencies = [ dependencies = [
"proc-macro-crate", "proc-macro-crate",
"proc-macro2", "proc-macro2",
"quote", "quote",
"syn 3.0.3", "syn 2.0.119",
"zbus_names", "zbus_names",
"zvariant", "zvariant",
"zvariant_utils", "zvariant_utils",
@ -2550,15 +2536,6 @@ dependencies = [
"zvariant", "zvariant",
] ]
[[package]]
name = "zcheapstr"
version = "1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d1afec51604565183aeb5c54c20aeab286120d4e4460f7f76e3e8bb8c0d99473"
dependencies = [
"serde",
]
[[package]] [[package]]
name = "zeroize" name = "zeroize"
version = "1.9.0" version = "1.9.0"
@ -2573,41 +2550,40 @@ checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b"
[[package]] [[package]]
name = "zvariant" name = "zvariant"
version = "5.14.0" version = "5.13.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b5e28c25bd8bb8da5a1f3e7065d0c156b9ee9a7973adf78b0e35eaefdf3b1b5c" checksum = "bee2a0bcd2a907786a456fff45aaaaf54c9ba5f50b71ae9ec1a4edd200c94911"
dependencies = [ dependencies = [
"endi", "endi",
"enumflags2", "enumflags2",
"serde", "serde",
"winnow 1.0.4", "winnow 1.0.4",
"zcheapstr",
"zvariant_derive", "zvariant_derive",
"zvariant_utils", "zvariant_utils",
] ]
[[package]] [[package]]
name = "zvariant_derive" name = "zvariant_derive"
version = "5.14.0" version = "5.13.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d496a145685283b67e232bd9e47377f6b60ad9d51e3601b23867f77c42477f96" checksum = "38a708216a18780796770bfe3f4739c7c83a3e8f789b755534bbbc06e4e23e12"
dependencies = [ dependencies = [
"proc-macro-crate", "proc-macro-crate",
"proc-macro2", "proc-macro2",
"quote", "quote",
"syn 3.0.3", "syn 2.0.119",
"zvariant_utils", "zvariant_utils",
] ]
[[package]] [[package]]
name = "zvariant_utils" name = "zvariant_utils"
version = "4.0.0" version = "3.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "629d80ece222cad20fe0e8741be493c4ab166acf3b85341bdc2cdbcfd8f3c2d6" checksum = "90cb9383f9b45290407a1258b202d3f8f01db719eb60b4e4055c6375af4fc7c7"
dependencies = [ dependencies = [
"proc-macro2", "proc-macro2",
"quote", "quote",
"serde", "serde",
"syn 3.0.3", "syn 2.0.119",
"winnow 1.0.4", "winnow 1.0.4",
] ]

View file

@ -16,7 +16,7 @@ discovery-driven emit half.
| Event | Data | When | | Event | Data | When |
|-------|------|------| |-------|------|------|
| `bread.cast.device_found` | `{ "id": "<mdns id or DLNA description URL>", "name": "<friendly name>", "model": "<model>", "protocol": "cast" \| "dlna" }` | A Chromecast/Google TV (mDNS) or DLNA/UPnP media renderer (SSDP) is discovered, or an already-known device changes (new host, new name). Re-resolutions that carry the same payload are **not** re-emitted — treat it as an upsert keyed by `id`, not an append-only log. `id` is protocol-specific and only unique *within* a protocol — a Cast device's mDNS id and a DLNA device's description URL share no namespace. | | `bread.cast.device_found` | `{ "id": "<mdns id or DLNA description URL>", "name": "<friendly name>", "model": "<model>", "protocol": "cast" \| "dlna" }` | A Chromecast/Google TV (mDNS) or DLNA/UPnP media renderer (SSDP) is discovered (or re-resolved) on the LAN. Fires on every re-resolution, not just the first sighting — treat it as an upsert keyed by `id`, not an append-only log. `id` is protocol-specific and only unique *within* a protocol — a Cast device's mDNS id and a DLNA device's description URL share no namespace. |
| `bread.cast.mirroring_started` | `{ "device_id": "<id>", "device_name": "<name>", "protocol": "cast" \| "dlna" }` | A mirroring session successfully started, whether triggered by `bread.command.cast.start`, the `breadcast` GTK popup, or (once wired) any other IPC client. | | `bread.cast.mirroring_started` | `{ "device_id": "<id>", "device_name": "<name>", "protocol": "cast" \| "dlna" }` | A mirroring session successfully started, whether triggered by `bread.command.cast.start`, the `breadcast` GTK popup, or (once wired) any other IPC client. |
| `bread.cast.mirroring_stopped` | `{}` | A mirroring session ended, whether via an explicit stop (command, IPC, or GTK popup) or unprompted (the portal picker's "stop sharing", the receiver dropping the connection, a DLNA renderer stopping playback from its own remote). There is no separate "stopped by whom" distinction in this event — `breadcastd`'s own logs have that detail if needed. | | `bread.cast.mirroring_stopped` | `{}` | A mirroring session ended, whether via an explicit stop (command, IPC, or GTK popup) or unprompted (the portal picker's "stop sharing", the receiver dropping the connection, a DLNA renderer stopping playback from its own remote). There is no separate "stopped by whom" distinction in this event — `breadcastd`'s own logs have that detail if needed. |
| `bread.cast.mirroring_failed` | `{ "device_id": "<id>", "error": "<message>" }` | A `start_cast`/`bread.command.cast.start` attempt failed before a session was established (device unreachable, portal capture denied, negotiation timeout, renderer rejected the stream, etc). | | `bread.cast.mirroring_failed` | `{ "device_id": "<id>", "error": "<message>" }` | A `start_cast`/`bread.command.cast.start` attempt failed before a session was established (device unreachable, portal capture denied, negotiation timeout, renderer rejected the stream, etc). |

View file

@ -1,9 +1,7 @@
name = "breadcast" name = "breadcast"
description = "Cast your screen to any Chromecast/Google TV or DLNA renderer — daemon + GTK4 popup" description = "Cast your screen to any Chromecast/Google TV or DLNA renderer — daemon + GTK4 popup"
binaries = ["breadcast", "breadcastd"] binaries = ["breadcast", "breadcastd"]
# gst-plugins-base: videoconvert/videoscale/videorate/appsink -- the # gst-plugin-va: the `vah264enc` element both encode pipelines use (see
# `gstreamer` package is only the core library. gst-plugin-va: the
# `vah264enc` element both encode pipelines use (see
# breadcast-core/src/pipeline/mod.rs) -- a separate Arch package from # breadcast-core/src/pipeline/mod.rs) -- a separate Arch package from
# gst-plugins-bad itself, not bundled into it. jsoncpp/openssl: runtime # gst-plugins-bad itself, not bundled into it. jsoncpp/openssl: runtime
# shared-library deps of breadcastd itself (not just a build dep of # shared-library deps of breadcastd itself (not just a build dep of
@ -16,7 +14,6 @@ system_deps = [
"gtk4", "gtk4",
"gtk4-layer-shell", "gtk4-layer-shell",
"gstreamer", "gstreamer",
"gst-plugins-base",
"gst-plugin-pipewire", "gst-plugin-pipewire",
"gst-plugins-bad", "gst-plugins-bad",
"gst-plugin-hlssink3", "gst-plugin-hlssink3",

View file

@ -209,20 +209,9 @@ CastStreamSender* breadcast_caststream_sender_create(
// than once process-wide only if ShutDown() was called first -- breadcast // than once process-wide only if ShutDown() was called first -- breadcast
// only ever has one active cast-streaming session at a time, so this // only ever has one active cast-streaming session at a time, so this
// assumption (baked into PlatformClientPosix's own singleton design) holds. // assumption (baked into PlatformClientPosix's own singleton design) holds.
//
// Create() itself OSP_CHECKs that no instance exists and aborts the
// process if a previous sender leaked (a failed start that never reached
// destroy()). Fail the new create instead of taking the daemon down.
if (openscreen::PlatformClientPosix::GetInstance() != nullptr) {
return nullptr;
}
openscreen::PlatformClientPosix::Create(std::chrono::milliseconds(50)); openscreen::PlatformClientPosix::Create(std::chrono::milliseconds(50));
openscreen::PlatformClientPosix* instance = openscreen::TaskRunner& task_runner =
openscreen::PlatformClientPosix::GetInstance(); openscreen::PlatformClientPosix::GetInstance()->GetTaskRunner();
if (instance == nullptr) {
return nullptr;
}
openscreen::TaskRunner& task_runner = instance->GetTaskRunner();
breadcast_caststream::VideoParams params; breadcast_caststream::VideoParams params;
params.width = width; params.width = width;
@ -266,13 +255,6 @@ CastStreamSender* breadcast_caststream_sender_create(
} }
void breadcast_caststream_sender_negotiate(CastStreamSender* sender) { void breadcast_caststream_sender_negotiate(CastStreamSender* sender) {
// environment is reset during destroy() on the TaskRunner thread; a call
// that races teardown (or arrives after a failed start) must not
// dereference the unique_ptr.
if (!sender || sender->shutting_down.load(std::memory_order_acquire) ||
!sender->environment) {
return;
}
sender->environment->task_runner().PostTask([sender] { sender->environment->task_runner().PostTask([sender] {
if (sender->shutting_down.load(std::memory_order_acquire) || if (sender->shutting_down.load(std::memory_order_acquire) ||
!sender->session) { !sender->session) {
@ -290,10 +272,6 @@ void breadcast_caststream_sender_on_message(CastStreamSender* sender,
size_t message_namespace_len, size_t message_namespace_len,
const char* message, const char* message,
size_t message_len) { size_t message_len) {
if (!sender || sender->shutting_down.load(std::memory_order_acquire) ||
!sender->environment) {
return;
}
auto source = std::make_shared<std::string>(source_id, source_id_len); auto source = std::make_shared<std::string>(source_id, source_id_len);
auto ns = std::make_shared<std::string>(message_namespace, message_namespace_len); auto ns = std::make_shared<std::string>(message_namespace, message_namespace_len);
auto body = std::make_shared<std::string>(message, message_len); auto body = std::make_shared<std::string>(message, message_len);
@ -311,10 +289,6 @@ int32_t breadcast_caststream_sender_enqueue_frame(CastStreamSender* sender,
size_t data_len, size_t data_len,
int32_t is_key_frame, int32_t is_key_frame,
int64_t capture_time_us) { int64_t capture_time_us) {
if (!sender || sender->shutting_down.load(std::memory_order_acquire) ||
!sender->environment) {
return -1;
}
if (!sender->negotiated.load(std::memory_order_acquire)) { if (!sender->negotiated.load(std::memory_order_acquire)) {
return -1; return -1;
} }
@ -404,12 +378,6 @@ int32_t breadcast_caststream_sender_enqueue_frame(CastStreamSender* sender,
switch (video_sender->EnqueueFrame(frame)) { switch (video_sender->EnqueueFrame(frame)) {
case Sender::OK: case Sender::OK:
sender->enqueue_ok.fetch_add(1, std::memory_order_relaxed); sender->enqueue_ok.fetch_add(1, std::memory_order_relaxed);
// A key frame that actually landed resyncs the decoder, so the
// "next frame must be an IDR" latch can clear. A rejected key
// frame leaves the flag set (the reject branches below).
if (is_key) {
sender->frame_chain_broken.store(false, std::memory_order_relaxed);
}
break; break;
case Sender::PAYLOAD_TOO_LARGE: case Sender::PAYLOAD_TOO_LARGE:
sender->enqueue_payload_too_large.fetch_add(1, std::memory_order_relaxed); sender->enqueue_payload_too_large.fetch_add(1, std::memory_order_relaxed);
@ -450,24 +418,10 @@ void breadcast_caststream_sender_take_stats(CastStreamSender* sender,
} }
int32_t breadcast_caststream_sender_needs_key_frame(CastStreamSender* sender) { int32_t breadcast_caststream_sender_needs_key_frame(CastStreamSender* sender) {
if (!sender) { return sender->needs_key_frame.load(std::memory_order_relaxed) ? 1 : 0;
return 0;
}
// `frame_chain_broken` is the drop-side latch SchedulePoll must not
// clobber (see the field comment). Load, don't exchange: the encoder
// may take several frames to honour a force-key-unit, and a consuming
// read here would forget the drop if the next poll happened before
// that IDR was actually EnqueueFrame'd (cleared on OK + is_key above).
return (sender->frame_chain_broken.load(std::memory_order_relaxed) ||
sender->needs_key_frame.load(std::memory_order_relaxed))
? 1
: 0;
} }
int32_t breadcast_caststream_sender_estimated_bandwidth_bps(CastStreamSender* sender) { int32_t breadcast_caststream_sender_estimated_bandwidth_bps(CastStreamSender* sender) {
if (!sender) {
return 0;
}
return sender->estimated_bandwidth_bps.load(std::memory_order_relaxed); return sender->estimated_bandwidth_bps.load(std::memory_order_relaxed);
} }
@ -483,10 +437,7 @@ void breadcast_caststream_sender_destroy(CastStreamSender* sender) {
// These must be torn down on the TaskRunner thread (they hold raw // These must be torn down on the TaskRunner thread (they hold raw
// references into it and into `environment`), so hop over there and block // references into it and into `environment`), so hop over there and block
// until it's done before shutting the TaskRunner itself down. A sender // until it's done before shutting the TaskRunner itself down.
// whose Environment never got built (create failed mid-flight) has no
// task runner to hop to.
if (sender->environment) {
std::promise<void> done; std::promise<void> done;
std::future<void> done_future = done.get_future(); std::future<void> done_future = done.get_future();
sender->environment->task_runner().PostTask([sender, &done] { sender->environment->task_runner().PostTask([sender, &done] {
@ -496,11 +447,8 @@ void breadcast_caststream_sender_destroy(CastStreamSender* sender) {
done.set_value(); done.set_value();
}); });
done_future.wait(); done_future.wait();
}
if (openscreen::PlatformClientPosix::GetInstance() != nullptr) {
openscreen::PlatformClientPosix::ShutDown(); openscreen::PlatformClientPosix::ShutDown();
}
delete sender; delete sender;
} }

View file

@ -88,18 +88,15 @@ void breadcast_caststream_sender_on_message(CastStreamSender* sender,
const char* message, const char* message,
size_t message_len); size_t message_len);
// Posts one encoded video access unit (Annex-B H.264) onto openscreen's // Enqueues one encoded video access unit (Annex-B H.264) for sending.
// TaskRunner for sending. `data` is copied before this returns, so the // `data` is copied before this returns, so the caller may reuse/free its
// caller may reuse/free its buffer immediately after. `capture_time_us` is // buffer immediately after. `capture_time_us` is only used to derive the
// only used to derive the RTP timestamp's relative spacing between frames // RTP timestamp's relative spacing between frames (it does not need to be
// (it does not need to be wall-clock-accurate, just monotonically // wall-clock-accurate, just monotonically increasing and proportional to
// increasing and proportional to real elapsed time between frames). // real elapsed time between frames). Returns 0 if queued, nonzero if the
// // session isn't negotiated yet or the frame was rejected (e.g. too large,
// Returns 0 if the frame was *posted* (not necessarily accepted -- // or the in-flight queue is full -- the caller should back off encoding
// Sender::EnqueueFrame runs later on the TaskRunner and may still reject // when this happens rather than treating it as fatal).
// it), or nonzero if the session isn't negotiated yet / is shutting down.
// Accept/reject outcomes are visible only via take_stats() and, for
// dropped frames that break the H.264 reference chain, needs_key_frame().
int32_t breadcast_caststream_sender_enqueue_frame(CastStreamSender* sender, int32_t breadcast_caststream_sender_enqueue_frame(CastStreamSender* sender,
const uint8_t* data, const uint8_t* data,
size_t data_len, size_t data_len,

View file

@ -1,8 +1,8 @@
//! Raw FFI bindings to `src/facade.h`/`src/facade.cc`, which wrap a pruned, //! Raw FFI bindings to `src/facade.h`/`src/facade.cc`, which wrap a pruned,
//! vendored subset of `chromium/openscreen`'s Cast Streaming sender (see //! vendored subset of `chromium/openscreen`'s Cast Streaming sender (see
//! `vendor/openscreen/PATCHES.md`). This crate is intentionally low-level and //! `vendor/openscreen/PATCHES.md`). This crate is intentionally low-level and
//! unsafe -- see `breadcast-core::caststream` for the ergonomic, thread-safe //! unsafe -- see `breadcast-caststream` (not this crate) for the ergonomic,
//! wrapper most callers should use instead. //! thread-safe wrapper most callers should use instead.
//! //!
//! # Threading contract //! # Threading contract
//! //!

View file

@ -21,10 +21,6 @@ gstreamer-video = "0.25"
tiny_http = "0.12" tiny_http = "0.12"
rupnp = "3.0.0" rupnp = "3.0.0"
futures-util = "0.3.33" futures-util = "0.3.33"
# Fallback LAN-IP enumeration used only when the `ip` command is unavailable
# (net.rs) -- already a transitive dep via mdns-sd, so this adds no new crate
# to the build.
if-addrs = "0.15"
breadcast-caststream-sys = { path = "../breadcast-caststream-sys" } breadcast-caststream-sys = { path = "../breadcast-caststream-sys" }
[dev-dependencies] [dev-dependencies]

View file

@ -117,10 +117,6 @@ impl CaptureSession {
.open(&token_path) .open(&token_path)
{ {
use std::io::Write; use std::io::Write;
use std::os::unix::fs::PermissionsExt;
// `mode()` only applies on create. A pre-existing
// world-readable token would keep its mode otherwise.
let _ = file.set_permissions(std::fs::Permissions::from_mode(0o600));
let _ = file.write_all(token.as_bytes()); let _ = file.write_all(token.as_bytes());
} }
} }

View file

@ -225,18 +225,16 @@ impl CastStreamSender {
} }
} }
/// Posts one encoded video access unit (Annex-B H.264) onto the C++ /// Enqueues one encoded video access unit (Annex-B H.264) for sending.
/// TaskRunner. `capture_time_us` only needs to be monotonically /// `capture_time_us` only needs to be monotonically increasing and
/// increasing and proportional to real elapsed time between frames -- /// proportional to real elapsed time between frames -- it does not need
/// it does not need to be wall-clock-accurate. /// to be wall-clock-accurate.
/// ///
/// Returns an error only if the session isn't negotiated (or is /// Returns an error if the session isn't negotiated yet or the frame
/// shutting down). A `Ok(())` means the frame was *posted*, not that /// was rejected under backpressure; callers should treat the latter as
/// `Sender::EnqueueFrame` accepted it -- accept/reject is visible via /// a dropped frame, not a fatal condition (see
/// [`Self::enqueue_stats`], and a reject that breaks the H.264 /// [`Self::needs_key_frame`]/[`Self::estimated_bandwidth_bps`] for how
/// reference chain latches [`Self::needs_key_frame`]. Treating this /// to react).
/// return as accept/reject is how an earlier version hid a frozen
/// picture behind a healthy "30fps enqueued" log line.
pub fn enqueue_frame(&self, data: &[u8], is_key_frame: bool, capture_time_us: i64) -> Result<()> { pub fn enqueue_frame(&self, data: &[u8], is_key_frame: bool, capture_time_us: i64) -> Result<()> {
let result = unsafe { let result = unsafe {
breadcast_caststream_sender_enqueue_frame( breadcast_caststream_sender_enqueue_frame(
@ -248,7 +246,7 @@ impl CastStreamSender {
) )
}; };
if result != 0 { if result != 0 {
bail!("frame not posted (session not negotiated or shutting down)"); bail!("frame not enqueued (session not negotiated yet)");
} }
Ok(()) Ok(())
} }

View file

@ -29,7 +29,7 @@ const ADDRESS_STALE_AFTER: Duration = Duration::from_secs(300);
/// to two IPv4 addresses (e.g. wired + wireless, or a DHCP lease change), /// to two IPv4 addresses (e.g. wired + wireless, or a DHCP lease change),
/// an unordered choice can silently pick a stale/unreachable one, and pick /// an unordered choice can silently pick a stale/unreachable one, and pick
/// a *different* one across otherwise-identical runs. /// a *different* one across otherwise-identical runs.
fn pick_address(addresses: &[(ScopedIp, Instant)]) -> Option<String> { fn pick_address(addresses: &[(ScopedIp, Instant)]) -> Option<IpAddr> {
let now = Instant::now(); let now = Instant::now();
let mut candidates: Vec<&(ScopedIp, Instant)> = addresses let mut candidates: Vec<&(ScopedIp, Instant)> = addresses
.iter() .iter()
@ -50,10 +50,7 @@ fn pick_address(addresses: &[(ScopedIp, Instant)]) -> Option<String> {
}) })
}) })
.or_else(|| candidates.first()) .or_else(|| candidates.first())
// Prefer ScopedIp's Display so a last-resort link-local IPv6 .map(|(ip, _)| ip.to_ip_addr())
// keeps its `%iface` zone -- `IpAddr` drops it and
// `TcpStream::connect("fe80::…")` then fails with EINVAL.
.map(|(ip, _)| ip.to_string())
} }
const SERVICE_TYPE: &str = "_googlecast._tcp.local."; const SERVICE_TYPE: &str = "_googlecast._tcp.local.";
@ -143,21 +140,10 @@ impl Discovery {
} }
} }
let Some(host) = pick_address(addrs) else { let Some(host) = pick_address(addrs).map(|ip| ip.to_string()) else {
warn!(%id, "cast device resolved with no usable addresses, skipping"); warn!(%id, "cast device resolved with no usable addresses, skipping");
continue; continue;
}; };
// First ServiceResolved is often IPv6-only (or a
// zoneless fe80::). rust_cast / Cast Streaming
// cannot connect to that. Wait for a later
// resolve that carries IPv4 or a scoped address.
if host.parse::<std::net::Ipv6Addr>().is_ok()
&& host.starts_with("fe80:")
&& !host.contains('%')
{
warn!(%id, %host, "cast device resolved to an unscoped link-local IPv6, waiting for a better address");
continue;
}
fullname_to_id.insert(info.get_fullname().to_string(), id.clone()); fullname_to_id.insert(info.get_fullname().to_string(), id.clone());

View file

@ -6,7 +6,7 @@ use tokio::sync::mpsc;
use tracing::{debug, warn}; use tracing::{debug, warn};
use super::device::DlnaDevice; use super::device::DlnaDevice;
use super::{AV_TRANSPORT, MEDIA_RENDERER}; use super::AV_TRANSPORT;
/// How often to re-issue an SSDP search burst. Unlike mDNS (continuous /// How often to re-issue an SSDP search burst. Unlike mDNS (continuous
/// multicast browsing via `mdns-sd` in [`crate::discovery`]), SSDP has no /// multicast browsing via `mdns-sd` in [`crate::discovery`]), SSDP has no
@ -69,7 +69,7 @@ impl DlnaDiscovery {
let mut known: HashMap<String, (DlnaDevice, u32)> = HashMap::new(); let mut known: HashMap<String, (DlnaDevice, u32)> = HashMap::new();
loop { loop {
let search_target = SearchTarget::URN(MEDIA_RENDERER); let search_target = SearchTarget::URN(AV_TRANSPORT);
match rupnp::discover(&search_target, SEARCH_TIMEOUT, None).await { match rupnp::discover(&search_target, SEARCH_TIMEOUT, None).await {
Ok(stream) => { Ok(stream) => {
use futures_util::StreamExt; use futures_util::StreamExt;
@ -86,10 +86,6 @@ impl DlnaDiscovery {
}; };
let url = device.url().to_string(); let url = device.url().to_string();
if device.find_service(&AV_TRANSPORT).is_none() {
debug!(%url, "UPnP device has no AVTransport, skipping");
continue;
}
confirmed.insert(url.clone()); confirmed.insert(url.clone());
if let Some(entry) = known.get_mut(&url) { if let Some(entry) = known.get_mut(&url) {

View file

@ -21,7 +21,3 @@ pub use discovery::{DlnaDiscovery, DlnaDiscoveryEvent};
pub use session::DlnaSession; pub use session::DlnaSession;
const AV_TRANSPORT: URN = URN::service("schemas-upnp-org", "AVTransport", 1); const AV_TRANSPORT: URN = URN::service("schemas-upnp-org", "AVTransport", 1);
/// Device-type search. Many TVs answer M-SEARCH for MediaRenderer but not
/// for the AVTransport *service* URN (Windows "Cast to Device" searches
/// this). After resolve we still require AVTransport.
const MEDIA_RENDERER: URN = URN::device("schemas-upnp-org", "MediaRenderer", 1);

View file

@ -1,6 +1,5 @@
use anyhow::{Context, Result}; use anyhow::{Context, Result};
use rupnp::Service; use rupnp::Service;
use std::time::Duration;
use super::device::DlnaDevice; use super::device::DlnaDevice;
use super::AV_TRANSPORT; use super::AV_TRANSPORT;
@ -57,50 +56,18 @@ impl DlnaSession {
/// renderer would reject with no useful diagnostic. /// renderer would reject with no useful diagnostic.
pub async fn load(&self, content_url: &str) -> Result<()> { pub async fn load(&self, content_url: &str) -> Result<()> {
let escaped = xml_escape(content_url); let escaped = xml_escape(content_url);
// Many Samsung/LG renderers reject an empty CurrentURIMetaData and
// need DIDL-Lite + protocolInfo before they will play a live HLS
// playlist. The DIDL itself is then XML-escaped for the SOAP body.
let didl = format!(
r#"<DIDL-Lite xmlns="urn:schemas-upnp-org:metadata-1-0/DIDL-Lite/" xmlns:dc="http://purl.org/dc/elements/1.1/" xmlns:upnp="urn:schemas-upnp-org:metadata-1-0/upnp/"><item id="0" parentID="-1" restricted="1"><dc:title>breadcast</dc:title><upnp:class>object.item.videoItem</upnp:class><res protocolInfo="http-get:*:application/vnd.apple.mpegurl:*">{escaped}</res></item></DIDL-Lite>"#
);
let metadata = xml_escape(&didl);
let set_uri_payload = format!( let set_uri_payload = format!(
"<InstanceID>0</InstanceID><CurrentURI>{escaped}</CurrentURI><CurrentURIMetaData>{metadata}</CurrentURIMetaData>" "<InstanceID>0</InstanceID><CurrentURI>{escaped}</CurrentURI><CurrentURIMetaData></CurrentURIMetaData>"
); );
self.service self.service
.action(&self.device_url, "SetAVTransportURI", &set_uri_payload) .action(&self.device_url, "SetAVTransportURI", &set_uri_payload)
.await .await
.context("SetAVTransportURI failed")?; .context("SetAVTransportURI failed")?;
// Some renderers auto-play after SetURI and then reject Play; self.service
// others stay TRANSITIONING for a moment. Either PLAYING state is
// success.
match self
.service
.action(&self.device_url, "Play", "<InstanceID>0</InstanceID><Speed>1</Speed>") .action(&self.device_url, "Play", "<InstanceID>0</InstanceID><Speed>1</Speed>")
.await .await
{ .context("Play failed")?;
Ok(_) => {}
Err(e) => {
// The renderer may already be auto-playing after SetURI (and
// reject a redundant Play), or may still be TRANSITIONING and
// only reach PLAYING a moment later -- either is success.
// A single immediate GetTransportInfo isn't enough for the
// latter, so give it a short bounded window before giving up
// on the Play error.
let mut attempts = 0u32;
loop {
if self.transport_state().await.ok().as_deref() == Some("PLAYING") {
break;
}
attempts += 1;
if attempts >= 5 {
return Err(e).context("Play failed");
}
tokio::time::sleep(Duration::from_millis(200)).await;
}
}
}
Ok(()) Ok(())
} }
@ -139,7 +106,7 @@ impl DlnaSession {
} }
} }
pub(crate) fn xml_escape(input: &str) -> String { fn xml_escape(input: &str) -> String {
let mut escaped = String::with_capacity(input.len()); let mut escaped = String::with_capacity(input.len());
for c in input.chars() { for c in input.chars() {
match c { match c {
@ -153,14 +120,3 @@ pub(crate) fn xml_escape(input: &str) -> String {
} }
escaped escaped
} }
#[cfg(test)]
mod tests {
use super::xml_escape;
#[test]
fn xml_escape_covers_the_five_markup_chars() {
assert_eq!(xml_escape(r#"a&b<c>d"e'f"#), "a&amp;b&lt;c&gt;d&quot;e&apos;f");
assert_eq!(xml_escape("http://10.0.0.1:1/t/playlist.m3u8"), "http://10.0.0.1:1/t/playlist.m3u8");
}
}

View file

@ -16,9 +16,8 @@ const WORKER_THREADS: usize = 8;
/// Serves `root` (the `hlssink3` output directory: `playlist.m3u8` + /// Serves `root` (the `hlssink3` output directory: `playlist.m3u8` +
/// `segment*.ts`) over plain HTTP. Runs a small fixed pool of worker /// `segment*.ts`) over plain HTTP. Runs a small fixed pool of worker
/// threads. [`HttpServer::shutdown`] (also invoked from `Drop`) unblocks /// threads; dropping the handle does not stop the server (there is no clean
/// those workers and closes the listener so a finished DLNA session does /// shutdown yet — matches the smoke-testing scope of the rest of Phase 2).
/// not keep the last screen-recording segments reachable on the LAN.
/// ///
/// Every servable path is namespaced under a random token /// Every servable path is namespaced under a random token
/// (`/<token>/playlist.m3u8`, etc. — see [`HttpServer::token`]) rather than /// (`/<token>/playlist.m3u8`, etc. — see [`HttpServer::token`]) rather than
@ -38,7 +37,6 @@ const WORKER_THREADS: usize = 8;
pub struct HttpServer { pub struct HttpServer {
addr: SocketAddr, addr: SocketAddr,
token: String, token: String,
server: Option<Arc<tiny_http::Server>>,
} }
impl HttpServer { impl HttpServer {
@ -76,19 +74,7 @@ impl HttpServer {
}); });
} }
Ok(Self { addr, token, server: Some(server) }) Ok(Self { addr, token })
}
/// Unblocks every worker and drops the listener. After this returns the
/// bind address is free and the path token no longer serves anything.
/// Safe to call more than once.
pub fn shutdown(&mut self) {
let Some(server) = self.server.take() else { return };
// `unblock` wakes one `recv()` at a time; wake every worker so
// they all observe the error and drop their Arc.
for _ in 0..WORKER_THREADS {
server.unblock();
}
} }
/// The bound address, e.g. `0.0.0.0:41823`. Combine with this /// The bound address, e.g. `0.0.0.0:41823`. Combine with this
@ -104,17 +90,7 @@ impl HttpServer {
/// doc for why that can't just be `self.addr()`), including the /// doc for why that can't just be `self.addr()`), including the
/// unguessable path token every request must carry. /// unguessable path token every request must carry.
pub fn url(&self, host: std::net::IpAddr, relative: &str) -> String { pub fn url(&self, host: std::net::IpAddr, relative: &str) -> String {
let port = self.addr.port(); format!("http://{host}:{}/{}/{relative}", self.addr.port(), self.token)
match host {
std::net::IpAddr::V6(v6) => format!("http://[{v6}]:{port}/{}/{relative}", self.token),
std::net::IpAddr::V4(v4) => format!("http://{v4}:{port}/{}/{relative}", self.token),
}
}
}
impl Drop for HttpServer {
fn drop(&mut self) {
self.shutdown();
} }
} }
@ -132,30 +108,12 @@ fn random_token() -> String {
.is_ok(); .is_ok();
if !read_ok { if !read_ok {
// Unreachable in practice on Linux, but better than a zero-entropy // Unreachable in practice on Linux, but better than a zero-entropy
// token if it ever happened. Mix several low-cost, not-directly- // token if it ever happened.
// foreseeable sources (two time reads a hair apart, pid, a process
// counter) into both hashes and write all 16 bytes, so the fallback
// never reduces to a single low-resolution clock tick or per-pid
// reuse -- still far weaker than /dev/urandom, which is why it stays
// a fallback, but not trivially guessable.
let mix = |salt: u64| {
use std::hash::{Hash, Hasher}; use std::hash::{Hash, Hasher};
let mut hasher = std::collections::hash_map::DefaultHasher::new(); let mut hasher = std::collections::hash_map::DefaultHasher::new();
std::time::SystemTime::now().hash(&mut hasher); std::time::SystemTime::now().hash(&mut hasher);
std::time::SystemTime::UNIX_EPOCH
.elapsed()
.ok()
.map(|d| d.as_nanos())
.unwrap_or_default()
.hash(&mut hasher);
std::process::id().hash(&mut hasher); std::process::id().hash(&mut hasher);
salt.hash(&mut hasher); bytes[..8].copy_from_slice(&hasher.finish().to_le_bytes());
hasher.finish()
};
let a = mix(0x9e3779b97f4a7c15);
let b = mix(0xdead_beef_dead_beef);
bytes[..8].copy_from_slice(&a.to_le_bytes());
bytes[8..].copy_from_slice(&b.to_le_bytes());
} }
bytes.iter().map(|b| format!("{b:02x}")).collect() bytes.iter().map(|b| format!("{b:02x}")).collect()
} }
@ -165,7 +123,7 @@ fn random_token() -> String {
/// multi-range (multipart ranges aren't needed for HLS segment fetches, and /// multi-range (multipart ranges aren't needed for HLS segment fetches, and
/// falling back to a full 200 response for those is always a valid /// falling back to a full 200 response for those is always a valid
/// response under the HTTP spec). /// response under the HTTP spec).
pub(crate) fn parse_range(value: &str, len: usize) -> Option<(usize, usize)> { fn parse_range(value: &str, len: usize) -> Option<(usize, usize)> {
let spec = value.strip_prefix("bytes=")?; let spec = value.strip_prefix("bytes=")?;
if spec.contains(',') || len == 0 { if spec.contains(',') || len == 0 {
return None; return None;
@ -275,11 +233,7 @@ fn handle_request(request: tiny_http::Request, root: &Path, token: &str) -> Resu
]; ];
let (status, body) = match range { let (status, body) = match range {
// RFC 7233: a last-byte-pos past the end is clamped, not 416. Some((start, end)) if start <= end && end < data.len() => {
// HLS clients often probe `bytes=0-1048575` against a ~400KB
// segment; answering 416 stalls playback with no encoder error.
Some((start, end)) if start < data.len() && start <= end => {
let end = end.min(data.len() - 1);
headers.push(("Content-Range".to_string(), format!("bytes {start}-{end}/{}", data.len()))); headers.push(("Content-Range".to_string(), format!("bytes {start}-{end}/{}", data.len())));
(206u16, data[start..=end].to_vec()) (206u16, data[start..=end].to_vec())
} }
@ -310,25 +264,3 @@ fn respond_status(request: tiny_http::Request, status: u16) -> Result<()> {
.respond(tiny_http::Response::empty(status)) .respond(tiny_http::Response::empty(status))
.context("failed to write HTTP error response") .context("failed to write HTTP error response")
} }
#[cfg(test)]
mod tests {
use super::parse_range;
#[test]
fn parse_range_accepts_the_usual_hls_shapes() {
assert_eq!(parse_range("bytes=0-99", 200), Some((0, 99)));
assert_eq!(parse_range("bytes=50-", 200), Some((50, 199)));
assert_eq!(parse_range("bytes=-20", 200), Some((180, 199)));
assert_eq!(parse_range("bytes=0-0", 200), Some((0, 0)));
}
#[test]
fn parse_range_rejects_malformed_or_multipart() {
assert_eq!(parse_range("bytes=", 200), None);
assert_eq!(parse_range("bytes=-", 200), None);
assert_eq!(parse_range("bytes=0-10,20-30", 200), None);
assert_eq!(parse_range("items=0-10", 200), None);
assert_eq!(parse_range("bytes=0-10", 0), None);
}
}

View file

@ -2,8 +2,6 @@ use std::net::{IpAddr, Ipv4Addr};
use anyhow::{Context, Result}; use anyhow::{Context, Result};
const EXCLUDED_PREFIXES: &[&str] = &["tailscale", "wg", "docker", "veth", "br-", "virbr", "lo"];
/// Finds this machine's LAN-reachable IPv4 address by enumerating network /// Finds this machine's LAN-reachable IPv4 address by enumerating network
/// interfaces directly, rather than the more common "UDP-connect to a /// interfaces directly, rather than the more common "UDP-connect to a
/// public address and read back the local endpoint" trick — that trick /// public address and read back the local endpoint" trick — that trick
@ -20,61 +18,8 @@ const EXCLUDED_PREFIXES: &[&str] = &["tailscale", "wg", "docker", "veth", "br-",
/// Shared by every casting protocol (Cast, DLNA, ...) — they all need to /// Shared by every casting protocol (Cast, DLNA, ...) — they all need to
/// embed this machine's own address in a URL handed to a receiver device. /// embed this machine's own address in a URL handed to a receiver device.
pub fn local_lan_ip() -> Result<IpAddr> { pub fn local_lan_ip() -> Result<IpAddr> {
pick_lan_ip(None) const EXCLUDED_PREFIXES: &[&str] = &["tailscale", "wg", "docker", "veth", "br-", "virbr", "lo"];
}
/// Like [`local_lan_ip`], but prefers the interface whose subnet contains
/// `peer`. Dual-homed machines (ethernet + wifi, two VLANs) otherwise
/// embed the wrong host in the HLS URL and the renderer cannot fetch it —
/// the same silent `LOAD FAILED` class as the Tailscale case above.
pub fn local_lan_ip_for(peer: IpAddr) -> Result<IpAddr> {
pick_lan_ip(Some(peer))
}
fn pick_lan_ip(peer: Option<IpAddr>) -> Result<IpAddr> {
let mut candidates = lan_ipv4_candidates()?;
if let Some(IpAddr::V4(peer_v4)) = peer {
if let Some((_, addr, _)) = candidates
.iter()
.filter(|(_, addr, prefix)| same_subnet(*addr, peer_v4, *prefix))
.max_by_key(|(_, _, prefix)| *prefix)
{
return Ok(IpAddr::V4(*addr));
}
}
// Prefer a conventionally-named physical/Wi-Fi interface when there's a
// choice, but any private, non-excluded address is acceptable.
candidates.sort_by_key(|(iface, _, _)| !(iface.starts_with("wl") || iface.starts_with("en") || iface.starts_with("eth")));
candidates
.into_iter()
.map(|(_, addr, _)| IpAddr::V4(addr))
.next()
.context("no LAN-reachable IPv4 address found (excluding loopback/VPN/virtual interfaces) — is this machine connected to a network?")
}
/// Enumerates this machine's private, LAN-reachable IPv4 addresses as
/// `(interface, address, prefix-len)` tuples. Primary path parses `ip -o
/// addr show` -- the formatting it relies on is stable, and it bakes in the
/// `scope global up` filter (excluding link-local 169.254/16 and down
/// interfaces) for free -- falling back to a `getifaddrs`-based enumeration
/// (`if-addrs`, already in the dependency tree via `mdns-sd`) when the `ip`
/// command is unavailable, so the LAN-IP discovery doesn't silently depend on
/// iproute2 being installed.
fn lan_ipv4_candidates() -> Result<Vec<(String, Ipv4Addr, u8)>> {
match ip_command_candidates() {
Ok(candidates) if !candidates.is_empty() => return Ok(candidates),
Ok(_) => {} // `ip` ran but found no private address -- genuine, fall through
Err(e) => {
tracing::debug!(error = ?e, "`ip addr show` unavailable, falling back to getifaddrs");
}
}
getifaddrs_candidates()
}
fn ip_command_candidates() -> Result<Vec<(String, Ipv4Addr, u8)>> {
let output = std::process::Command::new("ip") let output = std::process::Command::new("ip")
.args(["-4", "-o", "addr", "show", "scope", "global", "up"]) .args(["-4", "-o", "addr", "show", "scope", "global", "up"])
.output() .output()
@ -84,7 +29,7 @@ fn ip_command_candidates() -> Result<Vec<(String, Ipv4Addr, u8)>> {
} }
let text = String::from_utf8_lossy(&output.stdout); let text = String::from_utf8_lossy(&output.stdout);
let mut candidates: Vec<(String, Ipv4Addr, u8)> = Vec::new(); let mut candidates: Vec<(String, Ipv4Addr)> = Vec::new();
for line in text.lines() { for line in text.lines() {
// Format: "3: wlan0 inet 10.179.161.89/23 brd ... scope global dynamic wlan0" // Format: "3: wlan0 inet 10.179.161.89/23 brd ... scope global dynamic wlan0"
let mut fields = line.split_whitespace(); let mut fields = line.split_whitespace();
@ -97,65 +42,20 @@ fn ip_command_candidates() -> Result<Vec<(String, Ipv4Addr, u8)>> {
continue; continue;
} }
let Some(cidr) = fields.next() else { continue }; let Some(cidr) = fields.next() else { continue };
let mut parts = cidr.split('/'); let Some(addr) = cidr.split('/').next().and_then(|a| a.parse::<Ipv4Addr>().ok()) else { continue };
let Some(addr) = parts.next().and_then(|a| a.parse::<Ipv4Addr>().ok()) else { continue };
let prefix: u8 = parts.next().and_then(|p| p.parse().ok()).unwrap_or(32);
if !addr.is_private() { if !addr.is_private() {
continue; continue;
} }
candidates.push((iface.to_string(), addr, prefix.min(32))); candidates.push((iface.to_string(), addr));
} }
Ok(candidates)
}
/// `getifaddrs`-based counterpart to [`ip_command_candidates`], mirroring the // Prefer a conventionally-named physical/Wi-Fi interface when there's a
/// `ip` path's filters (up, non-loopback, private, not link-local, not an // choice, but any private, non-excluded address is acceptable.
/// excluded interface name) so the two produce equivalent candidate sets. candidates.sort_by_key(|(iface, _)| !(iface.starts_with("wl") || iface.starts_with("en") || iface.starts_with("eth")));
fn getifaddrs_candidates() -> Result<Vec<(String, Ipv4Addr, u8)>> {
let ifaces = if_addrs::get_if_addrs().context("failed to enumerate network interfaces (getifaddrs fallback)")?;
let mut candidates: Vec<(String, Ipv4Addr, u8)> = Vec::new();
for iface in ifaces {
if EXCLUDED_PREFIXES.iter().any(|p| iface.name.starts_with(p)) {
continue;
}
if !iface.is_oper_up() {
continue;
}
let if_addrs::IfAddr::V4(v4) = iface.addr else { continue };
// Mirrors `type(-4 -o addr show scope global up)`: drop loopback,
// link-local (169.254) and non-private networks.
if v4.ip.is_loopback() || v4.ip.is_link_local() || !v4.ip.is_private() {
continue;
}
candidates.push((iface.name.clone(), v4.ip, v4.prefixlen.min(32)));
}
Ok(candidates)
}
fn same_subnet(a: Ipv4Addr, b: Ipv4Addr, prefix: u8) -> bool { candidates
if prefix == 0 { .into_iter()
return true; .map(|(_, addr)| IpAddr::V4(addr))
} .next()
let mask = if prefix >= 32 { .context("no LAN-reachable IPv4 address found (excluding loopback/VPN/virtual interfaces) — is this machine connected to a network?")
u32::MAX
} else {
!((1u32 << (32 - prefix)) - 1)
};
(u32::from(a) & mask) == (u32::from(b) & mask)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn same_subnet_respects_prefix_length() {
let a: Ipv4Addr = "10.179.161.89".parse().unwrap();
let b: Ipv4Addr = "10.179.160.1".parse().unwrap();
let other: Ipv4Addr = "10.0.0.1".parse().unwrap();
assert!(same_subnet(a, b, 23));
assert!(!same_subnet(a, other, 23));
assert!(same_subnet(a, a, 32));
assert!(!same_subnet(a, b, 32));
}
} }

View file

@ -451,7 +451,6 @@ pub fn pull_encoded_frame(appsink: &gst_app::AppSink) -> Result<Option<(Vec<u8>,
let poll = gst::ClockTime::from_mseconds(CAPTURE_STALL_POLL.as_millis() as u64); let poll = gst::ClockTime::from_mseconds(CAPTURE_STALL_POLL.as_millis() as u64);
let mut stalled_for = Duration::ZERO; let mut stalled_for = Duration::ZERO;
loop { loop {
let pulled_at = std::time::Instant::now();
let Some(sample) = appsink.try_pull_sample(Some(poll)) else { let Some(sample) = appsink.try_pull_sample(Some(poll)) else {
// EOS is the ordinary end: the user hit "Stop sharing" in the // EOS is the ordinary end: the user hit "Stop sharing" in the
// portal, or the source went away. // portal, or the source went away.
@ -459,18 +458,16 @@ pub fn pull_encoded_frame(appsink: &gst_app::AppSink) -> Result<Option<(Vec<u8>,
return Ok(None); return Ok(None);
} }
// Teardown from another thread (`CastMirrorSession::stop` sets // Teardown from another thread (`CastMirrorSession::stop` sets
// the pipeline to Null) flushes the sink *before* // the pipeline to Null) makes the sink flush, and a flushing
// `current_state()` leaves Playing. A flushing sink returns // sink returns `None` *immediately* rather than after the
// `None` immediately; counting that as stall time used to // timeout. Treat that as a clean end too -- otherwise this would
// raise a spurious "capture stalled" on every normal stop. // busy-spin for the whole stall budget and then report a
if !matches!(appsink.current_state(), gst::State::Playing | gst::State::Paused) // spurious "capture stalled" on every normal stop.
|| matches!(appsink.pending_state(), gst::State::Null | gst::State::Ready) if !matches!(appsink.current_state(), gst::State::Playing | gst::State::Paused) {
|| pulled_at.elapsed() < Duration::from_millis(20)
{
return Ok(None); return Ok(None);
} }
stalled_for += pulled_at.elapsed(); stalled_for += CAPTURE_STALL_POLL;
if stalled_for < CAPTURE_STALL_TIMEOUT { if stalled_for < CAPTURE_STALL_TIMEOUT {
continue; continue;
} }
@ -532,56 +529,11 @@ pub enum RunOutcome {
Timeout, Timeout,
} }
/// Blocks the calling thread until the pipeline reports an error or EOS.
/// Unlike [`run_until_error_or_timeout`] this does not give up after a
/// fixed duration -- a live mirror session can last hours, and a 1-hour
/// leftover from the smoke-test helper was leaving GStreamer errors
/// unobserved for the rest of the cast. Also returns [`RunOutcome::Eos`]
/// once the pipeline has been torn down from another thread (`Null`), so
/// a daemon watcher does not sit forever after `stop()`.
pub fn run_until_eos_or_error(pipeline: &gst::Pipeline) -> Result<RunOutcome> {
let bus = pipeline.bus().context("pipeline has no bus")?;
loop {
let Some(msg) = bus.timed_pop_filtered(
gst::ClockTime::from_mseconds(500),
&[gst::MessageType::Error, gst::MessageType::Eos, gst::MessageType::Warning],
) else {
if !matches!(
pipeline.current_state(),
gst::State::Playing | gst::State::Paused | gst::State::Ready
) {
return Ok(RunOutcome::Eos);
}
continue;
};
use gst::MessageView;
match msg.view() {
MessageView::Error(e) => {
bail!(
"GStreamer pipeline error from {:?}: {} ({:?})",
e.src().map(|s| s.path_string()),
e.error(),
e.debug()
);
}
MessageView::Warning(w) => {
tracing::warn!(
src = ?w.src().map(|s| s.path_string()),
error = %w.error(),
"GStreamer pipeline warning"
);
}
MessageView::Eos(_) => return Ok(RunOutcome::Eos),
_ => {}
}
}
}
/// Blocks the calling thread until the pipeline reports an error or EOS, or /// Blocks the calling thread until the pipeline reports an error or EOS, or
/// `timeout` elapses (whichever first). Returns which of those happened, or /// `timeout` elapses (whichever first). Returns which of those happened, or
/// `Err` on a real pipeline error. Meant for smoke-testing from a /// `Err` on a real pipeline error. Meant for smoke-testing from a
/// synchronous `main`/example; the daemon uses [`run_until_eos_or_error`]. /// synchronous `main`/example; the real daemon will want an async/watch-based
/// version instead of blocking a thread.
pub fn run_until_error_or_timeout(pipeline: &gst::Pipeline, timeout: gst::ClockTime) -> Result<RunOutcome> { pub fn run_until_error_or_timeout(pipeline: &gst::Pipeline, timeout: gst::ClockTime) -> Result<RunOutcome> {
let bus = pipeline.bus().context("pipeline has no bus")?; let bus = pipeline.bus().context("pipeline has no bus")?;
let deadline = std::time::Instant::now() + std::time::Duration::from(timeout); let deadline = std::time::Instant::now() + std::time::Duration::from(timeout);

View file

@ -14,7 +14,7 @@ path = "src/main.rs"
anyhow = { workspace = true } anyhow = { workspace = true }
serde_json = { workspace = true } serde_json = { workspace = true }
breadcast-core = { path = "../breadcast-core" } breadcast-core = { path = "../breadcast-core" }
bread-theme = { git = "https://git.breadway.dev/Breadway/bread-ecosystem", tag = "v0.7.4", features = ["gtk"] } bread-theme = { git = "https://git.breadway.dev/Breadway/bread-ecosystem", tag = "v0.7.2", features = ["gtk"] }
bread-utils = { git = "https://git.breadway.dev/Breadway/bread-ecosystem", tag = "v0.7.2", features = ["gtk"] } bread-utils = { git = "https://git.breadway.dev/Breadway/bread-ecosystem", tag = "v0.7.2", features = ["gtk"] }
gtk4 = { version = "0.11", features = ["v4_12"] } gtk4 = { version = "0.11", features = ["v4_12"] }
gtk4-layer-shell = "0.8" gtk4-layer-shell = "0.8"

View file

@ -47,11 +47,6 @@ pub fn build_css(palette: &Palette) -> String {
color: {overlay}; color: {overlay};
padding: 24px 8px; padding: 24px 8px;
}} }}
.cast-error-label {{
color: #e35b5b;
font-size: 0.85em;
padding: 4px 8px;
}}
.cast-stop-button {{ .cast-stop-button {{
background-color: alpha(#e35b5b, 0.15); background-color: alpha(#e35b5b, 0.15);
color: #e35b5b; color: #e35b5b;

View file

@ -29,8 +29,6 @@ impl IpcClient {
let path = socket_path()?; let path = socket_path()?;
let write_half = UnixStream::connect(&path) let write_half = UnixStream::connect(&path)
.with_context(|| format!("failed to connect to breadcastd at {} — is it running?", path.display()))?; .with_context(|| format!("failed to connect to breadcastd at {} — is it running?", path.display()))?;
// A wedged daemon must not freeze the GTK main thread on write.
let _ = write_half.set_write_timeout(Some(std::time::Duration::from_secs(5)));
let read_half = write_half.try_clone().context("failed to duplicate the IPC socket handle")?; let read_half = write_half.try_clone().context("failed to duplicate the IPC socket handle")?;
let (tx, rx) = mpsc::channel(); let (tx, rx) = mpsc::channel();

View file

@ -13,9 +13,7 @@ use std::rc::Rc;
use bread_theme::load_palette; use bread_theme::load_palette;
use breadcast_core::ipc::{DeviceInfo, Protocol, ServerMessage, StateInfo}; use breadcast_core::ipc::{DeviceInfo, Protocol, ServerMessage, StateInfo};
use gtk4::prelude::*; use gtk4::prelude::*;
use gtk4::{ use gtk4::{Align, Application, Box as GBox, Button, Label, ListBox, Orientation, SelectionMode, glib};
Align, Application, Box as GBox, Button, EventControllerKey, Label, ListBox, Orientation, SelectionMode, glib,
};
use ipc_client::IpcClient; use ipc_client::IpcClient;
const PANEL_WIDTH: i32 = 360; const PANEL_WIDTH: i32 = 360;
@ -40,7 +38,6 @@ fn build_ui(app: &Application) {
bread_theme::gtk::apply_app_css(|| css::build_css(&load_palette())); bread_theme::gtk::apply_app_css(|| css::build_css(&load_palette()));
let window = bread_utils::gtk_popup::new_overlay_window(app, "breadcast"); let window = bread_utils::gtk_popup::new_overlay_window(app, "breadcast");
bread_theme::gtk::bind_window_auto(&window);
let panel = GBox::new(Orientation::Vertical, 8); let panel = GBox::new(Orientation::Vertical, 8);
panel.add_css_class("cast-panel"); panel.add_css_class("cast-panel");
@ -72,50 +69,12 @@ fn build_ui(app: &Application) {
empty_label.add_css_class("cast-empty-label"); empty_label.add_css_class("cast-empty-label");
panel.append(&empty_label); panel.append(&empty_label);
let error_label = Label::new(None);
error_label.add_css_class("cast-error-label");
error_label.set_wrap(true);
error_label.set_visible(false);
panel.append(&error_label);
window.set_child(Some(&panel)); window.set_child(Some(&panel));
bread_utils::gtk_popup::close_on_outside_click(&window, &panel, { bread_utils::gtk_popup::close_on_outside_click(&window, &panel, {
let window = window.clone(); let window = window.clone();
move || window.close() move || window.close()
}); });
let key_ctrl = EventControllerKey::new();
key_ctrl.set_propagation_phase(gtk4::PropagationPhase::Capture);
{
let window = window.clone();
let list = list.clone();
key_ctrl.connect_key_pressed(move |_, key, _, _| {
use gtk4::gdk::Key;
match key {
Key::Escape => {
window.close();
glib::Propagation::Stop
}
Key::Return | Key::KP_Enter => {
if let Some(row) = list.selected_row() {
row.activate();
}
glib::Propagation::Stop
}
Key::Down => {
bread_utils::gtk_popup::select_next_visible(&list);
glib::Propagation::Stop
}
Key::Up => {
bread_utils::gtk_popup::select_prev_visible(&list);
glib::Propagation::Stop
}
_ => glib::Propagation::Proceed,
}
});
}
window.add_controller(key_ctrl);
match IpcClient::connect() { match IpcClient::connect() {
Ok((client, rx)) => { Ok((client, rx)) => {
let client = Rc::new(RefCell::new(client)); let client = Rc::new(RefCell::new(client));
@ -124,16 +83,10 @@ fn build_ui(app: &Application) {
list.connect_row_activated({ list.connect_row_activated({
let client = client.clone(); let client = client.clone();
let window = window.clone();
move |_, row| { move |_, row| {
let Some(device_id) = (unsafe { row.data::<String>("device_id") }) else { return }; let Some(device_id) = (unsafe { row.data::<String>("device_id") }) else { return };
let device_id = unsafe { device_id.as_ref() }.clone(); let device_id = unsafe { device_id.as_ref() }.clone();
let _ = client.borrow_mut().send("start_cast", serde_json::json!({ "device_id": device_id })); let _ = client.borrow_mut().send("start_cast", serde_json::json!({ "device_id": device_id }));
// Close before the portal picker appears -- this overlay
// is KeyboardMode::Exclusive and otherwise steals
// Enter/Escape from the picker (and swallows clicks on
// the dimmed background).
window.close();
} }
}); });
@ -146,26 +99,11 @@ fn build_ui(app: &Application) {
let list = list.clone(); let list = list.clone();
let empty_label = empty_label.clone(); let empty_label = empty_label.clone();
let error_label = error_label.clone();
let status_pill = status_pill.clone(); let status_pill = status_pill.clone();
let stop_button = stop_button.clone(); let stop_button = stop_button.clone();
glib::timeout_add_local(std::time::Duration::from_millis(100), move || { glib::timeout_add_local(std::time::Duration::from_millis(100), move || {
loop { while let Ok(message) = rx.try_recv() {
match rx.try_recv() { handle_server_message(message, &list, &empty_label, &status_pill, &stop_button);
Ok(message) => {
handle_server_message(message, &list, &empty_label, &error_label, &status_pill, &stop_button);
}
Err(std::sync::mpsc::TryRecvError::Empty) => break,
Err(std::sync::mpsc::TryRecvError::Disconnected) => {
empty_label.set_label("breadcastd isn't running");
empty_label.set_visible(true);
list.set_visible(false);
stop_button.set_visible(false);
status_pill.set_label("Offline");
status_pill.remove_css_class("casting");
return glib::ControlFlow::Break;
}
}
} }
glib::ControlFlow::Continue glib::ControlFlow::Continue
}); });
@ -186,7 +124,6 @@ fn handle_server_message(
message: ServerMessage, message: ServerMessage,
list: &ListBox, list: &ListBox,
empty_label: &Label, empty_label: &Label,
error_label: &Label,
status_pill: &Label, status_pill: &Label,
stop_button: &Button, stop_button: &Button,
) { ) {
@ -200,8 +137,6 @@ fn handle_server_message(
} }
ServerMessage::Response { error: Some(error), .. } => { ServerMessage::Response { error: Some(error), .. } => {
eprintln!("breadcast: request failed: {error}"); eprintln!("breadcast: request failed: {error}");
error_label.set_label(&error);
error_label.set_visible(true);
None None
} }
_ => None, _ => None,
@ -209,10 +144,7 @@ fn handle_server_message(
let Some((event, data)) = payload else { return }; let Some((event, data)) = payload else { return };
match event.as_str() { match event.as_str() {
"device_list_changed" => update_device_list(list, empty_label, data), "device_list_changed" => update_device_list(list, empty_label, data),
"state_changed" => { "state_changed" => update_state(status_pill, stop_button, data),
error_label.set_visible(false);
update_state(status_pill, stop_button, data);
}
_ => {} _ => {}
} }
} }

View file

@ -83,11 +83,7 @@ impl CastMirrorSession {
/// receiver dropping the connection) back to the daemon actor, so it /// receiver dropping the connection) back to the daemon actor, so it
/// can transition back to `Idle` and notify GUI clients even if nobody /// can transition back to `Idle` and notify GUI clients even if nobody
/// called `stop()`. /// called `stop()`.
pub async fn start( pub async fn start(device: CastDevice, daemon_tx: tokio::sync::mpsc::Sender<DaemonCommand>) -> Result<Self> {
device: CastDevice,
daemon_tx: tokio::sync::mpsc::Sender<DaemonCommand>,
generation: u64,
) -> Result<Self> {
let capture = CaptureSession::start().await.context("failed to start portal screen capture")?; let capture = CaptureSession::start().await.context("failed to start portal screen capture")?;
let video_node_id = capture.video_node_id(); let video_node_id = capture.video_node_id();
@ -98,18 +94,13 @@ impl CastMirrorSession {
// OFFER below rather than re-derived there, so the advertised stream // OFFER below rather than re-derived there, so the advertised stream
// and the encoded stream cannot drift apart. // and the encoded stream cannot drift apart.
let (pipeline, appsink, encoder, video_params) = let (pipeline, appsink, encoder, video_params) =
match build_video_pipeline_for_streaming(video_node_id) { build_video_pipeline_for_streaming(video_node_id).context("failed to build the encode pipeline")?;
Ok(built) => built,
Err(e) => {
close_capture_bounded(capture).await;
return Err(e).context("failed to build the encode pipeline");
}
};
{ {
let pipeline_watch = pipeline.clone(); let pipeline_watch = pipeline.clone();
std::thread::spawn(move || { std::thread::spawn(move || {
match breadcast_core::pipeline::run_until_eos_or_error(&pipeline_watch) { match breadcast_core::pipeline::run_until_error_or_timeout(&pipeline_watch, gst::ClockTime::from_seconds(3600))
{
Ok(outcome) => tracing::debug!(?outcome, "encode pipeline bus watcher ended"), Ok(outcome) => tracing::debug!(?outcome, "encode pipeline bus watcher ended"),
Err(e) => tracing::error!(error = ?e, "encode pipeline error"), Err(e) => tracing::error!(error = ?e, "encode pipeline error"),
} }
@ -120,37 +111,19 @@ impl CastMirrorSession {
// (milliseconds on a LAN) but still blocking I/O -- run it off the // (milliseconds on a LAN) but still blocking I/O -- run it off the
// async worker thread pool rather than stalling it, even briefly. // async worker thread pool rather than stalling it, even briefly.
let device_for_connect = device.clone(); let device_for_connect = device.clone();
let connect = tokio::task::spawn_blocking(move || { let (session, _media_events, raw_messages) = tokio::task::spawn_blocking(move || {
CastSession::connect_app( CastSession::connect_app(
&device_for_connect, &device_for_connect,
CastDeviceApp::Custom(breadcast_core::caststream::MIRRORING_APP_ID.to_string()), CastDeviceApp::Custom(breadcast_core::caststream::MIRRORING_APP_ID.to_string()),
) )
}) })
.await; .await
let (session, _media_events, raw_messages) = match connect { .context("connect_app task panicked")?
Ok(Ok(connected)) => connected, .context("failed to connect and launch the Mirroring receiver")?;
Ok(Err(e)) => {
let _ = pipeline.set_state(gst::State::Null);
close_capture_bounded(capture).await;
return Err(e).context("failed to connect and launch the Mirroring receiver");
}
Err(e) => {
let _ = pipeline.set_state(gst::State::Null);
close_capture_bounded(capture).await;
return Err(e).context("connect_app task panicked");
}
};
let (sender, stream_events) = let (sender, stream_events) =
match CastStreamSender::start(&device.host, "sender-0", session.transport_id(), video_params) { CastStreamSender::start(&device.host, "sender-0", session.transport_id(), video_params)
Ok(started) => started, .context("failed to start the Cast Streaming session")?;
Err(e) => {
stop_session_bounded(&session).await;
let _ = pipeline.set_state(gst::State::Null);
close_capture_bounded(capture).await;
return Err(e).context("failed to start the Cast Streaming session");
}
};
let sender = Arc::new(sender); let sender = Arc::new(sender);
let message_pump = { let message_pump = {
@ -165,12 +138,9 @@ impl CastMirrorSession {
}; };
let negotiated = Arc::new(AtomicBool::new(false)); let negotiated = Arc::new(AtomicBool::new(false));
let failed = Arc::new(AtomicBool::new(false));
let event_pump = { let event_pump = {
let session = session.clone(); let session = session.clone();
let negotiated = negotiated.clone(); let negotiated = negotiated.clone();
let failed = failed.clone();
let daemon_tx = daemon_tx.clone();
std::thread::spawn(move || { std::thread::spawn(move || {
while let Ok(event) = stream_events.recv() { while let Ok(event) = stream_events.recv() {
match event { match event {
@ -180,68 +150,37 @@ impl CastMirrorSession {
} }
} }
CastStreamEvent::Negotiated => negotiated.store(true, Ordering::Release), CastStreamEvent::Negotiated => negotiated.store(true, Ordering::Release),
CastStreamEvent::Error(message) => { CastStreamEvent::Error(message) => tracing::warn!(%message, "Cast Streaming error"),
tracing::warn!(%message, "Cast Streaming error");
failed.store(true, Ordering::Release);
// After negotiation the frame pump is running
// and this is an unprompted death; before
// negotiation, start() itself observes `failed`
// and tears down.
if negotiated.load(Ordering::Acquire) {
let _ = daemon_tx.blocking_send(DaemonCommand::SessionEnded { generation });
}
}
CastStreamEvent::PictureLost => tracing::debug!("receiver reported picture loss"), CastStreamEvent::PictureLost => tracing::debug!("receiver reported picture loss"),
} }
} }
}) })
}; };
// From here every error path must use the same join/drop order as
// `stop()`. Building the session now and calling `stop()` on it is
// what keeps a leaked `CastStreamSender` from leaving
// PlatformClientPosix alive -- the next start would then hit
// OSP_CHECK(!instance_) and abort the daemon.
let mut started = Self {
pipeline,
session,
capture: Some(capture),
sender: Some(sender),
message_pump: Some(message_pump),
event_pump: Some(event_pump),
frame_pump: None,
};
tracing::info!(device = %device.name, "sending Cast Streaming OFFER"); tracing::info!(device = %device.name, "sending Cast Streaming OFFER");
if let Some(sender) = started.sender.as_ref() {
sender.negotiate(); sender.negotiate();
}
let deadline = tokio::time::Instant::now() + tokio::time::Duration::from_secs(10); let deadline = tokio::time::Instant::now() + tokio::time::Duration::from_secs(10);
while !negotiated.load(Ordering::Acquire) while !negotiated.load(Ordering::Acquire) && tokio::time::Instant::now() < deadline {
&& !failed.load(Ordering::Acquire)
&& tokio::time::Instant::now() < deadline
{
tokio::time::sleep(tokio::time::Duration::from_millis(50)).await; tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
} }
if failed.load(Ordering::Acquire) {
started.stop().await;
anyhow::bail!("Cast Streaming session error from {} during negotiation", device.name);
}
if !negotiated.load(Ordering::Acquire) { if !negotiated.load(Ordering::Acquire) {
started.stop().await; // Bounded for the same reason `Self::stop`'s calls are -- an
// unresponsive receiver (which is exactly what "negotiation
// timed out" implies) can wedge either of these forever
// otherwise, taking the whole single-threaded daemon actor
// down with it before this even gets to return an error.
stop_session_bounded(&session).await;
close_capture_bounded(capture).await;
anyhow::bail!("never received an ANSWER from {} (negotiation timed out)", device.name); anyhow::bail!("never received an ANSWER from {} (negotiation timed out)", device.name);
} }
if let Err(e) = started.pipeline.set_state(gst::State::Playing) { pipeline.set_state(gst::State::Playing).context("failed to start the encode pipeline")?;
started.stop().await;
return Err(e).context("failed to start the encode pipeline");
}
tracing::info!(device = %device.name, "mirroring started"); tracing::info!(device = %device.name, "mirroring started");
let frame_pump = { let frame_pump = {
let device_name = device.name.clone(); let device_name = device.name.clone();
let sender = started.sender.as_ref().expect("sender installed above").clone(); let sender = sender.clone();
std::thread::spawn(move || { std::thread::spawn(move || {
let result = frame_pump_loop(&appsink, &encoder, &sender); let result = frame_pump_loop(&appsink, &encoder, &sender);
if let Err(e) = result { if let Err(e) = result {
@ -251,12 +190,19 @@ impl CastMirrorSession {
// was, very recently) still alive. If the channel is full or // was, very recently) still alive. If the channel is full or
// closed, there's nothing more useful to do from this // closed, there's nothing more useful to do from this
// thread than drop the notification. // thread than drop the notification.
let _ = daemon_tx.blocking_send(DaemonCommand::SessionEnded { generation }); let _ = daemon_tx.blocking_send(DaemonCommand::SessionEnded);
}) })
}; };
started.frame_pump = Some(frame_pump);
Ok(started) Ok(Self {
pipeline,
session,
capture: Some(capture),
sender: Some(sender),
message_pump: Some(message_pump),
event_pump: Some(event_pump),
frame_pump: Some(frame_pump),
})
} }
/// Tears down the session. Order matters and is not interchangeable: /// Tears down the session. Order matters and is not interchangeable:

View file

@ -33,29 +33,7 @@ pub enum DaemonCommand {
/// dropping the connection or stopping playback) -- as opposed to /// dropping the connection or stopping playback) -- as opposed to
/// `StopCast` being called. Either way the daemon needs to forget the /// `StopCast` being called. Either way the daemon needs to forget the
/// (now-dead) session and go back to `Idle`. /// (now-dead) session and go back to `Idle`.
SessionEnded { generation: u64 }, SessionEnded,
/// Result of a `StartCast` that ran off this actor (the portal picker
/// and OFFER/ANSWER wait must not stall `list_devices` / `stop_cast`).
/// Ignored when `generation` no longer matches -- that means `StopCast`
/// cancelled the in-flight start.
StartFinished {
generation: u64,
outcome: Result<StartedSession, StartFailed>,
reply: oneshot::Sender<Result<(), String>>,
},
}
/// A session that finished starting, ready to be installed as `active_session`.
pub(crate) struct StartedSession {
pub session: ActiveSession,
pub device_id: String,
pub device_name: String,
pub protocol: Protocol,
}
pub(crate) struct StartFailed {
pub device_id: String,
pub error: String,
} }
pub fn spawn(events_tx: broadcast::Sender<ServerMessage>, bread_client: BreadClient) -> mpsc::Sender<DaemonCommand> { pub fn spawn(events_tx: broadcast::Sender<ServerMessage>, bread_client: BreadClient) -> mpsc::Sender<DaemonCommand> {
@ -67,8 +45,6 @@ pub fn spawn(events_tx: broadcast::Sender<ServerMessage>, bread_client: BreadCli
dlna_devices: HashMap::new(), dlna_devices: HashMap::new(),
state: StateInfo::Idle, state: StateInfo::Idle,
active_session: None, active_session: None,
starting: false,
start_generation: 0,
events_tx, events_tx,
bread_client, bread_client,
self_tx, self_tx,
@ -83,7 +59,7 @@ pub fn spawn(events_tx: broadcast::Sender<ServerMessage>, bread_client: BreadCli
/// The currently active mirroring session, if any -- exactly one of the two /// The currently active mirroring session, if any -- exactly one of the two
/// protocol-specific session types, chosen by which device map `StartCast` /// protocol-specific session types, chosen by which device map `StartCast`
/// found the requested device id in. /// found the requested device id in.
pub(crate) enum ActiveSession { enum ActiveSession {
Cast(CastMirrorSession), Cast(CastMirrorSession),
Dlna(Box<DlnaMirrorSession>), Dlna(Box<DlnaMirrorSession>),
} }
@ -97,16 +73,9 @@ impl ActiveSession {
} }
} }
/// `host` strings come straight from mDNS resolution, and mdns-sd renders a /// `host` strings come straight from mDNS resolution, so this just checks
/// link-local IPv6 it resolved via `ScopedIp` with a `%zone` suffix (e.g. /// the address, not whether a zone id is attached (mDNS never gives us one).
/// `fe80::…%wlan0`) so the address stays connectable. `IpAddr` can't parse
/// that suffix (std rejects zone-ids in `FromStr`), so the zone has to be
/// stripped before the network prefix can be examined -- otherwise a scoped
/// `fe80::…%iface` host would fail this check and, in `CastDeviceFound`, be
/// allowed to clobber a good routable host that `rust_cast` could actually
/// connect to.
fn is_link_local_v6(host: &str) -> bool { fn is_link_local_v6(host: &str) -> bool {
let host = host.split('%').next().unwrap_or(host);
matches!(host.parse::<std::net::IpAddr>(), Ok(std::net::IpAddr::V6(v6)) if (v6.segments()[0] & 0xffc0) == 0xfe80) matches!(host.parse::<std::net::IpAddr>(), Ok(std::net::IpAddr::V6(v6)) if (v6.segments()[0] & 0xffc0) == 0xfe80)
} }
@ -117,13 +86,6 @@ struct Daemon {
dlna_devices: HashMap<String, DlnaDevice>, dlna_devices: HashMap<String, DlnaDevice>,
state: StateInfo, state: StateInfo,
active_session: Option<ActiveSession>, active_session: Option<ActiveSession>,
/// True while a `StartCast` is running off this actor (portal picker /
/// negotiation). Distinct from `active_session` so a second start is
/// rejected before the first one has a session to install.
starting: bool,
/// Bumped by `StopCast` so a `StartFinished` from a cancelled start
/// tears its session down instead of installing it.
start_generation: u64,
events_tx: broadcast::Sender<ServerMessage>, events_tx: broadcast::Sender<ServerMessage>,
/// Used to publish `bread.cast.mirroring_started`/`.stopped`/`.failed` /// Used to publish `bread.cast.mirroring_started`/`.stopped`/`.failed`
/// on state transitions -- see `bread_events.rs`. A no-op if breadd /// on state transitions -- see `bread_events.rs`. A no-op if breadd
@ -149,12 +111,6 @@ impl Daemon {
self.start_cast(device_id, reply).await; self.start_cast(device_id, reply).await;
} }
DaemonCommand::StopCast { reply } => { DaemonCommand::StopCast { reply } => {
if self.starting {
// Invalidate the in-flight start so its StartFinished
// tears the session down instead of installing it.
self.starting = false;
self.start_generation = self.start_generation.wrapping_add(1);
}
if let Some(session) = self.active_session.take() { if let Some(session) = self.active_session.take() {
session.stop().await; session.stop().await;
bread_events::emit_mirroring_stopped(&self.bread_client); bread_events::emit_mirroring_stopped(&self.bread_client);
@ -163,48 +119,19 @@ impl Daemon {
self.broadcast_state(); self.broadcast_state();
let _ = reply.send(Ok(())); let _ = reply.send(Ok(()));
} }
DaemonCommand::SessionEnded { generation } => { DaemonCommand::SessionEnded => {
// Ignore a late notification from a session that already // The session already tore itself down (that's what
// stopped (or whose start was cancelled). An abandoned // triggered this) -- just drop our handle to it and update
// pump after a 5s join timeout used to take() the *next* // state. Ignored if this arrives after an explicit
// session and flip the UI to Idle while it was still live. // `StopCast` already cleared `active_session` (the
if generation != self.start_generation { // background pump/poll it came from may briefly outlive
return; // that call).
} if self.active_session.take().is_some() {
// The session's pump/poll noticed death, but the session
// handle itself has *not* been torn down -- there is no
// Drop impl. Dropping it here used to skip CastSession::stop
// (TV left on the last frame), skip portal close (PipeWire
// leak), and destroy the FFI sender while pump threads were
// still calling into it. Always run the ordered stop.
if let Some(session) = self.active_session.take() {
tracing::info!("mirror session ended on its own, returning to idle"); tracing::info!("mirror session ended on its own, returning to idle");
session.stop().await;
self.start_generation = self.start_generation.wrapping_add(1);
self.state = StateInfo::Idle; self.state = StateInfo::Idle;
self.broadcast_state(); self.broadcast_state();
bread_events::emit_mirroring_stopped(&self.bread_client); bread_events::emit_mirroring_stopped(&self.bread_client);
return;
} }
// No installed session, but a start is currently in flight
// (portal picker / OFFER-ANSWER). The event pump only reports
// SessionEnded once negotiation succeeded, so a matching
// generation here means the in-flight session died between
// `Negotiated` and `StartFinished` landing. If we ignore it,
// the pending `StartFinished` would install the dead session
// and leave the UI stuck on "Casting" until the frame pump
// happens to notice. Invalidate the pending start exactly as
// `StopCast` does -- bump the generation so
// `on_start_finished` tears the session down instead of
// installing it.
if self.starting {
tracing::debug!("in-flight mirror session ended before it was installed");
self.starting = false;
self.start_generation = self.start_generation.wrapping_add(1);
}
}
DaemonCommand::StartFinished { generation, outcome, reply } => {
self.on_start_finished(generation, outcome, reply).await;
} }
DaemonCommand::CastDeviceFound(device) => { DaemonCommand::CastDeviceFound(device) => {
// mDNS resolves one physical device on every local address // mDNS resolves one physical device on every local address
@ -220,8 +147,8 @@ impl Daemon {
); );
if should_replace { if should_replace {
self.cast_devices.insert(device.id.clone(), device); self.cast_devices.insert(device.id.clone(), device);
self.broadcast_devices();
} }
self.broadcast_devices();
} }
DaemonCommand::CastDeviceLost(id) => { DaemonCommand::CastDeviceLost(id) => {
self.cast_devices.remove(&id); self.cast_devices.remove(&id);
@ -239,100 +166,54 @@ impl Daemon {
} }
async fn start_cast(&mut self, device_id: String, reply: oneshot::Sender<Result<(), String>>) { async fn start_cast(&mut self, device_id: String, reply: oneshot::Sender<Result<(), String>>) {
if self.active_session.is_some() || self.starting { if self.active_session.is_some() {
let _ = reply.send(Err("already casting -- stop the current session first".to_string())); let _ = reply.send(Err("already casting -- stop the current session first".to_string()));
return; return;
} }
if let Some(device) = self.cast_devices.get(&device_id).cloned() { if let Some(device) = self.cast_devices.get(&device_id).cloned() {
self.starting = true; match CastMirrorSession::start(device.clone(), self.self_tx.clone()).await {
let generation = self.start_generation; Ok(session) => {
let daemon_tx = self.self_tx.clone(); self.active_session = Some(ActiveSession::Cast(session));
tokio::spawn(async move { self.state =
let outcome = match CastMirrorSession::start(device.clone(), daemon_tx.clone(), generation).await { StateInfo::Casting { device_id: device.id.clone(), device_name: device.name.clone(), protocol: Protocol::Cast };
Ok(session) => Ok(StartedSession { self.broadcast_state();
session: ActiveSession::Cast(session), bread_events::emit_mirroring_started(&self.bread_client, &device.id, &device.name, "cast");
device_id: device.id, let _ = reply.send(Ok(()));
device_name: device.name, }
protocol: Protocol::Cast, Err(e) => {
}),
// `{e:#}` (not `{e}`/`to_string()`) so the full anyhow // `{e:#}` (not `{e}`/`to_string()`) so the full anyhow
// context chain reaches the caller/GUI instead of just // context chain reaches the caller/GUI instead of just
// the outermost ".context()" message. // the outermost ".context()" message.
Err(e) => Err(StartFailed { device_id: device.id, error: format!("{e:#}") }), bread_events::emit_mirroring_failed(&self.bread_client, &device.id, &format!("{e:#}"));
}; let _ = reply.send(Err(format!("{e:#}")));
let _ = daemon_tx.send(DaemonCommand::StartFinished { generation, outcome, reply }).await; }
}); }
return; return;
} }
if let Some(device) = self.dlna_devices.get(&device_id).cloned() { if let Some(device) = self.dlna_devices.get(&device_id).cloned() {
self.starting = true; match DlnaMirrorSession::start(device.clone(), self.self_tx.clone()).await {
let generation = self.start_generation; Ok(session) => {
let daemon_tx = self.self_tx.clone(); self.active_session = Some(ActiveSession::Dlna(Box::new(session)));
tokio::spawn(async move {
let outcome = match DlnaMirrorSession::start(device.clone(), daemon_tx.clone(), generation).await {
Ok(session) => Ok(StartedSession {
session: ActiveSession::Dlna(Box::new(session)),
device_id: device.url,
device_name: device.friendly_name,
protocol: Protocol::Dlna,
}),
Err(e) => Err(StartFailed { device_id: device.url, error: format!("{e:#}") }),
};
let _ = daemon_tx.send(DaemonCommand::StartFinished { generation, outcome, reply }).await;
});
return;
}
let error = format!("unknown device id \"{device_id}\"");
bread_events::emit_mirroring_failed(&self.bread_client, &device_id, &error);
let _ = reply.send(Err(error));
}
async fn on_start_finished(
&mut self,
generation: u64,
outcome: Result<StartedSession, StartFailed>,
reply: oneshot::Sender<Result<(), String>>,
) {
if generation != self.start_generation || !self.starting {
// StopCast cancelled this start while the portal/negotiation
// was still running. Tear the session down if it succeeded
// anyway, so we don't leak a sender or leave the TV casting.
if let Ok(started) = outcome {
started.session.stop().await;
}
let _ = reply.send(Err("start cancelled".to_string()));
return;
}
self.starting = false;
match outcome {
Ok(started) => {
self.active_session = Some(started.session);
self.state = StateInfo::Casting { self.state = StateInfo::Casting {
device_id: started.device_id.clone(), device_id: device.url.clone(),
device_name: started.device_name.clone(), device_name: device.friendly_name.clone(),
protocol: started.protocol, protocol: Protocol::Dlna,
}; };
self.broadcast_state(); self.broadcast_state();
let protocol = match started.protocol { bread_events::emit_mirroring_started(&self.bread_client, &device.url, &device.friendly_name, "dlna");
Protocol::Cast => "cast",
Protocol::Dlna => "dlna",
};
bread_events::emit_mirroring_started(
&self.bread_client,
&started.device_id,
&started.device_name,
protocol,
);
let _ = reply.send(Ok(())); let _ = reply.send(Ok(()));
} }
Err(failed) => { Err(e) => {
bread_events::emit_mirroring_failed(&self.bread_client, &failed.device_id, &failed.error); bread_events::emit_mirroring_failed(&self.bread_client, &device.url, &format!("{e:#}"));
let _ = reply.send(Err(failed.error)); let _ = reply.send(Err(format!("{e:#}")));
} }
} }
return;
}
let _ = reply.send(Err(format!("unknown device id \"{device_id}\"")));
} }
fn device_list(&self) -> Vec<DeviceInfo> { fn device_list(&self) -> Vec<DeviceInfo> {
@ -360,159 +241,3 @@ impl Daemon {
let _ = self.events_tx.send(ServerMessage::Event { event: "device_list_changed".to_string(), data }); let _ = self.events_tx.send(ServerMessage::Event { event: "device_list_changed".to_string(), data });
} }
} }
#[cfg(test)]
mod tests {
use std::collections::HashMap;
use bread_utils::bread_client::BreadClient;
use tokio::sync::{broadcast, mpsc, oneshot};
use super::*;
/// A daemon parked mid-`StartCast`: `starting` is set and a concrete
/// generation is outstanding, but no `StartFinished` has landed yet.
/// This is exactly the window both races below target -- the portal
/// picker / OFFER/ANSWER negotiation is still running off the actor,
/// and the session that `StartFinished` would install is not installed.
///
/// A real `MirrorSession` can't be built in a unit test (it needs the
/// portal capture session, a GStreamer encode pipeline, and a live
/// renderer on the LAN), so the `Stale StartFinished` probes below use
/// an `Err(StartFailed)` outcome. That still exercises the whole point
/// of the guard: `on_start_finished` checks `generation`/`starting`
/// *before* branching on the outcome, so the stale event is rejected
/// with "start cancelled", never installs a session, and the UI stays
/// `Idle`. The `Ok` branch of that same guard calls
/// `started.session.stop()` to tear the dead session down -- the literal
/// teardown line -- which is the only part not separately asserted here.
fn daemon_with_in_flight_start(generation: u64) -> (Daemon, broadcast::Receiver<ServerMessage>) {
let (events_tx, events_rx) = broadcast::channel(16);
let (self_tx, _self_rx) = mpsc::channel(16);
let daemon = Daemon {
cast_devices: HashMap::new(),
dlna_devices: HashMap::new(),
state: StateInfo::Idle,
active_session: None,
starting: true,
start_generation: generation,
events_tx,
bread_client: BreadClient::connect("daemon-test"),
self_tx,
};
(daemon, events_rx)
}
/// A `StartFinished` carrying the *dead* session's generation -- the
/// probe both race tests replay after invalidating the in-flight start,
/// to prove it can't sneak a session back in.
fn stale_start_finished(reply: oneshot::Sender<Result<(), String>>) -> DaemonCommand {
DaemonCommand::StartFinished {
generation: 7,
outcome: Err(StartFailed {
device_id: "living-room-tv".to_string(),
error: "negotiation failed".to_string(),
}),
reply,
}
}
#[test]
fn link_local_v6_is_detected_with_and_without_a_zone() {
// Scoped form: mdns-sd's ScopedIp Display renders link-local IPv6 as
// `fe80::…%<iface>`, and IpAddr can't parse the `%` suffix -- this is
// the exact form the daemon-clobber bug shipped with.
assert!(is_link_local_v6("fe80::1234%wlan0"));
assert!(is_link_local_v6("fe80::1234%3"));
// Unscoped (bare)`fe80::…` also matches.
assert!(is_link_local_v6("fe80::1234"));
}
#[test]
fn non_link_local_addresses_are_not_marked_link_local() {
assert!(!is_link_local_v6("fe80"));
assert!(!is_link_local_v6("192.168.1.50"));
assert!(!is_link_local_v6("fd00::1")); // ULA
assert!(!is_link_local_v6("2606:4700::1111")); // public
assert!(!is_link_local_v6("2606:4700::1111%wlan0")); // public + spurious zone
assert!(!is_link_local_v6("not-an-ip"));
}
/// The session's event pump reports death (`SessionEnded`) *between*
/// negotiation succeeding and `StartFinished` landing -- the exact race
/// that used to install the already-dead session and leave the UI stuck
/// on "Casting" until the frame pump happened to notice.
#[tokio::test]
async fn session_ended_during_an_in_flight_start_invalidates_the_pending_start() {
let (mut daemon, mut events_rx) = daemon_with_in_flight_start(7);
daemon.handle(DaemonCommand::SessionEnded { generation: 7 }).await;
// Invalidate the pending start exactly as StopCast does: clear
// `starting` and bump the generation so on_start_finished tears the
// dead session down instead of installing it.
assert!(!daemon.starting, "in-flight start must be cancelled");
assert_eq!(daemon.start_generation, 8);
assert!(daemon.active_session.is_none());
assert!(matches!(daemon.state, StateInfo::Idle));
// The stale StartFinished from the dead session must not install it,
// and the caller must hear that the start was cancelled.
let (reply, reply_rx) = oneshot::channel();
daemon.handle(stale_start_finished(reply)).await;
assert_eq!(reply_rx.await.unwrap(), Err("start cancelled".to_string()));
assert!(daemon.active_session.is_none(), "dead session must not be installed");
assert!(matches!(daemon.state, StateInfo::Idle), "UI must return to / stay Idle");
// A cancelled in-flight start broadcasts nothing -- no bogus
// state_changed to a UI that is already Idle.
assert!(events_rx.try_recv().is_err());
}
/// `StopCast` while the portal/negotiation is still running must cancel
/// the in-flight start, so its eventual `StartFinished` is rejected
/// rather than installing a session the user just explicitly stopped.
#[tokio::test]
async fn stop_cast_cancels_an_in_flight_start() {
let (mut daemon, mut events_rx) = daemon_with_in_flight_start(7);
let (reply, reply_rx) = oneshot::channel();
daemon.handle(DaemonCommand::StopCast { reply }).await;
// StopCast still resolves Ok -- there was nothing installed to stop,
// only a start to cancel.
assert!(reply_rx.await.unwrap().is_ok());
assert!(!daemon.starting, "StopCast must cancel the in-flight start");
assert_eq!(daemon.start_generation, 8);
assert!(daemon.active_session.is_none());
assert!(matches!(daemon.state, StateInfo::Idle));
// A late StartFinished from the cancelled start never installs.
let (reply, reply_rx) = oneshot::channel();
daemon.handle(stale_start_finished(reply)).await;
assert_eq!(reply_rx.await.unwrap(), Err("start cancelled".to_string()));
assert!(daemon.active_session.is_none(), "cancelled start must not install a session");
assert!(matches!(daemon.state, StateInfo::Idle));
// StopCast itself broadcast the transition back to Idle.
assert!(events_rx.try_recv().is_ok());
}
/// Defensive: a `SessionEnded` whose generation no longer matches (a
/// late notification from an already-stopped or already-cancelled
/// session) must be ignored without touching the current generation, so
/// it can't confuse a brand-new later start.
#[tokio::test]
async fn a_stale_session_ended_while_idle_is_ignored_without_bumping_generation() {
let (mut daemon, _events_rx) = daemon_with_in_flight_start(0);
daemon.starting = false; // not starting -- the stale event's own forerunner already finished
daemon.handle(DaemonCommand::SessionEnded { generation: 99 }).await;
assert!(!daemon.starting);
assert_eq!(daemon.start_generation, 0, "stale SessionEnded must not bump a future start's generation");
assert!(daemon.active_session.is_none());
assert!(matches!(daemon.state, StateInfo::Idle));
}
}

View file

@ -13,11 +13,8 @@
use std::time::Duration; use std::time::Duration;
use anyhow::{Context, Result}; use anyhow::{Context, Result};
use breadcast_core::http_server::HttpServer;
use breadcast_core::net::local_lan_ip; use breadcast_core::net::local_lan_ip;
use breadcast_core::pipeline::{ use breadcast_core::pipeline::{build_video_pipeline, hls_output_dir, wait_for_playlist_segments};
build_video_pipeline, hls_output_dir, run_until_eos_or_error, wait_for_playlist_segments,
};
use breadcast_core::{CaptureSession, DlnaDevice, DlnaSession}; use breadcast_core::{CaptureSession, DlnaDevice, DlnaSession};
use gstreamer as gst; use gstreamer as gst;
use gstreamer::prelude::*; use gstreamer::prelude::*;
@ -30,20 +27,11 @@ use crate::daemon::DaemonCommand;
/// this is a poll, not a push. /// this is a poll, not a push.
const POLL_INTERVAL: Duration = Duration::from_secs(3); const POLL_INTERVAL: Duration = Duration::from_secs(3);
/// How long the playlist may sit unchanged before we treat the encode
/// side as dead. The Cast path has `pull_encoded_frame`'s stall watchdog;
/// this is the HLS equivalent -- portal/encoder stalls produce no bus
/// error, just a playlist that stops growing.
const PLAYLIST_STALL_TIMEOUT: Duration = Duration::from_secs(15);
pub struct DlnaMirrorSession { pub struct DlnaMirrorSession {
pipeline: gst::Pipeline, pipeline: gst::Pipeline,
session: DlnaSession, session: DlnaSession,
capture: Option<CaptureSession>, capture: Option<CaptureSession>,
http: HttpServer,
output_dir: std::path::PathBuf,
poll_task: tokio::task::JoinHandle<()>, poll_task: tokio::task::JoinHandle<()>,
stall_task: tokio::task::JoinHandle<()>,
} }
impl DlnaMirrorSession { impl DlnaMirrorSession {
@ -55,57 +43,42 @@ impl DlnaMirrorSession {
/// renderer stopping playback on its own, a GStreamer error, or the /// renderer stopping playback on its own, a GStreamer error, or the
/// renderer becoming unreachable) back to the daemon actor — mirrors /// renderer becoming unreachable) back to the daemon actor — mirrors
/// `CastMirrorSession::start`'s same use of it. /// `CastMirrorSession::start`'s same use of it.
pub async fn start( pub async fn start(device: DlnaDevice, daemon_tx: tokio::sync::mpsc::Sender<DaemonCommand>) -> Result<Self> {
device: DlnaDevice,
daemon_tx: tokio::sync::mpsc::Sender<DaemonCommand>,
generation: u64,
) -> Result<Self> {
let capture = CaptureSession::start().await.context("failed to start portal screen capture")?; let capture = CaptureSession::start().await.context("failed to start portal screen capture")?;
let video_node_id = capture.video_node_id(); let video_node_id = capture.video_node_id();
let output_dir = match hls_output_dir("dlna-mirror") { let output_dir = hls_output_dir("dlna-mirror")?;
Ok(dir) => dir, let pipeline =
Err(e) => { build_video_pipeline(video_node_id, &output_dir).context("failed to build the encode pipeline")?;
let _ = capture.close().await;
return Err(e);
}
};
let pipeline = match build_video_pipeline(video_node_id, &output_dir) {
Ok(p) => p,
Err(e) => {
let _ = capture.close().await;
let _ = std::fs::remove_dir_all(&output_dir);
return Err(e).context("failed to build the encode pipeline");
}
};
if let Err(e) = pipeline.set_state(gst::State::Playing) { // Fire-and-forget, same as `CastMirrorSession::start`'s identical
let _ = capture.close().await; // block: nothing joins this thread, it just self-terminates on
let _ = std::fs::remove_dir_all(&output_dir); // pipeline error, EOS, or its own 1-hour timeout, whichever is
return Err(e).context("failed to start the encode pipeline"); // first — see that function's doc comment for why that's fine.
{
let pipeline_watch = pipeline.clone();
std::thread::spawn(move || {
match breadcast_core::pipeline::run_until_error_or_timeout(&pipeline_watch, gst::ClockTime::from_seconds(3600))
{
Ok(outcome) => tracing::debug!(?outcome, "DLNA encode pipeline bus watcher ended"),
Err(e) => tracing::error!(error = ?e, "DLNA encode pipeline error"),
}
});
} }
let peer = host_ip_from_url(&device.url); pipeline.set_state(gst::State::Playing).context("failed to start the encode pipeline")?;
let lan_ip = match peer.map(breadcast_core::net::local_lan_ip_for).unwrap_or_else(local_lan_ip) {
Ok(ip) => ip, let lan_ip = local_lan_ip().context("failed to determine this machine's LAN-reachable IP")?;
Err(e) => {
abort_partial(&pipeline, capture, None, &output_dir).await;
return Err(e).context("failed to determine this machine's LAN-reachable IP");
}
};
// Bind an ephemeral port (`:0`) rather than a fixed one like the // Bind an ephemeral port (`:0`) rather than a fixed one like the
// `dlna_mirror_test` example uses -- the daemon may need to run // `dlna_mirror_test` example uses -- the daemon may need to run
// alongside that example, or a future concurrent-session mode, // alongside that example, or a future concurrent-session mode,
// without a bind conflict. The server is held on the session and // without a bind conflict. `HttpServer::start`'s worker threads
// shut down in `stop()` so the last screen-recording segments are // outlive this session once it stops (documented pre-existing
// not left reachable on the LAN. // limitation, see `http_server.rs` -- not something introduced
let http = match HttpServer::start("0.0.0.0:0", output_dir.clone()) { // here); one leaked idle listener per DLNA cast is an accepted
Ok(http) => http, // cost until that gets a real shutdown path.
Err(e) => { let http = breadcast_core::http_server::HttpServer::start("0.0.0.0:0", output_dir.clone())
abort_partial(&pipeline, capture, None, &output_dir).await; .context("failed to start the HLS HTTP server")?;
return Err(e).context("failed to start the HLS HTTP server");
}
};
let stream_url = http.url(lan_ip, "playlist.m3u8"); let stream_url = http.url(lan_ip, "playlist.m3u8");
// Two segments, not three: this is a "don't hand the renderer a 404 // Two segments, not three: this is a "don't hand the renderer a 404
@ -114,42 +87,30 @@ impl DlnaMirrorSession {
// `build_video_pipeline`'s note on HLS latency). Two is the minimum // `build_video_pipeline`'s note on HLS latency). Two is the minimum
// that still proves the encoder is genuinely producing output rather // that still proves the encoder is genuinely producing output rather
// than having emitted one segment and stalled. // than having emitted one segment and stalled.
if let Err(e) = wait_for_playlist_segments(&output_dir.join("playlist.m3u8"), 2, Duration::from_secs(20)).await wait_for_playlist_segments(&output_dir.join("playlist.m3u8"), 2, Duration::from_secs(20))
{ .await
abort_partial(&pipeline, capture, Some(http), &output_dir).await; .context("encode pipeline never produced playable HLS segments")?;
return Err(e).context("encode pipeline never produced playable HLS segments");
}
let session = match DlnaSession::connect(&device).await { let session = DlnaSession::connect(&device).await.context("failed to connect to the DLNA renderer")?;
Ok(session) => session, session.load(&stream_url).await.context("renderer rejected the stream load")?;
Err(e) => {
abort_partial(&pipeline, capture, Some(http), &output_dir).await;
return Err(e).context("failed to connect to the DLNA renderer");
}
};
if let Err(e) = session.load(&stream_url).await {
abort_partial(&pipeline, capture, Some(http), &output_dir).await;
return Err(e).context("renderer rejected the stream load");
}
tracing::info!(device = %device.friendly_name, %stream_url, "DLNA mirroring started"); tracing::info!(device = %device.friendly_name, %stream_url, "DLNA mirroring started");
let poll_task = { let poll_task = {
let session = session.clone(); let session = session.clone();
let device_name = device.friendly_name.clone(); let device_name = device.friendly_name.clone();
let daemon_tx = daemon_tx.clone();
tokio::spawn(async move { tokio::spawn(async move {
loop { loop {
tokio::time::sleep(POLL_INTERVAL).await; tokio::time::sleep(POLL_INTERVAL).await;
match session.transport_state().await { match session.transport_state().await {
Ok(state) if state == "STOPPED" || state == "NO_MEDIA_PRESENT" => { Ok(state) if state == "STOPPED" || state == "NO_MEDIA_PRESENT" => {
tracing::info!(device = %device_name, %state, "DLNA renderer ended playback on its own"); tracing::info!(device = %device_name, %state, "DLNA renderer ended playback on its own");
let _ = daemon_tx.send(DaemonCommand::SessionEnded { generation }).await; let _ = daemon_tx.send(DaemonCommand::SessionEnded).await;
return; return;
} }
Ok(_) => {} Ok(_) => {}
Err(e) => { Err(e) => {
tracing::warn!(device = %device_name, error = ?e, "DLNA transport state poll failed, treating renderer as gone"); tracing::warn!(device = %device_name, error = ?e, "DLNA transport state poll failed, treating renderer as gone");
let _ = daemon_tx.send(DaemonCommand::SessionEnded { generation }).await; let _ = daemon_tx.send(DaemonCommand::SessionEnded).await;
return; return;
} }
} }
@ -157,69 +118,17 @@ impl DlnaMirrorSession {
}) })
}; };
let stall_task = { Ok(Self { pipeline, session, capture: Some(capture), poll_task })
let playlist = output_dir.join("playlist.m3u8");
let daemon_tx = daemon_tx.clone();
tokio::spawn(async move {
let mut last_mtime = None;
let mut stalled_for = Duration::ZERO;
loop {
tokio::time::sleep(Duration::from_secs(1)).await;
let mtime = std::fs::metadata(&playlist).and_then(|m| m.modified()).ok();
if mtime != last_mtime {
last_mtime = mtime;
stalled_for = Duration::ZERO;
continue;
}
stalled_for += Duration::from_secs(1);
if stalled_for >= PLAYLIST_STALL_TIMEOUT {
tracing::warn!(
"DLNA encode stalled: playlist unchanged for {}s",
PLAYLIST_STALL_TIMEOUT.as_secs()
);
let _ = daemon_tx.send(DaemonCommand::SessionEnded { generation }).await;
return;
}
}
})
};
// Portal "stop sharing" / a real GStreamer error used to only log
// -- the bus watcher never told the daemon, so the UI stayed on
// Casting and the renderer kept looping stale segments. Notify.
{
let pipeline_watch = pipeline.clone();
let daemon_tx = daemon_tx.clone();
std::thread::spawn(move || {
match run_until_eos_or_error(&pipeline_watch) {
Ok(outcome) => tracing::debug!(?outcome, "DLNA encode pipeline bus watcher ended"),
Err(e) => tracing::error!(error = ?e, "DLNA encode pipeline error"),
}
let _ = daemon_tx.blocking_send(DaemonCommand::SessionEnded { generation });
});
}
Ok(Self {
pipeline,
session,
capture: Some(capture),
http,
output_dir,
poll_task,
stall_task,
})
} }
/// Tears down the session: stops polling, tells the renderer to stop, /// Tears down the session: stops polling, tells the renderer to stop,
/// stops the encode pipeline, shuts the HLS server, closes the portal /// stops the encode pipeline, and closes the portal capture session.
/// capture session, and deletes the recording directory.
pub async fn stop(mut self) { pub async fn stop(mut self) {
// A request, not a wait -- if the poll task is mid-poll and sends // A request, not a wait -- if the poll task is mid-poll and sends
// one more `SessionEnded` right as this races it, that's harmless: // one more `SessionEnded` right as this races it, that's harmless:
// `daemon.rs`'s handler already no-ops when `active_session` was // `daemon.rs`'s handler already no-ops when `active_session` was
// already cleared by this explicit stop. // already cleared by this explicit stop.
self.poll_task.abort(); self.poll_task.abort();
self.stall_task.abort();
if let Err(e) = self.session.stop().await { if let Err(e) = self.session.stop().await {
tracing::warn!(error = ?e, "failed to cleanly stop the DLNA session"); tracing::warn!(error = ?e, "failed to cleanly stop the DLNA session");
@ -227,45 +136,10 @@ impl DlnaMirrorSession {
if let Err(e) = self.pipeline.set_state(gst::State::Null) { if let Err(e) = self.pipeline.set_state(gst::State::Null) {
tracing::warn!(error = ?e, "failed to stop the encode pipeline cleanly"); tracing::warn!(error = ?e, "failed to stop the encode pipeline cleanly");
} }
self.http.shutdown();
if let Some(capture) = self.capture.take() { if let Some(capture) = self.capture.take() {
match tokio::time::timeout(Duration::from_secs(5), capture.close()).await { if let Err(e) = capture.close().await {
Ok(Err(e)) => tracing::warn!(error = ?e, "failed to cleanly close the portal capture session"), tracing::warn!(error = ?e, "failed to cleanly close the portal capture session");
Err(_) => tracing::warn!("portal capture session did not close within 5s -- abandoning it"),
Ok(Ok(())) => {}
} }
} }
if let Err(e) = std::fs::remove_dir_all(&self.output_dir) {
tracing::debug!(error = %e, dir = %self.output_dir.display(), "failed to remove HLS output dir");
}
} }
} }
async fn abort_partial(
pipeline: &gst::Pipeline,
capture: CaptureSession,
http: Option<HttpServer>,
output_dir: &std::path::Path,
) {
let _ = pipeline.set_state(gst::State::Null);
if let Some(mut http) = http {
http.shutdown();
}
match tokio::time::timeout(Duration::from_secs(5), capture.close()).await {
Ok(Err(e)) => tracing::warn!(error = ?e, "failed to close portal capture after a failed DLNA start"),
Err(_) => tracing::warn!("portal capture did not close within 5s after a failed DLNA start"),
Ok(Ok(())) => {}
}
let _ = std::fs::remove_dir_all(output_dir);
}
fn host_ip_from_url(url: &str) -> Option<std::net::IpAddr> {
let rest = url.split("://").nth(1)?;
let hostport = rest.split('/').next()?;
let host = if let Some(inside) = hostport.strip_prefix('[') {
inside.split(']').next()?
} else {
hostport.rsplit_once(':').map(|(h, _)| h).unwrap_or(hostport)
};
host.parse().ok()
}

View file

@ -13,8 +13,6 @@
//! a time, but nothing here assumes that) can connect concurrently; each //! a time, but nothing here assumes that) can connect concurrently; each
//! gets its own copy of every broadcast event. //! gets its own copy of every broadcast event.
use std::os::unix::fs::DirBuilderExt;
use anyhow::{Context, Result}; use anyhow::{Context, Result};
use breadcast_core::ipc::{ClientRequest, ServerMessage, socket_path}; use breadcast_core::ipc::{ClientRequest, ServerMessage, socket_path};
use serde_json::Value; use serde_json::Value;
@ -33,13 +31,7 @@ use crate::daemon::DaemonCommand;
pub async fn serve(daemon_tx: mpsc::Sender<DaemonCommand>, events_tx: broadcast::Sender<ServerMessage>) -> Result<()> { pub async fn serve(daemon_tx: mpsc::Sender<DaemonCommand>, events_tx: broadcast::Sender<ServerMessage>) -> Result<()> {
let socket_path = socket_path()?; let socket_path = socket_path()?;
if let Some(parent) = socket_path.parent() { if let Some(parent) = socket_path.parent() {
// 0700 even if umask is loose -- this directory holds the control std::fs::create_dir_all(parent)
// socket, and XDG_RUNTIME_DIR itself is 0700 but a recreate after
// a wiped runtime dir should not inherit a world-readable mode.
std::fs::DirBuilder::new()
.recursive(true)
.mode(0o700)
.create(parent)
.with_context(|| format!("failed to create socket dir {}", parent.display()))?; .with_context(|| format!("failed to create socket dir {}", parent.display()))?;
} }
// A stale socket file from an unclean previous exit makes bind() fail // A stale socket file from an unclean previous exit makes bind() fail
@ -53,18 +45,7 @@ pub async fn serve(daemon_tx: mpsc::Sender<DaemonCommand>, events_tx: broadcast:
tracing::info!(path = %socket_path.display(), "IPC socket listening"); tracing::info!(path = %socket_path.display(), "IPC socket listening");
loop { loop {
let (stream, _addr) = match listener.accept().await { let (stream, _addr) = listener.accept().await.context("failed to accept IPC connection")?;
Ok(accepted) => accepted,
Err(e) => {
// EMFILE / a single bad accept must not take the control
// socket down for the rest of the daemon's life -- discovery
// would keep running while every `breadcast` launch reports
// "isn't running".
tracing::warn!(error = %e, "IPC accept failed, retrying");
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
continue;
}
};
let daemon_tx = daemon_tx.clone(); let daemon_tx = daemon_tx.clone();
let events_rx = events_tx.subscribe(); let events_rx = events_tx.subscribe();
tokio::spawn(async move { tokio::spawn(async move {
@ -125,14 +106,10 @@ async fn handle_connection(
continue; continue;
} }
}; };
// Handle off the read loop so a long `start_cast` (portal picker)
// does not block `stop_cast` sitting in the same socket buffer.
let daemon_tx = daemon_tx.clone();
let writer_tx = writer_tx.clone();
tokio::spawn(async move {
let response = handle_request(request, &daemon_tx).await; let response = handle_request(request, &daemon_tx).await;
let _ = writer_tx.send(response); if writer_tx.send(response).is_err() {
}); break;
}
} }
forward_task.abort(); forward_task.abort();

View file

@ -73,17 +73,8 @@ async fn main() -> anyhow::Result<()> {
tracing::info!(?device, "Cast device found"); tracing::info!(?device, "Cast device found");
bread_events::emit_device_found(&bread_client, &device.id, &device.name, &device.model, "cast"); bread_events::emit_device_found(&bread_client, &device.id, &device.name, &device.model, "cast");
known_cast_devices.insert(device.id.clone(), device.clone()); known_cast_devices.insert(device.id.clone(), device.clone());
// Forward only on an actual change. The daemon
// actor is single-threaded and `Found` fires on
// every mDNS re-resolution (per address, per
// re-browse), so forwarding identical devices too
// would spam `device_list_changed` broadcasts and
// delay `list_devices`/`start_cast` replies that
// share the same actor queue. The daemon's own
// `CastDeviceLost`→refind logic still works, since
// a loss is a state change on its side too.
let _ = daemon_tx.send(DaemonCommand::CastDeviceFound(device)).await;
} }
let _ = daemon_tx.send(DaemonCommand::CastDeviceFound(device)).await;
} }
Some(DiscoveryEvent::Lost { id }) => { Some(DiscoveryEvent::Lost { id }) => {
tracing::info!(%id, "Cast device lost"); tracing::info!(%id, "Cast device lost");
@ -108,12 +99,8 @@ async fn main() -> anyhow::Result<()> {
tracing::info!(?device, "DLNA device found"); tracing::info!(?device, "DLNA device found");
bread_events::emit_device_found(&bread_client, &device.url, &device.friendly_name, "DLNA renderer", "dlna"); bread_events::emit_device_found(&bread_client, &device.url, &device.friendly_name, "DLNA renderer", "dlna");
known_dlna_devices.insert(device.url.clone(), device.clone()); known_dlna_devices.insert(device.url.clone(), device.clone());
// Same rationale as the Cast arm above: each SSDP
// poll only reports devices it *this* poll
// confirmed, but a device re-confirmed unchanged
// needs no daemon flush.
let _ = daemon_tx.send(DaemonCommand::DlnaDeviceFound(device)).await;
} }
let _ = daemon_tx.send(DaemonCommand::DlnaDeviceFound(device)).await;
} }
Some(DlnaDiscoveryEvent::Lost { url }) => { Some(DlnaDiscoveryEvent::Lost { url }) => {
tracing::info!(%url, "DLNA device lost"); tracing::info!(%url, "DLNA device lost");

View file

@ -7,7 +7,7 @@ PartOf=graphical-session.target
[Service] [Service]
Type=simple Type=simple
ExecStart=%h/.local/bin/breadcastd ExecStart=%h/.cargo/bin/breadcastd
Restart=on-failure Restart=on-failure
RestartSec=2 RestartSec=2

View file

@ -22,14 +22,11 @@ channel claims the namespace.
## The patch ## The patch
`src/lib.rs`: added `CastDevice::send_message(&self, namespace: &str, `src/lib.rs`: added `CastDevice::send_message<M: Serialize>(&self, namespace,
destination: &str, message: &str)`, built the same way destination, message)`, built the same way `ReceiverChannel::broadcast_message()`
`ReceiverChannel::broadcast_message()` is internally, but with a is internally, but with a caller-supplied `destination` instead of a
caller-supplied `destination` instead of a hardcoded `"*"`. The payload is hardcoded `"*"`. See the doc comment on that method for the exact rationale
a raw string (already-serialized JSON from the vendored openscreen C++), (marked "LOCAL PATCH (breadcast, not upstream)").
not a `Serialize` value -- a generic `M: Serialize` would double-encode
the OFFER. See the doc comment on that method (marked "LOCAL PATCH
(breadcast, not upstream)").
## Rolling the pin ## Rolling the pin