19 Commits
Author SHA1 Message Date
molluskandClaude Opus 4.7 0c9d8eb9f9 host: emit relay-only ticket (drop direct IP candidates)
The host ticket embedded every direct IP candidate the endpoint
discovered — on this machine that was 10 addrs, 7 of them useless
Docker-bridge gateways (172.16.0.0/12) plus LAN/public v4/v6. That
bloated the ticket to ~320 chars and leaked local network topology to
whoever received it.

Keep only the endpoint id + relay URL (~140 chars). The relay
coordinates hole-punching to a direct path after connect, so peer
reachability is unchanged; the direct addrs in the ticket only ever
shaved a moment off the first connection attempt, and n0 DNS discovery
already publishes the full addr keyed by id as a backstop.

Await endpoint.online() (15s cap) before building the ticket so the
relay URL is reliably populated; a relay outage degrades to a
possibly-incomplete ticket rather than a hang.

Experimental — isolated on feat/short-ticket pending an end-to-end
cross-machine connect test before merging to main.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-23 16:17:38 -04:00
molluskandClaude Opus 4.7 7fa5d410f9 cli: drop stale "+ ffmpeg" from --help about string
The Wayland path moved from a shelled-out ffmpeg to an in-process
GStreamer pipeline back in the 2026-05-16/18 pivot, but the clap
`about` string still advertised ffmpeg. Now reads "P2P screen sharing
over iroh".

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-23 15:44:02 -04:00
molluskandClaude Opus 4.7 8674f907f2 docs: sync README status with shipped audio + repair work
Per-app audio routing (--app), mic mixing (--mic), and --repair all
landed in recent commits but the README still listed the first and last
as stubs. Move them to Working, drop them from "Not yet working" (X11
capture is now the only remaining stub), and add an Audio section
documenting --app/--mic/--repair.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-23 15:37:23 -04:00
molluskandClaude Opus 4.7 25a5b597f7 repair: unload orphan pixelpass_capture_* sinks and paired loopbacks
Replaces the Phase-2 stub. Parses `pactl list short modules` for
`module-null-sink` entries whose `sink_name=pixelpass_capture_<pid>`
names a PID with no /proc/<pid>, and `module-loopback` entries whose
`sink=` names one of those orphan sinks. Unloads loopbacks first, then
sinks (mirrors Routing::shutdown order so PipeWire doesn't leave
zombie links).

Live PIDs — including this process and any other running pixelpass —
are skipped and reported. Same-tab parser is robust to multi-line
{ ... } argument blocks from other modules because continuation lines
never parse as a u32 module ID.

Verified with synthetic orphans against this build:
  - single dead orphan (sink + loopback) → both cleaned, count = 2
  - single live orphan (pid 1) → both preserved, message names the
    live count
  - mixed dead + live → dead pair cleaned, live pair preserved,
    output reports both

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-22 16:33:33 -04:00
molluskandClaude Opus 4.7 54ebe96ca1 host/audio: oscillate loopback on stream lifecycle (session 4 of 4)
Subscribe registry.global_remove so we know when routed stream nodes
vanish; drop them from routed_node_ids and emit LastRoutedStreamGone
on the N→0 transition. Tokio side re-runs `pactl load-module
module-loopback` with the same args as start, restoring the
default-sink monitor mirror so the viewer hears system audio again
instead of going silent when the routed app exits mid-session.

FirstRoutedStream now fires on every 0→N transition (not just the
first), so the pair oscillates cleanly: each app open/close cycle
unloads → re-loads the loopback.

Verified cross-machine 2026-05-22 16:29 EDT — host with Strawberry
picked, laptop viewer over mpv with YouTube playing on host as a
control. Strawberry audible on laptop, YouTube silent (route active).
Quit Strawberry → YouTube became audible (loopback restored).
Reopened Strawberry → routed again, YouTube dropped out (loopback
unloaded). Clean Ctrl+C teardown.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-22 16:29:28 -04:00
molluskandClaude Opus 4.7 a144665f41 host/audio: per-stream routing via libpipewire (session 3 of 4)
When opts.app is set, a dedicated OS thread runs a libpipewire
MainLoop, subscribes to the registry, and writes target.object to
the "default" metadata so WirePlumber reroutes matching streams to
our per-PID null-sink. Activation is now opts.app.is_some() OR the
existing PIXELPASS_AUDIO_VIA_NULL_SINK env var (kept for
no-filter dogfooding).

Threading: tokio side spawns a std::thread; the two sides bridge via
pipewire::channel for cmd→thread (Shutdown) and tokio::sync::mpsc
for event→tokio (FirstRoutedStream). Cross-thread quit goes through
the libpipewire channel so MainLoop is only mutated from its own
thread. Shutdown clears target.object on every routed stream before
quitting so WirePlumber doesn't log orphans.

Routing decisions:
- Filter is case-insensitive equality on application.name (predictable;
  no surprise matches from substring).
- target.object is written as Spa:Id with the sink's object.serial.
- Default-sink loopback stays loaded until the first stream is
  actually routed — avoids viewer silence if the user picks an app
  that isn't producing sound yet. On first route, the event task
  takes() the loopback module ID and unloads it.

Session 2 picker explainer + (app pick saved: ...) banner softening
both removed; banner is back to plain app-audio=NAME.

Verified end-to-end cross-machine: desktop host with Strawberry
selected, laptop viewer over mpv. Strawberry audible on the laptop;
YouTube playback started on the desktop was NOT audible on the
laptop. Routing isolates the filtered app.

Session 4 still open: recreate loopback when the last filtered stream
disappears (avoid silence), handle app-disappears-mid-session,
multi-instance, --repair coupling for orphan sink cleanup.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-22 15:56:44 -04:00
molluskandClaude Opus 4.7 339a9d49e4 host/audio: app enumeration + interactive picker (session 2 of 4)
list_playing_apps() shells out to `pactl -f json list sink-inputs`,
parses with serde_json, dedupes by application.name (BTreeMap for
stable ordering), returns Vec<App { name, stream_count }>.

Picker fires in interactive::run after preflight, before host::run.
Bypassed when --app NAME is on the CLI. Shows the apps with a
"per-app routing isn't live yet" explainer so users aren't surprised
that audio still captures system-wide. Empty-list path shows the
default + a "start your app first" hint so the feature stays
discoverable.

Banner softened to `system-audio (app pick saved: <name>)` when
opts.app is set — keeps the choice visible without lying about what
gets captured. Routing activation still gated on the
PIXELPASS_AUDIO_VIA_NULL_SINK env var (session 1's locked decision
#2); --app flips to that activation in session 3 once per-stream
filtering exists.

Verified end-to-end interactively: Strawberry shows up in the picker
during music playback, both default and app-pick paths advance into
the portal handshake, banner matches choice.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-22 15:32:00 -04:00
molluskandClaude Opus 4.7 8d32ded412 host/audio: per-PID PipeWire null-sink + loopback scaffolding
Session 1 of the per-app audio routing feature. Adds host/audio.rs
with a Routing struct that owns the lifecycle of two pactl-loaded
modules: a per-PID null-sink (pixelpass_capture_<pid>) and a loopback
mirroring @DEFAULT_SINK@.monitor into it at 20ms latency. Activated
by PIXELPASS_AUDIO_VIA_NULL_SINK=1 — kept hidden behind an env var
because without per-stream filtering (session 3) the user-facing
behavior of --app foo would be identical to no flag, which would
mislead users about what the flag does.

When the env var is set, wayland::start substitutes the gst pulsesrc
device from {DEFAULT_SINK}.monitor to pixelpass_capture_<pid>.monitor;
audio still works end-to-end via the loopback. CaptureHandle owns the
Routing alongside gst and serve; teardown order is gst → audio → serve
so streams unlink from the null-sink before the sink is destroyed.

Lifecycle is via pactl shell-outs rather than pipewire-rs. Null-sink
+ loopback are one-shot graph mutations with no event subscription;
the libpipewire route would mean dragging a MainLoop thread in for no
benefit until session 3 needs stream events.

Known cosmetic: the null-sink appears in Plasma's audio mixer as a
user-facing volume slider. Pactl's sink_properties= quoting is fiddly
enough that the device.hidden=true fix is parked for a follow-up.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-22 05:11:48 -04:00
molluskandClaude Opus 4.7 9625511c65 cleanup: demote clean-disconnect warn, drop dead --low-latency flag
handle_peer's `bridge ended with error: ...` log fired at WARN every
time a viewer cleanly closed — but bridge can only end three ways
(peer-close, local-socket-close, cancellation), none of which are real
errors. Collapsed to INFO for both Ok and Err arms; the message itself
still carries any error detail.

Also removed the `--low-latency` CLI flag and its HostOpts field. It
was a placeholder for an unimplemented Phase-2/3 SRT transport, never
read anywhere, and was generating a persistent dead_code warning. If
SRT ever happens, the flag can come back fresh.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-22 03:50:20 -04:00
molluskandClaude Opus 4.7 e1a018fdf7 host/serve: extract HTTP fanout from wayland.rs
The broadcast fanout, supervisor-facing listener bind, accept loop, and
per-viewer drain were all sitting inside host/wayland.rs even though
none of it is Wayland-specific. Move them to host/serve.rs so the X11
backend can share the same serving layer with a one-line constructor
call instead of copy-pasting (and drifting on) the fanout code.

No behavior change. Wayland's CaptureHandle now wraps a serve::Serve
instead of owning the listener/reader/server fields directly; gst
pipeline construction is unchanged. connect_to_capture moves alongside
Serve since it pairs with it.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-22 02:32:35 -04:00
molluskandClaude Opus 4.7 f939441e31 README: document multi-viewer + bandwidth pre-flight
Updates the status section to move multi-viewer out of "not yet
working", adds a Configuration section pointing at the new TOML config
at ~/.config/pixelpass/config.toml, and a Multi-viewer section
covering the lazy-sticky lifecycle, the --max-viewers cap, the
bandwidth-bitrate tradeoff, and how to fit more viewers by dropping
--bitrate. Known-limitations section gains "late joiners see ~2 s of
garbage" (expected behavior) and drops the now-stale "single viewer
per host" line.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-21 17:00:58 -04:00
molluskandClaude Opus 4.7 153febe078 pre-flight: bandwidth test + persistent config
First-run host launch now offers a one-time upstream measurement
against speed.cloudflare.com/__up via ureq (~5 MB POST, ~5s). The
result lives at ~/.config/pixelpass/config.toml under [bandwidth]
and feeds the default --max-viewers calculation on subsequent runs.

Sticky semantics for the dialog:
- Unmeasured: first-run prompt (Run / Skip)
- Measured / Skipped: silent — never re-prompts
- Failed: ask again on next launch (Retry / give up → Skipped)

`pixelpass --reconfigure` re-runs the test unconditionally for users
whose connection has changed (new ISP, moved house, etc.).

--max-viewers is now Option<u32>. When unset, host startup loads the
saved measurement, runs recommended_max_viewers(safe_mbps, bitrate),
and surfaces the source in the banner: "max viewers : N (auto: X.X
Mbps measured upstream)" — or user-specified / default fallback.

User verified end-to-end on 2026-05-21 16:54 EDT: first-run dialog,
skip path, run path, --reconfigure refresh, and banner integration
all work as expected.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-21 16:55:11 -04:00
molluskandClaude Opus 4.7 ffe5a90686 multi-viewer: broadcast fanout + supervisor lifecycle
One gst capture pipeline now fans out to N concurrent viewers via a
tokio::sync::broadcast<Arc<Vec<u8>>>. The HTTP listener accepts forever;
each accepted connection spawns a sender task draining its own
broadcast::Receiver. Slow consumers see Lagged and skip ahead — MPEG-TS
resyncs at the next keyframe.

