refactor: demangle lib and bin implementations

- remove bin options from lib options
- move default implementations into its own sub module

src/bin/main.rs

@@ -6,6 +6,7 @@ extern crate shlex;
 extern crate skim;
 extern crate time;
 
+use derive_builder::Builder;
 use std::env;
 use std::fs::File;
 use std::io::{BufRead, BufReader, BufWriter, Write};
@@ -226,17 +227,25 @@ fn real_main() -> Result<i32, std::io::Error> {
         writeln!(stdout, "{}", VERSION)?;
         return Ok(0);
     }
+    //------------------------------------------------------------------------------
+    // initialize collector
+    let collector_option = CollectorOption::default()
+        .ansi(opts.is_present("ansi"))
+        .delimiter(opts.values_of("delimiter").and_then(|vals| vals.last()).unwrap_or(""))
+        .with_nth(opts.values_of("with-nth").and_then(|vals| vals.last()).unwrap_or(""))
+        .nth(opts.values_of("nth").and_then(|vals| vals.last()).unwrap_or(""))
+        .read0(opts.is_present("read0"))
+        .replace_str(opts.values_of("replstr").and_then(|vals| vals.last()).unwrap_or(""))
+        .build();
+
+    let cmd_collector = Rc::new(RefCell::new(DefaultSkimCollector::new(collector_option)));
 
     //------------------------------------------------------------------------------
     // read in the history file
     let fz_query_histories = opts.values_of("history").and_then(|vals| vals.last());
     let cmd_query_histories = opts.values_of("cmd-history").and_then(|vals| vals.last());
-    debug!("query_history_file: {:?}", fz_query_histories);
-    debug!("cmd_history_file: {:?}", cmd_query_histories);
     let query_history = fz_query_histories.and_then(|filename| read_file_lines(filename).ok()).unwrap_or_else(|| vec![]);
     let cmd_history = cmd_query_histories.and_then(|filename| read_file_lines(filename).ok()).unwrap_or_else(|| vec![]);
