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

Another batch of removing async_stream (#1971)

Browse source
This commit is contained in:
Jonathan Turner 2020-06-12 16:40:23 -07:00 committed by GitHub
parent 935a5f6b9e
commit d82ce26b44
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
12 changed files with 286 additions and 304 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
crates/nu-cli/src/commands/split/command.rs
View file

@ -26,14 +26,10 @@ impl WholeStreamCommand for Command {
registry: &CommandRegistry,
) -> Result<OutputStream, ShellError> {
let registry = registry.clone();
let stream = async_stream! {
yield Ok(ReturnSuccess::Value(
Ok(OutputStream::one(Ok(ReturnSuccess::Value(
UntaggedValue::string(crate::commands::help::get_help(&Command, &registry))
.into_value(Tag::unknown()),
));
};
Ok(stream.to_output_stream())
))))
}
}

45
crates/nu-cli/src/commands/split/row.rs
View file

@ -35,41 +35,52 @@ impl WholeStreamCommand for SubCommand {
args: CommandArgs,
registry: &CommandRegistry,
) -> Result<OutputStream, ShellError> {
split_row(args, registry)
split_row(args, registry).await
}
}
fn split_row(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
async fn split_row(
args: CommandArgs,
registry: &CommandRegistry,
) -> Result<OutputStream, ShellError> {
let registry = registry.clone();
let stream = async_stream! {
let name = args.call_info.name_tag.clone();
let (SplitRowArgs { separator }, mut input) = args.process(&registry).await?;
while let Some(v) = input.next().await {
let (SplitRowArgs { separator }, input) = args.process(&registry).await?;
Ok(input
.flat_map(move |v| {
if let Ok(s) = v.as_string() {
let splitter = separator.item.replace("\\n", "\n");
trace!("splitting with {:?}", splitter);
let split_result: Vec<_> = s.split(&splitter).filter(|s| s.trim() != "").collect();
let split_result: Vec<String> = s
.split(&splitter)
.filter_map(|s| {
if s.trim() != "" {
Some(s.to_string())
} else {
None
}
})
.collect();
trace!("split result = {:?}", split_result);
for s in split_result {
yield ReturnSuccess::value(
UntaggedValue::Primitive(Primitive::String(s.into())).into_value(&v.tag),
);
}
futures::stream::iter(split_result.into_iter().map(move |s| {
ReturnSuccess::value(
UntaggedValue::Primitive(Primitive::String(s)).into_value(&v.tag),
)
}))
.to_output_stream()
} else {
yield Err(ShellError::labeled_error_with_secondary(
OutputStream::one(Err(ShellError::labeled_error_with_secondary(
"Expected a string from pipeline",
"requires string input",
name.span,
"value originates from here",
v.tag.span,
));
)))
}
}
};
Ok(stream.to_output_stream())
})
.to_output_stream())
}
#[cfg(test)]

43
crates/nu-cli/src/commands/str_/capitalize.rs
View file

@ -37,7 +37,7 @@ impl WholeStreamCommand for SubCommand {
args: CommandArgs,
registry: &CommandRegistry,
) -> Result<OutputStream, ShellError> {
operate(args, registry)
operate(args, registry).await
}
fn examples(&self) -> Vec<Example> {
@ -49,47 +49,44 @@ impl WholeStreamCommand for SubCommand {
}
}
fn operate(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
async fn operate(
args: CommandArgs,
registry: &CommandRegistry,
) -> Result<OutputStream, ShellError> {
let registry = registry.clone();
let stream = async_stream! {
let (Arguments { rest }, mut input) = args.process(&registry).await?;
let (Arguments { rest }, input) = args.process(&registry).await?;
let column_paths: Vec<_> = rest.iter().map(|x| x.clone()).collect();
let column_paths: Vec<_> = rest;
while let Some(v) = input.next().await {
Ok(input
.map(move |v| {
if column_paths.is_empty() {
match action(&v, v.tag()) {
Ok(out) => yield ReturnSuccess::value(out),
Err(err) => {
yield Err(err);
return;
}
Ok(out) => ReturnSuccess::value(out),
Err(err) => Err(err),
}
} else {
let mut ret = v.clone();
let mut ret = v;
for path in &column_paths {
let swapping = ret.swap_data_by_column_path(path, Box::new(move |old| action(old, old.tag())));
let swapping = ret.swap_data_by_column_path(
path,
Box::new(move |old| action(old, old.tag())),
);
match swapping {
Ok(new_value) => {
ret = new_value;
}
Err(err) => {
yield Err(err);
return;
}
Err(err) => return Err(err),
}
}
yield ReturnSuccess::value(ret);
ReturnSuccess::value(ret)
}
}
};
Ok(stream.to_output_stream())
})
.to_output_stream())
}
fn action(input: &Value, tag: impl Into<Tag>) -> Result<Value, ShellError> {

41
crates/nu-cli/src/commands/str_/downcase.rs
View file

@ -37,7 +37,7 @@ impl WholeStreamCommand for SubCommand {
args: CommandArgs,
registry: &CommandRegistry,
) -> Result<OutputStream, ShellError> {
operate(args, registry)
operate(args, registry).await
}
fn examples(&self) -> Vec<Example> {
@ -49,47 +49,46 @@ impl WholeStreamCommand for SubCommand {
}
}
fn operate(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
async fn operate(
args: CommandArgs,
registry: &CommandRegistry,
) -> Result<OutputStream, ShellError> {
let registry = registry.clone();
let stream = async_stream! {
let (Arguments { rest }, mut input) = args.process(&registry).await?;
let (Arguments { rest }, input) = args.process(&registry).await?;
let column_paths: Vec<_> = rest.iter().map(|x| x.clone()).collect();
let column_paths: Vec<_> = rest;
while let Some(v) = input.next().await {
Ok(input
.map(move |v| {
if column_paths.is_empty() {
match action(&v, v.tag()) {
Ok(out) => yield ReturnSuccess::value(out),
Err(err) => {
yield Err(err);
return;
}
Ok(out) => ReturnSuccess::value(out),
Err(err) => Err(err),
}
} else {
let mut ret = v.clone();
let mut ret = v;
for path in &column_paths {
let swapping = ret.swap_data_by_column_path(path, Box::new(move |old| action(old, old.tag())));
let swapping = ret.swap_data_by_column_path(
path,
Box::new(move |old| action(old, old.tag())),
);
match swapping {
Ok(new_value) => {
ret = new_value;
}
Err(err) => {
yield Err(err);
return;
return Err(err);
}
}
}
yield ReturnSuccess::value(ret);
ReturnSuccess::value(ret)
}
}
};
Ok(stream.to_output_stream())
})
.to_output_stream())
}
fn action(input: &Value, tag: impl Into<Tag>) -> Result<Value, ShellError> {

41
crates/nu-cli/src/commands/str_/set.rs
View file

@ -37,7 +37,7 @@ impl WholeStreamCommand for SubCommand {
args: CommandArgs,
registry: &CommandRegistry,
) -> Result<OutputStream, ShellError> {
operate(args, registry)
operate(args, registry).await
}
fn examples(&self) -> Vec<Example> {
@ -59,50 +59,49 @@ impl WholeStreamCommand for SubCommand {
#[derive(Clone)]
struct Replace(String);
fn operate(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
async fn operate(
args: CommandArgs,
registry: &CommandRegistry,
) -> Result<OutputStream, ShellError> {
let registry = registry.clone();
let stream = async_stream! {
let (Arguments { replace, rest }, mut input) = args.process(&registry).await?;
let (Arguments { replace, rest }, input) = args.process(&registry).await?;
let options = Replace(replace.item);
let column_paths: Vec<_> = rest.iter().map(|x| x.clone()).collect();
let column_paths: Vec<_> = rest;
while let Some(v) = input.next().await {
Ok(input
.map(move |v| {
if column_paths.is_empty() {
match action(&v, &options, v.tag()) {
Ok(out) => yield ReturnSuccess::value(out),
Err(err) => {
yield Err(err);
return;
}
Ok(out) => ReturnSuccess::value(out),
Err(err) => Err(err),
}
} else {
let mut ret = v.clone();
let mut ret = v;
for path in &column_paths {
let options = options.clone();
let swapping = ret.swap_data_by_column_path(path, Box::new(move |old| action(old, &options, old.tag())));
let swapping = ret.swap_data_by_column_path(
path,
Box::new(move |old| action(old, &options, old.tag())),
);
match swapping {
Ok(new_value) => {
ret = new_value;
}
Err(err) => {
yield Err(err);
return;
return Err(err);
}
}
}
yield ReturnSuccess::value(ret);
ReturnSuccess::value(ret)
}
}
};
Ok(stream.to_output_stream())
})
.to_output_stream())
}
fn action(_input: &Value, options: &Replace, tag: impl Into<Tag>) -> Result<Value, ShellError> {

58
crates/nu-cli/src/commands/str_/substring.rs
View file

@ -46,7 +46,7 @@ impl WholeStreamCommand for SubCommand {
args: CommandArgs,
registry: &CommandRegistry,
) -> Result<OutputStream, ShellError> {
operate(args, registry)
operate(args, registry).await
}
fn examples(&self) -> Vec<Example> {
@ -73,91 +73,83 @@ impl WholeStreamCommand for SubCommand {
#[derive(Clone)]
struct Substring(usize, usize);
fn operate(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
async fn operate(
args: CommandArgs,
registry: &CommandRegistry,
) -> Result<OutputStream, ShellError> {
let name = args.call_info.name_tag.clone();
let registry = registry.clone();
let stream = async_stream! {
let (Arguments { range, rest }, mut input) = args.process(&registry).await?;
let (Arguments { range, rest }, input) = args.process(&registry).await?;
let v: Vec<&str> = range.item.split(',').collect();
let start = match v[0] {
"" => 0,
_ => v[0]
.trim()
.parse()
.map_err(|_| {
_ => v[0].trim().parse().map_err(|_| {
ShellError::labeled_error(
"could not perform substring",
"could not perform substring",
name.span,
)
})?
})?,
};
let end = match v[1] {
"" => usize::max_value(),
_ => v[1]
.trim()
.parse()
.map_err(|_| {
_ => v[1].trim().parse().map_err(|_| {
ShellError::labeled_error(
"could not perform substring",
"could not perform substring",
name.span,
)
})?
})?,
};
if start > end {
yield Err(ShellError::labeled_error(
return Err(ShellError::labeled_error(
"End must be greater than or equal to Start",
"End must be greater than or equal to Start",
name.span,
));
return;
}
let options = Substring(start, end);
let column_paths: Vec<_> = rest.iter().map(|x| x.clone()).collect();
let column_paths: Vec<_> = rest;
while let Some(v) = input.next().await {
Ok(input
.map(move |v| {
if column_paths.is_empty() {
match action(&v, &options, v.tag()) {
Ok(out) => yield ReturnSuccess::value(out),
Err(err) => {
yield Err(err);
return;
}
Ok(out) => ReturnSuccess::value(out),
Err(err) => Err(err),
}
} else {
let mut ret = v.clone();
let mut ret = v;
for path in &column_paths {
let options = options.clone();
let swapping = ret.swap_data_by_column_path(path, Box::new(move |old| action(old, &options, old.tag())));
let swapping = ret.swap_data_by_column_path(
path,
Box::new(move |old| action(old, &options, old.tag())),
);
match swapping {
Ok(new_value) => {
ret = new_value;
}
Err(err) => {
yield Err(err);
return;
return Err(err);
}
}
}
yield ReturnSuccess::value(ret);
ReturnSuccess::value(ret)
}
}
};
Ok(stream.to_output_stream())
})
.to_output_stream())
}
fn action(input: &Value, options: &Substring, tag: impl Into<Tag>) -> Result<Value, ShellError> {

41
crates/nu-cli/src/commands/str_/to_decimal.rs
View file

@ -40,7 +40,7 @@ impl WholeStreamCommand for SubCommand {
args: CommandArgs,
registry: &CommandRegistry,
) -> Result<OutputStream, ShellError> {
operate(args, registry)
operate(args, registry).await
}
fn examples(&self) -> Vec<Example> {
@ -52,47 +52,46 @@ impl WholeStreamCommand for SubCommand {
}
}
fn operate(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
async fn operate(
args: CommandArgs,
registry: &CommandRegistry,
) -> Result<OutputStream, ShellError> {
let registry = registry.clone();
let stream = async_stream! {
let (Arguments { rest }, mut input) = args.process(&registry).await?;
let (Arguments { rest }, input) = args.process(&registry).await?;
let column_paths: Vec<_> = rest.iter().map(|x| x.clone()).collect();
let column_paths: Vec<_> = rest;
while let Some(v) = input.next().await {
Ok(input
.map(move |v| {
if column_paths.is_empty() {
match action(&v, v.tag()) {
Ok(out) => yield ReturnSuccess::value(out),
Err(err) => {
yield Err(err);
return;
}
Ok(out) => ReturnSuccess::value(out),
Err(err) => Err(err),
}
} else {
let mut ret = v.clone();
let mut ret = v;
for path in &column_paths {
let swapping = ret.swap_data_by_column_path(path, Box::new(move |old| action(old, old.tag())));
let swapping = ret.swap_data_by_column_path(
path,
Box::new(move |old| action(old, old.tag())),
);
match swapping {
Ok(new_value) => {
ret = new_value;
}
Err(err) => {
yield Err(err);
return;
return Err(err);
}
}
}
yield ReturnSuccess::value(ret);
ReturnSuccess::value(ret)
}
}
};
Ok(stream.to_output_stream())
})
.to_output_stream())
}
fn action(input: &Value, tag: impl Into<Tag>) -> Result<Value, ShellError> {

41
crates/nu-cli/src/commands/str_/to_integer.rs
View file

@ -40,7 +40,7 @@ impl WholeStreamCommand for SubCommand {
args: CommandArgs,
registry: &CommandRegistry,
) -> Result<OutputStream, ShellError> {
operate(args, registry)
operate(args, registry).await
}
fn examples(&self) -> Vec<Example> {
@ -52,47 +52,46 @@ impl WholeStreamCommand for SubCommand {
}
}
fn operate(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
async fn operate(
args: CommandArgs,
registry: &CommandRegistry,
) -> Result<OutputStream, ShellError> {
let registry = registry.clone();
let stream = async_stream! {
let (Arguments { rest }, mut input) = args.process(&registry).await?;
let (Arguments { rest }, input) = args.process(&registry).await?;
let column_paths: Vec<_> = rest.iter().map(|x| x.clone()).collect();
let column_paths: Vec<_> = rest;
while let Some(v) = input.next().await {
Ok(input
.map(move |v| {
if column_paths.is_empty() {
match action(&v, v.tag()) {
Ok(out) => yield ReturnSuccess::value(out),
Err(err) => {
yield Err(err);
return;
}
Ok(out) => ReturnSuccess::value(out),
Err(err) => Err(err),
}
} else {
let mut ret = v.clone();
let mut ret = v;
for path in &column_paths {
let swapping = ret.swap_data_by_column_path(path, Box::new(move |old| action(old, old.tag())));
let swapping = ret.swap_data_by_column_path(
path,
Box::new(move |old| action(old, old.tag())),
);
match swapping {
Ok(new_value) => {
ret = new_value;
}
Err(err) => {
yield Err(err);
return;
return Err(err);
}
}
}
yield ReturnSuccess::value(ret);
ReturnSuccess::value(ret)
}
}
};
Ok(stream.to_output_stream())
})
.to_output_stream())
}
fn action(input: &Value, tag: impl Into<Tag>) -> Result<Value, ShellError> {

41
crates/nu-cli/src/commands/str_/trim.rs
View file

@ -37,7 +37,7 @@ impl WholeStreamCommand for SubCommand {
args: CommandArgs,
registry: &CommandRegistry,
) -> Result<OutputStream, ShellError> {
operate(args, registry)
operate(args, registry).await
}
fn examples(&self) -> Vec<Example> {
@ -49,47 +49,46 @@ impl WholeStreamCommand for SubCommand {
}
}
fn operate(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
async fn operate(
args: CommandArgs,
registry: &CommandRegistry,
) -> Result<OutputStream, ShellError> {
let registry = registry.clone();
let stream = async_stream! {
let (Arguments { rest }, mut input) = args.process(&registry).await?;
let (Arguments { rest }, input) = args.process(&registry).await?;
let column_paths: Vec<_> = rest.iter().map(|x| x.clone()).collect();
let column_paths: Vec<_> = rest;
while let Some(v) = input.next().await {
Ok(input
.map(move |v| {
if column_paths.is_empty() {
match action(&v, v.tag()) {
Ok(out) => yield ReturnSuccess::value(out),
Err(err) => {
yield Err(err);
return;
}
Ok(out) => ReturnSuccess::value(out),
Err(err) => Err(err),
}
} else {
let mut ret = v.clone();
let mut ret = v;
for path in &column_paths {
let swapping = ret.swap_data_by_column_path(path, Box::new(move |old| action(old, old.tag())));
let swapping = ret.swap_data_by_column_path(
path,
Box::new(move |old| action(old, old.tag())),
);
match swapping {
Ok(new_value) => {
ret = new_value;
}
Err(err) => {
yield Err(err);
return;
return Err(err);
}
}
}
yield ReturnSuccess::value(ret);
ReturnSuccess::value(ret)
}
}
};
Ok(stream.to_output_stream())
})
.to_output_stream())
}
fn action(input: &Value, tag: impl Into<Tag>) -> Result<Value, ShellError> {

28
crates/nu-cli/src/commands/what.rs
View file

@ -2,7 +2,7 @@ use crate::commands::WholeStreamCommand;
use crate::prelude::*;
use nu_errors::ShellError;
use nu_protocol::{ReturnSuccess, ReturnValue, Signature, UntaggedValue};
use nu_protocol::{ReturnSuccess, Signature, UntaggedValue};
pub struct What;
@ -28,23 +28,23 @@ impl WholeStreamCommand for What {
args: CommandArgs,
registry: &CommandRegistry,
) -> Result<OutputStream, ShellError> {
// args.process(registry, what)?.run()
what(args, registry)
what(args, registry).await
}
}
pub fn what(args: CommandArgs, _registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
let stream = async_stream! {
let mut input = args.input;
while let Some(row) = input.next().await {
pub async fn what(
args: CommandArgs,
_registry: &CommandRegistry,
) -> Result<OutputStream, ShellError> {
Ok(args
.input
.map(|row| {
let name = value::format_type(&row, 100);
yield ReturnSuccess::value(UntaggedValue::string(name).into_value(Tag::unknown_anchor(row.tag.span)));
}
};
let stream: BoxStream<'static, ReturnValue> = stream.boxed();
Ok(OutputStream::from(stream))
ReturnSuccess::value(
UntaggedValue::string(name).into_value(Tag::unknown_anchor(row.tag.span)),
)
})
.to_output_stream())
}
#[cfg(test)]

22
crates/nu-cli/src/commands/wrap.rs
View file

@ -38,8 +38,7 @@ impl WholeStreamCommand for Wrap {
args: CommandArgs,
registry: &CommandRegistry,
) -> Result<OutputStream, ShellError> {
// args.process(registry, wrap)?.run()
wrap(args, registry)
wrap(args, registry).await
}
fn examples(&self) -> Vec<Example> {
@ -84,10 +83,9 @@ impl WholeStreamCommand for Wrap {
}
}
fn wrap(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
async fn wrap(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
let registry = registry.clone();
let stream = async_stream! {
let (WrapArgs { column }, mut input) = args.process(&registry).await?;
let mut result_table = vec![];
let mut are_all_rows = true;
@ -114,7 +112,6 @@ fn wrap(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, S
result_table.push(UntaggedValue::row(index_map).into_value(Tag::unknown()));
}
}
}
@ -130,18 +127,13 @@ fn wrap(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, S
let row = UntaggedValue::row(index_map).into_untagged_value();
yield ReturnSuccess::value(row);
Ok(OutputStream::one(ReturnSuccess::value(row)))
} else {
for item in result_table
.iter()
.map(|row| ReturnSuccess::value(row.clone())) {
yield item;
Ok(
futures::stream::iter(result_table.into_iter().map(ReturnSuccess::value))
.to_output_stream(),
)
}
}
};
Ok(stream.to_output_stream())
}
#[cfg(test)]

53
crates/nu-cli/src/shell/filesystem_shell.rs
View file

@ -562,54 +562,54 @@ impl Shell for FilesystemShell {
));
}
let stream = async_stream! {
for (f, tag) in all_targets.iter() {
Ok(
futures::stream::iter(all_targets.into_iter().map(move |(f, tag)| {
let is_empty = || match f.read_dir() {
Ok(mut p) => p.next().is_none(),
Err(_) => false
Err(_) => false,
};
if let Ok(metadata) = f.symlink_metadata() {
if metadata.is_file() || metadata.file_type().is_symlink() || recursive.item || is_empty() {
if metadata.is_file()
|| metadata.file_type().is_symlink()
|| recursive.item
|| is_empty()
{
let result;
#[cfg(feature = "trash-support")]
{
let rm_always_trash = config::config(Tag::unknown())?.get("rm_always_trash").map(|val| val.is_true()).unwrap_or(false);
let rm_always_trash = config::config(Tag::unknown())?
.get("rm_always_trash")
.map(|val| val.is_true())
.unwrap_or(false);
result = if _trash.item || (rm_always_trash && !_permanent.item) {
trash::remove(f)
.map_err(|e| f.to_string_lossy())
trash::remove(&f).map_err(|_| f.to_string_lossy())
} else if metadata.is_file() {
std::fs::remove_file(f)
.map_err(|e| f.to_string_lossy())
std::fs::remove_file(&f).map_err(|_| f.to_string_lossy())
} else {
std::fs::remove_dir_all(f)
.map_err(|e| f.to_string_lossy())
std::fs::remove_dir_all(&f).map_err(|_| f.to_string_lossy())
};
}
#[cfg(not(feature = "trash-support"))]
{
result = if metadata.is_file() {
std::fs::remove_file(f)
.map_err(|e| f.to_string_lossy())
std::fs::remove_file(&f).map_err(|_| f.to_string_lossy())
} else {
std::fs::remove_dir_all(f)
.map_err(|e| f.to_string_lossy())
std::fs::remove_dir_all(&f).map_err(|_| f.to_string_lossy())
};
}
if let Err(e) = result {
let msg = format!("Could not delete {:}", e);
yield Err(ShellError::labeled_error(msg, e, tag))
Err(ShellError::labeled_error(msg, e, tag))
} else {
let val = format!("deleted {:}", f.to_string_lossy()).into();
yield Ok(ReturnSuccess::Value(val))
Ok(ReturnSuccess::Value(val))
}
} else {
let msg = format!(
"Cannot remove {:}. try --recursive",
f.to_string_lossy()
);
yield Err(ShellError::labeled_error(
let msg =
format!("Cannot remove {:}. try --recursive", f.to_string_lossy());
Err(ShellError::labeled_error(
msg,
"cannot remove non-empty directory",
tag,
@ -617,16 +617,15 @@ impl Shell for FilesystemShell {
}
} else {
let msg = format!("no such file or directory: {:}", f.to_string_lossy());
yield Err(ShellError::labeled_error(
Err(ShellError::labeled_error(
msg,
"no such file or directory",
tag,
))
}
}
};
Ok(stream.to_output_stream())
}))
.to_output_stream(),
)
}
fn path(&self) -> String {