Add x command + fix fuzzer

crates/commands/x/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "rew_command_x"
+version.workspace = true
+edition.workspace = true
+
+[dependencies]
+anyhow.workspace = true
+bstr.workspace = true
+clap.workspace = true
+owo-colors.workspace = true
+rew_adapters = { path = "../../libs/adapters" }
+rew_core = { path = "../../libs/core" }
+rew_command = { path = "../../libs/command" }
+rew_exec = { path = "../../libs/exec" }
+rew_io = { path = "../../libs/io" }
+thiserror.workspace = true
+
+[dev-dependencies]
+claims.workspace = true
+rstest.workspace = true

crates/commands/x/src/lib.rs

@@ -0,0 +1,207 @@
+mod pattern;
+
+use crate::pattern::Item;
+pub use crate::pattern::Pattern;
+use anyhow::Result;
+use bstr::BStr;
+use bstr::BString;
+use clap::builder::ValueParser;
+use clap::command;
+use clap::Args;
+use rew_adapters::AdaptChannel;
+use rew_adapters::ConsumeToWrite;
+use rew_adapters::ProduceFromRead;
+use rew_adapters::ToBuffer;
+use rew_command::declare_command;
+use rew_command::CommandMeta;
+use rew_core::Channel;
+use rew_core::Consume;
+use rew_core::ProduceTo;
+use rew_exec::GlobalArgs;
+use rew_exec::InitWithGlobal;
+use rew_io::ByteSize;
+use rew_io::RecordReader;
+use rew_io::RecordWriter;
+use rew_io::Separator;
+use std::io::BufReader;
+use std::process::Child;
+use std::process::ChildStdin;
+use std::process::ChildStdout;
+use std::process::Stdio;
+use std::str::FromStr;
+
+declare_command! {
+    inner: Exec,
+    meta: ExecMeta,
+    args: ExecArgs,
+    adapter: AdaptChannel<Exec>,
+}
+
+/// Compose commands using a pattern
+#[derive(Args)]
+struct ExecArgs {
+    /// Composition pattern
+    pattern: Pattern,
+
+    /// Escape character for the pattern
+    #[arg(short, long, value_name = "CHAR", default_value_t = '\\')]
+    escape: char,
+}
+
+impl ExecArgs {
+    // We need to access `escape` option before parsing/validating the pattern.
+    // However, this is currently not supported by clap.
+    //
+    // So, we do the following trick instead:
+    //   1. Construct light-weight clap `Command` with only our subcommand available.
+    //   2. Disable validation of the `pattern` argument on the subcommand.
+    //   3. Parse program args using our modified `Command` and get `escape` from matches.
+    //
+    // We could also make the `pattern` a `String` and do the whole parsing after we have complete
+    // args but that would give us less consistent error message in case of an invalid pattern.
+    fn parse_escape() -> char {
+        *command!()
+            .subcommand(
+                ExecMeta::new()
+                    .build()
+                    .mut_arg("pattern", |arg| arg.value_parser(ValueParser::string())),
+            )
+            .get_matches()
+            .subcommand()
+            .and_then(|(_, matches)| matches.get_one::<char>("escape"))
+            .expect("could not parse `escape` option")
+    }
+}
+
+impl FromStr for Pattern {
+    type Err = pattern::Error;
+
+    fn from_str(input: &str) -> std::result::Result<Self, Self::Err> {
+        Pattern::parse(input, ExecArgs::parse_escape())
+    }
+}
+
+type ChildConsumer = ConsumeToWrite<RecordWriter, ChildStdin>;
+type ChildProducer = ProduceFromRead<BufReader<ChildStdout>, RecordReader>;
+
+struct Exec {
+    children: Vec<Child>,
+    max_line: ByteSize,
+    buffered: bool,
+}
+
+impl Exec {
+    fn new(children: Vec<Child>, max_line: ByteSize, buffered: bool) -> Self {
+        Self {
+            children,
+            max_line,
+            buffered,
+        }
+    }
+}
+
+impl InitWithGlobal<ExecArgs> for Exec {
+    fn init_with_global(args: ExecArgs, global_args: &GlobalArgs) -> Self {
+        let mut children: Vec<Child> = Vec::new();
+
+        for item in args.pattern.items() {
+            if let Item::Expression(pipeline) = item {
+                for command in pipeline {
+                    let child = std::process::Command::new(&command.name)
+                        .args(&command.args)
+                        .stdin(Stdio::piped())
+                        .stdout(Stdio::piped())
+                        .spawn()
+                        .unwrap();
+
+                    children.push(child);
+                }
+            }
+        }
+
+        Exec::new(
+            children,
+            global_args.max_line.clone(),
+            global_args.buffered(),
+        )
+    }
+}
+
+pub struct Consumer {
+    consumers: Vec<ChildConsumer>,
+}
+
+impl Consumer {
+    fn new(consumers: Vec<ChildConsumer>) -> Self {
+        Self { consumers }
+    }
+}
+
+impl Consume for Consumer {
+    fn consume(&mut self, value: &BStr) -> Result<()> {
+        for consumer in &mut self.consumers {
+            consumer.consume(value)?;
+        }
+        Ok(())
+    }
+
+    fn done(&mut self) -> Result<()> {
+        for consumer in &mut self.consumers {
+            consumer.done()?;
+        }
+        Ok(())
+    }
+}
+
+pub struct Producer {
+    producers: Vec<ChildProducer>,
+}
+
+impl Producer {
+    fn new(producers: Vec<ChildProducer>) -> Self {
+        Self { producers }
+    }
+}
+
+impl ProduceTo for Producer {
+    fn produce_to(&mut self, buffer: &mut BString) -> Result<bool> {
+        let mut some = false;
+        for producer in &mut self.producers {
+            if producer.produce_to(buffer)? {
+                some = true;
+            }
+        }
+        Ok(some)
+    }
+}
+
+impl Channel for Exec {
+    type Consumer = Consumer;
+    type Producer = ToBuffer<Producer>;
+
+    fn open(mut self) -> Result<(Self::Consumer, Self::Producer)> {
+        let consumers: Vec<ChildConsumer> = self
+            .children
+            .iter_mut()
+            .map(|child| {
+                let stdin = child.stdin.take().expect("Child has no stdin");
+                let consumer = RecordWriter::new(Separator::Newline, self.buffered);
+                ConsumeToWrite::new(consumer, stdin)
+            })
+            .collect();
+
+        let producers: Vec<ChildProducer> = self
+            .children
+            .iter_mut()
+            .map(|child| {
+                let stdout = child.stdout.take().expect("Child has no stdout");
+                let producer = RecordReader::new(Separator::Newline, self.max_line.clone());
+                ProduceFromRead::new(BufReader::new(stdout), producer)
+            })
+            .collect();
+
+        let consumer = Consumer::new(consumers);
+        let producer = ToBuffer::new(Producer::new(producers));
+        Ok((consumer, producer))
+    }
+}