-    debug!("query_history: {:?}", query_history);
-    debug!("cmd_history: {:?}", query_history);
 
     let mut options = parse_options(&opts);
     if fz_query_histories.is_some() || cmd_query_histories.is_some() {
@@ -246,16 +255,26 @@ fn real_main() -> Result<i32, std::io::Error> {
         options.bind.insert(0, "ctrl-p:previous-history,ctrl-n:next-history");
     }
 
+    options.cmd_collector = cmd_collector.clone();
+
     let options = options;
 
+    //------------------------------------------------------------------------------
+    let bin_options = BinOptionsBuilder::default()
+        .filter(opts.values_of("filter").and_then(|vals| vals.last()))
+        .print_query(opts.is_present("print_query"))
+        .print_cmd(opts.is_present("print_cmd"))
+        .output_ending(if opts.is_present("print0") { "\0" } else { "\n" })
+        .build()
+        .expect("");
+
     //------------------------------------------------------------------------------
     // read from pipe or command
     let stdin = std::io::stdin();
     let components_to_stop = Arc::new(AtomicUsize::new(0));
     let rx_item = match isatty(stdin.as_raw_fd()) {
         Ok(false) | Err(nix::Error::Sys(nix::errno::Errno::EINVAL)) => {
-            let collector_option = CollectorOption::with_options(&options);
+            let (rx_item, _) = cmd_collector.borrow().read_and_collect_from_command(components_to_stop, CollectorInput::Pipe(Box::new(BufReader::new(stdin))));
-            let (rx_item, _) = read_and_collect_from_command(components_to_stop, CollectorInput::Pipe(Box::new(BufReader::new(stdin))), collector_option);
             Some(rx_item)
         }
         Ok(true) | Err(_) => None,
@@ -264,12 +283,10 @@ fn real_main() -> Result<i32, std::io::Error> {
     //------------------------------------------------------------------------------
     // filter mode
     if opts.is_present("filter") {
-        return filter(&options, rx_item);
+        return filter(&bin_options, &options, rx_item);
     }
 
     //------------------------------------------------------------------------------
-    let output_ending = if options.print0 { "\0" } else { "\n" };
-
     let output = Skim::run_with(&options, rx_item);
     if output.is_none() {
         return Ok(130);
@@ -280,20 +297,20 @@ fn real_main() -> Result<i32, std::io::Error> {
     let output = output.unwrap();
 
     // output query
-    if options.print_query {
+    if bin_options.print_query {
-        write!(stdout, "{}{}", output.query, output_ending)?;
+        write!(stdout, "{}{}", output.query, bin_options.output_ending)?;
     }
 
-    if options.print_cmd {
+    if bin_options.print_cmd {
-        write!(stdout, "{}{}", output.cmd, output_ending)?;
+        write!(stdout, "{}{}", output.cmd, bin_options.output_ending)?;
     }
 
     if let Some(key) = output.accept_key {
-        write!(stdout, "{}{}", key, output_ending)?;
+        write!(stdout, "{}{}", key, bin_options.output_ending)?;
     }
 
     for item in output.selected_items.iter() {
-        write!(stdout, "{}{}", item.output(), output_ending)?;
+        write!(stdout, "{}{}", item.output(), bin_options.output_ending)?;
     }
 
     //------------------------------------------------------------------------------
@@ -327,15 +344,9 @@ fn parse_options<'a>(options: &'a ArgMatches) -> SkimOptions<'a> {
         .cmd(options.values_of("cmd").and_then(|vals| vals.last()))
         .query(options.values_of("query").and_then(|vals| vals.last()))
         .cmd_query(options.values_of("cmd-query").and_then(|vals| vals.last()))
-        .replstr(options.values_of("replstr").and_then(|vals| vals.last()))
         .interactive(options.is_present("interactive"))
         .prompt(options.values_of("prompt").and_then(|vals| vals.last()))
         .cmd_prompt(options.values_of("cmd-prompt").and_then(|vals| vals.last()))
-        .ansi(options.is_present("ansi"))
-        .delimiter(options.values_of("delimiter").and_then(|vals| vals.last()))
-        .with_nth(options.values_of("with-nth").and_then(|vals| vals.last()))
-        .nth(options.values_of("nth").and_then(|vals| vals.last()))
-        .read0(options.is_present("read0"))
         .bind(
             options
                 .values_of("bind")
@@ -350,9 +361,6 @@ fn parse_options<'a>(options: &'a ArgMatches) -> SkimOptions<'a> {
         })
         .layout(options.values_of("layout").and_then(|vals| vals.last()).unwrap_or(""))
         .reverse(options.is_present("reverse"))
-        .print0(options.is_present("print0"))
-        .print_query(options.is_present("print-query"))
-        .print_cmd(options.is_present("print-cmd"))
         .no_hscroll(options.is_present("no-hscroll"))
         .no_mouse(options.is_present("no-mouse"))
         .tabstop(options.values_of("tabstop").and_then(|vals| vals.last()))
@@ -371,7 +379,6 @@ fn parse_options<'a>(options: &'a ArgMatches) -> SkimOptions<'a> {
                 .unwrap_or(0),
         )
         .layout(options.values_of("layout").and_then(|vals| vals.last()).unwrap_or(""))
-        .filter(options.values_of("filter").and_then(|vals| vals.last()).unwrap_or(""))
         .algorithm(FuzzyAlgorithm::of(
             options.values_of("algorithm").and_then(|vals| vals.last()).unwrap(),
         ))
@@ -417,24 +424,35 @@ fn write_history_to_file(
     Ok(())
 }
 
-pub fn filter(options: &SkimOptions, source: Option<SkimItemReceiver>) -> Result<i32, std::io::Error> {
+#[derive(Builder)]
+pub struct BinOptions<'a> {
+    filter: Option<&'a str>,
+    output_ending: &'a str,
+    print_query: bool,
+    print_cmd: bool,
+}
+
+pub fn filter(
+    bin_option: &BinOptions,
+    options: &SkimOptions,
+    source: Option<SkimItemReceiver>,
+) -> Result<i32, std::io::Error> {
     let mut stdout = std::io::stdout();
 
-    let output_ending = if options.print0 { "\0" } else { "\n" };
-    let query = options.filter;
     let default_command = match env::var("SKIM_DEFAULT_COMMAND").as_ref().map(String::as_ref) {
         Ok("") | Err(_) => "find .".to_owned(),
         Ok(val) => val.to_owned(),
     };
+    let query = bin_option.filter.unwrap_or(&"");
     let cmd = options.cmd.unwrap_or(&default_command);
 
     // output query
-    if options.print_query {
+    if bin_option.print_query {
-        write!(stdout, "{}{}", query, output_ending)?;
+        write!(stdout, "{}{}", query, bin_option.output_ending)?;
     }
 
-    if options.print_cmd {
+    if bin_option.print_cmd {
-        write!(stdout, "{}{}", cmd, output_ending)?;
+        write!(stdout, "{}{}", cmd, bin_option.output_ending)?;
     }
 
     //------------------------------------------------------------------------------
@@ -454,11 +472,10 @@ pub fn filter(options: &SkimOptions, source: Option<SkimItemReceiver>) -> Result
     //------------------------------------------------------------------------------
     // start
     let components_to_stop = Arc::new(AtomicUsize::new(0));
-    let collector_option = CollectorOption::with_options(&options);
 
     let stream_of_item = source.unwrap_or_else(|| {
-        let collector_input = CollectorInput::Command(cmd.to_string());
+        let cmd_collector = options.cmd_collector.clone();
-        let (ret, _control) = read_and_collect_from_command(components_to_stop, collector_input, collector_option);
+        let (ret, _control) = cmd_collector.borrow_mut().invoke(cmd, components_to_stop);
         ret
     });
 
@@ -468,7 +485,7 @@ pub fn filter(options: &SkimOptions, source: Option<SkimItemReceiver>) -> Result
         .filter_map(|item| engine.match_item(item))
         .try_for_each(|matched| {
             num_matched += 1;
-            write!(stdout, "{}{}", matched.item.output(), output_ending)
+            write!(stdout, "{}{}", matched.item.output(), bin_option.output_ending)
         })?;
 
     Ok(if num_matched == 0 { 1 } else { 0 })

src/helper/item.rs

@@ -0,0 +1,112 @@
+use crate::ansi::ANSIParser;
+use crate::field::{parse_matching_fields, parse_transform_fields, FieldRange};
+use crate::{AnsiString, SkimItem};
+use regex::Regex;
+use std::borrow::Cow;
+
+//------------------------------------------------------------------------------
+/// An item will store everything that one line input will need to be operated and displayed.
+///
+/// What's special about an item?
+/// The simplest version of an item is a line of string, but things are getting more complex:
+/// - The conversion of lower/upper case is slow in rust, because it involds unicode.
+/// - We may need to interpret the ANSI codes in the text.
+/// - The text can be transformed and limited while searching.
+///
+/// About the ANSI, we made assumption that it is linewise, that means no ANSI codes will affect
+/// more than one line.
+#[derive(Debug)]
+pub struct DefaultSkimItem {
+    /// The text that will be output when user press `enter`
+    /// `Some(..)` => the original input is transformed, could not output `text` directly
+    /// `None` => that it is safe to output `text` directly
+    orig_text: Option<String>,
+
+    /// The text that will be shown on screen and matched.
+    text: AnsiString<'static>,
+
+    // Option<Box<_>> to reduce memory use in normal cases where no matching ranges are specified.
+    #[allow(clippy::box_vec)]
+    matching_ranges: Option<Box<Vec<(usize, usize)>>>,
+}
+
+impl<'a> DefaultSkimItem {
+    pub fn new(
+        orig_text: String,
+        ansi_enabled: bool,
+        trans_fields: &[FieldRange],
+        matching_fields: &[FieldRange],
+        delimiter: &Regex,
+    ) -> Self {
+        let using_transform_fields = !trans_fields.is_empty();
+
+        //        transformed | ANSI             | output
+        //------------------------------------------------------
+        //                    +- T -> trans+ANSI | ANSI
+        //                    |                  |
+        //      +- T -> trans +- F -> trans      | orig
+        // orig |                                |
+        //      +- F -> orig  +- T -> ANSI     ==| ANSI
+        //                    |                  |
+        //                    +- F -> orig       | orig
+
+        let mut ansi_parser: ANSIParser = Default::default();
+
+        let (orig_text, text) = if using_transform_fields && ansi_enabled {
+            // ansi and transform
+            let transformed = ansi_parser.parse_ansi(&parse_transform_fields(delimiter, &orig_text, trans_fields));
+            (Some(orig_text), transformed)
+        } else if using_transform_fields {
+            // transformed, not ansi
+            let transformed = parse_transform_fields(delimiter, &orig_text, trans_fields).into();
+            (Some(orig_text), transformed)
+        } else if ansi_enabled {
+            // not transformed, ansi
+            (None, ansi_parser.parse_ansi(&orig_text))
+        } else {
+            // normal case
+            (None, orig_text.into())
+        };
+
+        let matching_ranges = if !matching_fields.is_empty() {
+            Some(Box::new(parse_matching_fields(
+                delimiter,
+                text.stripped(),
+                matching_fields,
+            )))
+        } else {
+            None
+        };
+
+        DefaultSkimItem {
+            orig_text,
+            text,
+            matching_ranges,
+        }
+    }
+}
+
+impl SkimItem for DefaultSkimItem {
+    #[inline]
+    fn text(&self) -> Cow<str> {
+        Cow::Borrowed(self.text.stripped())
+    }
+
+    fn output(&self) -> Cow<str> {
+        if self.orig_text.is_some() {
+            if self.text.has_attrs() {
+                let mut ansi_parser: ANSIParser = Default::default();
+                let text = ansi_parser.parse_ansi(self.orig_text.as_ref().unwrap());
+                text.into_inner()
+            } else {
+                Cow::Borrowed(self.orig_text.as_ref().unwrap())
+            }
+        } else {
+            Cow::Borrowed(self.text.stripped())
+        }
+    }
+
+    fn get_matching_ranges(&self) -> Option<&[(usize, usize)]> {
+        self.matching_ranges.as_ref().map(|vec| vec as &[(usize, usize)])
+    }
+}

src/helper/item_collector.rs

@@ -0,0 +1,246 @@
+use crate::field::FieldRange;
+use crate::helper::item::DefaultSkimItem;
+use crate::reader::CommandCollector;
+use crate::{SkimItem, SkimItemReceiver, SkimItemSender};
+use crossbeam::channel::{bounded, Receiver, Sender};
+use regex::Regex;
+use std::env;
+use std::error::Error;
+use std::io::{BufRead, BufReader};
+use std::process::{Child, Command, Stdio};
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+use std::sync::Arc;
+use std::thread;
+
+const CMD_CHANNEL_SIZE: usize = 1024;
+const ITEM_CHANNEL_SIZE: usize = 10240;
+const DELIMITER_STR: &str = r"[\t\n ]+";
+const READ_BUFFER_SIZE: usize = 1024;
+
+#[derive(Clone, Debug)]
+pub struct CollectorOption {
+    use_ansi_color: bool,
+    transform_fields: Vec<FieldRange>,
+    matching_fields: Vec<FieldRange>,
+    delimiter: Regex,
+    replace_str: String,
+    line_ending: u8,
+}
+
+impl Default for CollectorOption {
+    fn default() -> Self {
+        Self {
+            use_ansi_color: false,
+            transform_fields: Vec::new(),
+            matching_fields: Vec::new(),
+            delimiter: Regex::new(DELIMITER_STR).unwrap(),
+            replace_str: "{}".to_string(),
+            line_ending: b'\n',
+        }
+    }
+}
+
+impl CollectorOption {
+    pub fn ansi(mut self, enable: bool) -> Self {
+        self.use_ansi_color = enable;
+        self
+    }
+
+    pub fn delimiter(mut self, delimiter: &str) -> Self {
+        if !delimiter.is_empty() {
+            self.delimiter = Regex::new(delimiter).unwrap_or_else(|_| Regex::new(DELIMITER_STR).unwrap());
+        }
+        self
+    }
+
+    pub fn with_nth(mut self, with_nth: &str) -> Self {
+        if !with_nth.is_empty() {
+            self.transform_fields = with_nth
+                .split(',')
+                .filter_map(|string| FieldRange::from_str(string))
+                .collect();
+        }
+        self
+    }
+
+    pub fn transform_fields(mut self, transform_fields: Vec<FieldRange>) -> Self {
+        self.transform_fields = transform_fields;
+        self
+    }
+
+    pub fn nth(mut self, nth: &str) -> Self {
+        if !nth.is_empty() {
+            self.matching_fields = nth
+                .split(',')
+                .filter_map(|string| FieldRange::from_str(string))
+                .collect();
+        }
+        self
+    }
+
+    pub fn matching_fields(mut self, matching_fields: Vec<FieldRange>) -> Self {
+        self.matching_fields = matching_fields;
+        self
+    }
+
+    pub fn replace_str(mut self, replace_str: &str) -> Self {
+        if !replace_str.is_empty() {
+            self.replace_str = replace_str.to_string();
+        }
+        self
+    }
+
+    pub fn read0(mut self, enable: bool) -> Self {
+        if enable {
+            self.line_ending = b'\0';
+        } else {
+            self.line_ending = b'\n';
+        }
+        self
+    }
+
+    pub fn build(self) -> Self {
+        self
+    }
+}
+
+pub enum CollectorInput {
+    Pipe(Box<dyn BufRead + Send>),
+    Command(String),
+}
+
+pub struct DefaultSkimCollector {
+    option: Arc<CollectorOption>,
+}
+
+impl DefaultSkimCollector {
+    pub fn new(option: CollectorOption) -> Self {
+        Self {
+            option: Arc::new(option),
+        }
+    }
+
+    /// components_to_stop == 0 => all the threads have been stopped
+    /// return (channel_for_receive_item, channel_to_stop_command)
+    pub fn read_and_collect_from_command(
+        &self,
+        components_to_stop: Arc<AtomicUsize>,
+        input: CollectorInput,
+    ) -> (Receiver<Arc<dyn SkimItem>>, Sender<i32>) {
+        let (command, mut source) = match input {
+            CollectorInput::Pipe(pipe) => (None, pipe),
+            CollectorInput::Command(cmd) => get_command_output(&cmd).expect("command not found"),
+        };
+
+        let (tx_interrupt, rx_interrupt) = bounded(CMD_CHANNEL_SIZE);
+        let (tx_item, rx_item): (SkimItemSender, SkimItemReceiver) = bounded(ITEM_CHANNEL_SIZE);
+
+        let started = Arc::new(AtomicBool::new(false));
+        let started_clone = started.clone();
+        let components_to_stop_clone = components_to_stop.clone();
+        let option = self.option.clone();
+        // listening to close signal and kill command if needed
+        thread::spawn(move || {
+            debug!("collector: command killer start");
+            components_to_stop_clone.fetch_add(1, Ordering::SeqCst);
+            started_clone.store(true, Ordering::SeqCst); // notify parent that it is started
+
+            let _ = rx_interrupt.recv(); // block waiting
+                                         // clean up resources
+            if let Some(mut x) = command {
+                let _ = x.kill();
+                let _ = x.wait();
+            }
+
+            components_to_stop_clone.fetch_sub(1, Ordering::SeqCst);
+            debug!("collector: command killer stop");
+        });
+
+        while !started.load(Ordering::SeqCst) {
+            // busy waiting for the thread to start. (components_to_stop is added)
+        }
+
+        let started = Arc::new(AtomicBool::new(false));
+        let started_clone = started.clone();
+        let tx_interrupt_clone = tx_interrupt.clone();
+        thread::spawn(move || {
+            debug!("collector: command collector start");
+            components_to_stop.fetch_add(1, Ordering::SeqCst);
+            started_clone.store(true, Ordering::SeqCst); // notify parent that it is started
+
+            let mut buffer = Vec::with_capacity(READ_BUFFER_SIZE);
+            loop {
+                buffer.clear();
+
+                // start reading
+                match source.read_until(option.line_ending, &mut buffer) {
+                    Ok(n) => {
+                        if n == 0 {
+                            break;
+                        }
+
+                        if buffer.ends_with(&[b'\r', b'\n']) {
+                            buffer.pop();
+                            buffer.pop();
+                        } else if buffer.ends_with(&[b'\n']) || buffer.ends_with(&[b'\0']) {
+                            buffer.pop();
+                        }
+
+                        let line = String::from_utf8_lossy(&buffer).to_string();
+
+                        let raw_item = DefaultSkimItem::new(
+                            line,
+                            option.use_ansi_color,
+                            &option.transform_fields,
+                            &option.matching_fields,
+                            &option.delimiter,
+                        );
+
+                        match tx_item.send(Arc::new(raw_item)) {
+                            Ok(_) => {}
+                            Err(_) => {
+                                debug!("collector: failed to send item, quit");
+                                break;
+                            }
+                        }
+                    }
+                    Err(_err) => {} // String not UTF8 or other error, skip.
+                }
+            }
+
+            let _ = tx_interrupt_clone.send(1); // ensure the waiting thread will exit
+            components_to_stop.fetch_sub(1, Ordering::SeqCst);
+            debug!("collector: command collector stop");
+        });
+
+        while !started.load(Ordering::SeqCst) {
+            // busy waiting for the thread to start. (components_to_stop is added)
+        }
+
+        (rx_item, tx_interrupt)
+    }
+}
+
+impl CommandCollector for DefaultSkimCollector {
+    fn invoke(&mut self, cmd: &str, components_to_stop: Arc<AtomicUsize>) -> (SkimItemReceiver, Sender<i32>) {
+        self.read_and_collect_from_command(components_to_stop, CollectorInput::Command(cmd.to_string()))
+    }
+}
+
+type CommandOutput = (Option<Child>, Box<dyn BufRead + Send>);
+fn get_command_output(cmd: &str) -> Result<CommandOutput, Box<dyn Error>> {
+    let shell = env::var("SHELL").unwrap_or_else(|_| "sh".to_string());
+    let mut command: Child = Command::new(shell)
+        .arg("-c")
+        .arg(cmd)
+        .stdout(Stdio::piped())
+        .stderr(Stdio::null())
+        .spawn()?;
+
+    let stdout = command
+        .stdout
+        .take()
+        .ok_or_else(|| "command output: unwrap failed".to_owned())?;
+
+    Ok((Some(command), Box::new(BufReader::new(stdout))))
+}

src/helper/mod.rs

@@ -0,0 +1,3 @@
+pub mod item;
+pub mod item_collector;
+pub mod string_reader;

src/helper/string_reader.rs

@@ -0,0 +1,71 @@
+use crate::prelude::bounded;
+use crate::{SkimItemReceiver, SkimItemSender};
+/// helper for turn a BufRead into a skim stream
+use std::io::BufRead;
+use std::sync::Arc;
+use std::thread;
+
+const ITEM_CHANNEL_SIZE: usize = 10240;
+
+pub struct SkimItemReader {
+    buf_size: usize,
+    line_ending: u8,
+}
+
+impl Default for SkimItemReader {
+    fn default() -> Self {
+        Self {
+            buf_size: ITEM_CHANNEL_SIZE,
+            line_ending: b'\n',
+        }
+    }
+}
+
+impl SkimItemReader {
+    pub fn buf_size(mut self, buf_size: usize) -> Self {
+        self.buf_size = buf_size;
+        self
+    }
+
+    pub fn line_ending(mut self, line_ending: u8) -> Self {
+        self.line_ending = line_ending;
+        self
+    }
+}
+
+impl SkimItemReader {
+    /// helper: convert bufread into SkimItemReceiver
+    pub fn of_bufread(&self, mut source: impl BufRead + Send + 'static) -> SkimItemReceiver {
+        let (tx_item, rx_item): (SkimItemSender, SkimItemReceiver) = bounded(self.buf_size);
+        let line_ending = self.line_ending;
+        thread::spawn(move || {
+            let mut buffer = Vec::with_capacity(1024);
+            loop {
+                buffer.clear();
+                // start reading
+                match source.read_until(line_ending, &mut buffer) {
+                    Ok(n) => {
+                        if n == 0 {
+                            break;
+                        }
+
+                        if buffer.ends_with(&[b'\r', b'\n']) {
+                            buffer.pop();
+                            buffer.pop();
+                        } else if buffer.ends_with(&[b'\n']) || buffer.ends_with(&[b'\0']) {
+                            buffer.pop();
+                        }
+
+                        let string = String::from_utf8_lossy(&buffer);
+                        let result = tx_item.send(Arc::new(string.into_owned()));
+                        if result.is_err() {
+                            break;
+                        }
+                    }
+                    Err(_err) => {} // String not UTF8 or other error, skip.
+                }
+            }
+        });
+        rx_item
+    }
+}

src/item.rs

@@ -1,126 +1,14 @@
 ///! An item is line of text that read from `find` command or stdin together with
 ///! the internal states, such as selected or not
-use std::borrow::Cow;
 use std::cmp::min;
 use std::default::Default;
 use std::ops::Deref;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
 
-use regex::Regex;
-
-use crate::ansi::{ANSIParser, AnsiString};
-use crate::field::{parse_matching_fields, parse_transform_fields, FieldRange};
 use crate::spinlock::{SpinLock, SpinLockGuard};
 use crate::SkimItem;
 
-//------------------------------------------------------------------------------
-/// An item will store everything that one line input will need to be operated and displayed.
-///
-/// What's special about an item?
-/// The simplest version of an item is a line of string, but things are getting more complex:
-/// - The conversion of lower/upper case is slow in rust, because it involds unicode.
-/// - We may need to interpret the ANSI codes in the text.
-/// - The text can be transformed and limited while searching.
-///
-/// About the ANSI, we made assumption that it is linewise, that means no ANSI codes will affect
-/// more than one line.
-#[derive(Debug)]
-pub struct DefaultSkimItem {
-    /// The text that will be output when user press `enter`
-    /// `Some(..)` => the original input is transformed, could not output `text` directly
-    /// `None` => that it is safe to output `text` directly
-    orig_text: Option<String>,
-
-    /// The text that will be shown on screen and matched.
-    text: AnsiString<'static>,
-
-    // Option<Box<_>> to reduce memory use in normal cases where no matching ranges are specified.
-    #[allow(clippy::box_vec)]
-    matching_ranges: Option<Box<Vec<(usize, usize)>>>,
-}
-
-impl<'a> DefaultSkimItem {
-    pub fn new(
-        orig_text: String,
-        ansi_enabled: bool,
-        trans_fields: &[FieldRange],
-        matching_fields: &[FieldRange],
-        delimiter: &Regex,
-    ) -> Self {
-        let using_transform_fields = !trans_fields.is_empty();
-
-        //        transformed | ANSI             | output
-        //------------------------------------------------------
-        //                    +- T -> trans+ANSI | ANSI
-        //                    |                  |
-        //      +- T -> trans +- F -> trans      | orig
-        // orig |                                |
-        //      +- F -> orig  +- T -> ANSI     ==| ANSI
-        //                    |                  |
-        //                    +- F -> orig       | orig
-
-        let mut ansi_parser: ANSIParser = Default::default();
-
-        let (orig_text, text) = if using_transform_fields && ansi_enabled {
-            // ansi and transform
-            let transformed = ansi_parser.parse_ansi(&parse_transform_fields(delimiter, &orig_text, trans_fields));
-            (Some(orig_text), transformed)
-        } else if using_transform_fields {
-            // transformed, not ansi
-            let transformed = parse_transform_fields(delimiter, &orig_text, trans_fields).into();
-            (Some(orig_text), transformed)
-        } else if ansi_enabled {
-            // not transformed, ansi
-            (None, ansi_parser.parse_ansi(&orig_text))
-        } else {
-            // normal case
-            (None, orig_text.into())
-        };
-
-        let matching_ranges = if !matching_fields.is_empty() {
-            Some(Box::new(parse_matching_fields(
-                delimiter,
-                text.stripped(),
-                matching_fields,
-            )))
-        } else {
-            None
-        };
-
-        DefaultSkimItem {
-            orig_text,
-            text,
-            matching_ranges,
-        }
-    }
-}
-
-impl SkimItem for DefaultSkimItem {
-    #[inline]
-    fn text(&self) -> Cow<str> {
-        Cow::Borrowed(self.text.stripped())
-    }
-
-    fn output(&self) -> Cow<str> {
-        if self.orig_text.is_some() {
-            if self.text.has_attrs() {
-                let mut ansi_parser: ANSIParser = Default::default();
-                let text = ansi_parser.parse_ansi(self.orig_text.as_ref().unwrap());
-                text.into_inner()
-            } else {
-                Cow::Borrowed(self.orig_text.as_ref().unwrap())
-            }
-        } else {
-            Cow::Borrowed(self.text.stripped())
-        }
-    }
-
-    fn get_matching_ranges(&self) -> Option<&[(usize, usize)]> {
-        self.matching_ranges.as_ref().map(|vec| vec as &[(usize, usize)])
-    }
-}
-
 //------------------------------------------------------------------------------
 pub type ItemIndex = (u32, u32);
 

src/item_collector.rs

@@ -1,268 +0,0 @@
-use crate::field::FieldRange;
-use crate::item::DefaultSkimItem;
-use crate::{SkimItem, SkimItemReceiver, SkimItemSender, SkimOptions};
-use crossbeam::channel::{bounded, Receiver, Sender};
-use regex::Regex;
-use std::env;
-use std::error::Error;
-use std::io::{BufRead, BufReader};
-use std::process::{Child, Command, Stdio};
-use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
-use std::sync::Arc;
-use std::thread;
-
-const CMD_CHANNEL_SIZE: usize = 1024;
-const ITEM_CHANNEL_SIZE: usize = 10240;
-const DELIMITER_STR: &str = r"[\t\n ]+";
-const READ_BUFFER_SIZE: usize = 1024;
-
-#[derive(Clone, Debug)]
-pub struct CollectorOption {
-    pub use_ansi_color: bool,
-    pub default_arg: String,
-    pub transform_fields: Vec<FieldRange>,
-    pub matching_fields: Vec<FieldRange>,
-    pub delimiter: Regex,
-    pub replace_str: String,
-    pub line_ending: u8,
-}
-
-impl Default for CollectorOption {
-    fn default() -> Self {
-        Self {
-            use_ansi_color: false,
-            default_arg: String::new(),
-            transform_fields: Vec::new(),
-            matching_fields: Vec::new(),
-            delimiter: Regex::new(DELIMITER_STR).unwrap(),
-            replace_str: "{}".to_string(),
-            line_ending: b'\n',
-        }
-    }
-}
-
-impl CollectorOption {
-    pub fn with_options(options: &SkimOptions) -> Self {
-        let mut reader_option = Self::default();
-        reader_option.parse_options(&options);
-        reader_option
-    }
-
-    fn parse_options(&mut self, options: &SkimOptions) {
-        if options.ansi {
-            self.use_ansi_color = true;
-        }
-
-        if let Some(delimiter) = options.delimiter {
-            self.delimiter = Regex::new(delimiter).unwrap_or_else(|_| Regex::new(DELIMITER_STR).unwrap());
-        }
-
-        if let Some(transform_fields) = options.with_nth {
-            self.transform_fields = transform_fields
-                .split(',')
-                .filter_map(|string| FieldRange::from_str(string))
-                .collect();
-        }
-
-        if let Some(matching_fields) = options.nth {
-            self.matching_fields = matching_fields
-                .split(',')
-                .filter_map(|string| FieldRange::from_str(string))
-                .collect();
-        }
-
-        if options.read0 {
-            self.line_ending = b'\0';
-        }
-    }
-}
-
-pub enum CollectorInput {
-    Pipe(Box<dyn BufRead + Send>),
-    Command(String),
-}
-
-/// components_to_stop == 0 => all the threads have been stopped
-/// return (channel_for_receive_item, channel_to_stop_command)
-pub fn read_and_collect_from_command(
-    components_to_stop: Arc<AtomicUsize>,
-    input: CollectorInput,
-    option: CollectorOption,
-) -> (Receiver<Arc<dyn SkimItem>>, Sender<i32>) {
-    let (command, mut source) = match input {
-        CollectorInput::Pipe(pipe) => (None, pipe),
-        CollectorInput::Command(cmd) => get_command_output(&cmd).expect("command not found"),
-    };
-
-    let (tx_interrupt, rx_interrupt) = bounded(CMD_CHANNEL_SIZE);
-    let (tx_item, rx_item): (SkimItemSender, SkimItemReceiver) = bounded(ITEM_CHANNEL_SIZE);
-
-    let started = Arc::new(AtomicBool::new(false));
-    let started_clone = started.clone();
-    let components_to_stop_clone = components_to_stop.clone();
-    // listening to close signal and kill command if needed
-    thread::spawn(move || {
-        debug!("collector: command killer start");
-        components_to_stop_clone.fetch_add(1, Ordering::SeqCst);
-        started_clone.store(true, Ordering::SeqCst); // notify parent that it is started
-
-        let _ = rx_interrupt.recv(); // block waiting
-                                     // clean up resources
-        if let Some(mut x) = command {
-            let _ = x.kill();
-            let _ = x.wait();
-        }
-
-        components_to_stop_clone.fetch_sub(1, Ordering::SeqCst);
-        debug!("collector: command killer stop");
-    });
-
-    while !started.load(Ordering::SeqCst) {
-        // busy waiting for the thread to start. (components_to_stop is added)
-    }
-
-    let started = Arc::new(AtomicBool::new(false));
-    let started_clone = started.clone();
-    let tx_interrupt_clone = tx_interrupt.clone();
-    thread::spawn(move || {
-        debug!("collector: command collector start");
-        components_to_stop.fetch_add(1, Ordering::SeqCst);
-        started_clone.store(true, Ordering::SeqCst); // notify parent that it is started
-
-        let opt = option;
-        // set the proper run number
-
-        let mut buffer = Vec::with_capacity(READ_BUFFER_SIZE);
-        loop {
-            buffer.clear();
-
-            // start reading
-            match source.read_until(opt.line_ending, &mut buffer) {
-                Ok(n) => {
-                    if n == 0 {
-                        break;
-                    }
-
-                    if buffer.ends_with(&[b'\r', b'\n']) {
-                        buffer.pop();
-                        buffer.pop();
-                    } else if buffer.ends_with(&[b'\n']) || buffer.ends_with(&[b'\0']) {
-                        buffer.pop();
-                    }
-
-                    let line = String::from_utf8_lossy(&buffer).to_string();
-
-                    let raw_item = DefaultSkimItem::new(
-                        line,
-                        opt.use_ansi_color,
-                        &opt.transform_fields,
-                        &opt.matching_fields,
-                        &opt.delimiter,
-                    );
-
-                    match tx_item.send(Arc::new(raw_item)) {
-                        Ok(_) => {}
-                        Err(_) => {
-                            debug!("collector: failed to send item, quit");
-                            break;
-                        }
-                    }
-                }
-                Err(_err) => {} // String not UTF8 or other error, skip.
-            }
-        }
-
-        let _ = tx_interrupt_clone.send(1); // ensure the waiting thread will exit
-        components_to_stop.fetch_sub(1, Ordering::SeqCst);
-        debug!("collector: command collector stop");
-    });
-
-    while !started.load(Ordering::SeqCst) {
-        // busy waiting for the thread to start. (components_to_stop is added)
-    }
-
-    (rx_item, tx_interrupt)
-}
-
-type CommandOutput = (Option<Child>, Box<dyn BufRead + Send>);
-fn get_command_output(cmd: &str) -> Result<CommandOutput, Box<dyn Error>> {
-    let shell = env::var("SHELL").unwrap_or_else(|_| "sh".to_string());
-    let mut command: Child = Command::new(shell)
-        .arg("-c")
-        .arg(cmd)
-        .stdout(Stdio::piped())
-        .stderr(Stdio::null())
-        .spawn()?;
-
-    let stdout = command
-        .stdout
-        .take()
-        .ok_or_else(|| "command output: unwrap failed".to_owned())?;
-
-    Ok((Some(command), Box::new(BufReader::new(stdout))))
-}
-
-//------------------------------------------------------------------------------
-// helper
-pub struct SkimItemReader {
-    buf_size: usize,
-    line_ending: u8,
-}
-
-impl Default for SkimItemReader {
-    fn default() -> Self {
-        Self {
-            buf_size: ITEM_CHANNEL_SIZE,
-            line_ending: b'\n',
-        }
-    }
-}
-
-impl SkimItemReader {
-    pub fn buf_size(mut self, buf_size: usize) -> Self {
-        self.buf_size = buf_size;
-        self
-    }
-
-    pub fn line_ending(mut self, line_ending: u8) -> Self {
-        self.line_ending = line_ending;
-        self
-    }
-}
-
-impl SkimItemReader {
-    /// helper: convert bufread into SkimItemReceiver
-    pub fn of_bufread(&self, mut source: impl BufRead + Send + 'static) -> SkimItemReceiver {
-        let (tx_item, rx_item): (SkimItemSender, SkimItemReceiver) = bounded(self.buf_size);
-        let line_ending = self.line_ending;
-        thread::spawn(move || {
-            let mut buffer = Vec::with_capacity(1024);
-            loop {
-                buffer.clear();
-                // start reading
-                match source.read_until(line_ending, &mut buffer) {
-                    Ok(n) => {
-                        if n == 0 {
-                            break;
-                        }
-
-                        if buffer.ends_with(&[b'\r', b'\n']) {
-                            buffer.pop();
-                            buffer.pop();
-                        } else if buffer.ends_with(&[b'\n']) || buffer.ends_with(&[b'\0']) {
-                            buffer.pop();
-                        }
-
-                        let string = String::from_utf8_lossy(&buffer);
-                        let result = tx_item.send(Arc::new(string.into_owned()));
-                        if result.is_err() {
-                            break;
-                        }
-                    }
-                    Err(_err) => {} // String not UTF8 or other error, skip.
-                }
-            }
-        });
-        rx_item
-    }
-}

src/lib.rs

@@ -25,12 +25,13 @@ use crate::reader::Reader;
 mod ansi;
 mod engine;
 mod event;
-mod field;
+/// provide default implementation
+pub mod field;
 mod global;
 mod header;
+mod helper;
 mod input;
 mod item;
-mod item_collector;
 mod matcher;
 mod model;
 mod options;

src/options.rs

@@ -2,7 +2,10 @@ use std::rc::Rc;
 
 use derive_builder::Builder;
 
+use crate::helper::item_collector::DefaultSkimCollector;
+use crate::reader::CommandCollector;
 use crate::{CaseMatching, FuzzyAlgorithm, MatchEngineFactory};
+use std::cell::RefCell;
 
 #[derive(Builder)]
 #[builder(build_fn(name = "final_build"))]
@@ -16,7 +19,6 @@ pub struct SkimOptions<'a> {
     pub tac: bool,
     pub nosort: bool,
     pub tiebreak: Option<String>,
-    pub ansi: bool,
     pub exact: bool,
     pub cmd: Option<&'a str>,
     pub interactive: bool,
@@ -24,8 +26,6 @@ pub struct SkimOptions<'a> {
     pub cmd_query: Option<&'a str>,
     pub regex: bool,
     pub delimiter: Option<&'a str>,
-    pub nth: Option<&'a str>,
-    pub with_nth: Option<&'a str>,
     pub replstr: Option<&'a str>,
     pub color: Option<&'a str>,
     pub margin: Option<&'a str>,
@@ -35,24 +35,19 @@ pub struct SkimOptions<'a> {
     pub preview: Option<&'a str>,
     pub preview_window: Option<&'a str>,
     pub reverse: bool,
-    pub read0: bool,
-    pub print0: bool,
     pub tabstop: Option<&'a str>,
-    pub print_query: bool,
-    pub print_cmd: bool,
-    pub print_score: bool,
     pub no_hscroll: bool,
     pub no_mouse: bool,
     pub inline_info: bool,
     pub header: Option<&'a str>,
     pub header_lines: usize,
     pub layout: &'a str,
-    pub filter: &'a str,
     pub algorithm: FuzzyAlgorithm,
     pub case: CaseMatching,
     pub engine_factory: Option<Rc<dyn MatchEngineFactory>>,
     pub query_history: &'a [String],
     pub cmd_history: &'a [String],
+    pub cmd_collector: Rc<RefCell<dyn CommandCollector>>,
 }
 
 impl<'a> Default for SkimOptions<'a> {
@@ -66,7 +61,6 @@ impl<'a> Default for SkimOptions<'a> {
             tac: false,
             nosort: false,
             tiebreak: None,
-            ansi: false,
             exact: false,
             cmd: None,
             interactive: false,
@@ -74,8 +68,6 @@ impl<'a> Default for SkimOptions<'a> {
             cmd_query: None,
             regex: false,
             delimiter: None,
-            nth: None,
-            with_nth: None,
             replstr: Some("{}"),
             color: None,
             margin: Some("0,0,0,0"),
@@ -85,24 +77,19 @@ impl<'a> Default for SkimOptions<'a> {
             preview: None,
             preview_window: Some("right:50%"),
             reverse: false,
-            read0: false,
-            print0: false,
             tabstop: None,
-            print_query: false,
-            print_cmd: false,
-            print_score: false,
             no_hscroll: false,
             no_mouse: false,
             inline_info: false,
             header: None,
             header_lines: 0,
             layout: "",
-            filter: "",
             algorithm: FuzzyAlgorithm::default(),
             case: CaseMatching::default(),
             engine_factory: None,
             query_history: &[],
             cmd_history: &[],
+            cmd_collector: Rc::new(RefCell::new(DefaultSkimCollector::new(Default::default()))),
         }
     }
 }

src/prelude.rs

@@ -1,10 +1,13 @@
 pub use crate::ansi::AnsiString;
 pub use crate::engine::{factory::*, fuzzy::FuzzyAlgorithm};
-pub use crate::item_collector::{read_and_collect_from_command, CollectorInput, CollectorOption, SkimItemReader};
+pub use crate::helper::item_collector::{CollectorInput, CollectorOption, DefaultSkimCollector};
+pub use crate::helper::string_reader::SkimItemReader;
 pub use crate::options::{SkimOptions, SkimOptionsBuilder};
 pub use crate::output::SkimOutput;
 pub use crate::*;
 pub use crossbeam::channel::{bounded, unbounded, Receiver, Sender};
 pub use std::borrow::Cow;
+pub use std::cell::RefCell;
+pub use std::rc::Rc;
 pub use std::sync::atomic::{AtomicUsize, Ordering};
 pub use std::sync::Arc;

src/reader.rs

@@ -2,17 +2,29 @@ use crate::global::mark_new_run;
 ///! Reader is used for reading items from datasource (e.g. stdin or command output)
 ///!
 ///! After reading in a line, reader will save an item into the pool(items)
-use crate::item_collector::{read_and_collect_from_command, CollectorInput, CollectorOption};
 use crate::options::SkimOptions;
 use crate::spinlock::SpinLock;
 use crate::{SkimItem, SkimItemReceiver};
 use crossbeam::channel::{bounded, select, Sender};
+use std::cell::RefCell;
+use std::rc::Rc;
 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::sync::Arc;
 use std::thread;
 
 const CHANNEL_SIZE: usize = 1024;
 
+pub trait CommandCollector {
+    /// execute the `cmd` and produce a
+    /// - skim item producer
+    /// - a channel sender, any message send would mean to terminate the `cmd` process (for now).
+    ///
+    /// Internally, the command collector may start several threads(components), the collector
+    /// should add `1` on every thread creation and sub `1` on thread termination. reader would use
+    /// this information to determine whether the collector had stopped or not.
+    fn invoke(&mut self, cmd: &str, components_to_stop: Arc<AtomicUsize>) -> (SkimItemReceiver, Sender<i32>);
+}
+
 pub struct ReaderControl {
     tx_interrupt: Sender<i32>,
     tx_interrupt_cmd: Option<Sender<i32>>,
@@ -46,14 +58,14 @@ impl ReaderControl {
 }
 
 pub struct Reader {
-    option: CollectorOption,
+    cmd_collector: Rc<RefCell<dyn CommandCollector>>,
     rx_item: Option<SkimItemReceiver>,
 }
 
 impl Reader {
     pub fn with_options(options: &SkimOptions) -> Self {
         Self {
-            option: CollectorOption::with_options(&options),
+            cmd_collector: options.cmd_collector.clone(),
             rx_item: None,
         }
     }
@@ -69,13 +81,10 @@ impl Reader {
         let components_to_stop: Arc<AtomicUsize> = Arc::new(AtomicUsize::new(0));
         let items = Arc::new(SpinLock::new(Vec::new()));
         let items_clone = items.clone();
-        let option_clone = self.option.clone();
-        let cmd = cmd.to_string();
 
         let (rx_item, tx_interrupt_cmd) = self.rx_item.take().map(|rx| (rx, None)).unwrap_or_else(|| {
             let components_to_stop_clone = components_to_stop.clone();
-            let (rx_item, tx_interrupt_cmd) =
+            let (rx_item, tx_interrupt_cmd) = self.cmd_collector.borrow_mut().invoke(cmd, components_to_stop_clone);
-                read_and_collect_from_command(components_to_stop_clone, CollectorInput::Command(cmd), option_clone);
             (rx_item, Some(tx_interrupt_cmd))
         });