Mirrors/nushell
2
0
Fork 0
mirror of https://github.com/nushell/nushell synced 2025-01-13 21:55:07 +00:00
Code Issues Projects Releases Packages Wiki Activity

Allow the table command to stream (#1278)

Browse source
This commit is contained in:
Jonathan Turner 2020-01-25 16:13:12 +13:00 committed by GitHub
parent a5e1372bc2
commit cdbfdf282f
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 131 additions and 150 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

224
src/commands/autoview.rs
View file

@ -1,15 +1,11 @@
use crate::commands::{RawCommandArgs, WholeStreamCommand}; use crate::commands::{RawCommandArgs, WholeStreamCommand};
use crate::prelude::*; use crate::prelude::*;
use futures::stream::TryStreamExt;
use nu_errors::ShellError; use nu_errors::ShellError;
use nu_parser::hir::{Expression, NamedArguments};
use nu_protocol::{Primitive, ReturnSuccess, Signature, UntaggedValue, Value}; use nu_protocol::{Primitive, ReturnSuccess, Signature, UntaggedValue, Value};
use std::sync::atomic::Ordering; use std::sync::atomic::Ordering;
pub struct Autoview; pub struct Autoview;
const STREAM_PAGE_SIZE: u64 = 50;
#[derive(Deserialize)] #[derive(Deserialize)]
pub struct AutoviewArgs {} pub struct AutoviewArgs {}
@ -45,22 +41,23 @@ pub fn autoview(
let table = context.get_command("table"); let table = context.get_command("table");
Ok(OutputStream::new(async_stream! { Ok(OutputStream::new(async_stream! {
let mut output_stream: OutputStream = context.input.into(); //let mut output_stream: OutputStream = context.input.into();
//let next = output_stream.try_next().await;
let next = output_stream.try_next().await; let mut input_stream = context.input;
match next { match input_stream.next().await {
Ok(Some(x)) => { Some(x) => {
match output_stream.try_next().await { match input_stream.next().await {
Ok(Some(y)) => { Some(y) => {
let ctrl_c = context.ctrl_c.clone(); let ctrl_c = context.ctrl_c.clone();
let stream = async_stream! { let stream = async_stream! {
yield Ok(x); yield Ok(x);
yield Ok(y); yield Ok(y);
loop { loop {
match output_stream.try_next().await { match input_stream.next().await {
Ok(Some(z)) => { Some(z) => {
if ctrl_c.load(Ordering::SeqCst) { if ctrl_c.load(Ordering::SeqCst) {
break; break;
} }
@ -70,142 +67,95 @@ pub fn autoview(
} }
} }
}; };
let stream = stream.to_input_stream();
if let Some(table) = table { if let Some(table) = table {
let mut new_output_stream: OutputStream = stream.to_output_stream(); let mut command_args = raw.with_input(stream);
let mut finished = false; let result = table.run(command_args, &context.commands);
let mut current_idx = 0; result.collect::<Vec<_>>().await;
loop {
let mut new_input = VecDeque::new();
for _ in 0..STREAM_PAGE_SIZE {
match new_output_stream.try_next().await {
Ok(Some(a)) => {
if let ReturnSuccess::Value(v) = a {
new_input.push_back(v);
}
}
_ => {
finished = true;
break;
}
}
}
let raw = raw.clone();
let input: Vec<Value> = new_input.into();
if input.len() > 0 && input.iter().all(|value| value.value.is_error()) {
let first = &input[0];
let mut host = context.host.clone();
let host = host.lock();
crate::cli::print_err(first.value.expect_error(), &*host, &context.source);
return;
}
let mut command_args = raw.with_input(input);
let mut named_args = NamedArguments::new();
named_args.insert_optional("start_number", Some(Expression::number(current_idx).into_expr(Span::unknown())));
command_args.call_info.args.named = Some(named_args);
let result = table.run(command_args, &context.commands);
result.collect::<Vec<_>>().await;
if finished {
break;
} else {
current_idx += STREAM_PAGE_SIZE;
}
}
} }
} }
_ => { _ => {
if let ReturnSuccess::Value(x) = x { match x {
match x { Value {
Value { value: UntaggedValue::Primitive(Primitive::String(ref s)),
value: UntaggedValue::Primitive(Primitive::String(ref s)), tag: Tag { anchor, span },
tag: Tag { anchor, span }, } if anchor.is_some() => {
} if anchor.is_some() => { if let Some(text) = text {
if let Some(text) = text { let mut stream = VecDeque::new();
let mut stream = VecDeque::new(); stream.push_back(UntaggedValue::string(s).into_value(Tag { anchor, span }));
stream.push_back(UntaggedValue::string(s).into_value(Tag { anchor, span })); let result = text.run(raw.with_input(stream), &context.commands);
let result = text.run(raw.with_input(stream.into()), &context.commands); result.collect::<Vec<_>>().await;
result.collect::<Vec<_>>().await; } else {
} else {
outln!("{}", s);
}
}
Value {
value: UntaggedValue::Primitive(Primitive::String(s)),
..
} => {
outln!("{}", s); outln!("{}", s);
} }
Value { }
value: UntaggedValue::Primitive(Primitive::Line(ref s)), Value {
tag: Tag { anchor, span }, value: UntaggedValue::Primitive(Primitive::String(s)),
} if anchor.is_some() => { ..
if let Some(text) = text { } => {
let mut stream = VecDeque::new(); outln!("{}", s);
stream.push_back(UntaggedValue::string(s).into_value(Tag { anchor, span })); }
let result = text.run(raw.with_input(stream.into()), &context.commands); Value {
result.collect::<Vec<_>>().await; value: UntaggedValue::Primitive(Primitive::Line(ref s)),
} else { tag: Tag { anchor, span },
outln!("{}\n", s); } if anchor.is_some() => {
} if let Some(text) = text {
} let mut stream = VecDeque::new();
Value { stream.push_back(UntaggedValue::string(s).into_value(Tag { anchor, span }));
value: UntaggedValue::Primitive(Primitive::Line(s)), let result = text.run(raw.with_input(stream), &context.commands);
.. result.collect::<Vec<_>>().await;
} => { } else {
outln!("{}\n", s); outln!("{}\n", s);
} }
Value { }
value: UntaggedValue::Primitive(Primitive::Path(s)), Value {
.. value: UntaggedValue::Primitive(Primitive::Line(s)),
} => { ..
outln!("{}", s.display()); } => {
} outln!("{}\n", s);
Value { }
value: UntaggedValue::Primitive(Primitive::Int(n)), Value {
.. value: UntaggedValue::Primitive(Primitive::Path(s)),
} => { ..
outln!("{}", n); } => {
} outln!("{}", s.display());
Value { }
value: UntaggedValue::Primitive(Primitive::Decimal(n)), Value {
.. value: UntaggedValue::Primitive(Primitive::Int(n)),
} => { ..
outln!("{}", n); } => {
} outln!("{}", n);
}
Value {
value: UntaggedValue::Primitive(Primitive::Decimal(n)),
..
} => {
outln!("{}", n);
}
Value { value: UntaggedValue::Primitive(Primitive::Binary(ref b)), .. } => { Value { value: UntaggedValue::Primitive(Primitive::Binary(ref b)), .. } => {
if let Some(binary) = binary { if let Some(binary) = binary {
let mut stream = VecDeque::new(); let mut stream = VecDeque::new();
stream.push_back(x); stream.push_back(x);
let result = binary.run(raw.with_input(stream.into()), &context.commands); let result = binary.run(raw.with_input(stream), &context.commands);
result.collect::<Vec<_>>().await; result.collect::<Vec<_>>().await;
} else { } else {
use pretty_hex::*; use pretty_hex::*;
outln!("{:?}", b.hex_dump()); outln!("{:?}", b.hex_dump());
}
} }
}
Value { value: UntaggedValue::Error(e), .. } => { Value { value: UntaggedValue::Error(e), .. } => {
yield Err(e); yield Err(e);
} }
Value { value: ref item, .. } => { Value { value: ref item, .. } => {
if let Some(table) = table { if let Some(table) = table {
let mut stream = VecDeque::new(); let mut stream = VecDeque::new();
stream.push_back(x); stream.push_back(x);
let result = table.run(raw.with_input(stream.into()), &context.commands); let result = table.run(raw.with_input(stream), &context.commands);
result.collect::<Vec<_>>().await; result.collect::<Vec<_>>().await;
} else { } else {
outln!("{:?}", item); outln!("{:?}", item);
}
} }
} }
} }

2
src/commands/command.rs
View file

@ -88,7 +88,7 @@ pub struct RawCommandArgs {
} }
impl RawCommandArgs { impl RawCommandArgs {
pub fn with_input(self, input: Vec<Value>) -> CommandArgs { pub fn with_input(self, input: impl Into<InputStream>) -> CommandArgs {
CommandArgs { CommandArgs {
host: self.host, host: self.host,
ctrl_c: self.ctrl_c, ctrl_c: self.ctrl_c,

16
src/commands/from_json.rs
View file

@ -106,10 +106,14 @@ fn from_json(
match from_json_string_to_value(json_str.to_string(), &name_tag) { match from_json_string_to_value(json_str.to_string(), &name_tag) {
Ok(x) => Ok(x) =>
yield ReturnSuccess::value(x), yield ReturnSuccess::value(x),
Err(_) => { Err(e) => {
if let Some(ref last_tag) = latest_tag { if let Some(ref last_tag) = latest_tag {
let mut message = "Could not parse as JSON (".to_string();
message.push_str(&e.to_string());
message.push_str(")");
yield Err(ShellError::labeled_error_with_secondary( yield Err(ShellError::labeled_error_with_secondary(
"Could nnot parse as JSON", message,
"input cannot be parsed as JSON", "input cannot be parsed as JSON",
&name_tag, &name_tag,
"value originates from here", "value originates from here",
@ -129,10 +133,14 @@ fn from_json(
} }
x => yield ReturnSuccess::value(x), x => yield ReturnSuccess::value(x),
} }
Err(_) => { Err(e) => {
if let Some(last_tag) = latest_tag { if let Some(last_tag) = latest_tag {
let mut message = "Could not parse as JSON (".to_string();
message.push_str(&e.to_string());
message.push_str(")");
yield Err(ShellError::labeled_error_with_secondary( yield Err(ShellError::labeled_error_with_secondary(
"Could not parse as JSON", message,
"input cannot be parsed as JSON", "input cannot be parsed as JSON",
name_tag, name_tag,
"value originates from here", "value originates from here",

39
src/commands/table.rs
View file

@ -4,6 +4,8 @@ use crate::prelude::*;
use nu_errors::ShellError; use nu_errors::ShellError;
use nu_protocol::{Primitive, ReturnSuccess, Signature, SyntaxShape, UntaggedValue, Value}; use nu_protocol::{Primitive, ReturnSuccess, Signature, SyntaxShape, UntaggedValue, Value};
const STREAM_PAGE_SIZE: usize = 100;
pub struct Table; pub struct Table;
impl WholeStreamCommand for Table { impl WholeStreamCommand for Table {
@ -33,11 +35,12 @@ impl WholeStreamCommand for Table {
} }
fn table(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> { fn table(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
let args = args.evaluate_once(registry)?; let mut args = args.evaluate_once(registry)?;
let mut finished = false;
let stream = async_stream! { let stream = async_stream! {
let host = args.host.clone(); let host = args.host.clone();
let start_number = match args.get("start_number") { let mut start_number = match args.get("start_number") {
Some(Value { value: UntaggedValue::Primitive(Primitive::Int(i)), .. }) => { Some(Value { value: UntaggedValue::Primitive(Primitive::Int(i)), .. }) => {
if let Some(num) = i.to_usize() { if let Some(num) = i.to_usize() {
num num
@ -51,15 +54,35 @@ fn table(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream,
} }
}; };
let input: Vec<Value> = args.input.into_vec().await; while !finished {
if input.len() > 0 { let mut new_input = VecDeque::new();
let mut host = host.lock();
let view = TableView::from_list(&input, start_number);
if let Some(view) = view { for _ in 0..STREAM_PAGE_SIZE {
handle_unexpected(&mut *host, |host| crate::format::print_view(&view, host)); match args.input.next().await {
Some(a) => {
new_input.push_back(a);
}
_ => {
finished = true;
break;
}
}
} }
let input: Vec<Value> = new_input.into();
if input.len() > 0 {
let mut host = host.lock();
let view = TableView::from_list(&input, start_number);
if let Some(view) = view {
handle_unexpected(&mut *host, |host| crate::format::print_view(&view, host));
}
}
start_number += STREAM_PAGE_SIZE;
} }
// Needed for async_stream to type check // Needed for async_stream to type check
if false { if false {
yield ReturnSuccess::value(UntaggedValue::nothing().into_value(Tag::unknown())); yield ReturnSuccess::value(UntaggedValue::nothing().into_value(Tag::unknown()));