From e77ecb447e3c5146b0f587323c2fae746414f5b3 Mon Sep 17 00:00:00 2001 From: "J. Champagne" Date: Sun, 15 Mar 2026 01:33:38 -0400 Subject: [PATCH] feat: Add streaming pipe support for async loading in. --- crates/pikl-core/src/model/event.rs | 3 + crates/pikl-core/src/runtime/input.rs | 44 +++++++ crates/pikl-core/src/runtime/menu.rs | 63 ++++++++++ .../pikl-core/src/script/action_fd/parse.rs | 9 ++ crates/pikl-tui/src/lib.rs | 31 ++++- crates/pikl/src/main.rs | 117 +++++++++++++++--- docs/DEVPLAN.md | 33 +++-- docs/use-cases/wallpaper-picker.md | 2 +- 8 files changed, 271 insertions(+), 31 deletions(-) diff --git a/crates/pikl-core/src/model/event.rs b/crates/pikl-core/src/model/event.rs index eed4283..b5690ee 100644 --- a/crates/pikl-core/src/model/event.rs +++ b/crates/pikl-core/src/model/event.rs @@ -53,6 +53,7 @@ pub enum Action { RemoveItems(Vec), ProcessHookResponse(HookResponse), CloseMenu, + StreamingDone, } /// Broadcast from the menu loop to all subscribers @@ -81,6 +82,8 @@ pub struct ViewState { pub mode: Mode, pub selection_count: usize, pub multi_enabled: bool, + /// True while stdin is still streaming items in. + pub streaming: bool, /// Column headers for table mode. None when not in /// table mode. pub columns: Option>, diff --git a/crates/pikl-core/src/runtime/input.rs b/crates/pikl-core/src/runtime/input.rs index d9881b2..d873648 100644 --- a/crates/pikl-core/src/runtime/input.rs +++ b/crates/pikl-core/src/runtime/input.rs @@ -8,6 +8,15 @@ use tokio::io::AsyncBufReadExt; use crate::error::PiklError; use crate::item::Item; +/// Parse a single line into a JSON Value. Tries JSON first, +/// falls back to wrapping as a string. Used by the streaming +/// stdin reader in the CLI binary, which sends raw Values +/// via `Action::AddItems`. +pub fn parse_line_to_value(line: &str) -> serde_json::Value { + serde_json::from_str::(line) + .unwrap_or_else(|_| serde_json::Value::String(line.to_string())) +} + /// Try to parse a line as JSON. Falls back to wrapping /// it as a plain-text string. The `label_key` controls /// which JSON key is used as the display label for object @@ -116,4 +125,39 @@ mod tests { assert_eq!(items.len(), 1); assert_eq!(items[0].label(), "quoted string"); } + + #[test] + fn parse_line_to_value_plain_text() { + let v = parse_line_to_value("hello world"); + assert_eq!(v, serde_json::Value::String("hello world".to_string())); + } + + #[test] + fn parse_line_to_value_json_object() { + let v = parse_line_to_value(r#"{"label": "foo", "count": 42}"#); + assert!(v.is_object()); + assert_eq!(v["label"], "foo"); + assert_eq!(v["count"], 42); + } + + #[test] + fn parse_line_to_value_invalid_json() { + let v = parse_line_to_value("{not valid json}"); + assert_eq!( + v, + serde_json::Value::String("{not valid json}".to_string()) + ); + } + + #[test] + fn parse_line_to_value_json_string() { + let v = parse_line_to_value(r#""quoted""#); + assert_eq!(v, serde_json::Value::String("quoted".to_string())); + } + + #[test] + fn parse_line_to_value_json_number() { + let v = parse_line_to_value("42"); + assert_eq!(v, serde_json::json!(42)); + } } diff --git a/crates/pikl-core/src/runtime/menu.rs b/crates/pikl-core/src/runtime/menu.rs index 3ea34f3..f316ae6 100644 --- a/crates/pikl-core/src/runtime/menu.rs +++ b/crates/pikl-core/src/runtime/menu.rs @@ -154,6 +154,7 @@ pub struct MenuRunner { generation: u64, selection: SelectionState, visual_anchor: Option, + streaming: bool, } impl MenuRunner { @@ -180,6 +181,7 @@ impl MenuRunner { generation: 0, selection: SelectionState::new(), visual_anchor: None, + streaming: false, }; (runner, action_tx) } @@ -312,6 +314,7 @@ impl MenuRunner { mode: self.mode, selection_count: self.selection.count(), multi_enabled: self.selection.multi_enabled, + streaming: self.streaming, columns, generation: self.generation, } @@ -557,6 +560,11 @@ impl MenuRunner { self.apply_action(action) } Action::CloseMenu => ActionOutcome::Closed, + Action::StreamingDone => { + debug!("streaming done"); + self.streaming = false; + ActionOutcome::Broadcast + } } } @@ -566,6 +574,11 @@ impl MenuRunner { self.mode = mode; } + /// Mark the menu as actively streaming items from stdin. + pub fn set_streaming(&mut self, v: bool) { + self.streaming = v; + } + /// Run the menu event loop. Consumes actions and /// broadcasts events. /// @@ -1824,4 +1837,54 @@ mod tests { // City "toronto" (7) vs header "City" (4): width = 7 assert_eq!(cols[2].width, 7); } + + // -- Streaming tests -- + + #[test] + fn streaming_done_sets_streaming_false() { + let (mut m, _tx) = test_menu(); + m.set_streaming(true); + m.run_filter(); + assert!(m.build_view_state().streaming); + let outcome = m.apply_action(Action::StreamingDone); + assert!(matches!(outcome, ActionOutcome::Broadcast)); + assert!(!m.build_view_state().streaming); + } + + #[test] + fn add_items_on_empty_streaming_menu() { + let (mut m, _tx) = MenuRunner::new(JsonMenu::new(vec![], "label".to_string())); + m.set_streaming(true); + m.run_filter(); + m.apply_action(Action::Resize { height: 20 }); + + assert_eq!(m.menu.total(), 0); + assert!(m.build_view_state().streaming); + + m.apply_action(Action::AddItems(vec![ + Value::String("one".into()), + Value::String("two".into()), + ])); + assert_eq!(m.menu.total(), 2); + assert_eq!(m.menu.filtered_count(), 2); + + // Confirm on non-empty streaming menu works + let outcome = m.apply_action(Action::Confirm); + assert!(matches!(outcome, ActionOutcome::Selected { .. })); + } + + #[test] + fn confirm_on_empty_streaming_menu_is_noop() { + let (mut m, _tx) = MenuRunner::new(JsonMenu::new(vec![], "label".to_string())); + m.set_streaming(true); + m.run_filter(); + let outcome = m.apply_action(Action::Confirm); + assert!(matches!(outcome, ActionOutcome::NoOp)); + } + + #[test] + fn view_state_streaming_defaults_false() { + let mut m = ready_menu(); + assert!(!m.build_view_state().streaming); + } } diff --git a/crates/pikl-core/src/script/action_fd/parse.rs b/crates/pikl-core/src/script/action_fd/parse.rs index 7c81c5c..62c98dd 100644 --- a/crates/pikl-core/src/script/action_fd/parse.rs +++ b/crates/pikl-core/src/script/action_fd/parse.rs @@ -161,6 +161,7 @@ pub fn parse_action(line_number: usize, line: &str) -> Result Ok(ScriptAction::ShowUi), "show-tui" => Ok(ScriptAction::ShowTui), "show-gui" => Ok(ScriptAction::ShowGui), + "streaming-done" => Ok(ScriptAction::Core(Action::StreamingDone)), _ => Err(ScriptError { line: line_number, source_line: line.to_string(), @@ -678,6 +679,14 @@ mod tests { ); } + #[test] + fn parse_streaming_done() { + assert_eq!( + parse_action(1, "streaming-done").unwrap_or(ScriptAction::Comment), + ScriptAction::Core(Action::StreamingDone) + ); + } + #[test] fn parse_set_mode_wrong_case() { assert!(parse_action(1, "set-mode Insert").is_err()); diff --git a/crates/pikl-tui/src/lib.rs b/crates/pikl-tui/src/lib.rs index 5846027..8346a02 100644 --- a/crates/pikl-tui/src/lib.rs +++ b/crates/pikl-tui/src/lib.rs @@ -374,7 +374,10 @@ fn render_menu( ), Span::raw(filter_text), Span::styled( - format!(" {filtered_count}/{total_count}"), + format!( + " {filtered_count}/{total_count}{}", + if vs.streaming { "..." } else { "" } + ), Style::default().fg(Color::DarkGray), ), ]; @@ -691,6 +694,7 @@ mod tests { mode: Mode::Insert, selection_count: 0, multi_enabled: false, + streaming: false, columns: None, generation: 1, } @@ -1166,6 +1170,7 @@ mod tests { mode: Mode::Insert, selection_count: 0, multi_enabled: false, + streaming: false, columns: None, generation: 1, }; @@ -1175,6 +1180,29 @@ mod tests { assert!(prompt.contains("0/0")); } + #[test] + fn streaming_indicator_shows_ellipsis() { + let mut vs = sample_view_state(); + vs.streaming = true; + let backend = render_to_backend(40, 5, &vs, ""); + let prompt = line_text(&backend, 0); + assert!( + prompt.contains("..."), + "streaming indicator should show '...': {prompt}" + ); + } + + #[test] + fn no_streaming_indicator_when_done() { + let vs = sample_view_state(); + let backend = render_to_backend(40, 5, &vs, ""); + let prompt = line_text(&backend, 0); + assert!( + !prompt.contains("..."), + "should not show '...' when not streaming: {prompt}" + ); + } + #[test] fn narrow_viewport_truncates() { let vs = sample_view_state(); @@ -1269,6 +1297,7 @@ mod tests { mode: Mode::Insert, selection_count: 0, multi_enabled: false, + streaming: false, columns: Some(vec![ ColumnHeader { display_name: "name".into(), diff --git a/crates/pikl/src/main.rs b/crates/pikl/src/main.rs index 3013988..e7e5771 100644 --- a/crates/pikl/src/main.rs +++ b/crates/pikl/src/main.rs @@ -16,7 +16,7 @@ use pikl_core::error::PiklError; use pikl_core::event::{Action, MenuResult, Mode}; use pikl_core::format::FormatTemplate; use pikl_core::hook::{HookEventKind, HookHandler}; -use pikl_core::input::read_items_sync; +use pikl_core::input::{parse_line_to_value, read_items_sync}; use pikl_core::item::Item; use pikl_core::json_menu::JsonMenu; use pikl_core::menu::MenuRunner; @@ -201,10 +201,6 @@ fn main() { std::process::exit(2); } - let timeout = cli - .stdin_timeout - .unwrap_or(if script.is_some() { 30 } else { 0 }); - let input_format = match InputFormat::from_str_opt(&cli.input_format) { Some(f) => f, None => { @@ -217,22 +213,55 @@ fn main() { } }; - let (items, csv_headers) = match read_stdin_with_timeout(timeout, &cli.label_key, input_format) - { - Ok(r) => r, - Err(e) => { - let _ = writeln!(std::io::stderr().lock(), "pikl: {e}"); + // Detect streaming mode: piped stdin, no action-fd, not CSV/TSV. + let streaming = script.is_none() + && !std::io::stdin().is_terminal() + && !matches!(input_format, InputFormat::Csv | InputFormat::Tsv); + + // Save the original stdin fd for the streaming reader before + // we dup2 /dev/tty onto fd 0. + #[cfg(unix)] + let saved_stdin_fd: Option = if streaming { + let fd = unsafe { libc::dup(libc::STDIN_FILENO) }; + if fd < 0 { + let _ = writeln!( + std::io::stderr().lock(), + "pikl: failed to dup stdin: {}", + std::io::Error::last_os_error() + ); std::process::exit(2); } + Some(fd) + } else { + None }; - if items.is_empty() { - let _ = writeln!( - std::io::stderr().lock(), - "pikl: empty input. nothing to pick from" - ); - std::process::exit(2); - } + let (items, csv_headers) = if streaming { + // Streaming mode: start with empty items, read in background later. + (Vec::new(), None) + } else { + let timeout = cli + .stdin_timeout + .unwrap_or(if script.is_some() { 30 } else { 0 }); + + let (items, csv_headers) = + match read_stdin_with_timeout(timeout, &cli.label_key, input_format) { + Ok(r) => r, + Err(e) => { + let _ = writeln!(std::io::stderr().lock(), "pikl: {e}"); + std::process::exit(2); + } + }; + + if items.is_empty() { + let _ = writeln!( + std::io::stderr().lock(), + "pikl: empty input. nothing to pick from" + ); + std::process::exit(2); + } + (items, csv_headers) + }; // Resolve column config: explicit --columns, or auto-generate from CSV headers let column_config = if let Some(ref cols) = cli.columns { @@ -288,12 +317,18 @@ fn main() { rt.block_on(run_headless(items, &cli, script, start_mode, column_config)) } else { let history_entries = history.as_ref().map(|h| h.entries().to_vec()); + #[cfg(unix)] + let saved_fd = saved_stdin_fd; + #[cfg(not(unix))] + let saved_fd: Option = None; rt.block_on(run_interactive( items, &cli, start_mode, column_config, history_entries, + streaming, + saved_fd, )) }; @@ -461,11 +496,14 @@ async fn run_interactive( start_mode: Mode, column_config: Option, filter_history: Option>, + streaming: bool, + saved_stdin_fd: Option, ) -> Result { let (mut menu, action_tx) = MenuRunner::new(build_menu(items, cli, column_config)); menu.set_initial_mode(start_mode); menu.set_multi(cli.multi); menu.set_selection_order(cli.selection_order); + menu.set_streaming(streaming); if let Some((_handler, dispatcher)) = build_hook_handler(cli, &action_tx) { menu.set_dispatcher(dispatcher); @@ -509,6 +547,51 @@ async fn run_interactive( let _ = signal_tx.send(Action::Cancel).await; }); + // Spawn background stdin reader for streaming mode. + if let Some(fd) = saved_stdin_fd { + let stream_tx = action_tx.clone(); + tokio::task::spawn_blocking(move || { + #[cfg(unix)] + { + use std::io::BufRead; + use std::os::unix::io::FromRawFd; + + let file = unsafe { std::fs::File::from_raw_fd(fd) }; + let mut reader = std::io::BufReader::new(file); + let mut batch = Vec::with_capacity(100); + let mut line = String::new(); + + loop { + line.clear(); + match reader.read_line(&mut line) { + Ok(0) => break, + Ok(_) => { + let trimmed = line.trim_end(); + if !trimmed.is_empty() { + batch.push(parse_line_to_value(trimmed)); + } + if (batch.len() >= 100 || reader.buffer().is_empty()) + && !batch.is_empty() + { + let _ = stream_tx.blocking_send(Action::AddItems( + std::mem::take(&mut batch), + )); + } + } + Err(e) => { + tracing::warn!(%e, "stdin read error"); + break; + } + } + } + if !batch.is_empty() { + let _ = stream_tx.blocking_send(Action::AddItems(batch)); + } + let _ = stream_tx.blocking_send(Action::StreamingDone); + } + }); + } + let tui_handle = tokio::spawn(async move { pikl_tui::run(action_tx, event_rx, filter_history).await }); diff --git a/docs/DEVPLAN.md b/docs/DEVPLAN.md index 913a763..5130f87 100644 --- a/docs/DEVPLAN.md +++ b/docs/DEVPLAN.md @@ -170,7 +170,7 @@ through hooks and structured I/O. A handler hook can receive hover events and emit commands to modify menu state. -## Phase 4: Multi-Select +## Phase 4: Multi-Select ✓ Selection with undo/redo. No marks or registers: filtering already handles navigation, and selections surviving filter @@ -275,21 +275,30 @@ pikl instance over the socket, push items, read state, and subscribe to events. A named session remembers filter history across invocations. -## Phase 7: Streaming & Watched Sources +## Phase 7: Streaming Stdin ✓ -Live, dynamic menus. +Auto-detected streaming when stdin is a pipe (not action-fd, +not CSV/TSV). The menu opens immediately with zero items and +streams them in as they arrive. + +`--watch` was dropped from this phase. External tools like +`inotifywait -m ~/walls | pikl` compose naturally with +streaming stdin. No built-in file watcher needed. **Deliverables:** -- Async/streaming stdin (items arrive over time, list - updates progressively) -- Streaming output (on-hover events emitted to stdout) -- `--watch path` for file/directory watching -- `--watch-extensions` filter -- notify crate integration (inotify on Linux, FSEvents - on macOS) +- Auto-detected streaming stdin (items arrive over time, + list updates progressively) +- `StreamingDone` action signals end of input +- `streaming` flag on ViewState for frontend indicators +- TUI shows `...` suffix on the count while streaming +- Background reader with batched `AddItems` (up to 100 + items per batch, flushes immediately for slow sources) +- `streaming-done` action-fd command for test scripts +- No new CLI flags: streaming is auto-detected -**Done when:** `find / -name '*.log' | pikl` populates -progressively. A watched directory updates the list live. +**Done when:** `seq 1 10000 | pikl` opens immediately and +items stream in. Slow sources show items appearing one at +a time with a `...` indicator. ## Phase 8: GUI Frontend (Wayland + X11) diff --git a/docs/use-cases/wallpaper-picker.md b/docs/use-cases/wallpaper-picker.md index 5033c79..d055c25 100644 --- a/docs/use-cases/wallpaper-picker.md +++ b/docs/use-cases/wallpaper-picker.md @@ -126,7 +126,7 @@ bind = $mod, B, exec, wallpicker copy | Hook debouncing | 3 | Don't flood hyprctl | | Fuzzy filtering | 2 | Filter by filename | | Manifest files | 3 | Reusable config | -| Watched sources | 7 | Live update on new files | +| Streaming stdin | 7 | Progressive load from find | | Sessions | 6 | Remember last position | ## Open Questions