Compare commits
3
Commits
74b4101d4f
...
f939441e31
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f939441e31 | ||
|
|
153febe078 | ||
|
|
ffe5a90686 |
Generated
+37
@@ -3074,6 +3074,7 @@ dependencies = [
|
|||||||
"anyhow",
|
"anyhow",
|
||||||
"arboard",
|
"arboard",
|
||||||
"ashpd",
|
"ashpd",
|
||||||
|
"chrono",
|
||||||
"clap",
|
"clap",
|
||||||
"dialoguer",
|
"dialoguer",
|
||||||
"directories",
|
"directories",
|
||||||
@@ -3086,8 +3087,10 @@ dependencies = [
|
|||||||
"thiserror 2.0.18",
|
"thiserror 2.0.18",
|
||||||
"tokio",
|
"tokio",
|
||||||
"tokio-util",
|
"tokio-util",
|
||||||
|
"toml",
|
||||||
"tracing",
|
"tracing",
|
||||||
"tracing-subscriber",
|
"tracing-subscriber",
|
||||||
|
"ureq",
|
||||||
"uuid",
|
"uuid",
|
||||||
"x11rb",
|
"x11rb",
|
||||||
]
|
]
|
||||||
@@ -4448,6 +4451,34 @@ version = "0.9.0"
|
|||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1"
|
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]]
|
[[package]]
|
||||||
name = "url"
|
name = "url"
|
||||||
version = "2.5.8"
|
version = "2.5.8"
|
||||||
@@ -4461,6 +4492,12 @@ dependencies = [
|
|||||||
"serde_derive",
|
"serde_derive",
|
||||||
]
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "utf8-zero"
|
||||||
|
version = "0.8.1"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "b8c0a043c9540bae7c578c88f91dda8bd82e59ae27c21baca69c8b191aaf5a6e"
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "utf8_iter"
|
name = "utf8_iter"
|
||||||
version = "1.0.4"
|
version = "1.0.4"
|
||||||
|
|||||||
@@ -30,6 +30,9 @@ uuid = { version = "1", features = ["v4"] }
|
|||||||
iroh-tickets = "1.0.0-rc.0"
|
iroh-tickets = "1.0.0-rc.0"
|
||||||
dialoguer = { version = "0.12", default-features = false }
|
dialoguer = { version = "0.12", default-features = false }
|
||||||
arboard = { version = "3", default-features = false, features = ["wayland-data-control"] }
|
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]
|
[profile.release]
|
||||||
lto = "thin"
|
lto = "thin"
|
||||||
|
|||||||
@@ -22,11 +22,15 @@ Working:
|
|||||||
- iroh QUIC bi-stream tunnel, direct-UDP and relay paths both verified
|
- iroh QUIC bi-stream tunnel, direct-UDP and relay paths both verified
|
||||||
- Interactive Host/View menu with clipboard auto-copy and mpv/VLC picker
|
- Interactive Host/View menu with clipboard auto-copy and mpv/VLC picker
|
||||||
- Headless mode for scripts (`pixelpass <ticket>`)
|
- 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:
|
Not yet working:
|
||||||
- X11 capture (stubbed, returns an error)
|
- X11 capture (stubbed, returns an error)
|
||||||
- Per-app audio routing (`--app <name>` is a flag stub)
|
- Per-app audio routing (`--app <name>` is a flag stub)
|
||||||
- Multi-viewer (single viewer per host by design right now)
|
|
||||||
- `--repair` (PipeWire orphan cleanup) is a stub
|
- `--repair` (PipeWire orphan cleanup) is a stub
|
||||||
|
|
||||||
## Quick start
|
## Quick start
|
||||||
@@ -39,7 +43,14 @@ pixelpass
|
|||||||
|
|
||||||
On the host machine: pick "Host", share a monitor via the portal dialog,
|
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
|
the ticket lands on your clipboard. Send it to your viewer however you
|
||||||
like (chat, email, paste in a note).
|
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,
|
On the viewer machine: run `pixelpass`, pick "View", paste the ticket,
|
||||||
pick mpv or VLC. The player launches detached and the stream starts.
|
pick mpv or VLC. The player launches detached and the stream starts.
|
||||||
@@ -154,6 +165,42 @@ relay path otherwise. Both have been verified end-to-end.
|
|||||||
- **VAAPI H.264 over x264**: ~5% of one CPU core instead of ~50% on
|
- **VAAPI H.264 over x264**: ~5% of one CPU core instead of ~50% on
|
||||||
the host's hardware.
|
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.
|
||||||
|
|
||||||
|
## 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
|
## Known limitations and gotchas
|
||||||
|
|
||||||
- **VLC needs `vlc-plugin-dvb` and `vlc-plugin-ffmpeg`** on Arch-family
|
- **VLC needs `vlc-plugin-dvb` and `vlc-plugin-ffmpeg`** on Arch-family
|
||||||
@@ -164,9 +211,8 @@ relay path otherwise. Both have been verified end-to-end.
|
|||||||
- **Audio echo** if the host plays the stream through speakers and
|
- **Audio echo** if the host plays the stream through speakers and
|
||||||
captures system audio — expected, the mic / monitor picks up the
|
captures system audio — expected, the mic / monitor picks up the
|
||||||
playback. Headphones bypass it.
|
playback. Headphones bypass it.
|
||||||
- **Single viewer per host** by design right now. Restarting the player
|
- **Late joiners see ~2 s of garbage** before the next keyframe lets
|
||||||
against the same URL fails with "connection refused"; restart the
|
their decoder lock. Expected behavior, not a bug.
|
||||||
viewer too.
|
|
||||||
- **VAAPI driver must be package-tracked**, not an orphaned `.so` on
|
- **VAAPI driver must be package-tracked**, not an orphaned `.so` on
|
||||||
disk. mpv's `--hwdec=auto` silently falls back to software decode
|
disk. mpv's `--hwdec=auto` silently falls back to software decode
|
||||||
otherwise, which then chokes on a low-power viewer.
|
otherwise, which then chokes on a low-power viewer.
|
||||||
|
|||||||
+15
@@ -45,6 +45,13 @@ pub struct Cli {
|
|||||||
#[arg(long)]
|
#[arg(long)]
|
||||||
pub low_latency: bool,
|
pub low_latency: bool,
|
||||||
|
|
||||||
|
/// 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 max_viewers: Option<u32>,
|
||||||
|
|
||||||
// ── viewer options ────────────────────────────────────────────────
|
// ── viewer options ────────────────────────────────────────────────
|
||||||
/// Local TCP port for the viewer to expose (default: random).
|
/// Local TCP port for the viewer to expose (default: random).
|
||||||
#[arg(long, default_value_t = 0)]
|
#[arg(long, default_value_t = 0)]
|
||||||
@@ -58,6 +65,12 @@ pub struct Cli {
|
|||||||
/// Clean up orphaned PipeWire state from a crashed host run, then exit.
|
/// Clean up orphaned PipeWire state from a crashed host run, then exit.
|
||||||
#[arg(long)]
|
#[arg(long)]
|
||||||
pub repair: bool,
|
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)]
|
#[derive(ValueEnum, Clone, Copy, Debug)]
|
||||||
@@ -76,6 +89,7 @@ pub struct HostOpts {
|
|||||||
pub framerate: u32,
|
pub framerate: u32,
|
||||||
pub no_hwencode: bool,
|
pub no_hwencode: bool,
|
||||||
pub low_latency: bool,
|
pub low_latency: bool,
|
||||||
|
pub max_viewers: Option<u32>,
|
||||||
pub interactive: bool,
|
pub interactive: bool,
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -96,6 +110,7 @@ impl Cli {
|
|||||||
framerate: self.framerate,
|
framerate: self.framerate,
|
||||||
no_hwencode: self.no_hwencode,
|
no_hwencode: self.no_hwencode,
|
||||||
low_latency: self.low_latency,
|
low_latency: self.low_latency,
|
||||||
|
max_viewers: self.max_viewers,
|
||||||
interactive,
|
interactive,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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 }
|
||||||
|
}
|
||||||
@@ -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(())
|
||||||
|
}
|
||||||
@@ -1,4 +1,6 @@
|
|||||||
pub mod alpn;
|
pub mod alpn;
|
||||||
|
pub mod bandwidth;
|
||||||
|
pub mod config;
|
||||||
pub mod deps;
|
pub mod deps;
|
||||||
pub mod display;
|
pub mod display;
|
||||||
pub mod process;
|
pub mod process;
|
||||||
|
|||||||
+221
-38
@@ -5,10 +5,28 @@ use anyhow::{Result, bail};
|
|||||||
use iroh::Endpoint;
|
use iroh::Endpoint;
|
||||||
use iroh::endpoint::{Connection, presets};
|
use iroh::endpoint::{Connection, presets};
|
||||||
use iroh_tickets::endpoint::EndpointTicket;
|
use iroh_tickets::endpoint::EndpointTicket;
|
||||||
|
use std::time::Duration;
|
||||||
|
use tokio::sync::{mpsc, oneshot};
|
||||||
use tokio_util::sync::CancellationToken;
|
use tokio_util::sync::CancellationToken;
|
||||||
|
|
||||||
use crate::cli::HostOpts;
|
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<()> {
|
pub async fn run(opts: HostOpts) -> Result<()> {
|
||||||
let display = DisplayServer::resolve(opts.display_server);
|
let display = DisplayServer::resolve(opts.display_server);
|
||||||
@@ -21,6 +39,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 cancel = signal::install_ctrl_c();
|
||||||
|
|
||||||
let endpoint = Endpoint::builder(presets::N0)
|
let endpoint = Endpoint::builder(presets::N0)
|
||||||
@@ -31,73 +54,177 @@ pub async fn run(opts: HostOpts) -> Result<()> {
|
|||||||
let addr = endpoint.addr();
|
let addr = endpoint.addr();
|
||||||
let ticket = EndpointTicket::new(addr);
|
let ticket = EndpointTicket::new(addr);
|
||||||
let clipboard_ok = opts.interactive && copy_to_clipboard(&ticket.to_string());
|
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;
|
endpoint.close().await;
|
||||||
result
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn accept_loop(
|
async fn accept_loop(
|
||||||
endpoint: &Endpoint,
|
endpoint: &Endpoint,
|
||||||
display: DisplayServer,
|
sup_tx: mpsc::Sender<SupervisorMsg>,
|
||||||
opts: &HostOpts,
|
|
||||||
cancel: CancellationToken,
|
cancel: CancellationToken,
|
||||||
) -> Result<()> {
|
) {
|
||||||
tokio::select! {
|
loop {
|
||||||
_ = cancel.cancelled() => {
|
tokio::select! {
|
||||||
tracing::info!("cancellation requested before any peer connected");
|
_ = cancel.cancelled() => {
|
||||||
Ok(())
|
tracing::info!("cancellation requested — closing accept loop");
|
||||||
}
|
return;
|
||||||
accepted = endpoint.accept() => {
|
}
|
||||||
let Some(incoming) = accepted else {
|
accepted = endpoint.accept() => {
|
||||||
bail!("endpoint stopped accepting connections");
|
let Some(incoming) = accepted else {
|
||||||
};
|
tracing::info!("endpoint stopped accepting connections");
|
||||||
let conn = incoming.await?;
|
return;
|
||||||
let remote = conn.remote_id();
|
};
|
||||||
tracing::info!(%remote, "peer connected");
|
let conn = match incoming.await {
|
||||||
eprintln!("\n[pixelpass] peer connected: {remote}\n");
|
Ok(c) => c,
|
||||||
handle_peer(conn, display, opts, cancel).await
|
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(
|
async fn handle_peer(
|
||||||
conn: Connection,
|
conn: Connection,
|
||||||
display: DisplayServer,
|
sup_tx: mpsc::Sender<SupervisorMsg>,
|
||||||
opts: &HostOpts,
|
|
||||||
cancel: CancellationToken,
|
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 (reply_tx, reply_rx) = oneshot::channel();
|
||||||
let port = capture_handle.local_port();
|
if sup_tx.send(SupervisorMsg::AddViewer(reply_tx)).await.is_err() {
|
||||||
let tcp = wayland::connect_to_capture(port, std::time::Duration::from_secs(5)).await?;
|
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 bridge = crate::common::tunnel::bridge(quic_send, quic_recv, tcp);
|
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 wayland::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! {
|
tokio::select! {
|
||||||
res = bridge => {
|
res = bridge => {
|
||||||
if let Err(e) = res {
|
if let Err(e) = res {
|
||||||
tracing::warn!("bridge ended with error: {e:#}");
|
tracing::warn!(%remote, "bridge ended with error: {e:#}");
|
||||||
} else {
|
} else {
|
||||||
tracing::info!("bridge closed cleanly");
|
tracing::info!(%remote, "bridge closed cleanly");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
_ = cancel.cancelled() => {
|
_ = cancel.cancelled() => {
|
||||||
tracing::info!("cancellation requested during stream");
|
tracing::info!(%remote, "cancellation during stream");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
capture_handle.shutdown().await;
|
eprintln!("[pixelpass] viewer disconnected: {remote}");
|
||||||
Ok(())
|
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(
|
fn print_host_banner(
|
||||||
ticket: &EndpointTicket,
|
ticket: &EndpointTicket,
|
||||||
display: DisplayServer,
|
display: DisplayServer,
|
||||||
opts: &HostOpts,
|
opts: &HostOpts,
|
||||||
|
resolution: &MaxViewersResolution,
|
||||||
clipboard_ok: bool,
|
clipboard_ok: bool,
|
||||||
) {
|
) {
|
||||||
eprintln!();
|
eprintln!();
|
||||||
@@ -106,23 +233,79 @@ fn print_host_banner(
|
|||||||
eprintln!("│ capture : {}", capture_summary(opts));
|
eprintln!("│ capture : {}", capture_summary(opts));
|
||||||
eprintln!("│ bitrate / fps : {} kbps @ {} fps", opts.bitrate, opts.framerate);
|
eprintln!("│ bitrate / fps : {} kbps @ {} fps", opts.bitrate, opts.framerate);
|
||||||
eprintln!("│ hw encode : {}", if opts.no_hwencode { "off" } else { "auto (VAAPI if available)" });
|
eprintln!("│ hw encode : {}", if opts.no_hwencode { "off" } else { "auto (VAAPI if available)" });
|
||||||
|
eprintln!("│ max viewers : {} ({})", resolution.value, resolution.source.label());
|
||||||
eprintln!("│");
|
eprintln!("│");
|
||||||
if clipboard_ok {
|
if clipboard_ok {
|
||||||
eprintln!("│ Your share code has been copied to your clipboard.");
|
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.)");
|
eprintln!("│ code is also shown below for manual copy.)");
|
||||||
} else {
|
} else {
|
||||||
eprintln!("│ Share this ticket with your viewer:");
|
eprintln!("│ Share this ticket with your viewer(s):");
|
||||||
}
|
}
|
||||||
eprintln!("│");
|
eprintln!("│");
|
||||||
eprintln!("│ pixelpass {ticket}");
|
eprintln!("│ pixelpass {ticket}");
|
||||||
eprintln!("│");
|
eprintln!("│");
|
||||||
eprintln!("│ Capture will not start until the viewer connects.");
|
eprintln!("│ Capture starts when the first viewer connects, runs while");
|
||||||
eprintln!("│ Press Ctrl+C to stop.");
|
eprintln!("│ any viewer is connected, and tears down when the last one");
|
||||||
|
eprintln!("│ leaves. Press Ctrl+C to stop the host entirely.");
|
||||||
eprintln!("└────────────────────────────────────────────────────────────");
|
eprintln!("└────────────────────────────────────────────────────────────");
|
||||||
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 {
|
fn copy_to_clipboard(text: &str) -> bool {
|
||||||
match arboard::Clipboard::new().and_then(|mut cb| cb.set_text(text.to_owned())) {
|
match arboard::Clipboard::new().and_then(|mut cb| cb.set_text(text.to_owned())) {
|
||||||
Ok(()) => true,
|
Ok(()) => true,
|
||||||
|
|||||||
+94
-17
@@ -1,7 +1,10 @@
|
|||||||
//! Wayland capture: ashpd ScreenCast portal → PipeWire fd → gst-launch
|
//! Wayland capture: ashpd ScreenCast portal → PipeWire fd → gst-launch
|
||||||
//! pipewiresrc → MPEG-TS on gst stdout → in-process HTTP server bound on a
|
//! 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
|
//! random localhost port. One gst child feeds a tokio::sync::broadcast channel;
|
||||||
//! pumps bytes to QUIC.
|
//! the HTTP listener accepts multiple connections and each one drains its own
|
||||||
|
//! fresh broadcast::Receiver — so a single capture pipeline fans out to N
|
||||||
|
//! concurrent viewers. Slow consumers see Lagged and skip ahead; the MPEG-TS
|
||||||
|
//! stream resyncs at the next keyframe.
|
||||||
|
|
||||||
use anyhow::{Context, Result, bail};
|
use anyhow::{Context, Result, bail};
|
||||||
use ashpd::{
|
use ashpd::{
|
||||||
@@ -16,18 +19,30 @@ use nix::sys::signal::{Signal, kill};
|
|||||||
use nix::unistd::{Pid, close};
|
use nix::unistd::{Pid, close};
|
||||||
use std::os::fd::{AsFd, IntoRawFd, OwnedFd, RawFd};
|
use std::os::fd::{AsFd, IntoRawFd, OwnedFd, RawFd};
|
||||||
use std::process::Stdio;
|
use std::process::Stdio;
|
||||||
|
use std::sync::Arc;
|
||||||
use std::time::Duration;
|
use std::time::Duration;
|
||||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||||
use tokio::net::{TcpListener, TcpStream};
|
use tokio::net::{TcpListener, TcpStream};
|
||||||
use tokio::process::{Child, ChildStdout, Command};
|
use tokio::process::{Child, ChildStdout, Command};
|
||||||
|
use tokio::sync::broadcast;
|
||||||
use tokio::task::JoinHandle;
|
use tokio::task::JoinHandle;
|
||||||
use tokio::time::{Instant, sleep, timeout};
|
use tokio::time::{Instant, sleep, timeout};
|
||||||
|
|
||||||
use crate::cli::HostOpts;
|
use crate::cli::HostOpts;
|
||||||
|
|
||||||
|
/// Broadcast-channel capacity in chunks. Each chunk is up to 64 KiB from gst
|
||||||
|
/// stdout, so 16 chunks ≈ 1 MiB ≈ ~2 s of buffered jitter at the default
|
||||||
|
/// 4 Mbps bitrate. A viewer that falls behind by more than this gets Lagged
|
||||||
|
/// and skips ahead — MPEG-TS resyncs at the next keyframe.
|
||||||
|
const FANOUT_CAPACITY: usize = 16;
|
||||||
|
|
||||||
|
/// Size of each chunk read from gst stdout.
|
||||||
|
const READ_CHUNK: usize = 64 * 1024;
|
||||||
|
|
||||||
pub struct CaptureHandle {
|
pub struct CaptureHandle {
|
||||||
port: u16,
|
port: u16,
|
||||||
gst: Option<Child>,
|
gst: Option<Child>,
|
||||||
|
reader: Option<JoinHandle<()>>,
|
||||||
server: Option<JoinHandle<()>>,
|
server: Option<JoinHandle<()>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -37,8 +52,8 @@ impl CaptureHandle {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Graceful teardown: SIGTERM gst, give it ~1s to exit, then SIGKILL, then
|
/// Graceful teardown: SIGTERM gst, give it ~1s to exit, then SIGKILL, then
|
||||||
/// abort the HTTP server task. Call this before dropping; Drop only fires
|
/// abort the reader + accept-loop tasks. Call this before dropping; Drop
|
||||||
/// the kill backstop.
|
/// only fires the kill backstop.
|
||||||
pub async fn shutdown(mut self) {
|
pub async fn shutdown(mut self) {
|
||||||
if let Some(child) = self.gst.as_mut()
|
if let Some(child) = self.gst.as_mut()
|
||||||
&& let Some(pid) = child.id()
|
&& let Some(pid) = child.id()
|
||||||
@@ -49,6 +64,9 @@ impl CaptureHandle {
|
|||||||
let _ = timeout(Duration::from_millis(1000), child.wait()).await;
|
let _ = timeout(Duration::from_millis(1000), child.wait()).await;
|
||||||
let _ = child.start_kill();
|
let _ = child.start_kill();
|
||||||
}
|
}
|
||||||
|
if let Some(task) = self.reader.take() {
|
||||||
|
task.abort();
|
||||||
|
}
|
||||||
if let Some(task) = self.server.take() {
|
if let Some(task) = self.server.take() {
|
||||||
task.abort();
|
task.abort();
|
||||||
}
|
}
|
||||||
@@ -60,6 +78,9 @@ impl Drop for CaptureHandle {
|
|||||||
if let Some(child) = self.gst.as_mut() {
|
if let Some(child) = self.gst.as_mut() {
|
||||||
let _ = child.start_kill();
|
let _ = child.start_kill();
|
||||||
}
|
}
|
||||||
|
if let Some(task) = self.reader.as_ref() {
|
||||||
|
task.abort();
|
||||||
|
}
|
||||||
if let Some(task) = self.server.as_ref() {
|
if let Some(task) = self.server.as_ref() {
|
||||||
task.abort();
|
task.abort();
|
||||||
}
|
}
|
||||||
@@ -199,28 +220,68 @@ pub async fn start(opts: &HostOpts) -> Result<CaptureHandle> {
|
|||||||
.take()
|
.take()
|
||||||
.context("gst-launch-1.0 stdout pipe unavailable")?;
|
.context("gst-launch-1.0 stdout pipe unavailable")?;
|
||||||
|
|
||||||
// 4. Spawn the HTTP server task. It owns the listener + gst stdout: it
|
// 4. Set up the broadcast fanout. The reader task pumps gst stdout chunks
|
||||||
// accepts one client (the host's bridge socket via connect_to_capture),
|
// into the channel; the accept-loop task spawns one sender task per
|
||||||
// drains the HTTP request, writes a fixed MPEG-TS response, then
|
// accepted TCP connection, each draining a fresh broadcast::Receiver.
|
||||||
// copies gst stdout to the socket forever.
|
let (tx, _) = broadcast::channel::<Arc<Vec<u8>>>(FANOUT_CAPACITY);
|
||||||
let server = tokio::spawn(serve_capture(listener, gst_stdout));
|
|
||||||
|
let reader = tokio::spawn(pump_gst_to_broadcast(gst_stdout, tx.clone()));
|
||||||
|
let server = tokio::spawn(run_accept_loop(listener, tx));
|
||||||
|
|
||||||
Ok(CaptureHandle {
|
Ok(CaptureHandle {
|
||||||
port,
|
port,
|
||||||
gst: Some(gst),
|
gst: Some(gst),
|
||||||
|
reader: Some(reader),
|
||||||
server: Some(server),
|
server: Some(server),
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn serve_capture(listener: TcpListener, mut gst_stdout: ChildStdout) {
|
/// Reads gst's stdout in chunks and broadcasts each to all current subscribers.
|
||||||
let mut sock = match listener.accept().await {
|
/// `broadcast::send` returns Err when there are no receivers; we ignore it and
|
||||||
Ok((s, _)) => s,
|
/// keep reading so gst doesn't backpressure waiting for a viewer.
|
||||||
Err(e) => {
|
async fn pump_gst_to_broadcast(
|
||||||
tracing::warn!("capture HTTP accept failed: {e}");
|
mut gst_stdout: ChildStdout,
|
||||||
return;
|
tx: broadcast::Sender<Arc<Vec<u8>>>,
|
||||||
|
) {
|
||||||
|
let mut buf = vec![0u8; READ_CHUNK];
|
||||||
|
loop {
|
||||||
|
match gst_stdout.read(&mut buf).await {
|
||||||
|
Ok(0) => {
|
||||||
|
tracing::info!("gst stdout EOF — fanout reader exiting");
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
Ok(n) => {
|
||||||
|
let chunk = Arc::new(buf[..n].to_vec());
|
||||||
|
let _ = tx.send(chunk);
|
||||||
|
}
|
||||||
|
Err(e) => {
|
||||||
|
tracing::warn!("gst stdout read error: {e}");
|
||||||
|
return;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
};
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Accepts TCP connections on the local capture port forever. Each accepted
|
||||||
|
/// connection becomes its own viewer-serving task with a private receiver.
|
||||||
|
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));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Drains the HTTP request, writes a fixed 200 OK, then pumps broadcast
|
||||||
|
/// chunks to the socket until the channel closes or the socket errors out.
|
||||||
|
/// On Lagged (slow consumer), skip ahead — MPEG-TS recovers at next keyframe.
|
||||||
|
async fn serve_one_viewer(mut sock: TcpStream, mut rx: broadcast::Receiver<Arc<Vec<u8>>>) {
|
||||||
if !drain_http_request(&mut sock).await {
|
if !drain_http_request(&mut sock).await {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
@@ -234,7 +295,23 @@ async fn serve_capture(listener: TcpListener, mut gst_stdout: ChildStdout) {
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
let _ = tokio::io::copy(&mut gst_stdout, &mut sock).await;
|
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 {
|
async fn drain_http_request(sock: &mut TcpStream) -> bool {
|
||||||
|
|||||||
+137
-1
@@ -4,6 +4,7 @@ use iroh_tickets::endpoint::EndpointTicket;
|
|||||||
use std::str::FromStr;
|
use std::str::FromStr;
|
||||||
|
|
||||||
use crate::cli::Cli;
|
use crate::cli::Cli;
|
||||||
|
use crate::common::{bandwidth, config};
|
||||||
use crate::{host, viewer};
|
use crate::{host, viewer};
|
||||||
|
|
||||||
pub async fn run(cli: Cli) -> Result<()> {
|
pub async fn run(cli: Cli) -> Result<()> {
|
||||||
@@ -20,7 +21,10 @@ pub async fn run(cli: Cli) -> Result<()> {
|
|||||||
.interact()?;
|
.interact()?;
|
||||||
|
|
||||||
match choice {
|
match choice {
|
||||||
0 => host::run(cli.into_host_opts(true)).await,
|
0 => {
|
||||||
|
preflight_if_needed(&theme).await;
|
||||||
|
host::run(cli.into_host_opts(true)).await
|
||||||
|
}
|
||||||
_ => {
|
_ => {
|
||||||
let ticket = prompt_ticket(&theme)?;
|
let ticket = prompt_ticket(&theme)?;
|
||||||
viewer::run(ticket, cli.into_viewer_opts(true)).await
|
viewer::run(ticket, cli.into_viewer_opts(true)).await
|
||||||
@@ -28,6 +32,138 @@ pub async fn run(cli: Cli) -> Result<()> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// `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() {
|
fn print_welcome() {
|
||||||
eprintln!();
|
eprintln!();
|
||||||
eprintln!("Welcome to PixelPass.");
|
eprintln!("Welcome to PixelPass.");
|
||||||
|
|||||||
@@ -20,6 +20,10 @@ async fn main() -> Result<()> {
|
|||||||
return repair::run().await;
|
return repair::run().await;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if cli.reconfigure {
|
||||||
|
return interactive::run_reconfigure().await;
|
||||||
|
}
|
||||||
|
|
||||||
match cli.ticket.as_deref() {
|
match cli.ticket.as_deref() {
|
||||||
Some(s) => {
|
Some(s) => {
|
||||||
let ticket: EndpointTicket = s.parse().map_err(|e| {
|
let ticket: EndpointTicket = s.parse().map_err(|e| {
|
||||||
|
|||||||
Reference in New Issue
Block a user