Extract ps and sys subcrates. Move helper methods to UntaggedValue

crates/nu_plugin_ps/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+name = "nu_plugin_ps"
+version = "0.1.0"
+authors = ["Yehuda Katz <wycats@gmail.com>", "Jonathan Turner <jonathan.d.turner@gmail.com>", "Andrés N. Robalino <andres@androbtech.com>"]
+edition = "2018"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+nu-protocol = { path = "../nu-protocol" }
+nu-source = { path = "../nu-source" }
+nu-errors = { path = "../nu-errors" }
+futures-preview = { version = "=0.3.0-alpha.19", features = ["compat", "io-compat"] }
+heim = "0.0.8"
+futures-timer = "2.0.0"
+pin-utils = "0.1.0-alpha.4"
+
+[build-dependencies]
+nu-build = { version = "0.1.0", path = "../nu-build" }

crates/nu_plugin_ps/build.rs

@@ -0,0 +1,3 @@
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    nu_build::build()
+}

crates/nu_plugin_ps/src/main.rs

@@ -0,0 +1,83 @@
+use futures::executor::block_on;
+use futures::stream::{StreamExt, TryStreamExt};
+
+use heim::process::{self as process, Process, ProcessResult};
+use heim::units::{ratio, Ratio};
+use std::usize;
+
+use nu_errors::ShellError;
+use nu_protocol::{
+    serve_plugin, CallInfo, Plugin, ReturnSuccess, ReturnValue, Signature, TaggedDictBuilder,
+    UntaggedValue, Value,
+};
+use nu_source::Tag;
+
+use std::time::Duration;
+
+struct Ps;
+impl Ps {
+    fn new() -> Ps {
+        Ps
+    }
+}
+
+async fn usage(process: Process) -> ProcessResult<(process::Process, Ratio)> {
+    let usage_1 = process.cpu_usage().await?;
+    futures_timer::Delay::new(Duration::from_millis(100)).await;
+    let usage_2 = process.cpu_usage().await?;
+
+    Ok((process, usage_2 - usage_1))
+}
+
+async fn ps(tag: Tag) -> Vec<Value> {
+    let processes = process::processes()
+        .map_ok(|process| {
+            // Note that there is no `.await` here,
+            // as we want to pass the returned future
+            // into the `.try_buffer_unordered`.
+            usage(process)
+        })
+        .try_buffer_unordered(usize::MAX);
+    pin_utils::pin_mut!(processes);
+
+    let mut output = vec![];
+    while let Some(res) = processes.next().await {
+        if let Ok((process, usage)) = res {
+            let mut dict = TaggedDictBuilder::new(&tag);
+            dict.insert_untagged("pid", UntaggedValue::int(process.pid()));
+            if let Ok(name) = process.name().await {
+                dict.insert_untagged("name", UntaggedValue::string(name));
+            }
+            if let Ok(status) = process.status().await {
+                dict.insert_untagged("status", UntaggedValue::string(format!("{:?}", status)));
+            }
+            dict.insert_untagged("cpu", UntaggedValue::decimal(usage.get::<ratio::percent>()));
+            output.push(dict.into_value());
+        }
+    }
+
+    output
+}
+
+impl Plugin for Ps {
+    fn config(&mut self) -> Result<Signature, ShellError> {
+        Ok(Signature::build("ps")
+            .desc("View information about system processes.")
+            .filter())
+    }
+
+    fn begin_filter(&mut self, callinfo: CallInfo) -> Result<Vec<ReturnValue>, ShellError> {
+        Ok(block_on(ps(callinfo.name_tag))
+            .into_iter()
+            .map(ReturnSuccess::value)
+            .collect())
+    }
+
+    fn filter(&mut self, _: Value) -> Result<Vec<ReturnValue>, ShellError> {
+        Ok(vec![])
+    }
+}
+
+fn main() {
+    serve_plugin(&mut Ps::new());
+}