6 changed files with 1110 additions and 140 deletions
Generated
+970 -120
View File
File diff suppressed because it is too large Load Diff
+4 -1
View File
@@ -34,6 +34,9 @@ ureq = { version = "3", default-features = false, features = ["rustls"] }
toml = "1"
chrono = { version = "0.4", default-features = false, features = ["clock", "serde"] }
eframe = { version = "0.34.2", default-features = false, features = ["glow", "default_fonts", "wayland", "x11"], optional = true }
tray-icon = { version = "0.24.0", optional = true }
notify-rust = { version = "4.17.0", optional = true }
gtk = { version = "0.18.2", optional = true }
[profile.release]
lto = "thin"
@@ -43,4 +46,4 @@ strip = "symbols"
[features]
# Opt-in graphical front-end (pixelpass --gui). Default-off so the headless
# build never pulls the GUI toolkit tree.
gui = ["dep:eframe"]
gui = ["dep:eframe", "dep:tray-icon", "dep:notify-rust", "dep:gtk"]
+4 -2
View File
@@ -42,8 +42,10 @@ pub enum Event<'a> {
max_viewers: u32,
max_viewers_source: &'a str,
},
/// Active viewer count changed.
ViewerCount { active: u32, max: u32 },
/// A new viewer joined.
ViewerJoined { id: &'a str, active: u32, max: u32 },
/// A viewer disconnected.
ViewerLeft { id: &'a str, active: u32, max: u32 },
/// Capture pipeline lifecycle (spawned on first viewer, torn down on last).
Capture { state: CaptureState },
/// A viewer was turned away (host full, or capture spawn failed).
+26 -5
View File
@@ -35,7 +35,13 @@ pub enum ChildEvent {
max_viewers: u32,
max_viewers_source: String,
},
ViewerCount {
ViewerJoined {
id: String,
active: u32,
max: u32,
},
ViewerLeft {
id: String,
active: u32,
max: u32,
},
@@ -63,6 +69,7 @@ pub struct ChildProc {
child: Child,
pub rx: Receiver<ChildEvent>,
stderr_tail: Arc<Mutex<Vec<String>>>,
stdin: std::process::ChildStdin,
}
impl ChildProc {
@@ -72,12 +79,13 @@ impl ChildProc {
let exe = std::env::current_exe()?;
let mut child = Command::new(exe)
.args(args)
.stdin(Stdio::null())
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()?;
let (tx, rx) = std::sync::mpsc::channel();
let stdin = child.stdin.take().expect("stdin piped");
let stdout = child.stdout.take().expect("stdout piped");
std::thread::spawn(move || {
let reader = BufReader::new(stdout);
@@ -114,6 +122,7 @@ impl ChildProc {
child,
rx,
stderr_tail,
stdin,
})
}
@@ -127,6 +136,14 @@ impl ChildProc {
self.stderr_tail.lock().unwrap().join("\n")
}
/// Send a newline-terminated command to the child.
pub fn send_command(&mut self, cmd: &str) {
use std::io::Write;
if let Err(e) = writeln!(self.stdin, "{cmd}") {
tracing::warn!("failed to send command to child: {e}");
}
}
/// Gracefully stop the child: SIGINT (so the host runs its ctrl-c teardown
/// — tears down capture, closes the endpoint), with a ~2 s grace period
/// before a hard kill. Idempotent.
@@ -205,10 +222,14 @@ mod tests {
}
#[test]
fn viewer_count_round_trips() {
fn viewer_events_round_trips() {
assert!(matches!(
parse(Event::ViewerCount { active: 2, max: 4 }),
ChildEvent::ViewerCount { active: 2, max: 4 }
parse(Event::ViewerJoined { id: "nodeXYZ", active: 2, max: 4 }),
ChildEvent::ViewerJoined { id, active: 2, max: 4 } if id == "nodeXYZ"
));
assert!(matches!(
parse(Event::ViewerLeft { id: "nodeXYZ", active: 1, max: 4 }),
ChildEvent::ViewerLeft { id, active: 1, max: 4 } if id == "nodeXYZ"
));
}
+63 -2
View File
@@ -168,6 +168,8 @@ struct HostState {
copied: bool,
last_refusal: Option<String>,
error: Option<String>,
viewers: Vec<String>,
tray_icon: Option<tray_icon::TrayIcon>,
}
impl Default for HostState {
@@ -186,6 +188,8 @@ impl Default for HostState {
copied: false,
last_refusal: None,
error: None,
viewers: Vec::new(),
tray_icon: None,
}
}
}
@@ -231,6 +235,11 @@ impl eframe::App for PixelPassApp {
self.pump_host_events();
self.pump_viewer_events();
if let Ok(_event) = tray_icon::menu::MenuEvent::receiver().try_recv() {
// Only one menu item right now: "Stop Hosting"
self.stop_host();
}
match self.screen {
Screen::Menu => self.menu(ui),
Screen::Host => self.host(ui),
@@ -361,6 +370,19 @@ impl PixelPassApp {
ui.add_space(6.0);
ui.label(format!("Viewers: {} / {}", self.host.active, self.host.max));
if !self.host.viewers.is_empty() {
ui.add_space(4.0);
for id in &self.host.viewers.clone() {
ui.horizontal(|ui| {
ui.label(format!("{}", short_id(id)));
if ui.button("Kick").clicked() {
if let Some(p) = &mut self.host.proc {
p.send_command(&format!("kick {id}"));
}
}
});
}
}
if let Some(info) = &self.host.info {
ui.label(
@@ -472,7 +494,28 @@ impl PixelPassApp {
}
match ChildProc::spawn(&args, ctx) {
Ok(p) => self.host.proc = Some(p),
Ok(p) => {
self.host.proc = Some(p);
#[cfg(target_os = "linux")]
if let Err(e) = gtk::init() {
tracing::warn!("Failed to initialize GTK for system tray: {e}");
}
use tray_icon::{TrayIconBuilder, menu::{Menu, MenuItem}};
let menu = Menu::new();
let _ = menu.append(&MenuItem::new("Stop Hosting", true, None));
let icon = tray_icon::Icon::from_rgba(vec![255, 0, 0, 255], 1, 1).unwrap();
if let Ok(tray) = TrayIconBuilder::new()
.with_title("PixelPass")
.with_tooltip("PixelPass Screen Sharing")
.with_icon(icon)
.with_menu(Box::new(menu))
.build()
{
self.host.tray_icon = Some(tray);
}
}
Err(e) => self.host.error = Some(format!("Couldn't start host: {e}")),
}
}
@@ -483,6 +526,8 @@ impl PixelPassApp {
self.host.capturing = false;
self.host.ticket = None;
self.host.copied = false;
self.host.viewers.clear();
self.host.tray_icon = None;
}
/// Drain the host child's event channel into state, and detect an
@@ -543,9 +588,25 @@ impl PixelPassApp {
cap_source: max_viewers_source,
});
}
ChildEvent::ViewerCount { active, max } => {
ChildEvent::ViewerJoined { id, active, max } => {
self.host.active = active;
self.host.max = max;
if !self.host.viewers.contains(&id) {
self.host.viewers.push(id.clone());
let _ = notify_rust::Notification::new()
.summary("PixelPass Viewer Connected")
.body(&format!("Viewer {} joined the stream.", short_id(&id)))
.show();
}
}
ChildEvent::ViewerLeft { id, active, max } => {
self.host.active = active;
self.host.max = max;
self.host.viewers.retain(|v| v != &id);
let _ = notify_rust::Notification::new()
.summary("PixelPass Viewer Disconnected")
.body(&format!("Viewer {} left the stream.", short_id(&id)))
.show();
}
ChildEvent::Capture { state } => {
self.host.capturing = matches!(state, child::CaptureState::Started);
+43 -10
View File
@@ -10,7 +10,9 @@ use anyhow::{Result, bail};
use iroh::endpoint::{Connection, presets};
use iroh::{Endpoint, EndpointAddr};
use iroh_tickets::endpoint::EndpointTicket;
use std::collections::HashMap;
use std::time::Duration;
use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::sync::{mpsc, oneshot};
use tokio_util::sync::CancellationToken;
@@ -28,10 +30,16 @@ 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>>),
AddViewer {
id: String,
cancel: CancellationToken,
reply: oneshot::Sender<Result<u16, String>>,
},
/// A viewer's session ended. Supervisor decrements the count and tears
/// down capture if it just hit zero.
RemoveViewer,
RemoveViewer { id: String },
/// Request to kick a specific viewer.
KickViewer { id: String },
}
pub async fn run(opts: HostOpts) -> Result<()> {
@@ -113,6 +121,17 @@ pub async fn run(opts: HostOpts) -> Result<()> {
sup_rx,
));
// Stdin listener for "kick <id>"
let stdin_sup_tx = sup_tx.clone();
tokio::spawn(async move {
let mut lines = BufReader::new(tokio::io::stdin()).lines();
while let Ok(Some(line)) = lines.next_line().await {
if let Some(id) = line.strip_prefix("kick ") {
let _ = stdin_sup_tx.send(SupervisorMsg::KickViewer { id: id.trim().to_string() }).await;
}
}
});
accept_loop(&endpoint, sup_tx.clone(), cancel.clone()).await;
drop(sup_tx);
@@ -159,9 +178,11 @@ async fn handle_peer(
cancel: CancellationToken,
) {
let remote = conn.remote_id();
let id_str = remote.to_string();
let peer_cancel = CancellationToken::new();
let (reply_tx, reply_rx) = oneshot::channel();
if sup_tx.send(SupervisorMsg::AddViewer(reply_tx)).await.is_err() {
if sup_tx.send(SupervisorMsg::AddViewer { id: id_str.clone(), cancel: peer_cancel.clone(), reply: reply_tx }).await.is_err() {
tracing::warn!(%remote, "supervisor channel closed; dropping peer");
return;
}
@@ -182,7 +203,7 @@ async fn handle_peer(
Ok(s) => s,
Err(e) => {
tracing::warn!(%remote, "accept_bi failed: {e:#}");
let _ = sup_tx.send(SupervisorMsg::RemoveViewer).await;
let _ = sup_tx.send(SupervisorMsg::RemoveViewer { id: id_str }).await;
return;
}
};
@@ -193,7 +214,7 @@ async fn handle_peer(
Ok(t) => t,
Err(e) => {
tracing::warn!(%remote, "connect_to_capture failed: {e:#}");
let _ = sup_tx.send(SupervisorMsg::RemoveViewer).await;
let _ = sup_tx.send(SupervisorMsg::RemoveViewer { id: id_str }).await;
return;
}
};
@@ -207,10 +228,13 @@ async fn handle_peer(
_ = cancel.cancelled() => {
tracing::info!(%remote, "cancellation during stream");
}
_ = peer_cancel.cancelled() => {
tracing::info!(%remote, "kicked by host");
}
}
eprintln!("[pixelpass] viewer disconnected: {remote}");
let _ = sup_tx.send(SupervisorMsg::RemoveViewer).await;
let _ = sup_tx.send(SupervisorMsg::RemoveViewer { id: id_str }).await;
}
/// Owns the single shared CaptureHandle and the active viewer count. Spawns
@@ -226,10 +250,11 @@ async fn supervise(
) {
let mut handle: Option<CaptureHandle> = None;
let mut count: u32 = 0;
let mut viewers: HashMap<String, CancellationToken> = HashMap::new();
while let Some(msg) = rx.recv().await {
match msg {
SupervisorMsg::AddViewer(reply) => {
SupervisorMsg::AddViewer { id, cancel, reply } => {
if count >= max_viewers {
let reason =
format!("host is full ({count} of {max_viewers} viewers connected)");
@@ -256,13 +281,15 @@ async fn supervise(
let port = handle.as_ref().expect("handle was just set").local_port();
count += 1;
viewers.insert(id.clone(), cancel);
let _ = reply.send(Ok(port));
output::emit(output::Event::ViewerCount { active: count, max: max_viewers });
output::emit(output::Event::ViewerJoined { id: &id, active: count, max: max_viewers });
tracing::info!(active = count, cap = max_viewers, "viewer joined");
}
SupervisorMsg::RemoveViewer => {
SupervisorMsg::RemoveViewer { id } => {
viewers.remove(&id);
count = count.saturating_sub(1);
output::emit(output::Event::ViewerCount { active: count, max: max_viewers });
output::emit(output::Event::ViewerLeft { id: &id, active: count, max: max_viewers });
tracing::info!(active = count, cap = max_viewers, "viewer left");
if count == 0
&& let Some(h) = handle.take()
@@ -274,6 +301,12 @@ async fn supervise(
});
}
}
SupervisorMsg::KickViewer { id } => {
if let Some(cancel) = viewers.get(&id) {
tracing::info!(%id, "kicking viewer");
cancel.cancel();
}
}
}
}