Reorganize plugin API around commands (#12170)

[Context on
Discord](https://discord.com/channels/601130461678272522/855947301380947968/1216517833312309419)

# Description
This is a significant breaking change to the plugin API, but one I think
is worthwhile. @ayax79 mentioned on Discord that while trying to start
on a dataframes plugin, he was a little disappointed that more wasn't
provided in terms of code organization for commands, particularly since
there are *a lot* of `dfr` commands.

This change treats plugins more like miniatures of the engine, with
dispatch of the command name being handled inherently, each command
being its own type, and each having their own signature within the trait
impl for the command type rather than having to find a way to centralize
it all into one `Vec`.

For the example plugins that have multiple commands, I definitely like
how this looks a lot better. This encourages doing code organization the
right way and feels very good.

For the plugins that have only one command, it's just a little bit more
boilerplate - but still worth it, in my opinion.

The `Box<dyn PluginCommand<Plugin = Self>>` type in `commands()` is a
little bit hairy, particularly for Rust beginners, but ultimately not so
bad, and it gives the desired flexibility for shared state for a whole
plugin + the individual commands.

# User-Facing Changes
Pretty big breaking change to plugin API, but probably one that's worth
making.

```rust
use nu_plugin::*;
use nu_protocol::{PluginSignature, PipelineData, Type, Value};

struct LowercasePlugin;
struct Lowercase;

// Plugins can now have multiple commands
impl PluginCommand for Lowercase {
    type Plugin = LowercasePlugin;

    // The signature lives with the command
    fn signature(&self) -> PluginSignature {
        PluginSignature::build("lowercase")
            .usage("Convert each string in a stream to lowercase")
            .input_output_type(Type::List(Type::String.into()), Type::List(Type::String.into()))
    }

    // We also provide SimplePluginCommand which operates on Value like before
    fn run(
        &self,
        plugin: &LowercasePlugin,
        engine: &EngineInterface,
        call: &EvaluatedCall,
        input: PipelineData,
    ) -> Result<PipelineData, LabeledError> {
        let span = call.head;
        Ok(input.map(move |value| {
            value.as_str()
                .map(|string| Value::string(string.to_lowercase(), span))
                // Errors in a stream should be returned as values.
                .unwrap_or_else(|err| Value::error(err, span))
        }, None)?)
    }
}

// Plugin now just has a list of commands, and the custom value op stuff still goes here
impl Plugin for LowercasePlugin {
    fn commands(&self) -> Vec<Box<dyn PluginCommand<Plugin=Self>>> {
        vec![Box::new(Lowercase)]
    }
}

fn main() {
    serve_plugin(&LowercasePlugin{}, MsgPackSerializer)
}
```

Time this however you like - we're already breaking stuff for 0.92, so
it might be good to do it now, but if it feels like a lot all at once,
it could wait.

# Tests + Formatting
- 🟢 `toolkit fmt`
- 🟢 `toolkit clippy`
- 🟢 `toolkit test`
- 🟢 `toolkit test stdlib`

# After Submitting
- [ ] Update examples in the book
- [x] Fix #12088 to match - this change would actually simplify it a
lot, because the methods are currently just duplicated between `Plugin`
and `StreamingPlugin`, but they only need to be on `Plugin` with this
change

crates/nu_plugin_stream_example/README.md

@@ -1,7 +1,6 @@
 # Streaming Plugin Example
 
-Crate with a simple example of the `StreamingPlugin` trait that needs to be implemented
-in order to create a binary that can be registered into nushell declaration list
+Crate with a simple example of a plugin with commands that produce streams
 
 ## `stream_example seq`
 

crates/nu_plugin_stream_example/src/commands/collect_external.rs

@@ -0,0 +1,51 @@
+use nu_plugin::{EngineInterface, EvaluatedCall, LabeledError, PluginCommand};
+use nu_protocol::{Category, PipelineData, PluginExample, PluginSignature, RawStream, Type, Value};
+
+use crate::StreamExample;
+
+/// `<list<string>> | stream_example collect-external`
+pub struct CollectExternal;
+
+impl PluginCommand for CollectExternal {
+    type Plugin = StreamExample;
+
+    fn signature(&self) -> PluginSignature {
+        PluginSignature::build("stream_example collect-external")
+            .usage("Example transformer to raw external stream")
+            .search_terms(vec!["example".into()])
+            .input_output_types(vec![
+                (Type::List(Type::String.into()), Type::String),
+                (Type::List(Type::Binary.into()), Type::Binary),
+            ])
+            .plugin_examples(vec![PluginExample {
+                example: "[a b] | stream_example collect-external".into(),
+                description: "collect strings into one stream".into(),
+                result: Some(Value::test_string("ab")),
+            }])
+            .category(Category::Experimental)
+    }
+
+    fn run(
+        &self,
+        _plugin: &StreamExample,
+        _engine: &EngineInterface,
+        call: &EvaluatedCall,
+        input: PipelineData,
+    ) -> Result<PipelineData, LabeledError> {
+        let stream = input.into_iter().map(|value| {
+            value
+                .as_str()
+                .map(|str| str.as_bytes())
+                .or_else(|_| value.as_binary())
+                .map(|bin| bin.to_vec())
+        });
+        Ok(PipelineData::ExternalStream {
+            stdout: Some(RawStream::new(Box::new(stream), None, call.head, None)),
+            stderr: None,
+            exit_code: None,
+            span: call.head,
+            metadata: None,
+            trim_end_newline: false,
+        })
+    }
+}

crates/nu_plugin_stream_example/src/commands/for_each.rs

@@ -0,0 +1,45 @@
+use nu_plugin::{EngineInterface, EvaluatedCall, LabeledError, PluginCommand};
+use nu_protocol::{Category, PipelineData, PluginExample, PluginSignature, SyntaxShape, Type};
+
+use crate::StreamExample;
+
+/// `<list> | stream_example for-each { |value| ... }`
+pub struct ForEach;
+
+impl PluginCommand for ForEach {
+    type Plugin = StreamExample;
+
+    fn signature(&self) -> PluginSignature {
+        PluginSignature::build("stream_example for-each")
+            .usage("Example execution of a closure with a stream")
+            .extra_usage("Prints each value the closure returns to stderr")
+            .input_output_type(Type::ListStream, Type::Nothing)
+            .required(
+                "closure",
+                SyntaxShape::Closure(Some(vec![SyntaxShape::Any])),
+                "The closure to run for each input value",
+            )
+            .plugin_examples(vec![PluginExample {
+                example: "ls | get name | stream_example for-each { |f| ^file $f }".into(),
+                description: "example with an external command".into(),
+                result: None,
+            }])
+            .category(Category::Experimental)
+    }
+
+    fn run(
+        &self,
+        _plugin: &StreamExample,
+        engine: &EngineInterface,
+        call: &EvaluatedCall,
+        input: PipelineData,
+    ) -> Result<PipelineData, LabeledError> {
+        let closure = call.req(0)?;
+        let config = engine.get_config()?;
+        for value in input {
+            let result = engine.eval_closure(&closure, vec![value.clone()], Some(value))?;
+            eprintln!("{}", result.to_expanded_string(", ", &config));
+        }
+        Ok(PipelineData::Empty)
+    }
+}

crates/nu_plugin_stream_example/src/commands/generate.rs

@@ -0,0 +1,79 @@
+use nu_plugin::{EngineInterface, EvaluatedCall, LabeledError, PluginCommand};
+use nu_protocol::{
+    Category, IntoInterruptiblePipelineData, PipelineData, PluginExample, PluginSignature,
+    SyntaxShape, Type, Value,
+};
+
+use crate::StreamExample;
+
+/// `stream_example generate <initial> { |previous| {out: ..., next: ...} }`
+pub struct Generate;
+
+impl PluginCommand for Generate {
+    type Plugin = StreamExample;
+
+    fn signature(&self) -> PluginSignature {
+        PluginSignature::build("stream_example generate")
+            .usage("Example execution of a closure to produce a stream")
+            .extra_usage("See the builtin `generate` command")
+            .input_output_type(Type::Nothing, Type::ListStream)
+            .required(
+                "initial",
+                SyntaxShape::Any,
+                "The initial value to pass to the closure",
+            )
+            .required(
+                "closure",
+                SyntaxShape::Closure(Some(vec![SyntaxShape::Any])),
+                "The closure to run to generate values",
+            )
+            .plugin_examples(vec![PluginExample {
+                example:
+                    "stream_example generate 0 { |i| if $i <= 10 { {out: $i, next: ($i + 2)} } }"
+                        .into(),
+                description: "Generate a sequence of numbers".into(),
+                result: Some(Value::test_list(
+                    [0, 2, 4, 6, 8, 10]
+                        .into_iter()
+                        .map(Value::test_int)
+                        .collect(),
+                )),
+            }])
+            .category(Category::Experimental)
+    }
+
+    fn run(
+        &self,
+        _plugin: &StreamExample,
+        engine: &EngineInterface,
+        call: &EvaluatedCall,
+        _input: PipelineData,
+    ) -> Result<PipelineData, LabeledError> {
+        let engine = engine.clone();
+        let call = call.clone();
+        let initial: Value = call.req(0)?;
+        let closure = call.req(1)?;
+
+        let mut next = (!initial.is_nothing()).then_some(initial);
+
+        Ok(std::iter::from_fn(move || {
+            next.take()
+                .and_then(|value| {
+                    engine
+                        .eval_closure(&closure, vec![value.clone()], Some(value))
+                        .and_then(|record| {
+                            if record.is_nothing() {
+                                Ok(None)
+                            } else {
+                                let record = record.as_record()?;
+                                next = record.get("next").cloned();
+                                Ok(record.get("out").cloned())
+                            }
+                        })
+                        .transpose()
+                })
+                .map(|result| result.unwrap_or_else(|err| Value::error(err, call.head)))
+        })
+        .into_pipeline_data(None))
+    }
+}

crates/nu_plugin_stream_example/src/commands/mod.rs

@@ -0,0 +1,11 @@
+mod collect_external;
+mod for_each;
+mod generate;
+mod seq;
+mod sum;
+
+pub use collect_external::CollectExternal;
+pub use for_each::ForEach;
+pub use generate::Generate;
+pub use seq::Seq;
+pub use sum::Sum;

crates/nu_plugin_stream_example/src/commands/seq.rs

@@ -0,0 +1,47 @@
+use nu_plugin::{EngineInterface, EvaluatedCall, LabeledError, PluginCommand};
+use nu_protocol::{
+    Category, ListStream, PipelineData, PluginExample, PluginSignature, SyntaxShape, Type, Value,
+};
+
+use crate::StreamExample;
+
+/// `stream_example seq <first> <last>`
+pub struct Seq;
+
+impl PluginCommand for Seq {
+    type Plugin = StreamExample;
+
+    fn signature(&self) -> PluginSignature {
+        PluginSignature::build("stream_example seq")
+            .usage("Example stream generator for a list of values")
+            .search_terms(vec!["example".into()])
+            .required("first", SyntaxShape::Int, "first number to generate")
+            .required("last", SyntaxShape::Int, "last number to generate")
+            .input_output_type(Type::Nothing, Type::List(Type::Int.into()))
+            .plugin_examples(vec![PluginExample {
+                example: "stream_example seq 1 3".into(),
+                description: "generate a sequence from 1 to 3".into(),
+                result: Some(Value::test_list(vec![
+                    Value::test_int(1),
+                    Value::test_int(2),
+                    Value::test_int(3),
+                ])),
+            }])
+            .category(Category::Experimental)
+    }
+
+    fn run(
+        &self,
+        _plugin: &StreamExample,
+        _engine: &EngineInterface,
+        call: &EvaluatedCall,
+        _input: PipelineData,
+    ) -> Result<PipelineData, LabeledError> {
+        let first: i64 = call.req(0)?;
+        let last: i64 = call.req(1)?;
+        let span = call.head;
+        let iter = (first..=last).map(move |number| Value::int(number, span));
+        let list_stream = ListStream::from_stream(iter, None);
+        Ok(PipelineData::ListStream(list_stream, None))
+    }
+}

crates/nu_plugin_stream_example/src/commands/sum.rs

@@ -0,0 +1,91 @@
+use nu_plugin::{EngineInterface, EvaluatedCall, LabeledError, PluginCommand};
+use nu_protocol::{Category, PipelineData, PluginExample, PluginSignature, Span, Type, Value};
+
+use crate::StreamExample;
+
+/// `<list> | stream_example sum`
+pub struct Sum;
+
+impl PluginCommand for Sum {
+    type Plugin = StreamExample;
+
+    fn signature(&self) -> PluginSignature {
+        PluginSignature::build("stream_example sum")
+            .usage("Example stream consumer for a list of values")
+            .search_terms(vec!["example".into()])
+            .input_output_types(vec![
+                (Type::List(Type::Int.into()), Type::Int),
+                (Type::List(Type::Float.into()), Type::Float),
+            ])
+            .plugin_examples(vec![PluginExample {
+                example: "seq 1 5 | stream_example sum".into(),
+                description: "sum values from 1 to 5".into(),
+                result: Some(Value::test_int(15)),
+            }])
+            .category(Category::Experimental)
+    }
+
+    fn run(
+        &self,
+        _plugin: &StreamExample,
+        _engine: &EngineInterface,
+        call: &EvaluatedCall,
+        input: PipelineData,
+    ) -> Result<PipelineData, LabeledError> {
+        let mut acc = IntOrFloat::Int(0);
+        let span = input.span();
+        for value in input {
+            if let Ok(n) = value.as_i64() {
+                acc.add_i64(n);
+            } else if let Ok(n) = value.as_f64() {
+                acc.add_f64(n);
+            } else {
+                return Err(LabeledError {
+                    label: "Stream only accepts ints and floats".into(),
+                    msg: format!("found {}", value.get_type()),
+                    span,
+                });
+            }
+        }
+        Ok(PipelineData::Value(acc.to_value(call.head), None))
+    }
+}
+
+/// Accumulates numbers into either an int or a float. Changes type to float on the first
+/// float received.
+#[derive(Clone, Copy)]
+enum IntOrFloat {
+    Int(i64),
+    Float(f64),
+}
+
+impl IntOrFloat {
+    pub(crate) fn add_i64(&mut self, n: i64) {
+        match self {
+            IntOrFloat::Int(ref mut v) => {
+                *v += n;
+            }
+            IntOrFloat::Float(ref mut v) => {
+                *v += n as f64;
+            }
+        }
+    }
+
+    pub(crate) fn add_f64(&mut self, n: f64) {
+        match self {
+            IntOrFloat::Int(v) => {
+                *self = IntOrFloat::Float(*v as f64 + n);
+            }
+            IntOrFloat::Float(ref mut v) => {
+                *v += n;
+            }
+        }
+    }
+
+    pub(crate) fn to_value(self, span: Span) -> Value {
+        match self {
+            IntOrFloat::Int(v) => Value::int(v, span),
+            IntOrFloat::Float(v) => Value::float(v, span),
+        }
+    }
+}

crates/nu_plugin_stream_example/src/example.rs

@@ -1,115 +0,0 @@
-use nu_plugin::{EngineInterface, EvaluatedCall, LabeledError};
-use nu_protocol::{IntoInterruptiblePipelineData, ListStream, PipelineData, RawStream, Value};
-
-pub struct Example;
-
-mod int_or_float;
-use self::int_or_float::IntOrFloat;
-
-impl Example {
-    pub fn seq(
-        &self,
-        call: &EvaluatedCall,
-        _input: PipelineData,
-    ) -> Result<PipelineData, LabeledError> {
-        let first: i64 = call.req(0)?;
-        let last: i64 = call.req(1)?;
-        let span = call.head;
-        let iter = (first..=last).map(move |number| Value::int(number, span));
-        let list_stream = ListStream::from_stream(iter, None);
-        Ok(PipelineData::ListStream(list_stream, None))
-    }
-
-    pub fn sum(
-        &self,
-        call: &EvaluatedCall,
-        input: PipelineData,
-    ) -> Result<PipelineData, LabeledError> {
-        let mut acc = IntOrFloat::Int(0);
-        let span = input.span();
-        for value in input {
-            if let Ok(n) = value.as_i64() {
-                acc.add_i64(n);
-            } else if let Ok(n) = value.as_f64() {
-                acc.add_f64(n);
-            } else {
-                return Err(LabeledError {
-                    label: "Stream only accepts ints and floats".into(),
-                    msg: format!("found {}", value.get_type()),
-                    span,
-                });
-            }
-        }
-        Ok(PipelineData::Value(acc.to_value(call.head), None))
-    }
-
-    pub fn collect_external(
-        &self,
-        call: &EvaluatedCall,
-        input: PipelineData,
-    ) -> Result<PipelineData, LabeledError> {
-        let stream = input.into_iter().map(|value| {
-            value
-                .as_str()
-                .map(|str| str.as_bytes())
-                .or_else(|_| value.as_binary())
-                .map(|bin| bin.to_vec())
-        });
-        Ok(PipelineData::ExternalStream {
-            stdout: Some(RawStream::new(Box::new(stream), None, call.head, None)),
-            stderr: None,
-            exit_code: None,
-            span: call.head,
-            metadata: None,
-            trim_end_newline: false,
-        })
-    }
-
-    pub fn for_each(
-        &self,
-        engine: &EngineInterface,
-        call: &EvaluatedCall,
-        input: PipelineData,
-    ) -> Result<PipelineData, LabeledError> {
-        let closure = call.req(0)?;
-        let config = engine.get_config()?;
-        for value in input {
-            let result = engine.eval_closure(&closure, vec![value.clone()], Some(value))?;
-            eprintln!("{}", result.to_expanded_string(", ", &config));
-        }
-        Ok(PipelineData::Empty)
-    }
-
-    pub fn generate(
-        &self,
-        engine: &EngineInterface,
-        call: &EvaluatedCall,
-    ) -> Result<PipelineData, LabeledError> {
-        let engine = engine.clone();
-        let call = call.clone();
-        let initial: Value = call.req(0)?;
-        let closure = call.req(1)?;
-
-        let mut next = (!initial.is_nothing()).then_some(initial);
-
-        Ok(std::iter::from_fn(move || {
-            next.take()
-                .and_then(|value| {
-                    engine
-                        .eval_closure(&closure, vec![value.clone()], Some(value))
-                        .and_then(|record| {
-                            if record.is_nothing() {
-                                Ok(None)
-                            } else {
-                                let record = record.as_record()?;
-                                next = record.get("next").cloned();
-                                Ok(record.get("out").cloned())
-                            }
-                        })
-                        .transpose()
-                })
-                .map(|result| result.unwrap_or_else(|err| Value::error(err, call.head)))
-        })
-        .into_pipeline_data(None))
-    }
-}

crates/nu_plugin_stream_example/src/example/int_or_float.rs

@@ -1,42 +0,0 @@
-use nu_protocol::Value;
-
-use nu_protocol::Span;
-
-/// Accumulates numbers into either an int or a float. Changes type to float on the first
-/// float received.
-#[derive(Clone, Copy)]
-pub(crate) enum IntOrFloat {
-    Int(i64),
-    Float(f64),
-}
-
-impl IntOrFloat {
-    pub(crate) fn add_i64(&mut self, n: i64) {
-        match self {
-            IntOrFloat::Int(ref mut v) => {
-                *v += n;
-            }
-            IntOrFloat::Float(ref mut v) => {
-                *v += n as f64;
-            }
-        }
-    }
-
-    pub(crate) fn add_f64(&mut self, n: f64) {
-        match self {
-            IntOrFloat::Int(v) => {
-                *self = IntOrFloat::Float(*v as f64 + n);
-            }
-            IntOrFloat::Float(ref mut v) => {
-                *v += n;
-            }
-        }
-    }
-
-    pub(crate) fn to_value(self, span: Span) -> Value {
-        match self {
-            IntOrFloat::Int(v) => Value::int(v, span),
-            IntOrFloat::Float(v) => Value::float(v, span),
-        }
-    }
-}

crates/nu_plugin_stream_example/src/lib.rs

@@ -1,4 +1,50 @@
-mod example;
-mod nu;
+use nu_plugin::{
+    EngineInterface, EvaluatedCall, LabeledError, Plugin, PluginCommand, SimplePluginCommand,
+};
+use nu_protocol::{Category, PluginSignature, Value};
 
-pub use example::Example;
+mod commands;
+pub use commands::*;
+
+pub struct StreamExample;
+
+impl Plugin for StreamExample {
+    fn commands(&self) -> Vec<Box<dyn PluginCommand<Plugin = Self>>> {
+        vec![
+            Box::new(Main),
+            Box::new(Seq),
+            Box::new(Sum),
+            Box::new(CollectExternal),
+            Box::new(ForEach),
+            Box::new(Generate),
+        ]
+    }
+}
+
+/// `stream_example`
+pub struct Main;
+
+impl SimplePluginCommand for Main {
+    type Plugin = StreamExample;
+
+    fn signature(&self) -> PluginSignature {
+        PluginSignature::build("stream_example")
+            .usage("Examples for streaming plugins")
+            .search_terms(vec!["example".into()])
+            .category(Category::Experimental)
+    }
+
+    fn run(
+        &self,
+        _plugin: &StreamExample,
+        _engine: &EngineInterface,
+        call: &EvaluatedCall,
+        _input: &Value,
+    ) -> Result<Value, LabeledError> {
+        Err(LabeledError {
+            label: "No subcommand provided".into(),
+            msg: "add --help here to see usage".into(),
+            span: Some(call.head.past()),
+        })
+    }
+}

crates/nu_plugin_stream_example/src/main.rs

@@ -1,12 +1,12 @@
 use nu_plugin::{serve_plugin, MsgPackSerializer};
-use nu_plugin_stream_example::Example;
+use nu_plugin_stream_example::StreamExample;
 
 fn main() {
     // When defining your plugin, you can select the Serializer that could be
     // used to encode and decode the messages. The available options are
     // MsgPackSerializer and JsonSerializer. Both are defined in the serializer
     // folder in nu-plugin.
-    serve_plugin(&Example {}, MsgPackSerializer {})
+    serve_plugin(&StreamExample {}, MsgPackSerializer {})
 
     // Note
     // When creating plugins in other languages one needs to consider how a plugin

crates/nu_plugin_stream_example/src/nu/mod.rs

@@ -1,125 +0,0 @@
-use crate::Example;
-use nu_plugin::{EngineInterface, EvaluatedCall, LabeledError, StreamingPlugin};
-use nu_protocol::{
-    Category, PipelineData, PluginExample, PluginSignature, Span, SyntaxShape, Type, Value,
-};
-
-impl StreamingPlugin for Example {
-    fn signature(&self) -> Vec<PluginSignature> {
-        let span = Span::unknown();
-        vec![
-            PluginSignature::build("stream_example")
-                .usage("Examples for streaming plugins")
-                .search_terms(vec!["example".into()])
-                .category(Category::Experimental),
-            PluginSignature::build("stream_example seq")
-                .usage("Example stream generator for a list of values")
-                .search_terms(vec!["example".into()])
-                .required("first", SyntaxShape::Int, "first number to generate")
-                .required("last", SyntaxShape::Int, "last number to generate")
-                .input_output_type(Type::Nothing, Type::List(Type::Int.into()))
-                .plugin_examples(vec![PluginExample {
-                    example: "stream_example seq 1 3".into(),
-                    description: "generate a sequence from 1 to 3".into(),
-                    result: Some(Value::list(
-                        vec![
-                            Value::int(1, span),
-                            Value::int(2, span),
-                            Value::int(3, span),
-                        ],
-                        span,
-                    )),
-                }])
-                .category(Category::Experimental),
-            PluginSignature::build("stream_example sum")
-                .usage("Example stream consumer for a list of values")
-                .search_terms(vec!["example".into()])
-                .input_output_types(vec![
-                    (Type::List(Type::Int.into()), Type::Int),
-                    (Type::List(Type::Float.into()), Type::Float),
-                ])
-                .plugin_examples(vec![PluginExample {
-                    example: "seq 1 5 | stream_example sum".into(),
-                    description: "sum values from 1 to 5".into(),
-                    result: Some(Value::int(15, span)),
-                }])
-                .category(Category::Experimental),
-            PluginSignature::build("stream_example collect-external")
-                .usage("Example transformer to raw external stream")
-                .search_terms(vec!["example".into()])
-                .input_output_types(vec![
-                    (Type::List(Type::String.into()), Type::String),
-                    (Type::List(Type::Binary.into()), Type::Binary),
-                ])
-                .plugin_examples(vec![PluginExample {
-                    example: "[a b] | stream_example collect-external".into(),
-                    description: "collect strings into one stream".into(),
-                    result: Some(Value::string("ab", span)),
-                }])
-                .category(Category::Experimental),
-            PluginSignature::build("stream_example for-each")
-                .usage("Example execution of a closure with a stream")
-                .extra_usage("Prints each value the closure returns to stderr")
-                .input_output_type(Type::ListStream, Type::Nothing)
-                .required(
-                    "closure",
-                    SyntaxShape::Closure(Some(vec![SyntaxShape::Any])),
-                    "The closure to run for each input value",
-                )
-                .plugin_examples(vec![PluginExample {
-                    example: "ls | get name | stream_example for-each { |f| ^file $f }".into(),
-                    description: "example with an external command".into(),
-                    result: None,
-                }])
-                .category(Category::Experimental),
-            PluginSignature::build("stream_example generate")
-                .usage("Example execution of a closure to produce a stream")
-                .extra_usage("See the builtin `generate` command")
-                .input_output_type(Type::Nothing, Type::ListStream)
-                .required(
-                    "initial",
-                    SyntaxShape::Any,
-                    "The initial value to pass to the closure"
-                )
-                .required(
-                    "closure",
-                    SyntaxShape::Closure(Some(vec![SyntaxShape::Any])),
-                    "The closure to run to generate values",
-                )
-                .plugin_examples(vec![PluginExample {
-                    example: "stream_example generate 0 { |i| if $i <= 10 { {out: $i, next: ($i + 2)} } }".into(),
-                    description: "Generate a sequence of numbers".into(),
-                    result: Some(Value::test_list(
-                        [0, 2, 4, 6, 8, 10].into_iter().map(Value::test_int).collect(),
-                    )),
-                }])
-                .category(Category::Experimental),
-        ]
-    }
-
-    fn run(
-        &self,
-        name: &str,
-        engine: &EngineInterface,
-        call: &EvaluatedCall,
-        input: PipelineData,
-    ) -> Result<PipelineData, LabeledError> {
-        match name {
-            "stream_example" => Err(LabeledError {
-                label: "No subcommand provided".into(),
-                msg: "add --help here to see usage".into(),
-                span: Some(call.head)
-            }),
-            "stream_example seq" => self.seq(call, input),
-            "stream_example sum" => self.sum(call, input),
-            "stream_example collect-external" => self.collect_external(call, input),
-            "stream_example for-each" => self.for_each(engine, call, input),
-            "stream_example generate" => self.generate(engine, call),
-            _ => Err(LabeledError {
-                label: "Plugin call with wrong name signature".into(),
-                msg: "the signature used to call the plugin does not match any name in the plugin signature vector".into(),
-                span: Some(call.head),
-            }),
-        }
-    }
-}