crates/knut-pipelines/src/gui.rs
//! Desktop viewer (feature `gui`): an egui/eframe app that renders the
//! dataflow DAG as a live **node graph** with [`egui_snarl`], alongside a
//! Spark **job-progress** panel and a scrolling **event log**.
//!
//! The whole app is a pure function of one [`ViewerState`]: nodes are coloured
//! by each dataset's run status, the progress bars reflect the latest Spark
//! `ExecutionProgress`, and the log is the event stream. With the `grpc`
//! feature also on, [`launch`] can connect to a Spark Connect server and feed
//! the state from a live run on a background thread; without it, the viewer is
//! a static graph explorer.
//!
//! Everything shown here is also reachable from the CLI
//! ([`crate::viewer::dump_state`]) — the GUI is an extra lens on the same
//! model, never the only way to see a pipeline.
use eframe::egui::{self, Color32};
use egui_snarl::ui::{PinInfo, SnarlPin, SnarlStyle, SnarlViewer};
use egui_snarl::{InPin, NodeId, OutPin, Snarl};
use crate::model::OutputType;
#[cfg(feature = "grpc")]
use crate::progress::ViewerUpdate;
use crate::progress::{NodeStatus, ViewerState};
/// Colour for a node's run status — also used for its pins and wires.
fn status_color(status: NodeStatus) -> Color32 {
match status {
NodeStatus::Idle => Color32::from_gray(130),
NodeStatus::Running => Color32::from_rgb(232, 184, 48),
NodeStatus::Done => Color32::from_rgb(78, 192, 102),
NodeStatus::Failed => Color32::from_rgb(222, 74, 74),
}
}
/// A single colour per output kind, for the type chip in a node body.
fn type_color(t: OutputType) -> Color32 {
match t {
OutputType::Table => Color32::from_rgb(96, 160, 230),
OutputType::MaterializedView => Color32::from_rgb(150, 130, 230),
OutputType::TemporaryView => Color32::from_rgb(120, 200, 200),
OutputType::Sink => Color32::from_rgb(210, 150, 90),
OutputType::Unspecified => Color32::from_gray(150),
}
}
/// What each snarl node carries. Refreshed from [`ViewerState`] every frame so
/// the graph reflects live status without rebuilding the topology.
#[derive(Clone)]
struct NodeData {
name: String,
output_type: OutputType,
status: NodeStatus,
}
/// The [`SnarlViewer`] that draws each dataset node: one input pin (upstream
/// reads) and one output pin (downstream consumers), coloured by status.
struct DagViewer;
impl SnarlViewer<NodeData> for DagViewer {
fn title(&mut self, node: &NodeData) -> String {
node.name.clone()
}
fn inputs(&mut self, _node: &NodeData) -> usize {
1
}
fn outputs(&mut self, _node: &NodeData) -> usize {
1
}
// egui-snarl 0.10 dropped the `scale: f32` parameter from the pin/body hooks (the
// viewer no longer scales by hand — `SnarlStyle` owns it) and widened the pin return
// type from the concrete `PinInfo` to `impl SnarlPin + 'static`. `PinInfo` still
// implements `SnarlPin` (`egui-snarl-0.10.0/src/ui/pin.rs:224`), so the bodies below
// are unchanged — only the signatures moved.
fn show_input(
&mut self,
pin: &InPin,
_ui: &mut egui::Ui,
snarl: &mut Snarl<NodeData>,
) -> impl SnarlPin + 'static {
let color = snarl
.get_node(pin.id.node)
.map(|n| status_color(n.status))
.unwrap_or(Color32::GRAY);
PinInfo::circle().with_fill(color)
}
fn show_output(
&mut self,
pin: &OutPin,
_ui: &mut egui::Ui,
snarl: &mut Snarl<NodeData>,
) -> impl SnarlPin + 'static {
let color = snarl
.get_node(pin.id.node)
.map(|n| status_color(n.status))
.unwrap_or(Color32::GRAY);
PinInfo::circle().with_fill(color)
}
fn has_body(&mut self, _node: &NodeData) -> bool {
true
}
fn show_body(
&mut self,
node: NodeId,
_inputs: &[InPin],
_outputs: &[OutPin],
ui: &mut egui::Ui,
snarl: &mut Snarl<NodeData>,
) {
if let Some(n) = snarl.get_node(node) {
ui.horizontal(|ui| {
ui.colored_label(type_color(n.output_type), n.output_type.label());
ui.separator();
ui.colored_label(status_color(n.status), n.status.label());
});
}
}
}
/// Build a snarl graph from a [`ViewerState`], laying nodes out in topological
/// layers (left → right), and return the name→[`NodeId`] map for later
/// status refreshes and edge wiring.
fn build_snarl(
state: &ViewerState,
) -> (Snarl<NodeData>, std::collections::BTreeMap<String, NodeId>) {
use std::collections::BTreeMap;
let mut snarl = Snarl::new();
let mut ids: BTreeMap<String, NodeId> = BTreeMap::new();
// Longest-path layer per dataset, following flow edges.
let order = state.graph.topo_order().unwrap_or_else(|| {
state
.graph
.datasets
.iter()
.map(|d| d.name.clone())
.collect()
});
let mut layer: BTreeMap<String, usize> = BTreeMap::new();
for name in &order {
let mut l = 0;
for f in state.graph.flows_into(name) {
for r in &f.reads {
if let Some(rl) = layer.get(r) {
l = l.max(rl + 1);
}
}
}
layer.insert(name.clone(), l);
}
// Position: x by layer, y by index within layer.
let mut per_layer: BTreeMap<usize, usize> = BTreeMap::new();
for d in &state.graph.datasets {
let l = *layer.get(&d.name).unwrap_or(&0);
let row = per_layer.entry(l).or_insert(0);
let pos = egui::pos2(l as f32 * 240.0, *row as f32 * 130.0);
*row += 1;
let id = snarl.insert_node(
pos,
NodeData {
name: d.name.clone(),
output_type: d.output_type,
status: state.status_of(&d.name),
},
);
ids.insert(d.name.clone(), id);
}
// Wire edges: for each flow, connect each read's output → target's input.
for f in &state.graph.flows {
let Some(&tgt) = ids.get(&f.target) else {
continue;
};
for r in &f.reads {
if let Some(&src) = ids.get(r) {
snarl.connect(
egui_snarl::OutPinId {
node: src,
output: 0,
},
egui_snarl::InPinId {
node: tgt,
input: 0,
},
);
}
}
}
(snarl, ids)
}
/// A live update channel fed by a background run (only with `grpc`).
#[cfg(feature = "grpc")]
type UpdateRx = std::sync::mpsc::Receiver<crate::Result<ViewerUpdate>>;
/// The egui application.
///
/// `pub` so a **headless robot test** can build one, apply synthetic
/// [`ViewerUpdate`](crate::ViewerUpdate)s, run real egui frames via
/// [`ViewerApp::draw`], and assert on [`ViewerApp::app_state_json`] — driving
/// the true snarl-viewer render path with no window and no Spark server. (egui
/// 0.28, which egui-snarl 0.4 pins, has no `egui_kittest` — that crate starts at
/// 0.30 — so the robot runs frames through `egui::Context::run_ui` directly, the
/// same headless frame-run kittest wraps internally.)
pub struct ViewerApp {
state: ViewerState,
snarl: Snarl<NodeData>,
ids: std::collections::BTreeMap<String, NodeId>,
style: SnarlStyle,
/// Connection URL, shown in the header.
connect: Option<String>,
/// Live updates from a background run, if connected.
#[cfg(feature = "grpc")]
rx: Option<UpdateRx>,
/// A terminal error from the live stream, if any.
error: Option<String>,
/// If set (`KNUT_SCREENSHOT=<path>`), capture one frame to a PNG and exit —
/// for generating viewer screenshots in the docs.
screenshot: Option<String>,
/// Frame counter, used to let the layout settle before the screenshot.
frame: u32,
/// Introspection: where the live UI-state JSON is written (`$KNUT_VIEWER_STATE`,
/// default `<tmp>/knut_viewer_state.json`) so the viewer can be robot-tested
/// headlessly — always on, overwritten in place ~3×/s.
state_path: std::path::PathBuf,
/// Last state-dump time, for the ~300 ms throttle.
last_dump: Option<std::time::Instant>,
/// Last click-trail line emitted, to suppress duplicate stderr lines.
last_trail: String,
}
impl ViewerApp {
/// Build the viewer app over `state` (and an optional connect endpoint,
/// shown in the header). Public so a headless robot test can construct the
/// real app and drive it frame-by-frame.
pub fn new(state: ViewerState, connect: Option<String>) -> Self {
let (snarl, ids) = build_snarl(&state);
ViewerApp {
state,
snarl,
ids,
style: SnarlStyle::new(),
connect,
#[cfg(feature = "grpc")]
rx: None,
error: None,
screenshot: std::env::var("KNUT_SCREENSHOT").ok(),
frame: 0,
state_path: std::env::var("KNUT_VIEWER_STATE")
.map(std::path::PathBuf::from)
.unwrap_or_else(|_| std::env::temp_dir().join("knut_viewer_state.json")),
last_dump: None,
last_trail: String::new(),
}
}
/// The full introspection snapshot: the [`ViewerState`] data layer
/// ([`ViewerState::state_json`]) plus the GUI-only fields (source endpoint,
/// derived connect status, terminal error). Mirrors nornir viz's
/// `state_json()`; **any new visible control must be added here.**
pub fn app_state_json(&self) -> serde_json::Value {
let mut v = self.state.state_json();
let source = self
.connect
.clone()
.unwrap_or_else(|| "static (no server)".into());
let connect_status = if let Some(e) = &self.error {
format!("error: {e}")
} else if let Some(url) = &self.connect {
format!("connected {url}")
} else {
"local (no server)".into()
};
if let Some(obj) = v.as_object_mut() {
obj.insert("source".into(), serde_json::json!(source));
obj.insert("connect_status".into(), serde_json::json!(connect_status));
obj.insert("error".into(), serde_json::json!(self.error));
}
v
}
/// Write the introspection JSON to `state_path`, throttled to ~300 ms — the
/// always-on external-state channel for headless robot-testing.
fn dump_state_if_due(&mut self) {
let now = std::time::Instant::now();
if let Some(prev) = self.last_dump
&& now.duration_since(prev) < std::time::Duration::from_millis(300)
{
return;
}
self.last_dump = Some(now);
if let Ok(json) = serde_json::to_vec_pretty(&self.app_state_json()) {
let _ = std::fs::write(&self.state_path, json);
}
}
/// The click/activity trail on stderr: one line per meaningful state change
/// (last event + the status summary), de-duplicated. The "what happened"
/// log a headless test or operator reads alongside the JSON dump.
fn emit_trail(&mut self) {
let (idle, running, done, failed) = self.state.status_counts();
let last = self
.state
.events
.last()
.map(|e| e.message.as_str())
.unwrap_or("—");
let line = format!(
"knut-viewer: {last} | {done} done · {running} running · {failed} failed · {idle} idle"
);
if line != self.last_trail {
eprintln!("{line}");
self.last_trail = line;
}
}
/// Drive the screenshot state machine: after a few frames request a capture;
/// when the reply arrives, save it as PNG and close the window.
fn handle_screenshot(&mut self, ctx: &egui::Context) {
let Some(path) = self.screenshot.clone() else {
return;
};
self.frame += 1;
// Let the snarl graph lay out + paint first.
if self.frame == 8 {
// egui 0.34's `Screenshot` carries a `UserData` payload so a caller can tag
// the request and match it on the way back out of `Event::Screenshot`. This
// shot is identified by frame number instead, so the payload is empty.
ctx.send_viewport_cmd(egui::ViewportCommand::Screenshot(egui::UserData::default()));
}
let shot = ctx.input(|i| {
i.raw.events.iter().find_map(|e| match e {
egui::Event::Screenshot { image, .. } => Some(image.clone()),
_ => None,
})
});
if let Some(image) = shot {
let [w, h] = image.size;
let mut rgba = Vec::with_capacity(w * h * 4);
for px in &image.pixels {
rgba.extend_from_slice(&[px.r(), px.g(), px.b(), px.a()]);
}
match image::RgbaImage::from_raw(w as u32, h as u32, rgba) {
Some(buf) => match buf.save(&path) {
Ok(()) => eprintln!("screenshot saved to {path} ({w}x{h})"),
Err(e) => eprintln!("screenshot save failed: {e}"),
},
None => eprintln!("screenshot buffer size mismatch"),
}
ctx.send_viewport_cmd(egui::ViewportCommand::Close);
} else {
ctx.request_repaint(); // keep frames advancing until the reply lands
}
}
/// Push the current per-node status from the state into the snarl nodes so
/// the colours track the live run without rebuilding the graph.
fn refresh_node_status(&mut self) {
for (name, id) in &self.ids {
if let Some(n) = self.snarl.get_node_mut(*id) {
n.status = self.state.status_of(name);
}
}
}
/// **Headless-drive seam** (robot test): apply one [`ViewerUpdate`] straight
/// into the viewer state — the synthetic, no-server equivalent of a live
/// update arriving over the wire — and refresh the snarl node colours so the
/// DAG reflects it, exactly as `pump_updates` does for a real stream. Returns
/// whether the model moved.
pub fn apply_update(&mut self, update: crate::ViewerUpdate) -> bool {
let changed = self.state.apply(update);
if changed {
self.refresh_node_status();
}
changed
}
/// The run status carried by the **snarl node** for `dataset` (as drawn in
/// the DAG viewer), or `None` if there is no such node. Lets a robot test
/// assert the graph node — not merely the backing state — transitioned.
pub fn snarl_status_of(&self, dataset: &str) -> Option<NodeStatus> {
let id = self.ids.get(dataset)?;
self.snarl.get_node(*id).map(|n| n.status)
}
/// Number of nodes currently in the snarl graph (one per dataset).
pub fn snarl_node_count(&self) -> usize {
self.snarl.nodes().count()
}
/// Render one real egui frame — every panel plus the snarl DAG — into the root
/// `ui`, with no window and no `eframe::Frame`. This is the exact body the
/// [`eframe::App::ui`] impl runs; a robot test calls it via
/// `egui::Context::run_ui` to drive the true render path headlessly.
///
/// egui 0.36 (2026-10-02): panels are `egui::Panel` and show inside a `Ui`
/// (the context-level `TopBottomPanel`/`SidePanel::show(ctx)` are gone), so
/// `draw` takes the root `Ui` instead of the `Context`.
pub fn draw(&mut self, ui: &mut egui::Ui) {
let ctx = ui.ctx().clone();
// Only push status into the snarl nodes when an update actually landed;
// an idle frame leaves the graph untouched (H12). The initial state is
// already reflected because `build_snarl` coloured the nodes from it.
if self.pump_updates() {
self.refresh_node_status();
self.emit_trail(); // stderr activity trail, only on a real change
}
// Always-on introspection: overwrite the state JSON (throttled) so the
// viewer is observable headlessly (robot-testing).
self.dump_state_if_due();
egui::Panel::top("header").show(ui, |ui| {
ui.horizontal(|ui| {
ui.heading("knut · Spark pipeline viewer");
ui.separator();
let id = self.state.graph.graph_id.as_deref().unwrap_or("<static>");
ui.label(format!("graph {id}"));
if let Some(url) = &self.connect {
ui.separator();
ui.label(format!("● {url}"));
}
});
});
egui::Panel::right("side")
.default_size(320.0)
.show(ui, |ui| {
let (idle, running, done, failed) = self.state.status_counts();
ui.heading("Run status");
ui.label(format!(
"{done} done · {running} running · {failed} failed · {idle} idle"
));
ui.separator();
ui.heading("Spark job progress");
if let Some(p) = &self.state.progress {
ui.add(egui::ProgressBar::new(p.fraction()).show_percentage());
ui.label(p.summary());
ui.add_space(4.0);
egui::ScrollArea::vertical()
.max_height(140.0)
.id_salt("stages")
.show(ui, |ui| {
for s in &p.stages {
ui.add(egui::ProgressBar::new(s.fraction()).text(format!(
"stage {} — {}/{}",
s.stage_id, s.num_completed_tasks, s.num_tasks
)));
}
});
} else {
ui.label("no job running");
}
ui.separator();
ui.heading("Events");
if let Some(err) = &self.error {
ui.colored_label(
status_color(NodeStatus::Failed),
format!("stream error: {err}"),
);
}
egui::ScrollArea::vertical()
.stick_to_bottom(true)
.show(ui, |ui| {
for ev in &self.state.events {
let color = match ev.level {
crate::model::EventLevel::Error => status_color(NodeStatus::Failed),
crate::model::EventLevel::Warn => status_color(NodeStatus::Running),
crate::model::EventLevel::Info => Color32::from_gray(200),
};
ui.colored_label(color, &ev.message);
}
});
});
egui::CentralPanel::default().show(ui, |ui| {
self.snarl.show(&mut DagViewer, &self.style, "dag", ui);
});
// Screenshot mode: capture one settled frame to PNG, then exit. This
// path drives its own repaints (so it must keep doing so below).
self.handle_screenshot(&ctx);
// Event-driven repaint (H12): the background thread calls
// `ctx.request_repaint()` after each `tx.send`, so the UI wakes exactly
// when a message arrives instead of busy-spinning at 10 Hz. A slow
// safety fallback covers any missed wakeup without burning CPU at idle.
#[cfg(feature = "grpc")]
if self.rx.is_some() {
ctx.request_repaint_after(std::time::Duration::from_secs(2));
}
}
/// Drain any pending live updates into the state. Returns `true` when at
/// least one update (or error) was applied, so the caller repaints/rebuilds
/// derived state only when the model actually moved (H12).
#[cfg(feature = "grpc")]
fn pump_updates(&mut self) -> bool {
let Some(rx) = &self.rx else { return false };
// Collect first to avoid borrowing self while mutating it.
let mut drained = Vec::new();
while let Ok(item) = rx.try_recv() {
drained.push(item);
}
let mut changed = false;
for item in drained {
match item {
Ok(update) => changed |= self.state.apply(update),
Err(e) => {
self.error = Some(e.to_string());
changed = true;
}
}
}
changed
}
#[cfg(not(feature = "grpc"))]
fn pump_updates(&mut self) -> bool {
false
}
}
impl eframe::App for ViewerApp {
// `ui`, not `update`. eframe 0.34 made `ui` the required entry point and marked
// `update` `#[deprecated = "Use Self::ui instead"]`, and it calls BOTH in sequence
// (`native/epi_integration.rs:294,299`) — so leaving the body in `update` would have
// kept working while quietly riding a deprecated hook, and implementing both would
// have drawn the graph twice per frame.
//
// `draw` takes the root `Ui` (egui 0.36 panels show inside a `Ui`), so the window
// loop and the headless robot keep rendering through the one `draw` — CLI/GUI
// parity's GUI-side twin, driven either by eframe's event loop or by a test's
// `egui::Context::run_ui`.
fn ui(&mut self, ui: &mut egui::Ui, _frame: &mut eframe::Frame) {
self.draw(ui);
}
}
/// A deferred background-spawn: given the live UI's [`egui::Context`] (so the
/// thread can wake the UI on each message), start the run and return its
/// receiver. Run inside the eframe creator closure, where the context exists.
#[cfg(feature = "grpc")]
type SpawnFn = Box<dyn FnOnce(egui::Context) -> UpdateRx>;
/// Launch the desktop viewer over `state`. When the `grpc` feature is enabled
/// and `connect` is `Some(url)`, a background thread drives a live run and
/// streams updates into the view.
pub fn launch(mut state: ViewerState, connect: Option<String>) -> crate::Result<()> {
// In screenshot mode, seed some run state so the captured frame shows the
// coloured DAG, job-progress bars, and an event log (not an all-idle graph).
if std::env::var("KNUT_SCREENSHOT").is_ok() {
seed_demo_run(&mut state);
}
let app = ViewerApp::new(state, connect.clone());
// The spawn is deferred until the eframe creator runs, so it can hand the
// live `egui::Context` to the background thread for repaint-on-message.
#[cfg(feature = "grpc")]
let spawn: Option<SpawnFn> =
connect.map(|url| Box::new(move |ctx: egui::Context| spawn_live(url, ctx)) as SpawnFn);
#[cfg(feature = "grpc")]
return run_app(app, spawn);
#[cfg(not(feature = "grpc"))]
run_app(app)
}
/// Launch the viewer over a **pipeline spec** and drive a live define+run in
/// the background (needs `grpc`). The spec graph renders immediately; the run's
/// events colour its nodes and fill the progress/event panels as they arrive.
#[cfg(feature = "grpc")]
pub fn launch_live_pipeline(
state: ViewerState,
url: String,
graph: crate::DataflowGraph,
storage: Option<String>,
) -> crate::Result<()> {
let app = ViewerApp::new(state, Some(url.clone()));
let spawn: SpawnFn =
Box::new(move |ctx: egui::Context| spawn_pipeline(url, graph, storage, ctx));
run_app(app, Some(spawn))
}
/// Run the eframe app to completion. With `grpc`, an optional deferred spawn is
/// run inside the creator closure (where the live [`egui::Context`] exists) so
/// the background thread can wake the UI on each message.
#[cfg(feature = "grpc")]
fn run_app(mut app: ViewerApp, spawn: Option<SpawnFn>) -> crate::Result<()> {
let native_options = eframe::NativeOptions::default();
eframe::run_native(
"knut-pipelines",
native_options,
Box::new(move |cc| {
if let Some(spawn) = spawn {
app.rx = Some(spawn(cc.egui_ctx.clone()));
}
Ok(Box::new(app))
}),
)
.map_err(|e| crate::PipelineError::Other(format!("eframe: {e}")))
}
/// Run the eframe app to completion (no live stream without `grpc`).
#[cfg(not(feature = "grpc"))]
fn run_app(app: ViewerApp) -> crate::Result<()> {
let native_options = eframe::NativeOptions::default();
eframe::run_native(
"knut-pipelines",
native_options,
Box::new(|_cc| Ok(Box::new(app))),
)
.map_err(|e| crate::PipelineError::Other(format!("eframe: {e}")))
}
/// Seed a viewer state with a representative mid-run snapshot — some nodes
/// done/running, a job-progress reading, and a few log lines — so a docs
/// screenshot shows the viewer doing its job rather than an empty graph.
fn seed_demo_run(state: &mut ViewerState) {
use crate::model::{EventLevel, PipelineRunEvent};
use crate::progress::{ExecutionProgress, StageProgress};
let names: Vec<String> = state
.graph
.datasets
.iter()
.map(|d| d.name.clone())
.collect();
if let Some(first) = names.first() {
state.apply(crate::ViewerUpdate::Event(PipelineRunEvent {
timestamp: None,
level: EventLevel::Info,
message: format!("Flow {first} has COMPLETED."),
element: Some(first.clone()),
}));
}
if let Some(second) = names.get(1) {
state.apply(crate::ViewerUpdate::Event(PipelineRunEvent {
timestamp: None,
level: EventLevel::Info,
message: format!("Flow {second} is RUNNING."),
element: Some(second.clone()),
}));
}
state.apply(crate::ViewerUpdate::Progress(ExecutionProgress {
stages: vec![
StageProgress {
stage_id: 1,
num_tasks: 16,
num_completed_tasks: 16,
input_bytes_read: 4_194_304,
done: true,
},
StageProgress {
stage_id: 2,
num_tasks: 16,
num_completed_tasks: 7,
input_bytes_read: 1_048_576,
done: false,
},
StageProgress {
stage_id: 3,
num_tasks: 1,
num_completed_tasks: 0,
input_bytes_read: 0,
done: false,
},
],
num_inflight_tasks: 9,
}));
}
/// Spawn a background thread that connects, **defines** the pipeline graph,
/// **runs** it, and forwards each [`ViewerUpdate`] over a channel.
#[cfg(feature = "grpc")]
fn spawn_pipeline(
url: String,
graph: crate::DataflowGraph,
storage: Option<String>,
ctx: egui::Context,
) -> UpdateRx {
use futures_util::StreamExt;
let (tx, rx) = std::sync::mpsc::channel();
std::thread::spawn(move || {
// Wake the UI whenever we hand it a message, so it can repaint without
// polling (H12).
let send = |item| {
let ok = tx.send(item).is_ok();
ctx.request_repaint();
ok
};
let rt = match tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
{
Ok(rt) => rt,
Err(e) => {
send(Err(crate::PipelineError::Other(format!("runtime: {e}"))));
return;
}
};
rt.block_on(async move {
let mut client = match crate::client::PipelinesClient::connect(&url).await {
Ok(c) => c,
Err(e) => {
send(Err(e));
return;
}
};
let graph_id = match client.define_graph(&graph).await {
Ok(id) => id,
Err(e) => {
send(Err(e));
return;
}
};
let run = crate::model::PipelineRun {
graph_id: Some(graph_id),
full_refresh_all: true,
storage,
..Default::default()
};
match client.run_updates(&run).await {
Ok(mut stream) => {
while let Some(item) = stream.next().await {
if !send(item) {
break;
}
}
}
Err(e) => {
send(Err(e));
}
}
});
});
rx
}
/// Spawn a background thread that connects to Spark Connect, starts a run, and
/// forwards each [`ViewerUpdate`] over a channel the UI drains each frame.
#[cfg(feature = "grpc")]
fn spawn_live(url: String, ctx: egui::Context) -> UpdateRx {
use futures_util::StreamExt;
let (tx, rx) = std::sync::mpsc::channel();
std::thread::spawn(move || {
// Wake the UI on every message so it repaints event-driven (H12).
let send = |item| {
let ok = tx.send(item).is_ok();
ctx.request_repaint();
ok
};
let rt = match tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
{
Ok(rt) => rt,
Err(e) => {
send(Err(crate::PipelineError::Other(format!("runtime: {e}"))));
return;
}
};
rt.block_on(async move {
let mut client = match crate::client::PipelinesClient::connect(&url).await {
Ok(c) => c,
Err(e) => {
send(Err(e));
return;
}
};
// A default run against whatever graph the server has active.
let run = crate::model::PipelineRun::default();
match client.run_updates(&run).await {
Ok(mut stream) => {
while let Some(item) = stream.next().await {
if !send(item) {
break; // UI gone
}
}
}
Err(e) => {
send(Err(e));
}
}
});
});
rx
}
#[cfg(test)]
mod tests {
use super::*;
use crate::model::{Dataset, Flow, OutputType};
fn state() -> ViewerState {
ViewerState::new(
crate::DataflowGraph::new()
.with_dataset(Dataset::new("a", OutputType::Table))
.with_dataset(Dataset::new("b", OutputType::MaterializedView))
.with_flow(Flow::new("f", "b", ["a"])),
)
}
#[test]
fn build_snarl_has_a_node_per_dataset() {
let (snarl, ids) = build_snarl(&state());
assert_eq!(ids.len(), 2);
assert!(ids.contains_key("a") && ids.contains_key("b"));
assert_eq!(snarl.nodes().count(), 2);
}
#[test]
fn status_colors_are_distinct() {
assert_ne!(
status_color(NodeStatus::Done),
status_color(NodeStatus::Failed)
);
assert_ne!(
status_color(NodeStatus::Idle),
status_color(NodeStatus::Running)
);
}
}