diff --git a/Cargo.toml b/Cargo.toml index 0907648f35..2167159c91 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -147,6 +147,8 @@ start = ["nu_plugin_start"] trash-support = ["nu-cli/trash-support", "nu-command/trash-support"] tree = ["nu_plugin_tree"] xpath = ["nu_plugin_xpath"] +#This is disabled in extra for now +table-pager = ["nu-command/table-pager"] [profile.release] #strip = "symbols" #Couldn't get working +nightly diff --git a/crates/nu-command/Cargo.toml b/crates/nu-command/Cargo.toml index 8ea8fc5d1f..7b83d4bd1f 100644 --- a/crates/nu-command/Cargo.toml +++ b/crates/nu-command/Cargo.toml @@ -62,6 +62,7 @@ itertools = "0.10.0" lazy_static = "1.*" log = "0.4.14" meval = "0.2.0" +minus = { optional = true, features = ["async_std_lib", "search"], git = "https://github.com/rezural/minus", branch = "add-finished-callback" } num-bigint = { version = "0.3.1", features = ["serde"] } num-format = { version = "0.4.0", features = ["with-num-bigint"] } num-traits = "0.2.14" @@ -130,3 +131,5 @@ stable = [] trash-support = ["trash"] directories = ["directories-next"] dirs = ["dirs-next"] +table-pager = ["minus"] + diff --git a/crates/nu-command/src/commands/table/command.rs b/crates/nu-command/src/commands/table/command.rs index d686a68a47..27cea6f2d5 100644 --- a/crates/nu-command/src/commands/table/command.rs +++ b/crates/nu-command/src/commands/table/command.rs @@ -8,6 +8,12 @@ use nu_protocol::{Primitive, Signature, SyntaxShape, UntaggedValue, Value}; use nu_table::{draw_table, Alignment, StyledString, TextStyle}; use std::collections::HashMap; use std::time::Instant; +#[cfg(feature = "table-pager")] +use { + futures::future::join, + minus::{ExitStrategy, Pager}, + std::fmt::Write, +}; const STREAM_PAGE_SIZE: usize = 1000; const STREAM_TIMEOUT_CHECK_INTERVAL: usize = 100; @@ -156,7 +162,7 @@ async fn table( args: CommandArgs, ) -> Result { let mut args = args.evaluate_once().await?; - let mut finished = false; + // Ideally, get_color_config would get all the colors configured in the config.toml // and create a style based on those settings. However, there are few places where // this just won't work right now, like header styling, because a style needs to know @@ -187,64 +193,123 @@ async fn table( let term_width = args.host.lock().width(); - while !finished { - let mut new_input: VecDeque = VecDeque::new(); + #[cfg(feature = "table-pager")] + let pager = Pager::new() + .set_exit_strategy(ExitStrategy::PagerQuit) + .set_searchable(true) + .set_page_if_havent_overflowed(false) + .finish(); - let start_time = Instant::now(); - for idx in 0..STREAM_PAGE_SIZE { - if let Some(val) = delay_slot { - new_input.push_back(val); - delay_slot = None; - } else { - match args.input.next().await { - Some(a) => { - if !new_input.is_empty() { - if let Some(descs) = new_input.get(0) { - let descs = descs.data_descriptors(); - let compare = a.data_descriptors(); - if descs != compare { - delay_slot = Some(a); - break; + let stream_data = async { + let finished = Arc::new(AtomicBool::new(false)); + // we are required to clone finished, for use within the callback, otherwise we get borrow errors + #[cfg(feature = "table-pager")] + let finished_within_callback = finished.clone(); + #[cfg(feature = "table-pager")] + { + // This is called when the pager finishes, to indicate to the + // while loop below to finish, in case of long running InputStream consumer + // that doesnt finish by the time the user quits out of the pager + pager + .lock() + .await + .on_finished_callbacks + .push(Box::new(move || { + finished_within_callback.store(true, Ordering::Relaxed); + })); + } + while !finished.clone().load(Ordering::Relaxed) { + let mut new_input: VecDeque = VecDeque::new(); + + let start_time = Instant::now(); + for idx in 0..STREAM_PAGE_SIZE { + if let Some(val) = delay_slot { + new_input.push_back(val); + delay_slot = None; + } else { + match args.input.next().await { + Some(a) => { + if !new_input.is_empty() { + if let Some(descs) = new_input.get(0) { + let descs = descs.data_descriptors(); + let compare = a.data_descriptors(); + if descs != compare { + delay_slot = Some(a); + break; + } else { + new_input.push_back(a); + } } else { new_input.push_back(a); } } else { new_input.push_back(a); } - } else { - new_input.push_back(a); + } + _ => { + finished.store(true, Ordering::Relaxed); + break; } } - _ => { - finished = true; - break; - } - } - // Check if we've gone over our buffering threshold - if (idx + 1) % STREAM_TIMEOUT_CHECK_INTERVAL == 0 { - let end_time = Instant::now(); + // Check if we've gone over our buffering threshold + if (idx + 1) % STREAM_TIMEOUT_CHECK_INTERVAL == 0 { + let end_time = Instant::now(); - // If we've been buffering over a second, go ahead and send out what we have so far - if (end_time - start_time).as_secs() >= 1 { - break; + // If we've been buffering over a second, go ahead and send out what we have so far + if (end_time - start_time).as_secs() >= 1 { + break; + } + + if finished.load(Ordering::Relaxed) { + break; + } } } } + + let input: Vec = new_input.into(); + + if !input.is_empty() { + let t = from_list(&input, &configuration, start_number, &color_hm); + let output = draw_table(&t, term_width, &color_hm); + #[cfg(feature = "table-pager")] + { + let mut pager = pager.lock().await; + writeln!(pager.lines, "{}", output).map_err(|_| { + ShellError::untagged_runtime_error("Error writing to pager") + })?; + } + + #[cfg(not(feature = "table-pager"))] + println!("{}", output); + } + + start_number += input.len(); } - let input: Vec = new_input.into(); - - if !input.is_empty() { - let t = from_list(&input, &configuration, start_number, &color_hm); - - let output = draw_table(&t, term_width, &color_hm); - println!("{}", output); + #[cfg(feature = "table-pager")] + { + let mut pager_lock = pager.lock().await; + pager_lock.data_finished(); } - start_number += input.len(); + Result::<_, ShellError>::Ok(()) + }; + + #[cfg(feature = "table-pager")] + { + let (minus_result, streaming_result) = + join(minus::async_std_updating(pager.clone()), stream_data).await; + minus_result.map_err(|_| ShellError::untagged_runtime_error("Error paging data"))?; + streaming_result?; } + #[cfg(not(feature = "table-pager"))] + stream_data + .await + .map_err(|_| ShellError::untagged_runtime_error("Error streaming data"))?; + Ok(OutputStream::empty()) }