pub mod app;
#[cfg(test)]
pub mod frames;
pub mod input;
pub mod link;
pub mod source;
pub mod tail;
pub mod term;
pub mod theme;
pub mod view;
use std::io::IsTerminal;
use std::path::Path;
use std::time::{Duration, Instant};
use futures_util::{Stream, StreamExt};
use ratatui::Terminal;
use ratatui::backend::{Backend, CrosstermBackend};
use shep_core::paths::ShepPaths;
use tokio::sync::mpsc;
use self::app::{App, Control, Effect, Msg, Sent};
use self::source::Shepherd;
use self::theme::Palette;
use crate::cli::LookoutArgs;
use crate::exit::ExitCode;
use crate::output::Streams;
pub const HEARTBEAT: Duration = Duration::from_secs(1);
pub const MIN_REDRAW: Duration = Duration::from_millis(33);
pub async fn lookout(streams: &mut Streams<'_>, paths: &ShepPaths, args: &LookoutArgs) -> ExitCode {
if !std::io::stdout().is_terminal() {
return streams.fail(
ExitCode::Usage,
"lookout needs a terminal; stdout is not one",
);
}
let mut shepherd = source::UnixShepherd::new(&paths.socket);
let opened = match shepherd.link().await {
Ok(opened) => opened,
Err(err) => {
let code = err.exit_code();
return streams.fail(code, &err.to_string());
}
};
let palette = Palette::detect(
std::env::var_os("NO_COLOR").as_deref(),
std::env::var_os("TERM").as_deref(),
std::env::var_os("COLORTERM").as_deref(),
);
let control = resolve_control(args.allow_control, &paths.kv);
let app = App::new(
palette,
control,
paths.home.to_string_lossy().into_owned(),
Instant::now(),
);
term::install_panic_hook();
let _guard = term::RestoreGuard::new();
let out = match term::enter() {
Ok(out) => out,
Err(err) => {
return streams.fail(
ExitCode::Failure,
&format!("could not put the terminal into raw mode: {err}"),
);
}
};
let terminal = match Terminal::new(CrosstermBackend::new(out)) {
Ok(terminal) => terminal,
Err(err) => {
return streams.fail(
ExitCode::Failure,
&format!("could not open the terminal: {err}"),
);
}
};
let (msg_tx, msg_rx) = mpsc::channel(1024);
let (poll_tx, poll_rx) = mpsc::channel(8);
let (request_tx, request_rx) = mpsc::channel(2);
let link = tokio::spawn(link::run_link(
shepherd,
opened,
msg_tx,
link::Channels {
polls: poll_rx,
requests: request_rx,
},
link::FLOCK_POLL,
));
let events = crossterm::event::EventStream::new();
let _ = run_ui(
app,
terminal,
events,
msg_rx,
poll_tx,
request_tx,
source::LocalReader::new(),
)
.await;
link.abort();
ExitCode::Success
}
#[must_use]
pub fn resolve_control(flag: bool, kv: &Path) -> Control {
if flag {
return Control::Allowed;
}
match shep_core::kv::get(kv, "lookout.allow_control") {
Ok(Some(value)) if value == "true" => Control::Allowed,
_ => Control::ReadOnly,
}
}
pub async fn run_ui<B: Backend, S, L>(
mut app: App,
mut terminal: Terminal<B>,
events: S,
mut msgs: mpsc::Receiver<Msg>,
polls: mpsc::Sender<()>,
requests: mpsc::Sender<self::app::Sent>,
mut local: L,
) -> Terminal<B>
where
S: Stream<Item = std::io::Result<crossterm::event::Event>> + Unpin,
L: source::Local,
{
let mut events = events;
let mut heartbeat = tokio::time::interval(HEARTBEAT);
heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
let mut sigterm =
tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()).ok();
let mut keys_done = false;
let mut link_done = false;
let mut dirty = true;
let mut feed_dirty = false;
let mut lambs_dirty = false;
let mut last_draw: Option<Instant> = None;
loop {
let may_draw = last_draw.is_none_or(|at| at.elapsed() >= MIN_REDRAW);
if feed_dirty && may_draw {
let tail = match app.selected_row() {
None => tail::Tail::default(),
Some(row) => {
let (out, err) = (row.info.out_file.clone(), row.info.err_file.clone());
local.tail(out.as_deref().map(Path::new), err.as_deref().map(Path::new))
}
};
let _ = app.update(Msg::Bleats { tail });
feed_dirty = false;
dirty = true;
}
if lambs_dirty && may_draw {
let height = terminal.size().map_or(0, |size| size.height);
if view::panes_for(height).detail
&& let Some(id) = app.selected()
{
let _ = requests.try_send(Sent::Lambs { id });
}
lambs_dirty = false;
}
if dirty && may_draw {
let _ = terminal.draw(|frame| view::draw(&app, frame));
dirty = false;
last_draw = Some(Instant::now());
}
let msg = tokio::select! {
biased;
() = async {
match sigterm.as_mut() {
Some(signal) => {
signal.recv().await;
}
None => std::future::pending().await,
}
} => break,
event = events.next(), if !keys_done => match event {
Some(Ok(crossterm::event::Event::Resize(..))) => Some(Msg::Resize),
Some(Ok(event)) => input::map_key(&event, app.mode()).map(Msg::Key),
Some(Err(_)) | None => {
keys_done = true;
None
}
},
msg = msgs.recv(), if !link_done => match msg {
Some(msg) => Some(msg),
None => {
link_done = true;
None
}
},
_ = heartbeat.tick() => {
let _ = app.update(Msg::Host { sample: local.host() });
Some(Msg::Tick { now: Instant::now() })
}
};
let Some(msg) = msg else { continue };
match app.update(msg) {
Effect::Quit => break,
Effect::PollNow => {
let _ = polls.try_send(());
lambs_dirty = true;
dirty = true;
}
Effect::RefreshFeed => {
feed_dirty = true;
dirty = true;
}
Effect::RefreshSelected => {
feed_dirty = true;
lambs_dirty = true;
dirty = true;
}
Effect::Send(sent) => {
if let Err(err) = requests.try_send(sent) {
let (mpsc::error::TrySendError::Full(sent)
| mpsc::error::TrySendError::Closed(sent)) = err;
let _ = app.update(Msg::Unsent { sent });
}
dirty = true;
}
Effect::None => dirty = true,
}
}
terminal
}
#[cfg(test)]
mod tests {
use std::path::Path;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use super::*;
use futures_util::stream;
use ratatui::Terminal;
use ratatui::backend::TestBackend;
use shep_core::protocol::{BusEvent, ProcessInfo};
use shep_core::status::ProcStatus;
use crate::lookout::app::{App, Control, KeyPress, Sent};
use crate::lookout::source::{HostSample, Local};
use crate::lookout::tail::Tail;
use crate::lookout::theme::Palette;
#[derive(Clone, Default)]
struct FakeLocal {
sample: Option<HostSample>,
hosts: Arc<AtomicUsize>,
tails: Arc<AtomicUsize>,
}
impl Local for FakeLocal {
fn host(&mut self) -> Option<HostSample> {
self.hosts.fetch_add(1, Ordering::Relaxed);
self.sample
}
fn tail(&mut self, _out: Option<&Path>, _err: Option<&Path>) -> Tail {
self.tails.fetch_add(1, Ordering::Relaxed);
Tail::default()
}
}
#[tokio::test(start_paused = true)]
async fn the_loop_draws_and_quits_on_a_keypress() {
let (msg_tx, msg_rx) = mpsc::channel(16);
let (poll_tx, _poll_rx) = mpsc::channel(1);
let (request_tx, _request_rx) = mpsc::channel(2);
let app = App::new(
Palette::detect(None, None, None),
Control::ReadOnly,
"/tmp/shep".to_string(),
Instant::now(),
);
let terminal = Terminal::new(TestBackend::new(80, 12)).unwrap();
let keys = stream::iter(vec![Ok(crossterm::event::Event::Key(
crossterm::event::KeyEvent::new(
crossterm::event::KeyCode::Char('q'),
crossterm::event::KeyModifiers::NONE,
),
))]);
drop(msg_tx);
let done = tokio::time::timeout(
Duration::from_secs(10),
run_ui(
app,
terminal,
keys,
msg_rx,
poll_tx,
request_tx,
FakeLocal::default(),
),
)
.await;
let terminal = done.expect("the loop left on `q` within ten seconds");
let frame = crate::lookout::frames::render_text(terminal.backend().buffer());
assert!(frame.contains("shep lookout"), "it drew at least once");
}
#[tokio::test(start_paused = true)]
async fn a_drop_forwards_a_poll_request_to_the_link_task() {
let (msg_tx, msg_rx) = mpsc::channel(16);
let (poll_tx, mut poll_rx) = mpsc::channel(4);
let (request_tx, _request_rx) = mpsc::channel(2);
msg_tx
.send(Msg::Event(BusEvent::Dropped { count: 4 }))
.await
.unwrap();
msg_tx.send(Msg::Key(KeyPress::Quit)).await.unwrap();
drop(msg_tx);
let app = App::new(
Palette::detect(None, None, None),
Control::ReadOnly,
"/tmp/shep".to_string(),
Instant::now(),
);
let terminal = Terminal::new(TestBackend::new(80, 12)).unwrap();
let _ = tokio::time::timeout(
Duration::from_secs(10),
run_ui(
app,
terminal,
stream::empty(),
msg_rx,
poll_tx,
request_tx,
FakeLocal::default(),
),
)
.await
.expect("the loop left within ten seconds");
assert_eq!(poll_rx.try_recv(), Ok(()), "the poll request was forwarded");
}
#[test]
fn the_flag_wins_over_the_store_and_the_store_is_read() {
let dir = tempfile::tempdir().unwrap();
let kv = dir.path().join("kv.json");
assert_eq!(resolve_control(false, &kv), Control::ReadOnly);
shep_core::kv::set(&kv, "lookout.allow_control", "true").unwrap();
assert_eq!(resolve_control(false, &kv), Control::Allowed);
assert_eq!(resolve_control(true, &kv), Control::Allowed);
shep_core::kv::set(&kv, "lookout.allow_control", "false").unwrap();
assert_eq!(resolve_control(false, &kv), Control::ReadOnly);
assert_eq!(
resolve_control(true, &kv),
Control::Allowed,
"the flag wins over a store that says no"
);
}
#[tokio::test(start_paused = true)]
async fn the_heartbeat_asks_the_local_reader_for_a_host_sample() {
let (msg_tx, msg_rx) = mpsc::channel(64);
let (poll_tx, _poll_rx) = mpsc::channel(4);
let (request_tx, _request_rx) = mpsc::channel(2);
let local = FakeLocal::default();
let hosts = Arc::clone(&local.hosts);
tokio::spawn(async move {
tokio::time::sleep(Duration::from_millis(1_500)).await;
let _ = msg_tx.send(Msg::Key(KeyPress::Quit)).await;
});
let app = App::new(
Palette::detect(None, None, None),
Control::ReadOnly,
"/tmp/shep".to_string(),
Instant::now(),
);
let terminal = Terminal::new(TestBackend::new(120, 24)).unwrap();
let _ = tokio::time::timeout(
Duration::from_secs(10),
run_ui(
app,
terminal,
stream::empty(),
msg_rx,
poll_tx,
request_tx,
local,
),
)
.await
.expect("the loop left within ten seconds");
assert!(
hosts.load(Ordering::Relaxed) >= 1,
"the heartbeat fired and never sampled the host"
);
}
#[tokio::test]
async fn a_heartbeat_puts_the_host_strip_on_the_frame() {
let (msg_tx, msg_rx) = mpsc::channel(64);
let (poll_tx, _poll_rx) = mpsc::channel(4);
let (request_tx, _request_rx) = mpsc::channel(2);
let local = FakeLocal {
sample: Some(crate::lookout::view::fixtures::sample()),
..FakeLocal::default()
};
tokio::spawn(async move {
tokio::time::sleep(MIN_REDRAW * 3).await;
let _ = msg_tx.send(Msg::Resize).await;
tokio::time::sleep(MIN_REDRAW).await;
let _ = msg_tx.send(Msg::Key(KeyPress::Quit)).await;
});
let app = App::new(
Palette::detect(None, None, None),
Control::ReadOnly,
"/tmp/shep".to_string(),
Instant::now(),
);
let terminal = Terminal::new(TestBackend::new(120, 24)).unwrap();
let terminal = tokio::time::timeout(
Duration::from_secs(10),
run_ui(
app,
terminal,
stream::empty(),
msg_rx,
poll_tx,
request_tx,
local,
),
)
.await
.expect("the loop left within ten seconds");
let frame = crate::lookout::frames::render_text(terminal.backend().buffer());
assert!(
frame.contains("host load 2.31 4.10 3.88 / 10 cores"),
"the strip drew the sample the heartbeat took: {frame}"
);
assert!(
!frame.contains("not read yet"),
"and not the pre-heartbeat sentence"
);
}
#[tokio::test]
async fn a_burst_of_selection_moves_costs_one_read_and_not_one_per_key() {
let (msg_tx, msg_rx) = mpsc::channel(64);
let (poll_tx, _poll_rx) = mpsc::channel(4);
let (request_tx, _request_rx) = mpsc::channel(2);
let local = FakeLocal::default();
let tails = Arc::clone(&local.tails);
let at = Instant::now();
msg_tx
.send(Msg::Snapshot {
rows: (0..8)
.map(|id| {
ProcessInfo::builder(id, format!("sheep-{id}"), ProcStatus::Online).build()
})
.collect(),
at,
})
.await
.unwrap();
for _ in 0..20 {
msg_tx.send(Msg::Key(KeyPress::SelectDown)).await.unwrap();
}
tokio::spawn(async move {
tokio::time::sleep(MIN_REDRAW * 3).await;
let _ = msg_tx.send(Msg::Resize).await;
tokio::time::sleep(MIN_REDRAW).await;
let _ = msg_tx.send(Msg::Key(KeyPress::Quit)).await;
});
let app = App::new(
Palette::detect(None, None, None),
Control::ReadOnly,
"/tmp/shep".to_string(),
Instant::now(),
);
let terminal = Terminal::new(TestBackend::new(120, 24)).unwrap();
let _ = tokio::time::timeout(
Duration::from_secs(5),
run_ui(
app,
terminal,
stream::empty(),
msg_rx,
poll_tx,
request_tx,
local,
),
)
.await
.expect("the loop left within five seconds");
assert_eq!(
tails.load(Ordering::Relaxed),
1,
"a snapshot and twenty selection moves must coalesce into one read"
);
}
#[tokio::test]
async fn a_burst_of_selection_moves_costs_one_lamb_request() {
let (msg_tx, msg_rx) = mpsc::channel(64);
let (poll_tx, _poll_rx) = mpsc::channel(4);
let (request_tx, mut request_rx) = mpsc::channel(2);
let local = FakeLocal::default();
let at = Instant::now();
msg_tx
.send(Msg::Snapshot {
rows: (0..8)
.map(|id| {
ProcessInfo::builder(id, format!("sheep-{id}"), ProcStatus::Online).build()
})
.collect(),
at,
})
.await
.unwrap();
for _ in 0..20 {
msg_tx.send(Msg::Key(KeyPress::SelectDown)).await.unwrap();
}
tokio::spawn(async move {
tokio::time::sleep(MIN_REDRAW * 3).await;
let _ = msg_tx.send(Msg::Resize).await;
tokio::time::sleep(MIN_REDRAW).await;
let _ = msg_tx.send(Msg::Key(KeyPress::Quit)).await;
});
let app = App::new(
Palette::detect(None, None, None),
Control::ReadOnly,
"/tmp/shep".to_string(),
Instant::now(),
);
let terminal = Terminal::new(TestBackend::new(120, 24)).unwrap();
let _ = tokio::time::timeout(
Duration::from_secs(5),
run_ui(
app,
terminal,
stream::empty(),
msg_rx,
poll_tx,
request_tx,
local,
),
)
.await
.expect("the loop left within five seconds");
let mut asked = 0;
while let Ok(sent) = request_rx.try_recv() {
assert!(matches!(sent, Sent::Lambs { .. }));
asked += 1;
}
assert_eq!(asked, 1, "twenty moves, one Describe");
}
#[tokio::test]
async fn no_lambs_are_requested_when_the_detail_pane_is_not_drawn() {
let (msg_tx, msg_rx) = mpsc::channel(64);
let (poll_tx, _poll_rx) = mpsc::channel(4);
let (request_tx, mut request_rx) = mpsc::channel(2);
let local = FakeLocal::default();
let at = Instant::now();
msg_tx
.send(Msg::Snapshot {
rows: (0..8)
.map(|id| {
ProcessInfo::builder(id, format!("sheep-{id}"), ProcStatus::Online).build()
})
.collect(),
at,
})
.await
.unwrap();
for _ in 0..20 {
msg_tx.send(Msg::Key(KeyPress::SelectDown)).await.unwrap();
}
tokio::spawn(async move {
tokio::time::sleep(MIN_REDRAW * 3).await;
let _ = msg_tx.send(Msg::Resize).await;
tokio::time::sleep(MIN_REDRAW).await;
let _ = msg_tx.send(Msg::Key(KeyPress::Quit)).await;
});
let app = App::new(
Palette::detect(None, None, None),
Control::ReadOnly,
"/tmp/shep".to_string(),
Instant::now(),
);
let terminal = Terminal::new(TestBackend::new(120, 20)).unwrap();
let _ = tokio::time::timeout(
Duration::from_secs(5),
run_ui(
app,
terminal,
stream::empty(),
msg_rx,
poll_tx,
request_tx,
local,
),
)
.await
.expect("the loop left within five seconds");
assert!(
request_rx.try_recv().is_err(),
"no lamb request when the detail pane is not drawn"
);
}
}