Host runtime is now lazy + sticky: a supervisor task owns the capture
handle and viewer count. First viewer triggers capture::spawn; last
viewer triggers shutdown. Subsequent reconnects re-trigger the portal
dialog as expected. --max-viewers (default 2) caps concurrent viewers;
additional connections get a "host is full" refusal and are dropped.

Banner updated to reflect the new lifecycle and viewer cap.

NOT YET RUNTIME-VERIFIED. cargo build is clean and the pipeline-level
smoke test still passes, but the multi-viewer behavior (cap enforcement,
lazy-sticky restart, concurrent fanout) requires manual end-to-end
testing with the portal dialog + multiple mpv instances.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-21 16:11:43 -04:00
molluskandClaude Opus 4.7 74b4101d4f vlc-plugin-ffmpeg: extend docs + runtime check
The previous vlc-plugin-dvb diagnosis was incomplete. On a laptop with
only vlc-plugin-dvb installed, VLC reads the MPEG-TS container, sees
the H.264 stream type in the PMT, then errors "Codec h264 ... is not
supported" because libavcodec_plugin.so is also a split package and
also wasn't pulled in by the base `vlc` install.

Installing vlc-plugin-ffmpeg (which pulls ffmpeg4.4 as a compat dep)
on the laptop made VLC play pixelpass cleanly via Intel iHD hardware
decode.

- README: list both plugin packages under requirements; rewrite the
  known-limitations line.
- interactive.rs: extend the launch-time check to also probe for
  libavcodec_plugin.so; combine both into one warning that lists
  every missing piece and the single pacman invocation to fix.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-21 06:21:32 -04:00
molluskandClaude Opus 4.7 6e4d30bfa9 vlc-plugin-dvb: document and warn at launch
VLC's MPEG-TS demuxer (libts_plugin.so) ships in a separate package on
Arch / CachyOS (vlc-plugin-dvb). Without it, VLC silently falls back
to the PS demuxer and misidentifies our H.264 stream — the symptom is
a green screen. mpv doesn't share this dependency.

- README: list vlc-plugin-dvb under requirements, replace the
  "green screen, not yet diagnosed" gotcha with the diagnosis.
- interactive.rs: when the user picks VLC, check for
  /usr/lib/vlc/plugins/demux/libts_plugin.so and print a warning to
  stderr if it's missing. Soft warning, not a hard error — VLC still
  spawns so the user can confirm the symptom for themselves.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-21 05:28:19 -04:00
molluskandClaude Opus 4.7 15766834f1 Revert "viewer HTTP: Content-Type application/octet-stream, not video/mp2t"
This reverts commit 0a253bd919.

The Content-Type change was a misdiagnosis. The real cause of VLC's
"no demux modules matched" was a missing `vlc-plugin-dvb` package on
the test machine — Arch/CachyOS ship the MPEG-TS demuxer plugin
(`libts_plugin.so`) in a separate package from `vlc`. Without it, VLC
falls through to the PS demuxer and misidentifies the H.264 stream.
With the package installed, `video/mp2t` opens cleanly.

`video/mp2t` is the correct Content-Type for an MPEG-TS stream and is
what we should be sending. Documentation of the package requirement
and a runtime check follow in a separate commit.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-21 05:26:53 -04:00
molluskandClaude Opus 4.7 0a253bd919 viewer HTTP: Content-Type application/octet-stream, not video/mp2t
VLC parses Content-Type before invoking the demuxer chain. With
video/mp2t it commits to demux="ts" by MIME alone, bypassing
byte-probing; when the ts demuxer's Open fails on the live HTTP stream
("no demux modules matched"), the input never opens. mpv probes
regardless of Content-Type.

