diff --git a/src/bin/main.rs b/src/bin/main.rs index 121b9246..14ec09d3 100644 --- a/src/bin/main.rs +++ b/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 { 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 { 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, _) = read_and_collect_from_command(components_to_stop, CollectorInput::Pipe(Box::new(BufReader::new(stdin))), collector_option); + let (rx_item, _) = cmd_collector.borrow().read_and_collect_from_command(components_to_stop, CollectorInput::Pipe(Box::new(BufReader::new(stdin)))); Some(rx_item) } Ok(true) | Err(_) => None, @@ -264,12 +283,10 @@ fn real_main() -> Result { //------------------------------------------------------------------------------ // 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 { let output = output.unwrap(); // output query - if options.print_query { - write!(stdout, "{}{}", output.query, output_ending)?; + if bin_options.print_query { + write!(stdout, "{}{}", output.query, bin_options.output_ending)?; } - if options.print_cmd { - write!(stdout, "{}{}", output.cmd, output_ending)?; + if bin_options.print_cmd { + 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) -> Result { +#[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, +) -> Result { 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 { - write!(stdout, "{}{}", query, output_ending)?; + if bin_option.print_query { + write!(stdout, "{}{}", query, bin_option.output_ending)?; } - if options.print_cmd { - write!(stdout, "{}{}", cmd, output_ending)?; + if bin_option.print_cmd { + write!(stdout, "{}{}", cmd, bin_option.output_ending)?; } //------------------------------------------------------------------------------ @@ -454,11 +472,10 @@ pub fn filter(options: &SkimOptions, source: Option) -> 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 (ret, _control) = read_and_collect_from_command(components_to_stop, collector_input, collector_option); + let cmd_collector = options.cmd_collector.clone(); + let (ret, _control) = cmd_collector.borrow_mut().invoke(cmd, components_to_stop); ret }); @@ -468,7 +485,7 @@ pub fn filter(options: &SkimOptions, source: Option) -> 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 }) diff --git a/src/helper/item.rs b/src/helper/item.rs new file mode 100644 index 00000000..2a02fc98 --- /dev/null +++ b/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, + + /// The text that will be shown on screen and matched. + text: AnsiString<'static>, + + // Option> to reduce memory use in normal cases where no matching ranges are specified. + #[allow(clippy::box_vec)] + matching_ranges: Option>>, +} + +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 { + Cow::Borrowed(self.text.stripped()) + } + + fn output(&self) -> Cow { + 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)]) + } +} diff --git a/src/helper/item_collector.rs b/src/helper/item_collector.rs new file mode 100644 index 00000000..347bdcea --- /dev/null +++ b/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, + matching_fields: Vec, + 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) -> 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) -> 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), + Command(String), +} + +pub struct DefaultSkimCollector { + option: Arc, +} + +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, + input: CollectorInput, + ) -> (Receiver>, Sender) { + 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) -> (SkimItemReceiver, Sender) { + self.read_and_collect_from_command(components_to_stop, CollectorInput::Command(cmd.to_string())) + } +} + +type CommandOutput = (Option, Box); +fn get_command_output(cmd: &str) -> Result> { + 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)))) +} diff --git a/src/helper/mod.rs b/src/helper/mod.rs new file mode 100644 index 00000000..d736e3fc --- /dev/null +++ b/src/helper/mod.rs @@ -0,0 +1,3 @@ +pub mod item; +pub mod item_collector; +pub mod string_reader; diff --git a/src/helper/string_reader.rs b/src/helper/string_reader.rs new file mode 100644 index 00000000..cca13816 --- /dev/null +++ b/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 + } +} diff --git a/src/item.rs b/src/item.rs index bc1ccb52..5eed2238 100644 --- a/src/item.rs +++ b/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, - - /// The text that will be shown on screen and matched. - text: AnsiString<'static>, - - // Option> to reduce memory use in normal cases where no matching ranges are specified. - #[allow(clippy::box_vec)] - matching_ranges: Option>>, -} - -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 { - Cow::Borrowed(self.text.stripped()) - } - - fn output(&self) -> Cow { - 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); diff --git a/src/item_collector.rs b/src/item_collector.rs deleted file mode 100644 index 3ed3673a..00000000 --- a/src/item_collector.rs +++ /dev/null @@ -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, - pub matching_fields: Vec, - 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), - 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, - input: CollectorInput, - option: CollectorOption, -) -> (Receiver>, Sender) { - 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, Box); -fn get_command_output(cmd: &str) -> Result> { - 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 - } -} diff --git a/src/lib.rs b/src/lib.rs index 4716a410..f5092a01 100644 --- a/src/lib.rs +++ b/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; diff --git a/src/options.rs b/src/options.rs index 603636ad..09f948e5 100644 --- a/src/options.rs +++ b/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, - 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>, pub query_history: &'a [String], pub cmd_history: &'a [String], + pub cmd_collector: Rc>, } 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()))), } } } diff --git a/src/prelude.rs b/src/prelude.rs index 75eeb76a..5efe7ea1 100644 --- a/src/prelude.rs +++ b/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; diff --git a/src/reader.rs b/src/reader.rs index 1ea19da0..b5978ab2 100644 --- a/src/reader.rs +++ b/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) -> (SkimItemReceiver, Sender); +} + pub struct ReaderControl { tx_interrupt: Sender, tx_interrupt_cmd: Option>, @@ -46,14 +58,14 @@ impl ReaderControl { } pub struct Reader { - option: CollectorOption, + cmd_collector: Rc>, rx_item: Option, } 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 = 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) = - read_and_collect_from_command(components_to_stop_clone, CollectorInput::Command(cmd), option_clone); + let (rx_item, tx_interrupt_cmd) = self.cmd_collector.borrow_mut().invoke(cmd, components_to_stop_clone); (rx_item, Some(tx_interrupt_cmd)) });