Reproduced deterministically with a Python shim that mimics our
response headers byte-for-byte: only the Content-Type matters.
Changing it to application/octet-stream (or any non-video MIME, or
omitting the header) makes VLC fall back to byte-probing, which
finds the TS sync pattern and opens cleanly. mpv unaffected.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-20 16:30:46 -04:00
molluskandClaude Opus 4.7 3aa8d73ea0 Cargo.toml: drop ffmpeg from package description
ffmpeg was removed from the Wayland path on 2026-05-16 (commit 7b8b6bc).
The description was stale.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-20 16:07:48 -04:00
molluskandClaude Opus 4.7 8619df10d5 Add README
Covers v0.1 status, quick-start (interactive + headless), system
deps, build, architecture diagram, design rationale, and known
limitations. No README existed before — this fills the gap now that
v0.1 is verified.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-20 16:04:36 -04:00
14 changed files with 1928 additions and 161 deletions
Generated
+37
View File
@@ -3074,6 +3074,7 @@ dependencies = [
"anyhow",
"arboard",
"ashpd",
"chrono",
"clap",
"dialoguer",
"directories",
@@ -3086,8 +3087,10 @@ dependencies = [
"thiserror 2.0.18",
"tokio",
"tokio-util",
"toml",
"tracing",
"tracing-subscriber",
"ureq",
"uuid",
"x11rb",
]
@@ -4448,6 +4451,34 @@ version = "0.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1"
[[package]]
name = "ureq"
version = "3.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dea7109cdcd5864d4eeb1b58a1648dc9bf520360d7af16ec26d0a9354bafcfc0"
dependencies = [
"base64",
"log",
"percent-encoding",
"rustls",
"rustls-pki-types",
"ureq-proto",
"utf8-zero",
"webpki-roots",
]
[[package]]
name = "ureq-proto"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e994ba84b0bd1b1b0cf92878b7ef898a5c1760108fe7b6010327e274917a808c"
dependencies = [
"base64",
"http",
"httparse",
"log",
]
[[package]]
name = "url"
version = "2.5.8"
@@ -4461,6 +4492,12 @@ dependencies = [
"serde_derive",
]
[[package]]
name = "utf8-zero"
version = "0.8.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b8c0a043c9540bae7c578c88f91dda8bd82e59ae27c21baca69c8b191aaf5a6e"
[[package]]
name = "utf8_iter"
version = "1.0.4"
+4 -1
View File
@@ -2,7 +2,7 @@
name = "pixelpass"
version = "0.1.0"
edition = "2024"
description = "P2P screen sharing CLI over iroh + ffmpeg"
description = "P2P screen sharing CLI over iroh"
license = "MIT OR Apache-2.0"
publish = false
@@ -30,6 +30,9 @@ uuid = { version = "1", features = ["v4"] }
iroh-tickets = "1.0.0-rc.0"
dialoguer = { version = "0.12", default-features = false }
arboard = { version = "3", default-features = false, features = ["wayland-data-control"] }
ureq = { version = "3", default-features = false, features = ["rustls"] }
toml = "1"
chrono = { version = "0.4", default-features = false, features = ["clock", "serde"] }
[profile.release]
lto = "thin"
+240
View File
@@ -0,0 +1,240 @@
# pixelpass
P2P screen sharing CLI for Linux. Single binary, hole-punched over
[iroh](https://www.iroh.computer/) — no port forwarding, no signup, no
server-side accounts. Hardware-encoded H.264 + AAC audio, viewed in
mpv or VLC.
Built for people who just want to show their screen to a friend
without spinning up a Discord call or fighting with NAT.
## Status
**v0.1.0** — verified end-to-end on the public internet (LTE relay path,
~2s latency, real carrier-grade NAT) as of 2026-05-20.
Working:
- Wayland capture via the screencast portal (KDE Plasma 6 confirmed; other
Wayland compositors with the portal should work but are untested)
- VAAPI H.264 encode in GStreamer (RDNA3 confirmed; other VAAPI-capable
GPUs should work)
- Audio capture of the default sink's monitor, with optional per-app
routing (`--app <name>`) and microphone mixing (`--mic`)
- `--repair` cleanup of orphaned PipeWire state left by a crashed host
- iroh QUIC bi-stream tunnel, direct-UDP and relay paths both verified
- Interactive Host/View menu with clipboard auto-copy and mpv/VLC picker
- Headless mode for scripts (`pixelpass <ticket>`)
- Multi-viewer fanout (default 2, configurable via `--max-viewers`;
shared gst pipeline, one broadcast channel per host)
- First-run upstream bandwidth pre-flight, persisted to
`~/.config/pixelpass/config.toml` and used to auto-size the default
viewer cap
Not yet working:
- X11 capture (stubbed, returns an error — Phase 2 follow-up)
## Quick start
### Interactive (recommended)
```sh
pixelpass
```
On the host machine: pick "Host", share a monitor via the portal dialog,
the ticket lands on your clipboard. Send it to your viewer however you
like (chat, email, paste in a note). The same ticket works for multiple
viewers up to your `--max-viewers` cap.
The very first host launch offers a one-time upstream bandwidth test
(~5 s, ~5 MB to Cloudflare's open speed-test endpoint) so it can pick
a sensible default for the viewer cap. You can skip it and a
conservative default (2 viewers) is used; re-run it later with
`pixelpass --reconfigure`.
On the viewer machine: run `pixelpass`, pick "View", paste the ticket,
pick mpv or VLC. The player launches detached and the stream starts.
### Headless
```sh
# host: prints a ticket on stdout, waits for a peer
pixelpass
# viewer: skips the menu
pixelpass <ticket>
# then run the printed mpv command in another terminal
```
## Requirements
- Linux (Wayland session for now; X11 stubbed)
- A VAAPI-capable GPU and the right driver:
- AMD: `libva-mesa-driver`
- Intel: `intel-media-driver` (modern iGPUs) or `intel-vaapi-driver` (older)
- NVIDIA: `libva-nvidia-driver` (untested)
- `vainfo` from `libva-utils` should list at least one H.264 entrypoint
- GStreamer with these plugin packages installed:
- `gstreamer`, `gst-plugins-base`, `gst-plugins-good`, `gst-plugins-bad`,
`gst-plugins-ugly`, `gst-libav`, `gst-plugin-va`, `gst-plugin-pipewire`
- A player: `mpv` (recommended) or `vlc`
- If you use VLC, two split plugin packages are also needed on Arch-family
distros — the base `vlc` package does not pull them in:
- `vlc-plugin-dvb` — provides the MPEG-TS demuxer (`libts_plugin.so`).
Without it, VLC can't parse the container.
- `vlc-plugin-ffmpeg` — provides the H.264 decoder
(`libavcodec_plugin.so`). Without it, VLC parses the container,
identifies the codec as H.264, then errors with
`Codec h264 ... is not supported`.
mpv ships its own decoder stack and doesn't share either dependency.
- PipeWire (for screencast portal + audio capture)
On Arch / CachyOS / EndeavourOS:
```sh
sudo pacman -S gstreamer gst-plugins-base gst-plugins-good gst-plugins-bad \
gst-plugins-ugly gst-libav gst-plugin-va gst-plugin-pipewire \
libva-utils mpv
# plus your GPU's VAAPI driver
# plus, if you want to use VLC instead of mpv:
sudo pacman -S vlc vlc-plugin-dvb vlc-plugin-ffmpeg
```
If the viewer is running on battery, set the CPU governor to performance
or balanced — power-saver can choke even hardware-decoded 1080p H.264.
## Build
```sh
cargo build --release
./target/release/pixelpass --help
```
`rustc` 1.95+ / edition 2024.
## How it works
```
Host Viewer
──── ──────
Wayland portal (ashpd) ──> PipeWire fd
gst-launch: pipewiresrc -> videorate -> vah264enc ->
h264parse -> mpegtsmux
(audio: pulsesrc <sink>.monitor ->
avenc_aac -> aacparse ─┘)
│ stdout
tokio HTTP server (in-process, ~30 lines)
iroh QUIC bi-stream (ALPN pixelpass/0) ◄══════════►
tokio TcpListener
on 127.0.0.1:rand
mpv / VLC HTTP client
```
The viewer's player connects to a localhost HTTP server, which is
just one end of the iroh tunnel. The host's HTTP server sits on the
other end and streams GStreamer's stdout (an MPEG-TS containing
hardware-encoded H.264 + AAC) through with no demux or remux.
iroh handles NAT traversal: direct UDP if hole-punching succeeds,
relay path otherwise. Both have been verified end-to-end.
## Why these choices
- **iroh over Holesail / dumbpipe / Tailscale**: single Rust dep, no Node
runtime, no signup, no daemon — fits the "one self-contained binary"
goal.
- **GStreamer for capture/encode, not ffmpeg**: stride/format pitfalls
when bridging raw video between processes; one in-process pipeline
sidesteps them.
- **In-process Rust HTTP server, not ffmpeg-as-server**: ffmpeg's
`-listen 1` is one-shot and probe-budget-sensitive; the Rust task is
pure passthrough with no codec assumptions.
- **MPEG-TS over fragmented MP4**: every player on Linux handles it
out of the box. AV1-in-MPEG-TS was tried and is unworkable through
libavformat — if AV1 ever comes back, it has to ride a different
container.
- **VAAPI H.264 over x264**: ~5% of one CPU core instead of ~50% on
the host's hardware.
## Configuration
`pixelpass` keeps a small TOML config at `~/.config/pixelpass/config.toml`
(or the XDG equivalent). Right now it only stores the result of the
bandwidth pre-flight:
```toml
[bandwidth]
status = "measured" # measured | skipped | failed | unmeasured
upstream_mbps = 8.78 # safe estimate (raw * 0.8)
measured_at = "2026-05-21T20:41:16Z"
```
- `pixelpass --reconfigure` re-runs the test (e.g. after an ISP change).
- Deleting the file resets pixelpass to first-run state.
- Skip is sticky — once you skip the test, pixelpass won't ask again
unless you reconfigure.
## Audio
By default pixelpass captures the default sink's monitor — the viewer
hears whatever the host hears. Two flags adjust this:
- `--app <name>` routes only a single application's audio. pixelpass
creates a per-PID null-sink and uses libpipewire to reroute matching
`Stream/Output/Audio` nodes (by `application.name`) into it, so the
viewer hears just that app instead of the whole desktop. In the
interactive menu you can pick the app from a list of what's currently
playing.
- `--mic` mixes the default microphone source into the stream alongside
system audio.
If a host crashes mid-session it can leave orphaned `pixelpass_capture_*`
null-sinks and their paired loopbacks loaded in PipeWire. Run
`pixelpass --repair` to unload them and exit.
## Multi-viewer
One gst capture pipeline fans out to N concurrent viewers via a
`tokio::sync::broadcast` channel. The same ticket is reusable: as long
as a viewer is connected, capture stays alive; when the last one
leaves, the pipeline tears down and the portal stops streaming. A new
viewer connecting after that re-triggers the portal dialog.
Capacity is bounded by upstream bandwidth (each viewer is its own
encrypted egress). The default cap comes from the bandwidth pre-flight
result; `--max-viewers <N>` overrides it. When the cap is hit,
additional connections are politely refused with a "host is full"
message and the host keeps running.
For more viewers, drop the per-viewer bitrate: e.g. `pixelpass
--bitrate 2500 --max-viewers 4` fits four 2.5 Mbps streams in roughly
12 Mbps of upstream.
## Known limitations and gotchas
- **VLC needs `vlc-plugin-dvb` and `vlc-plugin-ffmpeg`** on Arch-family
distros — the base `vlc` package doesn't pull these in, and missing
either one breaks playback (the first kills the demuxer, the second
kills the H.264 decoder). pixelpass warns at player-launch time if
either plugin isn't on disk. mpv doesn't share these dependencies.
- **Audio echo** if the host plays the stream through speakers and
captures system audio — expected, the mic / monitor picks up the
playback. Headphones bypass it.
- **Late joiners see ~2 s of garbage** before the next keyframe lets
their decoder lock. Expected behavior, not a bug.
- **VAAPI driver must be package-tracked**, not an orphaned `.so` on
disk. mpv's `--hwdec=auto` silently falls back to software decode
otherwise, which then chokes on a low-power viewer.
## License
MIT OR Apache-2.0, your pick.
+14 -5
View File
@@ -4,7 +4,7 @@ use clap::{Parser, ValueEnum};
#[command(
name = "pixelpass",
version,
about = "P2P screen sharing over iroh + ffmpeg",
about = "P2P screen sharing over iroh",
long_about = "Run with no arguments for an interactive Host/View menu. \
Pass a ticket positionally to skip the menu and view headlessly."
)]
@@ -41,9 +41,12 @@ pub struct Cli {
#[arg(long)]
pub no_hwencode: bool,
/// Use low-latency SRT transport instead of HTTP MPEG-TS (Phase 2/3).
/// Maximum number of concurrent viewers. Additional connections are
/// politely refused with a "host full" message. Defaults to the
/// connection-aware recommendation from the bandwidth pre-flight if
/// available, otherwise 2.
#[arg(long)]
pub low_latency: bool,
pub max_viewers: Option<u32>,
// ── viewer options ────────────────────────────────────────────────
/// Local TCP port for the viewer to expose (default: random).
@@ -58,6 +61,12 @@ pub struct Cli {
/// Clean up orphaned PipeWire state from a crashed host run, then exit.
#[arg(long)]
pub repair: bool,
/// Re-run the bandwidth pre-flight test, save the result, then exit.
/// Use this if your connection has changed (new ISP, moved house, etc.)
/// or if the previously saved test result is stale.
#[arg(long)]
pub reconfigure: bool,
}
#[derive(ValueEnum, Clone, Copy, Debug)]
@@ -75,7 +84,7 @@ pub struct HostOpts {
pub bitrate: u32,
pub framerate: u32,
pub no_hwencode: bool,
pub low_latency: bool,
pub max_viewers: Option<u32>,
pub interactive: bool,
}
@@ -95,7 +104,7 @@ impl Cli {
bitrate: self.bitrate,
framerate: self.framerate,
no_hwencode: self.no_hwencode,
low_latency: self.low_latency,
max_viewers: self.max_viewers,
interactive,
}
}
+73
View File
@@ -0,0 +1,73 @@
//! One-shot upstream bandwidth measurement against Cloudflare's open
//! speed-test endpoint. POST a fixed payload, time it, derive Mbps.
//!
//! Run via `tokio::task::spawn_blocking` from async contexts — ureq is a
//! blocking client and we don't want to wedge the tokio runtime during
//! the test.
use anyhow::{Context, Result};
use std::time::{Duration, Instant};
const ENDPOINT: &str = "https://speed.cloudflare.com/__up";
const PAYLOAD_BYTES: usize = 5 * 1024 * 1024; // 5 MiB
const HTTP_TIMEOUT: Duration = Duration::from_secs(30);
/// Multiplier applied to the raw measurement. TCP slow-start, ramp-up, and
/// real-world contention all mean a one-shot upstream test slightly
/// overestimates sustainable throughput; clamp to 80% for headroom.
const SAFETY_FACTOR: f64 = 0.80;
/// Result of a successful measurement.
#[derive(Debug, Clone)]
pub struct Measurement {
/// Raw measured throughput in megabits per second.
pub raw_mbps: f64,
/// `raw_mbps * SAFETY_FACTOR` — the value to use when sizing things.
pub safe_mbps: f64,
/// How long the upload took.
pub elapsed: Duration,
}
/// Blocking upload-speed test. Call from a `spawn_blocking` task.
pub fn measure_upstream_blocking() -> Result<Measurement> {
let payload = vec![0u8; PAYLOAD_BYTES];
let agent = ureq::Agent::config_builder()
.timeout_global(Some(HTTP_TIMEOUT))
.build()
.new_agent();
let start = Instant::now();
let response = agent
.post(ENDPOINT)
.content_type("application/octet-stream")
.send(&payload[..])
.context("upload request to Cloudflare failed")?;
let elapsed = start.elapsed();
let status = response.status();
if !status.is_success() {
anyhow::bail!("Cloudflare returned HTTP {status}");
}
let bits = (PAYLOAD_BYTES as f64) * 8.0;
let seconds = elapsed.as_secs_f64().max(0.001);
let raw_mbps = bits / seconds / 1_000_000.0;
let safe_mbps = raw_mbps * SAFETY_FACTOR;
Ok(Measurement {
raw_mbps,
safe_mbps,
elapsed,
})
}
/// Convert a safe-upstream Mbps figure plus the host's per-viewer bitrate
/// (kbps for video, ignoring audio + protocol overhead which we account for
/// via SAFETY_FACTOR) into a recommended viewer count. Floors to at least 1.
pub fn recommended_max_viewers(safe_mbps: f64, bitrate_kbps: u32) -> u32 {
let per_viewer_mbps = (bitrate_kbps as f64) / 1000.0;
if per_viewer_mbps <= 0.0 {
return 1;
}
let n = (safe_mbps / per_viewer_mbps).floor();
if n < 1.0 { 1 } else { n as u32 }
}
+101
View File
@@ -0,0 +1,101 @@
//! Persistent user-level config at `~/.config/pixelpass/config.toml`.
//!
//! Right now this only tracks the bandwidth pre-flight result. Future
//! preferences (default player, default bitrate, etc.) can hang off the
//! same file under their own `[section]`.
use anyhow::{Context, Result};
use chrono::{DateTime, Utc};
use directories::ProjectDirs;
use serde::{Deserialize, Serialize};
use std::fs;
use std::io::Write;
use std::path::PathBuf;
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct Config {
#[serde(default)]
pub bandwidth: BandwidthEntry,
}
/// Result of the first-run upstream measurement.
///
/// `status = "unmeasured"` means we've never asked the user — show the
/// first-run dialog. `"measured"` means we have a number. `"skipped"`
/// means the user opted out (sticky — don't ask again). `"failed"`
/// means the last attempt errored and we should ask the user on next
/// interactive launch whether to retry.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct BandwidthEntry {
#[serde(default = "default_status")]
pub status: BandwidthStatus,
#[serde(default)]
pub upstream_mbps: Option<f64>,
#[serde(default)]
pub measured_at: Option<DateTime<Utc>>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum BandwidthStatus {
Unmeasured,
Measured,
Skipped,
Failed,
}
impl Default for BandwidthStatus {
fn default() -> Self {
Self::Unmeasured
}
}
fn default_status() -> BandwidthStatus {
BandwidthStatus::Unmeasured
}
/// Returns `~/.config/pixelpass/config.toml` (or the XDG equivalent on other
/// platforms). The parent directory is created lazily by [`save`].
pub fn config_path() -> Result<PathBuf> {
let dirs = ProjectDirs::from("", "", "pixelpass")
.context("could not locate a config directory for pixelpass")?;
Ok(dirs.config_dir().join("config.toml"))
}
/// Returns the loaded config, or a `Default` instance if the file doesn't
/// exist yet. Bubble up parse errors so we don't silently overwrite a
/// hand-edited config the user is debugging.
pub fn load() -> Result<Config> {
let path = config_path()?;
match fs::read_to_string(&path) {
Ok(s) => toml::from_str::<Config>(&s)
.with_context(|| format!("failed to parse {}", path.display())),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(Config::default()),
Err(e) => Err(e).with_context(|| format!("failed to read {}", path.display())),
}
}
/// Atomic write via tempfile-in-same-dir + rename.
pub fn save(cfg: &Config) -> Result<()> {
let path = config_path()?;
let parent = path
.parent()
.context("config path has no parent directory")?;
fs::create_dir_all(parent)
.with_context(|| format!("failed to create {}", parent.display()))?;
let serialized =
toml::to_string_pretty(cfg).context("failed to serialize config to TOML")?;
let tmp = parent.join(format!(".config.toml.tmp.{}", std::process::id()));
{
let mut f = fs::File::create(&tmp)
.with_context(|| format!("failed to create {}", tmp.display()))?;
f.write_all(serialized.as_bytes())
.with_context(|| format!("failed to write {}", tmp.display()))?;
f.sync_all().ok();
}
fs::rename(&tmp, &path)
.with_context(|| format!("failed to rename {} -> {}", tmp.display(), path.display()))?;
Ok(())
}
+2
View File
@@ -1,4 +1,6 @@
pub mod alpn;
pub mod bandwidth;
pub mod config;
pub mod deps;
pub mod display;
pub mod process;
+539
View File
@@ -0,0 +1,539 @@
//! Per-app audio routing.
//!
//! Two cooperating layers:
//!
//! - **Null-sink + loopback** (pactl shell-out): a per-PID null-sink
//! `pixelpass_capture_<pid>` plus a `module-loopback` that mirrors the
//! default sink's monitor into it. gst captures from the null-sink's
//! monitor, so the viewer hears whatever the user hears — by default.
//!
//! - **Per-stream rerouting** (libpipewire on a dedicated OS thread):
//! when [`HostOpts::app`] is set, a [`StreamRouter`] subscribes to the
//! PipeWire registry, finds `Stream/Output/Audio` nodes whose
//! `application.name` matches the filter, and writes
//! `target.object` to the "default" metadata so WirePlumber reroutes
//! them to our null-sink. Once at least one stream is actually routed,
//! the loopback is unloaded — otherwise the viewer would hear the
//! filtered audio twice (once via the routed stream, once via the
//! default-sink monitor loopback).
//!
//! pactl is the right tool for the one-shot null-sink/loopback graph
//! mutations. libpipewire is dragged in only when per-stream filtering
//! is requested, because that needs registry-event subscription.
use anyhow::{Context, Result, bail};
use std::cell::RefCell;
use std::collections::BTreeMap;
use std::process::Command;
use std::rc::Rc;
use std::sync::{Arc, Mutex};
use std::thread::JoinHandle;
use crate::cli::HostOpts;
/// Owns the pactl-loaded modules plus, when filtering is active, the
/// libpipewire stream-router thread. Drop unloads modules as a backstop;
/// prefer [`Routing::shutdown`] explicitly so failures get logged.
pub struct Routing {
sink_module: Option<u32>,
/// Shared with the event task so it can `take()` and unload on the
/// first successful route. `Routing::shutdown` unloads whatever
/// remains.
loopback_module: Arc<Mutex<Option<u32>>>,
sink_name: String,
stream_router: Option<StreamRouter>,
event_task: Option<tokio::task::JoinHandle<()>>,
}
impl Routing {
/// Create the per-PID null-sink + loopback. If `opts.app` is set,
/// also spawn the libpipewire thread that reroutes matching streams.
pub async fn start(opts: &HostOpts) -> Result<Self> {
let pid = std::process::id();
let sink_name = format!("pixelpass_capture_{pid}");
let sink_module =
load_module(&["module-null-sink", &format!("sink_name={sink_name}")])
.context("failed to load module-null-sink")?;
// 20ms loopback latency keeps the mirrored audio tight; pactl's
// default of 200ms is enough to be perceptible.
let loopback_module = load_module(&[
"module-loopback",
"source=@DEFAULT_SINK@.monitor",
&format!("sink={sink_name}"),
"latency_msec=20",
])
.context("failed to load module-loopback (null-sink will be cleaned up on Drop)")?;
tracing::info!(
sink_module,
loopback_module,
%sink_name,
"audio routing: null-sink + loopback ready"
);
let loopback_arc = Arc::new(Mutex::new(Some(loopback_module)));
let mut routing = Self {
sink_module: Some(sink_module),
loopback_module: Arc::clone(&loopback_arc),
sink_name: sink_name.clone(),
stream_router: None,
event_task: None,
};
if let Some(app) = &opts.app {
let (router, mut event_rx) = StreamRouter::spawn(app.clone(), sink_name.clone())?;
let loopback_for_task = Arc::clone(&loopback_arc);
let sink_name_for_task = sink_name.clone();
let event_task = tokio::spawn(async move {
while let Some(ev) = event_rx.recv().await {
match ev {
Event::FirstRoutedStream => {
let mid = loopback_for_task.lock().unwrap().take();
if let Some(id) = mid {
tracing::info!(
"audio routing: first stream routed → unloading default-sink loopback"
);
unload_module(id);
}
}
Event::LastRoutedStreamGone => {
// Routed app exited mid-session. Restore the
// default-sink loopback so the viewer hears
// system audio again instead of silence.
if loopback_for_task.lock().unwrap().is_some() {
continue;
}
tracing::info!(
"audio routing: last routed stream gone → restoring default-sink loopback"
);
match load_module(&[
"module-loopback",
"source=@DEFAULT_SINK@.monitor",
&format!("sink={sink_name_for_task}"),
"latency_msec=20",
]) {
Ok(id) => {
*loopback_for_task.lock().unwrap() = Some(id);
}
Err(e) => {
tracing::warn!(
"audio routing: failed to re-load loopback: {e:#}"
);
}
}
}
}
}
});
routing.stream_router = Some(router);
routing.event_task = Some(event_task);
}
Ok(routing)
}
pub fn sink_name(&self) -> &str {
&self.sink_name
}
/// Stop the stream router (if any), then unload loopback (if still
/// loaded), then unload the null-sink. Order matters: PipeWire can
/// leave zombie links if you destroy a sink with active inputs.
pub fn shutdown(mut self) {
if let Some(router) = self.stream_router.take() {
router.shutdown();
}
if let Some(task) = self.event_task.take() {
task.abort();
}
if let Some(id) = self.loopback_module.lock().unwrap().take() {
unload_module(id);
}
if let Some(id) = self.sink_module.take() {
unload_module(id);
}
}
}
impl Drop for Routing {
fn drop(&mut self) {
if let Some(router) = self.stream_router.take() {
router.shutdown();
}
if let Some(task) = self.event_task.take() {
task.abort();
}
if let Some(id) = self.loopback_module.lock().unwrap().take() {
unload_module(id);
}
if let Some(id) = self.sink_module.take() {
unload_module(id);
}
}
}
// ──────────────────────────────────────────────────────────────────────
// App enumeration (interactive picker source)
// ──────────────────────────────────────────────────────────────────────
/// One deduplicated app currently producing audio. The picker in
/// interactive mode shows these as the per-app capture choices.
#[derive(Debug, Clone)]
pub struct App {
pub name: String,
pub stream_count: u32,
}
/// Enumerate apps currently sending audio to any sink, deduplicated by
/// `application.name`. Returns an empty Vec if nothing is playing.
pub fn list_playing_apps() -> Result<Vec<App>> {
let output = Command::new("pactl")
.args(["-f", "json", "list", "sink-inputs"])
.output()
.context("failed to run `pactl -f json list sink-inputs`")?;
if !output.status.success() {
bail!(
"pactl list sink-inputs failed: {}",
String::from_utf8_lossy(&output.stderr).trim()
);
}
parse_sink_inputs(&output.stdout)
}
fn parse_sink_inputs(stdout: &[u8]) -> Result<Vec<App>> {
let entries: Vec<SinkInput> =
serde_json::from_slice(stdout).context("pactl returned unparseable JSON")?;
let mut counts: BTreeMap<String, u32> = BTreeMap::new();
for entry in entries {
let Some(name) = entry.properties.application_name else { continue };
let trimmed = name.trim();
if trimmed.is_empty() {
continue;
}
*counts.entry(trimmed.to_string()).or_insert(0) += 1;
}
Ok(counts
.into_iter()
.map(|(name, stream_count)| App { name, stream_count })
.collect())
}
#[derive(serde::Deserialize)]
struct SinkInput {
properties: SinkInputProperties,
}
#[derive(serde::Deserialize)]
struct SinkInputProperties {
#[serde(rename = "application.name")]
application_name: Option<String>,
}
// ──────────────────────────────────────────────────────────────────────
// pactl module helpers
// ──────────────────────────────────────────────────────────────────────
fn load_module(args: &[&str]) -> Result<u32> {
let output = Command::new("pactl")
.arg("load-module")
.args(args)
.output()
.context("failed to run pactl load-module")?;
if !output.status.success() {
bail!(
"pactl load-module failed: {}",
String::from_utf8_lossy(&output.stderr).trim()
);
}
let id_str = String::from_utf8(output.stdout)
.context("pactl returned non-UTF-8")?
.trim()
.to_string();
id_str
.parse::<u32>()
.with_context(|| format!("pactl returned unexpected module ID: {id_str:?}"))
}
fn unload_module(id: u32) {
let result = Command::new("pactl")
.arg("unload-module")
.arg(id.to_string())
.output();
match result {
Ok(output) if output.status.success() => {
tracing::info!(module = id, "audio routing: unloaded pactl module");
}
Ok(output) => {
tracing::warn!(
module = id,
stderr = %String::from_utf8_lossy(&output.stderr).trim(),
"audio routing: pactl unload-module exited non-zero"
);
}
Err(e) => {
tracing::warn!(
module = id,
"audio routing: failed to run pactl unload-module: {e}"
);
}
}
}
// ──────────────────────────────────────────────────────────────────────
// Per-stream routing (libpipewire thread)
// ──────────────────────────────────────────────────────────────────────
/// Command from tokio → libpipewire thread.
enum Cmd {
/// Clear `target.object` for everything we routed, then quit the
/// MainLoop so the thread joins.
Shutdown,
}
/// Event from libpipewire thread → tokio. The pair drives loopback
/// oscillation: unload on `FirstRoutedStream`, re-load on
/// `LastRoutedStreamGone`. Both fire on count-transitions (0→N and N→0
/// respectively), not on every change.
enum Event {
/// At least one stream is now routed to our sink. Receiver unloads
/// the default-sink loopback so the filtered audio isn't doubled.
FirstRoutedStream,
/// The last routed stream just disappeared (app closed, paused,
/// switched output). Receiver re-loads the default-sink loopback so
/// the viewer doesn't go silent.
LastRoutedStreamGone,
}
/// Handle to the libpipewire stream-router thread.
pub struct StreamRouter {
cmd_tx: pipewire::channel::Sender<Cmd>,
thread: Option<JoinHandle<()>>,
}
impl StreamRouter {
/// Spawn the libpipewire thread. Returns the router handle and the
/// event receiver tokio side polls.
fn spawn(
filter_name: String,
sink_name: String,
) -> Result<(Self, tokio::sync::mpsc::UnboundedReceiver<Event>)> {
let (cmd_tx, cmd_rx) = pipewire::channel::channel::<Cmd>();
let (event_tx, event_rx) = tokio::sync::mpsc::unbounded_channel::<Event>();
let thread = std::thread::Builder::new()
.name("pixelpass-pw-router".to_string())
.spawn(move || {
if let Err(e) = run_router(filter_name, sink_name, cmd_rx, event_tx) {
tracing::warn!("audio routing: libpipewire thread exited with error: {e:#}");
}
})
.context("failed to spawn libpipewire router thread")?;
Ok((
Self {
cmd_tx,
thread: Some(thread),
},
event_rx,
))
}
fn shutdown(mut self) {
// Best-effort: if the send fails the thread is already gone.
let _ = self.cmd_tx.send(Cmd::Shutdown);
if let Some(t) = self.thread.take()
&& let Err(e) = t.join()
{
tracing::warn!("audio routing: pw thread join failed: {e:?}");
}
}
}
/// Body of the libpipewire thread. Owns MainLoop, registry listener, and
/// all PipeWire proxies for the duration of the routing session.
fn run_router(
filter_name: String,
sink_name: String,
cmd_rx: pipewire::channel::Receiver<Cmd>,
event_tx: tokio::sync::mpsc::UnboundedSender<Event>,
) -> Result<()> {
use pipewire::{self as pw, types::ObjectType};
let main_loop = pw::main_loop::MainLoopRc::new(None)
.context("pw main loop construction failed")?;
let context = pw::context::ContextRc::new(&main_loop, None)
.context("pw context construction failed")?;
let core = context
.connect_rc(None)
.context("pw core connect failed (is the daemon running?)")?;
let registry = core
.get_registry_rc()
.context("pw get_registry failed")?;
let state = Rc::new(RefCell::new(RouterState {
sink_serial: None,
default_metadata: None,
routed_node_ids: Vec::new(),
pending: Vec::new(),
}));
// Cmd handler: clear metadata for routed streams, then quit.
let main_loop_for_cmd = main_loop.clone();
let state_for_cmd = Rc::clone(&state);
let _cmd_recv = cmd_rx.attach(main_loop.loop_(), move |cmd| match cmd {
Cmd::Shutdown => {
let s = state_for_cmd.borrow();
if let Some(meta) = &s.default_metadata {
for &nid in &s.routed_node_ids {
meta.set_property(nid, "target.object", None, None);
}
if !s.routed_node_ids.is_empty() {
tracing::info!(
n = s.routed_node_ids.len(),
"audio routing: cleared target.object on routed streams before quitting"
);
}
}
main_loop_for_cmd.quit();
}
});
let filter_lower = filter_name.to_ascii_lowercase();
let sink_name_owned = sink_name.clone();
let registry_weak = registry.downgrade();
let state_for_reg = Rc::clone(&state);
let event_tx_for_reg = event_tx.clone();
let state_for_remove = Rc::clone(&state);
let event_tx_for_remove = event_tx.clone();
let _reg_listener = registry
.add_listener_local()
.global(move |obj| {
let Some(reg) = registry_weak.upgrade() else { return };
match obj.type_ {
ObjectType::Node => {
let Some(props) = obj.props.as_ref() else { return };
if props.get("node.name") == Some(sink_name_owned.as_str()) {
if let Some(serial) = props
.get("object.serial")
.and_then(|s| s.parse::<u32>().ok())
{
state_for_reg.borrow_mut().sink_serial = Some(serial);
tracing::info!(
serial,
"audio routing: pixelpass sink registered"
);
try_flush(&state_for_reg, &event_tx_for_reg);
}
return;
}
if props.get("media.class") != Some("Stream/Output/Audio") {
return;
}
let Some(app) = props.get("application.name") else { return };
if !app.eq_ignore_ascii_case(&filter_lower) {
return;
}
tracing::info!(
node_id = obj.id,
%app,
"audio routing: matched stream, queued for route"
);
state_for_reg.borrow_mut().pending.push(obj.id);
try_flush(&state_for_reg, &event_tx_for_reg);
}
ObjectType::Metadata => {
let Some(props) = obj.props.as_ref() else { return };
if props.get("metadata.name") != Some("default") {
return;
}
let metadata: pw::metadata::Metadata = match reg.bind(obj) {
Ok(m) => m,
Err(e) => {
tracing::warn!("audio routing: bind default metadata failed: {e}");
return;
}
};
state_for_reg.borrow_mut().default_metadata = Some(metadata);
tracing::info!("audio routing: default metadata bound");
try_flush(&state_for_reg, &event_tx_for_reg);
}
_ => {}
}
})
.global_remove(move |id| {
handle_global_remove(&state_for_remove, &event_tx_for_remove, id);
})
.register();
tracing::info!(filter = %filter_name, "audio routing: pw thread running");
main_loop.run();
tracing::info!("audio routing: pw thread exiting");
Ok(())
}
struct RouterState {
sink_serial: Option<u32>,
default_metadata: Option<pipewire::metadata::Metadata>,
routed_node_ids: Vec<u32>,
pending: Vec<u32>,
}
/// Drop the vanished node from `routed_node_ids` and `pending`. If it
/// was the last routed stream, emit `LastRoutedStreamGone` so the
/// tokio side restores the default-sink loopback.
fn handle_global_remove(
state: &Rc<RefCell<RouterState>>,
event_tx: &tokio::sync::mpsc::UnboundedSender<Event>,
id: u32,
) {
let mut s = state.borrow_mut();
let was_routed = !s.routed_node_ids.is_empty();
s.routed_node_ids.retain(|&x| x != id);
s.pending.retain(|&x| x != id);
if was_routed && s.routed_node_ids.is_empty() {
tracing::info!(
node_id = id,
"audio routing: last routed stream disappeared"
);
let _ = event_tx.send(Event::LastRoutedStreamGone);
}
}
/// Drain pending streams to the sink, but only once both prerequisites
/// (sink serial known + default metadata bound) are in place. Emits
/// `FirstRoutedStream` when routed count crosses 0→N (so it fires
/// each time the count comes back up from zero, not just the first
/// time — pairs with `LastRoutedStreamGone` to oscillate the loopback).
fn try_flush(
state: &Rc<RefCell<RouterState>>,
event_tx: &tokio::sync::mpsc::UnboundedSender<Event>,
) {
let mut s = state.borrow_mut();
let Some(serial) = s.sink_serial else { return };
if s.default_metadata.is_none() {
return;
}
if s.pending.is_empty() {
return;
}
let was_empty = s.routed_node_ids.is_empty();
let serial_str = serial.to_string();
let pending = std::mem::take(&mut s.pending);
if let Some(meta) = &s.default_metadata {
for nid in &pending {
meta.set_property(*nid, "target.object", Some("Spa:Id"), Some(&serial_str));
tracing::info!(
node_id = *nid,
sink_serial = serial,
"audio routing: stream routed to pixelpass sink"
);
}
}
s.routed_node_ids.extend(pending);
if was_empty && !s.routed_node_ids.is_empty() {
let _ = event_tx.send(Event::FirstRoutedStream);
}
}
+247 -47
View File
@@ -1,14 +1,34 @@
pub mod audio;
mod capture;
mod serve;
mod wayland;
use anyhow::{Result, bail};
use iroh::Endpoint;
use iroh::endpoint::{Connection, presets};
use iroh::{Endpoint, EndpointAddr};
use iroh_tickets::endpoint::EndpointTicket;
use std::time::Duration;
use tokio::sync::{mpsc, oneshot};
use tokio_util::sync::CancellationToken;
use crate::cli::HostOpts;
use crate::common::{alpn::ALPN, deps, display::DisplayServer, signal};
use crate::common::{
alpn::ALPN, bandwidth, config, config::BandwidthStatus, deps, display::DisplayServer, signal,
tunnel,
};
use self::capture::CaptureHandle;
/// Messages from per-viewer tasks to the capture supervisor.
enum SupervisorMsg {
/// A new viewer wants in. Supervisor replies with the local capture
/// HTTP port to connect to, or an error string if the host is full or
/// capture spawn failed.
AddViewer(oneshot::Sender<Result<u16, String>>),
/// A viewer's session ended. Supervisor decrements the count and tears
/// down capture if it just hit zero.
RemoveViewer,
}
pub async fn run(opts: HostOpts) -> Result<()> {
let display = DisplayServer::resolve(opts.display_server);
@@ -21,6 +41,11 @@ pub async fn run(opts: HostOpts) -> Result<()> {
);
}
let resolution = resolve_max_viewers(&opts);
if resolution.value == 0 {
bail!("--max-viewers must be at least 1");
}
let cancel = signal::install_ctrl_c();
let endpoint = Endpoint::builder(presets::N0)
@@ -28,76 +53,195 @@ pub async fn run(opts: HostOpts) -> Result<()> {
.bind()
.await?;
// Relay-only ticket: wait for the home relay to connect, then keep only
// the endpoint id + relay URL and drop the direct IP candidates. The relay
// coordinates hole-punching to a direct path right after connect, so this
// doesn't change whether peers can reach each other — it just keeps the
// ticket short (~140 vs ~320 chars) and stops it from leaking LAN /
// Docker-bridge addresses to whoever receives the ticket. Awaiting online()
// first guarantees the relay URL is actually present (addr() right after
// bind can return before the relay handshake completes); the 15s cap means
// a relay outage degrades to a possibly-incomplete ticket rather than a hang
// (n0 DNS discovery still resolves the id in that case).
if tokio::time::timeout(Duration::from_secs(15), endpoint.online())
.await
.is_err()
{
tracing::warn!("home relay not connected within 15s; ticket may be incomplete");
}
let addr = endpoint.addr();
let ticket = EndpointTicket::new(addr);
let relay_only =
EndpointAddr::new(addr.id).with_addrs(addr.addrs.iter().filter(|a| a.is_relay()).cloned());
let ticket = EndpointTicket::new(relay_only);
let clipboard_ok = opts.interactive && copy_to_clipboard(&ticket.to_string());
print_host_banner(&ticket, display, &opts, clipboard_ok);
print_host_banner(&ticket, display, &opts, &resolution, clipboard_ok);
let result = accept_loop(&endpoint, display, &opts, cancel.clone()).await;
let (sup_tx, sup_rx) = mpsc::channel::<SupervisorMsg>(16);
let supervisor = tokio::spawn(supervise(opts.clone(), display, resolution.value, sup_rx));
accept_loop(&endpoint, sup_tx.clone(), cancel.clone()).await;
drop(sup_tx);
let _ = supervisor.await;
endpoint.close().await;
result
Ok(())
}
async fn accept_loop(
endpoint: &Endpoint,
display: DisplayServer,
opts: &HostOpts,
sup_tx: mpsc::Sender<SupervisorMsg>,
cancel: CancellationToken,
) -> Result<()> {
tokio::select! {
_ = cancel.cancelled() => {
tracing::info!("cancellation requested before any peer connected");
Ok(())
}
accepted = endpoint.accept() => {
let Some(incoming) = accepted else {
bail!("endpoint stopped accepting connections");
};
let conn = incoming.await?;
let remote = conn.remote_id();
tracing::info!(%remote, "peer connected");
eprintln!("\n[pixelpass] peer connected: {remote}\n");
handle_peer(conn, display, opts, cancel).await
) {
loop {
tokio::select! {
_ = cancel.cancelled() => {
tracing::info!("cancellation requested — closing accept loop");
return;
}
accepted = endpoint.accept() => {
let Some(incoming) = accepted else {
tracing::info!("endpoint stopped accepting connections");
return;
};
let conn = match incoming.await {
Ok(c) => c,
Err(e) => {
tracing::warn!("incoming connection failed: {e:#}");
continue;
}
};
let sup_tx = sup_tx.clone();
let cancel = cancel.clone();
tokio::spawn(handle_peer(conn, sup_tx, cancel));
}
}
}
}
async fn handle_peer(
conn: Connection,
display: DisplayServer,
opts: &HostOpts,
sup_tx: mpsc::Sender<SupervisorMsg>,
cancel: CancellationToken,
) -> Result<()> {
let (quic_send, quic_recv) = conn.accept_bi().await?;
) {
let remote = conn.remote_id();
let capture_handle = capture::spawn(display, opts).await?;
let port = capture_handle.local_port();
let tcp = wayland::connect_to_capture(port, std::time::Duration::from_secs(5)).await?;
let bridge = crate::common::tunnel::bridge(quic_send, quic_recv, tcp);
tokio::select! {
res = bridge => {
if let Err(e) = res {
tracing::warn!("bridge ended with error: {e:#}");
} else {
tracing::info!("bridge closed cleanly");
}
let (reply_tx, reply_rx) = oneshot::channel();
if sup_tx.send(SupervisorMsg::AddViewer(reply_tx)).await.is_err() {
tracing::warn!(%remote, "supervisor channel closed; dropping peer");
return;
}
let port = match reply_rx.await {
Ok(Ok(p)) => p,
Ok(Err(reason)) => {
tracing::warn!(%remote, %reason, "refusing viewer");
eprintln!("[pixelpass] refusing viewer {remote}: {reason}");
return;
}
Err(_) => {
tracing::warn!(%remote, "supervisor reply dropped; dropping peer");
return;
}
};
let (quic_send, quic_recv) = match conn.accept_bi().await {
Ok(s) => s,
Err(e) => {
tracing::warn!(%remote, "accept_bi failed: {e:#}");
let _ = sup_tx.send(SupervisorMsg::RemoveViewer).await;
return;
}
};
eprintln!("[pixelpass] viewer connected: {remote}");
let tcp = match serve::connect_to_capture(port, Duration::from_secs(5)).await {
Ok(t) => t,
Err(e) => {
tracing::warn!(%remote, "connect_to_capture failed: {e:#}");
let _ = sup_tx.send(SupervisorMsg::RemoveViewer).await;
return;
}
};
let bridge = tunnel::bridge(quic_send, quic_recv, tcp);
tokio::select! {
res = bridge => match res {
Ok(()) => tracing::info!(%remote, "bridge closed cleanly"),
Err(e) => tracing::info!(%remote, "bridge ended: {e:#}"),
},
_ = cancel.cancelled() => {
tracing::info!("cancellation requested during stream");
tracing::info!(%remote, "cancellation during stream");
}
}
capture_handle.shutdown().await;
Ok(())
eprintln!("[pixelpass] viewer disconnected: {remote}");
let _ = sup_tx.send(SupervisorMsg::RemoveViewer).await;
}
/// Owns the single shared CaptureHandle and the active viewer count. Spawns
/// capture lazily on the first AddViewer; tears it down when the count drops
/// back to zero. Enforces the max-viewers cap by refusing AddViewer when
/// the count is already at the cap.
async fn supervise(
opts: HostOpts,
display: DisplayServer,
max_viewers: u32,
mut rx: mpsc::Receiver<SupervisorMsg>,
) {
let mut handle: Option<CaptureHandle> = None;
let mut count: u32 = 0;
while let Some(msg) = rx.recv().await {
match msg {
SupervisorMsg::AddViewer(reply) => {
if count >= max_viewers {
let _ = reply.send(Err(format!(
"host is full ({count} of {max_viewers} viewers connected)"
)));
continue;
}
if handle.is_none() {
tracing::info!("first viewer arriving — spawning capture");
match capture::spawn(display, &opts).await {
Ok(h) => handle = Some(h),
Err(e) => {
let _ = reply.send(Err(format!("capture spawn failed: {e:#}")));
continue;
}
}
}
let port = handle.as_ref().expect("handle was just set").local_port();
count += 1;
let _ = reply.send(Ok(port));
tracing::info!(active = count, cap = max_viewers, "viewer joined");
}
SupervisorMsg::RemoveViewer => {
count = count.saturating_sub(1);
tracing::info!(active = count, cap = max_viewers, "viewer left");
if count == 0
&& let Some(h) = handle.take()
{
tracing::info!("last viewer left — tearing down capture");
h.shutdown().await;
}
}
}
}
if let Some(h) = handle.take() {
tracing::info!("host shutdown — tearing down capture");
h.shutdown().await;
}
}
fn print_host_banner(
ticket: &EndpointTicket,
display: DisplayServer,
opts: &HostOpts,
resolution: &MaxViewersResolution,
clipboard_ok: bool,
) {
eprintln!();
@@ -106,23 +250,79 @@ fn print_host_banner(
eprintln!("│ capture : {}", capture_summary(opts));
eprintln!("│ bitrate / fps : {} kbps @ {} fps", opts.bitrate, opts.framerate);
eprintln!("│ hw encode : {}", if opts.no_hwencode { "off" } else { "auto (VAAPI if available)" });
eprintln!("│ max viewers : {} ({})", resolution.value, resolution.source.label());
eprintln!("");
if clipboard_ok {
eprintln!("│ Your share code has been copied to your clipboard.");
eprintln!("│ Send it to your viewer. (If clipboard didn't work, the");
eprintln!("│ Send it to your viewer(s). (If clipboard didn't work, the");
eprintln!("│ code is also shown below for manual copy.)");
} else {
eprintln!("│ Share this ticket with your viewer:");
eprintln!("│ Share this ticket with your viewer(s):");
}
eprintln!("");
eprintln!("│ pixelpass {ticket}");
eprintln!("");
eprintln!("│ Capture will not start until the viewer connects.");
eprintln!("Press Ctrl+C to stop.");
eprintln!("│ Capture starts when the first viewer connects, runs while");
eprintln!("any viewer is connected, and tears down when the last one");
eprintln!("│ leaves. Press Ctrl+C to stop the host entirely.");
eprintln!("└────────────────────────────────────────────────────────────");
eprintln!();
}
/// How we arrived at the final viewer cap. Surfaced in the banner so the
/// user can tell at a glance whether the number is what they specified,
/// what their measured upstream supports, or just the fallback default.
struct MaxViewersResolution {
value: u32,
source: MaxViewersSource,
}
enum MaxViewersSource {
/// User passed --max-viewers explicitly.
UserFlag,
/// Derived from the saved bandwidth measurement.
BandwidthMeasurement { safe_mbps: f64 },
/// No flag, no measurement — falling back.
DefaultFallback,
}
impl MaxViewersSource {
fn label(&self) -> String {
match self {
MaxViewersSource::UserFlag => "user-specified".to_string(),
MaxViewersSource::BandwidthMeasurement { safe_mbps } => {
format!("auto: {safe_mbps:.1} Mbps measured upstream")
}
MaxViewersSource::DefaultFallback => {
"default — run `pixelpass --reconfigure` for a connection-aware value".to_string()
}
}
}
}
fn resolve_max_viewers(opts: &HostOpts) -> MaxViewersResolution {
if let Some(n) = opts.max_viewers {
return MaxViewersResolution {
value: n,
source: MaxViewersSource::UserFlag,
};
}
if let Ok(cfg) = config::load()
&& cfg.bandwidth.status == BandwidthStatus::Measured
&& let Some(upstream) = cfg.bandwidth.upstream_mbps
{
let n = bandwidth::recommended_max_viewers(upstream, opts.bitrate);
return MaxViewersResolution {
value: n,
source: MaxViewersSource::BandwidthMeasurement { safe_mbps: upstream },
};
}
MaxViewersResolution {
value: 2,
source: MaxViewersSource::DefaultFallback,
}
}
fn copy_to_clipboard(text: &str) -> bool {
match arboard::Clipboard::new().and_then(|mut cb| cb.set_text(text.to_owned())) {
Ok(()) => true,
+191
View File
@@ -0,0 +1,191 @@
//! Display-server-agnostic serving layer: takes a capture child's stdout
//! producing MPEG-TS bytes and fans them out to N concurrent HTTP viewers
//! on a localhost port. One reader task pumps stdout chunks into a
//! tokio::sync::broadcast channel; the accept loop spawns one drain task
//! per accepted TCP connection. Slow consumers see Lagged and skip ahead;
//! MPEG-TS resyncs at the next keyframe.
//!
//! Backends (host/wayland.rs, future host/x11.rs) build their own gst
//! pipeline and hand the resulting ChildStdout to [`Serve::bind`].
use anyhow::{Context, Result, bail};
use std::sync::Arc;
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::process::ChildStdout;
use tokio::sync::broadcast;
use tokio::task::JoinHandle;
use tokio::time::{Instant, sleep};
/// Broadcast-channel capacity in chunks. Each chunk is up to 64 KiB from
/// the capture child's stdout, so 16 chunks ≈ 1 MiB ≈ ~2 s of buffered
/// jitter at typical bitrates. A viewer that falls behind by more than
/// this gets Lagged and skips ahead — MPEG-TS recovers at the next
/// keyframe.
const FANOUT_CAPACITY: usize = 16;
/// Size of each chunk read from the capture child's stdout.
const READ_CHUNK: usize = 64 * 1024;
/// Owns the localhost HTTP listener and the two long-running tasks that
/// pump bytes from a capture child to all connected viewers.
pub struct Serve {
port: u16,
reader: Option<JoinHandle<()>>,
server: Option<JoinHandle<()>>,
}
impl Serve {
/// Bind a localhost listener on a random port, set up the broadcast
/// fanout, and spawn the reader + accept-loop tasks. The provided
/// `stdout` is assumed to produce MPEG-TS bytes.
pub async fn bind(stdout: ChildStdout) -> Result<Self> {
let listener = TcpListener::bind("127.0.0.1:0")
.await
.context("could not bind local capture HTTP listener")?;
let port = listener.local_addr()?.port();
let (tx, _) = broadcast::channel::<Arc<Vec<u8>>>(FANOUT_CAPACITY);
let reader = tokio::spawn(pump_to_broadcast(stdout, tx.clone()));
let server = tokio::spawn(run_accept_loop(listener, tx));
Ok(Self {
port,
reader: Some(reader),
server: Some(server),
})
}
pub fn local_port(&self) -> u16 {
self.port
}
/// Abort the reader and accept-loop tasks. Backends typically call this
/// after killing their capture child so the reader sees stdout EOF and
/// exits on its own; the abort is a backstop.
pub async fn shutdown(mut self) {
if let Some(task) = self.reader.take() {
task.abort();
}
if let Some(task) = self.server.take() {
task.abort();
}
}
}
impl Drop for Serve {
fn drop(&mut self) {
if let Some(task) = self.reader.as_ref() {
task.abort();
}
if let Some(task) = self.server.as_ref() {
task.abort();
}
}
}
/// Connect to the local capture HTTP listener, retrying until it's up or
/// we time out. Returns the connected socket — the bridge layer pipes
/// QUIC↔this socket once it's open.
pub async fn connect_to_capture(port: u16, max_wait: Duration) -> Result<TcpStream> {
let deadline = Instant::now() + max_wait;
loop {
match TcpStream::connect(("127.0.0.1", port)).await {
Ok(stream) => return Ok(stream),
Err(_) if Instant::now() < deadline => {
sleep(Duration::from_millis(50)).await;
}
Err(e) => bail!("capture HTTP listener never came up on 127.0.0.1:{port}: {e}"),
}
}
}
/// Read the capture child's stdout in chunks and broadcast each to all
/// current subscribers. `broadcast::send` returns Err when there are no
/// receivers; we ignore it so the capture child isn't backpressured
/// waiting for a viewer.
async fn pump_to_broadcast(mut stdout: ChildStdout, tx: broadcast::Sender<Arc<Vec<u8>>>) {
let mut buf = vec![0u8; READ_CHUNK];
loop {
match stdout.read(&mut buf).await {
Ok(0) => {
tracing::info!("capture stdout EOF — fanout reader exiting");
return;
}
Ok(n) => {
let chunk = Arc::new(buf[..n].to_vec());
let _ = tx.send(chunk);
}
Err(e) => {
tracing::warn!("capture stdout read error: {e}");
return;
}
}
}
}
async fn run_accept_loop(listener: TcpListener, tx: broadcast::Sender<Arc<Vec<u8>>>) {
loop {
let sock = match listener.accept().await {
Ok((s, _)) => s,
Err(e) => {
tracing::warn!("capture HTTP accept failed: {e}");
return;
}
};
let rx = tx.subscribe();
tokio::spawn(serve_one_viewer(sock, rx));
}
}
async fn serve_one_viewer(mut sock: TcpStream, mut rx: broadcast::Receiver<Arc<Vec<u8>>>) {
if !drain_http_request(&mut sock).await {
return;
}
const RESPONSE: &[u8] = b"HTTP/1.1 200 OK\r\n\
Content-Type: video/mp2t\r\n\
Cache-Control: no-cache, no-store\r\n\
Connection: close\r\n\
\r\n";
if sock.write_all(RESPONSE).await.is_err() {
return;
}
loop {
match rx.recv().await {
Ok(chunk) => {
if sock.write_all(&chunk).await.is_err() {
return;
}
}
Err(broadcast::error::RecvError::Lagged(skipped)) => {
tracing::warn!(
skipped,
"viewer fanout lagged — MPEG-TS will resync at next keyframe"
);
continue;
}
Err(broadcast::error::RecvError::Closed) => return,
}
}
}
async fn drain_http_request(sock: &mut TcpStream) -> bool {
let mut buf = [0u8; 1024];
let mut total = Vec::with_capacity(512);
loop {
match sock.read(&mut buf).await {
Ok(0) => return false,
Ok(n) => total.extend_from_slice(&buf[..n]),
Err(_) => return false,
}
if total.windows(4).any(|w| w == b"\r\n\r\n") {
return true;
}
if total.len() > 16 * 1024 {
return false;
}
}
}
+53 -96
View File
@@ -1,7 +1,6 @@
//! Wayland capture: ashpd ScreenCast portal → PipeWire fd → gst-launch
//! pipewiresrc → MPEG-TS on gst stdout → in-process HTTP server bound on a
//! random localhost port. The host bridge TCP-connects to that server and
//! pumps bytes to QUIC.
//! Wayland capture: ashpd ScreenCast portal → PipeWire fd → gst-launch.
//! Builds the gst pipeline that produces MPEG-TS on stdout, then hands
//! that stdout to [`super::serve::Serve`] which handles the HTTP fanout.
use anyhow::{Context, Result, bail};
use ashpd::{
@@ -17,28 +16,31 @@ use nix::unistd::{Pid, close};
use std::os::fd::{AsFd, IntoRawFd, OwnedFd, RawFd};
use std::process::Stdio;
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::process::{Child, ChildStdout, Command};
use tokio::task::JoinHandle;
use tokio::time::{Instant, sleep, timeout};
use tokio::process::{Child, Command};
use tokio::time::timeout;
use super::audio::Routing;
use super::serve::Serve;
use crate::cli::HostOpts;
pub struct CaptureHandle {
port: u16,
gst: Option<Child>,
server: Option<JoinHandle<()>>,
audio: Option<Routing>,
serve: Option<Serve>,
}
impl CaptureHandle {
pub fn local_port(&self) -> u16 {
self.port
self.serve
.as_ref()
.expect("serve is always Some until shutdown")
.local_port()
}
/// Graceful teardown: SIGTERM gst, give it ~1s to exit, then SIGKILL, then
/// abort the HTTP server task. Call this before dropping; Drop only fires
/// the kill backstop.
/// Graceful teardown: SIGTERM gst, give it ~1s to exit, then SIGKILL,
/// unload audio routing (if any), then tear down the serve layer.
/// The serve reader will see EOF on gst stdout and exit on its own;
/// serve.shutdown() is the backstop.
pub async fn shutdown(mut self) {
if let Some(child) = self.gst.as_mut()
&& let Some(pid) = child.id()
@@ -49,8 +51,11 @@ impl CaptureHandle {
let _ = timeout(Duration::from_millis(1000), child.wait()).await;
let _ = child.start_kill();
}
if let Some(task) = self.server.take() {
task.abort();
if let Some(audio) = self.audio.take() {
audio.shutdown();
}
if let Some(serve) = self.serve.take() {
serve.shutdown().await;
}
}
}
@@ -60,9 +65,7 @@ impl Drop for CaptureHandle {
if let Some(child) = self.gst.as_mut() {
let _ = child.start_kill();
}
if let Some(task) = self.server.as_ref() {
task.abort();
}
// Routing's and Serve's own Drop impls handle the rest.
}
}
@@ -109,20 +112,35 @@ pub async fn start(opts: &HostOpts) -> Result<CaptureHandle> {
clear_cloexec(&pw_fd)?;
let raw_fd: RawFd = pw_fd.into_raw_fd();
// 2. Bind the in-process HTTP listener on a random localhost port.
let listener = TcpListener::bind("127.0.0.1:0")
.await
.context("could not bind local capture HTTP listener")?;
let port = listener.local_addr()?.port();
// 3. Spawn gst-launch with the full pipeline: video AND audio captured,
// 2. Spawn gst-launch with the full pipeline: video AND audio captured,
// encoded, and muxed into MPEG-TS inside gst. Output goes to stdout,
// which we pipe straight to our HTTP server task — no demux/remux,
// which the serve layer pipes to its HTTP fanout — no demux/remux,
// no codec assumptions.
let key_interval = (opts.framerate * 2).to_string();
let bitrate = opts.bitrate.to_string();
let audio_monitor = default_audio_monitor().await?;
let audio_device = format!("device={audio_monitor}");
// Audio routing activates when either:
// - `opts.app` is set (per-stream rerouting to a per-PID null-sink),
// - or `PIXELPASS_AUDIO_VIA_NULL_SINK=1` is set (no app filter, just
// captures everything via the null-sink → useful for development
// and dogfooding the loopback path before app filtering is picked).
let routing_requested =
opts.app.is_some() || std::env::var_os("PIXELPASS_AUDIO_VIA_NULL_SINK").is_some();
let audio_routing = if routing_requested {
Some(
Routing::start(opts)
.await
.context("audio routing setup failed")?,
)
} else {
None
};
let audio_device = if let Some(r) = &audio_routing {
format!("device={}.monitor", r.sink_name())
} else {
let default = default_audio_monitor().await?;
format!("device={default}")
};
let mut gst_cmd = Command::new("gst-launch-1.0");
gst_cmd
.args([
@@ -199,62 +217,17 @@ pub async fn start(opts: &HostOpts) -> Result<CaptureHandle> {
.take()
.context("gst-launch-1.0 stdout pipe unavailable")?;
// 4. Spawn the HTTP server task. It owns the listener + gst stdout: it
// accepts one client (the host's bridge socket via connect_to_capture),
// drains the HTTP request, writes a fixed MPEG-TS response, then
// copies gst stdout to the socket forever.
let server = tokio::spawn(serve_capture(listener, gst_stdout));
// 3. Hand stdout to the serve layer, which binds the localhost HTTP
// listener and runs the broadcast fanout.
let serve = Serve::bind(gst_stdout).await?;
Ok(CaptureHandle {
port,
gst: Some(gst),
server: Some(server),
audio: audio_routing,
serve: Some(serve),
})
}
async fn serve_capture(listener: TcpListener, mut gst_stdout: ChildStdout) {
let mut sock = match listener.accept().await {
Ok((s, _)) => s,
Err(e) => {
tracing::warn!("capture HTTP accept failed: {e}");
return;
}
};
if !drain_http_request(&mut sock).await {
return;
}
const RESPONSE: &[u8] = b"HTTP/1.1 200 OK\r\n\
Content-Type: video/mp2t\r\n\
Cache-Control: no-cache, no-store\r\n\
Connection: close\r\n\
\r\n";
if sock.write_all(RESPONSE).await.is_err() {
return;
}
let _ = tokio::io::copy(&mut gst_stdout, &mut sock).await;
}
async fn drain_http_request(sock: &mut TcpStream) -> bool {
let mut buf = [0u8; 1024];
let mut total = Vec::with_capacity(512);
loop {
match sock.read(&mut buf).await {
Ok(0) => return false,
Ok(n) => total.extend_from_slice(&buf[..n]),
Err(_) => return false,
}
if total.windows(4).any(|w| w == b"\r\n\r\n") {
return true;
}
if total.len() > 16 * 1024 {
return false;
}
}
}
fn clear_cloexec(fd: &impl AsFd) -> Result<()> {
let flags_int = fcntl(fd.as_fd(), FcntlArg::F_GETFD).context("F_GETFD on pipewire fd")?;
let mut flags = FdFlag::from_bits_truncate(flags_int);
@@ -263,22 +236,6 @@ fn clear_cloexec(fd: &impl AsFd) -> Result<()> {
Ok(())
}
/// Connect to the in-process capture HTTP listener, retrying until it's up or
/// we time out. Returns the connected socket — the listener accepts exactly
/// one connection (the bridge socket), so this stream IS the bridge socket.
pub async fn connect_to_capture(port: u16, max_wait: Duration) -> Result<TcpStream> {
let deadline = Instant::now() + max_wait;
loop {
match TcpStream::connect(("127.0.0.1", port)).await {
Ok(stream) => return Ok(stream),
Err(_) if Instant::now() < deadline => {
sleep(Duration::from_millis(50)).await;
}
Err(e) => bail!("capture HTTP listener never came up on 127.0.0.1:{port}: {e}"),
}
}
}
async fn default_audio_monitor() -> Result<String> {
let output = Command::new("pactl")
.arg("get-default-sink")
+227 -5
View File
@@ -4,6 +4,7 @@ use iroh_tickets::endpoint::EndpointTicket;
use std::str::FromStr;
use crate::cli::Cli;
use crate::common::{bandwidth, config};
use crate::{host, viewer};
pub async fn run(cli: Cli) -> Result<()> {
@@ -20,7 +21,14 @@ pub async fn run(cli: Cli) -> Result<()> {
.interact()?;
match choice {
0 => host::run(cli.into_host_opts(true)).await,
0 => {
preflight_if_needed(&theme).await;
let mut cli = cli;
if cli.app.is_none() {
cli.app = pick_app(&theme)?;
}
host::run(cli.into_host_opts(true)).await
}
_ => {
let ticket = prompt_ticket(&theme)?;
viewer::run(ticket, cli.into_viewer_opts(true)).await
@@ -28,6 +36,182 @@ pub async fn run(cli: Cli) -> Result<()> {
}
}
/// Picker for the per-app audio capture choice. Lists apps currently
/// producing audio (deduped by `application.name`); user picks one or
/// the "all system audio" default. Bypassed when `--app NAME` was given
/// on the CLI.
fn pick_app(theme: &ColorfulTheme) -> Result<Option<String>> {
let apps = match host::audio::list_playing_apps() {
Ok(a) => a,
Err(e) => {
tracing::warn!("could not enumerate playing apps: {e:#}");
return Ok(None);
}
};
eprintln!();
eprintln!("Audio capture");
eprintln!("─────────────");
if apps.is_empty() {
eprintln!("No other apps are currently producing audio.");
eprintln!("Start your game / music / call first if you want to pick it specifically.");
}
eprintln!();
let mut items = vec!["Capture all system audio (default)".to_string()];
for app in &apps {
items.push(if app.stream_count == 1 {
app.name.clone()
} else {
format!("{} ({} streams)", app.name, app.stream_count)
});
}
let choice = Select::with_theme(theme)
.with_prompt("What audio should the viewer hear?")
.items(&items)
.default(0)
.interact()?;
if choice == 0 {
Ok(None)
} else {
Ok(Some(apps[choice - 1].name.clone()))
}
}
/// `pixelpass --reconfigure` entry point: unconditionally re-run the
/// bandwidth pre-flight test, save the result, and return. Used to
/// refresh a stale measurement (e.g. user moved house, changed ISP).
pub async fn run_reconfigure() -> Result<()> {
eprintln!();
eprintln!("Re-running bandwidth pre-flight test…");
let mut cfg = config::load().unwrap_or_default();
run_bandwidth_test(&mut cfg).await;
Ok(())
}
/// First-run pre-flight gate. Called once, when the user picks "Host" in
/// the interactive menu. Behavior by saved status:
/// - Unmeasured (first ever launch): explain + offer Run / Skip
/// - Failed (previous attempt errored): offer Retry / give-up-and-skip
/// - Measured or Skipped: silent — never re-prompts
async fn preflight_if_needed(theme: &ColorfulTheme) {
let mut cfg = config::load().unwrap_or_default();
match cfg.bandwidth.status {
config::BandwidthStatus::Measured | config::BandwidthStatus::Skipped => return,
config::BandwidthStatus::Unmeasured => {
eprintln!();
eprintln!("First-time setup");
eprintln!("────────────────");
eprintln!("PixelPass can measure your upload speed to recommend a safe");
eprintln!("default for how many viewers your connection can handle.");
eprintln!("The test takes about 5 seconds and uploads ~5 MB to");
eprintln!("Cloudflare's open speed-test endpoint.");
eprintln!();
eprintln!("If you skip, a conservative default (2 viewers) is used.");
eprintln!("You can run the test later with `pixelpass --reconfigure`.");
eprintln!();
let Ok(choice) = Select::with_theme(theme)
.with_prompt("What would you like to do?")
.items(&[
"Run the bandwidth test (recommended)",
"Skip — use the conservative default",
])
.default(0)
.interact()
else {
return;
};
if choice == 1 {
cfg.bandwidth = config::BandwidthEntry {
status: config::BandwidthStatus::Skipped,
upstream_mbps: None,
measured_at: None,
};
let _ = config::save(&cfg);
eprintln!("Pre-flight skipped.");
return;
}
run_bandwidth_test(&mut cfg).await;
}
config::BandwidthStatus::Failed => {
eprintln!();
let Ok(choice) = Select::with_theme(theme)
.with_prompt("Last bandwidth test failed. Try again?")
.items(&[
"Yes — retry now",
"No — use the conservative default",
])
.default(0)
.interact()
else {
return;
};
if choice == 1 {
cfg.bandwidth = config::BandwidthEntry {
status: config::BandwidthStatus::Skipped,
upstream_mbps: None,
measured_at: None,
};
let _ = config::save(&cfg);
eprintln!("OK — using the conservative default.");
return;
}
run_bandwidth_test(&mut cfg).await;
}
}
}
async fn run_bandwidth_test(cfg: &mut config::Config) {
eprintln!();
eprintln!("Measuring upstream…");
let result = tokio::task::spawn_blocking(bandwidth::measure_upstream_blocking).await;
let measurement = match result {
Ok(Ok(m)) => m,
Ok(Err(e)) => {
eprintln!("Test failed: {e:#}");
eprintln!("Marking as failed — you'll be asked again on next launch.");
cfg.bandwidth = config::BandwidthEntry {
status: config::BandwidthStatus::Failed,
upstream_mbps: None,
measured_at: None,
};
let _ = config::save(cfg);
return;
}
Err(join_err) => {
eprintln!("Test task panicked: {join_err}");
cfg.bandwidth = config::BandwidthEntry {
status: config::BandwidthStatus::Failed,
upstream_mbps: None,
measured_at: None,
};
let _ = config::save(cfg);
return;
}
};
eprintln!(
"Measured {:.2} Mbps up (safe estimate {:.2} Mbps, took {:.1}s).",
measurement.raw_mbps,
measurement.safe_mbps,
measurement.elapsed.as_secs_f64()
);
cfg.bandwidth = config::BandwidthEntry {
status: config::BandwidthStatus::Measured,
upstream_mbps: Some(measurement.safe_mbps),
measured_at: Some(chrono::Utc::now()),
};
if let Err(e) = config::save(cfg) {
eprintln!("Warning: failed to save result: {e:#}");
}
}
fn print_welcome() {
eprintln!();
eprintln!("Welcome to PixelPass.");
@@ -68,14 +252,52 @@ impl Player {
url,
],
),
Player::Vlc => crate::common::process::spawn_detached(
"vlc",
&["--network-caching=200", "--live-caching=200", url],
),
Player::Vlc => {
warn_if_vlc_plugins_missing();
crate::common::process::spawn_detached(
"vlc",
&["--network-caching=200", "--live-caching=200", url],
)
}
}
}
}
// On Arch-family distros, the base `vlc` package omits two plugins
// pixelpass needs: the MPEG-TS demuxer (`vlc-plugin-dvb`) and the
// libavcodec-based H.264 decoder (`vlc-plugin-ffmpeg`). Missing either
// produces a confusing error chain — warn at launch.
fn warn_if_vlc_plugins_missing() {
const REQUIRED: &[(&str, &str)] = &[
(
"/usr/lib/vlc/plugins/demux/libts_plugin.so",
"vlc-plugin-dvb",
),
(
"/usr/lib/vlc/plugins/codec/libavcodec_plugin.so",
"vlc-plugin-ffmpeg",
),
];
let missing: Vec<&(&str, &str)> = REQUIRED
.iter()
.filter(|(p, _)| !std::path::Path::new(p).exists())
.collect();
if missing.is_empty() {
return;
}
eprintln!();
eprintln!("Warning: VLC is missing plugins pixelpass needs:");
for (path, pkg) in &missing {
eprintln!(" - {path} (install `{pkg}`)");
}
eprintln!("On Arch / CachyOS / EndeavourOS: `sudo pacman -S {}`.", {
let names: Vec<&str> = missing.iter().map(|(_, p)| *p).collect();
names.join(" ")
});
eprintln!("mpv is unaffected.");
eprintln!();
}
pub fn prompt_player() -> Result<Player> {
let theme = ColorfulTheme::default();
let choice = Select::with_theme(&theme)
+8
View File
@@ -16,10 +16,18 @@ async fn main() -> Result<()> {
let cli = Cli::parse();
init_tracing(cli.verbose);
// libpipewire requires global init before any pw_* call. Idempotent;
// safe to call even when the per-app audio thread never spawns.
pipewire::init();
if cli.repair {
return repair::run().await;
}
if cli.reconfigure {
return interactive::run_reconfigure().await;
}
match cli.ticket.as_deref() {
Some(s) => {
let ticket: EndpointTicket = s.parse().map_err(|e| {
+192 -7
View File
@@ -1,12 +1,197 @@
//! `--repair`: clean up any null sinks / loopbacks that a crashed pixelpass
//! host left behind. Phase 2 will scan PipeWire for nodes tagged with the
//! `pixelpass.session = <uuid>` property and destroy them.
//! `--repair`: clean up null-sinks and loopbacks left behind by a crashed
//! pixelpass host. Identifies orphans by the `pixelpass_capture_<pid>`
//! name pattern + dead-PID check, then unloads paired loopbacks first
//! (mirrors `Routing::shutdown`'s order so PipeWire doesn't leave zombie
//! links). Live PIDs — including this process and any other running
//! pixelpass — are left alone.
use anyhow::Result;
use anyhow::{Context, Result, bail};
use std::collections::HashSet;
use std::path::Path;
use std::process::Command;
const SINK_NAME_PREFIX: &str = "pixelpass_capture_";
pub async fn run() -> Result<()> {
eprintln!("[pixelpass] --repair: PipeWire scan not yet implemented (Phase 2).");
eprintln!(" Run `pactl list short sinks | grep pixelpass` to spot orphans,");
eprintln!(" and `pactl unload-module <id>` to remove them manually.");
let modules = list_modules().context("failed to list pactl modules")?;
let mut dead_sinks: Vec<OrphanSink> = Vec::new();
let mut dead_pids: HashSet<u32> = HashSet::new();
let mut live_skipped: u32 = 0;
for m in &modules {
if m.name != "module-null-sink" {
continue;
}
let Some(sink_name) = extract_kv(&m.args, "sink_name") else {
continue;
};
let Some(pid_str) = sink_name.strip_prefix(SINK_NAME_PREFIX) else {
continue;
};
let Ok(pid) = pid_str.parse::<u32>() else {
continue;
};
if is_pid_alive(pid) {
live_skipped += 1;
continue;
}
dead_pids.insert(pid);
dead_sinks.push(OrphanSink {
id: m.id,
sink_name: sink_name.to_string(),
pid,
});
}
let mut dead_loopbacks: Vec<u32> = Vec::new();
for m in &modules {
if m.name != "module-loopback" {
continue;
}
let Some(sink) = extract_kv(&m.args, "sink") else {
continue;
};
let Some(pid_str) = sink.strip_prefix(SINK_NAME_PREFIX) else {
continue;
};
let Ok(pid) = pid_str.parse::<u32>() else {
continue;
};
if dead_pids.contains(&pid) {
dead_loopbacks.push(m.id);
}
}
if dead_sinks.is_empty() && dead_loopbacks.is_empty() {
if live_skipped > 0 {
println!(
"[pixelpass] --repair: nothing to clean up ({live_skipped} live pixelpass host(s) left alone)."
);
} else {
println!("[pixelpass] --repair: nothing to clean up.");
}
return Ok(());
}
let mut unloaded = 0u32;
let mut failed = 0u32;
for id in &dead_loopbacks {
match unload_module(*id) {
Ok(()) => {
println!("[pixelpass] --repair: unloaded loopback module #{id}");
unloaded += 1;
}
Err(e) => {
eprintln!("[pixelpass] --repair: failed to unload loopback #{id}: {e:#}");
failed += 1;
}
}
}
for orphan in &dead_sinks {
match unload_module(orphan.id) {
Ok(()) => {
println!(
"[pixelpass] --repair: unloaded {} (orphaned from pid {})",
orphan.sink_name, orphan.pid
);
unloaded += 1;
}
Err(e) => {
eprintln!(
"[pixelpass] --repair: failed to unload {} (#{}): {e:#}",
orphan.sink_name, orphan.id
);
failed += 1;
}
}
}
if live_skipped > 0 {
println!(
"[pixelpass] --repair: left {live_skipped} live pixelpass host(s) alone."
);
}
if failed > 0 {
bail!("--repair: {failed} module(s) failed to unload (see errors above)");
}
println!("[pixelpass] --repair: cleaned up {unloaded} module(s).");
Ok(())
}
struct Module {
id: u32,
name: String,
args: String,
}
struct OrphanSink {
id: u32,
sink_name: String,
pid: u32,
}
fn list_modules() -> Result<Vec<Module>> {
let output = Command::new("pactl")
.args(["list", "short", "modules"])
.output()
.context("failed to run `pactl list short modules`")?;
if !output.status.success() {
bail!(
"pactl list short modules failed: {}",
String::from_utf8_lossy(&output.stderr).trim()
);
}
let text = String::from_utf8(output.stdout).context("pactl returned non-UTF-8")?;
let mut modules = Vec::new();
// `pactl list short modules` is tab-separated, but some modules have
// multi-line `{ ... }` argument blocks that wrap onto continuation
// lines starting with whitespace. The wrap lines never parse as a
// u32 ID, so the simple per-line + parse-id filter is robust.
for line in text.lines() {
let mut parts = line.splitn(4, '\t');
let Some(id_str) = parts.next() else { continue };
let Ok(id) = id_str.parse::<u32>() else { continue };
let Some(name) = parts.next() else { continue };
let args = parts.next().unwrap_or("").to_string();
modules.push(Module {
id,
name: name.to_string(),
args,
});
}
Ok(modules)
}
fn extract_kv<'a>(args: &'a str, key: &str) -> Option<&'a str> {
for token in args.split_whitespace() {
if let Some(rest) = token.strip_prefix(key)
&& let Some(value) = rest.strip_prefix('=')
{
return Some(value);
}
}
None
}
fn is_pid_alive(pid: u32) -> bool {
Path::new(&format!("/proc/{pid}")).exists()
}
fn unload_module(id: u32) -> Result<()> {
let output = Command::new("pactl")
.arg("unload-module")
.arg(id.to_string())
.output()
.context("failed to run pactl unload-module")?;
if !output.status.success() {
bail!(
"pactl unload-module #{id}: {}",
String::from_utf8_lossy(&output.stderr).trim()
);
}
Ok(())
}