asp/keep
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor: simplify filter plugin interface to use &mut dyn Read/Write

Browse Source
This commit is contained in:
Andrew Phillips
2025-09-15 17:42:35 -03:00
committed by Andrew Phillips (aider)
parent a8871a9575
commit a72395fe83
17 changed files with 102 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
src/filter_plugin/exec.rs
View File

@@ -93,7 +93,7 @@ impl FilterPlugin for ExecFilter {
/// // In filter context:
/// filter.filter(Box::new(&mut input), Box::new(&mut output)).unwrap();
/// ```
fn filter(&mut self, mut reader: Box<&mut dyn Read>, mut writer: Box<&mut dyn Write>) -> Result<()> {
fn filter(&mut self, reader: &mut dyn Read, writer: &mut dyn Write) -> Result<()> {
if !self.supported {
return Err(std::io::Error::new(
std::io::ErrorKind::NotFound,
@@ -103,6 +103,10 @@ impl FilterPlugin for ExecFilter {
debug!("FILTER_EXEC: Executing command: {} {:?}", self.program, self.args);
// Read all input first
let mut input_data = Vec::new();
std::io::copy(reader, &mut input_data)?;
let mut child = Command::new(&self.program)
.args(&self.args)
.stdin(Stdio::piped())
@@ -116,86 +120,45 @@ impl FilterPlugin for ExecFilter {
)
})?;
let stdin = child.stdin.take().ok_or_else(|| {
let mut stdin = child.stdin.take().ok_or_else(|| {
std::io::Error::new(
std::io::ErrorKind::Other,
"Failed to capture stdin from child process",
)
})?;
let stdout = child.stdout.take().ok_or_else(|| {
// Write input to child stdin
stdin.write_all(&input_data)?;
drop(stdin); // Close stdin to signal EOF
let mut stdout = child.stdout.take().ok_or_else(|| {
std::io::Error::new(
std::io::ErrorKind::Other,
"Failed to capture stdout from child process",
)
})?;
self.stdin_writer = Some(stdin);
self.stdout_reader = Some(stdout);
self.child_process = Some(child);
let mut stdin_writer = self.stdin_writer.as_mut().unwrap();
let mut stdout_reader = self.stdout_reader.as_mut().unwrap();
// Thread to copy from input reader to child stdin
let input_thread = std::thread::spawn(move || {
std::io::copy(&mut *reader, &mut *stdin_writer)
.map_err(|e| {
std::io::Error::new(
std::io::ErrorKind::Other,
format!("Failed to write to process stdin: {}", e),
)
})?;
// Close stdin to signal EOF to the child process
drop(stdin_writer);
Ok(())
});
// Thread to copy from child stdout to output writer
let output_thread = std::thread::spawn(move || {
std::io::copy(&mut *stdout_reader, &mut *writer)
.map_err(|e| {
std::io::Error::new(
std::io::ErrorKind::Other,
format!("Failed to read from process stdout: {}", e),
)
})?;
Ok(())
});
// Wait for both threads to complete
input_thread.join().map_err(|e| {
std::io::Error::new(
std::io::ErrorKind::Other,
format!("Input thread panicked: {:?}", e),
)
})?;
output_thread.join().map_err(|e| {
std::io::Error::new(
std::io::ErrorKind::Other,
format!("Output thread panicked: {:?}", e),
)
})?;
// Copy stdout to writer
std::io::copy(&mut stdout, writer)?;
// Wait for the child process to finish
if let Some(mut child) = self.child_process.take() {
let output = child.wait_with_output()
.map_err(|e| {
std::io::Error::new(
std::io::ErrorKind::Other,
format!("Failed to wait on child process: {}", e),
)
})?;
if !output.status.success() {
let stderr = String::from_utf8_lossy(&output.stderr);
if !stderr.is_empty() {
warn!("FILTER_EXEC: Process stderr: {}", stderr);
}
return Err(std::io::Error::new(
let output = child.wait_with_output()
.map_err(|e| {
std::io::Error::new(
std::io::ErrorKind::Other,
format!("Process exited with error: {:?}", output.status),
));
format!("Failed to wait on child process: {}", e),
)
})?;
if !output.status.success() {
let stderr = String::from_utf8_lossy(&output.stderr);
if !stderr.is_empty() {
warn!("FILTER_EXEC: Process stderr: {}", stderr);
}
return Err(std::io::Error::new(
std::io::ErrorKind::Other,
format!("Process exited with error: {:?}", output.status),
));
}
debug!("FILTER_EXEC: Process completed successfully");

6
src/filter_plugin/grep.rs
View File

@@ -70,12 +70,12 @@ impl GrepFilter {
/// filter.filter(&mut input, &mut output)?;
/// ```
impl FilterPlugin for GrepFilter {
fn filter(&mut self, mut reader: Box<dyn Read>, mut writer: Box<dyn Write>) -> Result<()> {
let mut buf_reader = std::io::BufReader::new(reader.as_mut());
fn filter(&mut self, reader: &mut dyn Read, writer: &mut dyn Write) -> Result<()> {
let mut buf_reader = std::io::BufReader::new(reader);
for line in buf_reader.by_ref().lines() {
let line = line?;
if self.regex.is_match(&line) {
writeln!(writer.as_mut(), "{}", line)?;
writeln!(writer, "{}", line)?;
}
}
Ok(())

45
src/filter_plugin/head.rs
View File

@@ -72,7 +72,7 @@ impl HeadBytesFilter {
/// // Input "Hello World" becomes "Hello"
/// ```
impl FilterPlugin for HeadBytesFilter {
fn filter(&mut self, mut reader: Box<dyn Read>, mut writer: Box<dyn Write>) -> Result<()> {
fn filter(&mut self, reader: &mut dyn Read, writer: &mut dyn Write) -> Result<()> {
if self.remaining == 0 {
return Ok(());
}
@@ -80,11 +80,11 @@ impl FilterPlugin for HeadBytesFilter {
let mut buffer = vec![0; PIPESIZE];
while self.remaining > 0 {
let to_read = std::cmp::min(self.remaining, PIPESIZE);
let bytes_read = reader.as_mut().read(&mut buffer[..to_read])?;
let bytes_read = reader.read(&mut buffer[..to_read])?;
if bytes_read == 0 {
break;
}
writer.as_mut().write_all(&buffer[..bytes_read])?;
writer.write_all(&buffer[..bytes_read])?;
self.remaining -= bytes_read;
}
Ok(())
@@ -159,38 +159,39 @@ impl HeadLinesFilter {
}
/// Filters input by reading only the first N lines and writing them to the output.
///
///
/// Uses buffered line reading to process input line-by-line until the limit or EOF.
///
///
/// # Arguments
///
///
/// * `reader` - Mutable reference to the input data stream.
/// * `writer` - Mutable reference to the output stream.
///
///
/// # Returns
///
///
/// * `Result<()>` - Success if filtering completes, or I/O error.
///
///
/// # Errors
///
///
/// * `io::Error` from line reading or writing operations.
///
///
/// # Examples
///
///
/// ```
/// // Assuming a filter chain with head_lines(2)
/// // Input "Line1\nLine2\nLine3" becomes "Line1\nLine2\n"
/// // Input: "Line1\nLine2\nLine3" becomes "Line1\nLine2\n"
/// ```
impl FilterPlugin for HeadLinesFilter {
fn filter(&mut self, mut reader: Box<dyn Read>, mut writer: Box<dyn Write>) -> Result<()> {
fn filter(&mut self, reader: &mut dyn Read, writer: &mut dyn Write) -> Result<()> {
if self.remaining == 0 {
return Ok(());
}
let mut buf_reader = std::io::BufReader::new(reader.as_mut());
let mut buf_reader = std::io::BufReader::new(reader);
for line in buf_reader.by_ref().lines() {
let line = line?;
writeln!(writer.as_mut(), "{}", line)?;
writeln!(writer, "{}", line)?;
self.remaining -= 1;
if self.remaining == 0 {
break;
@@ -200,11 +201,11 @@ impl FilterPlugin for HeadLinesFilter {
}
/// Clones this filter into a new boxed instance.
///
///
/// Creates an independent copy with the same configuration.
///
///
/// # Returns
///
///
/// A new `Box<dyn FilterPlugin>` clone.
fn clone_box(&self) -> Box<dyn FilterPlugin> {
Box::new(Self {
@@ -213,11 +214,11 @@ impl FilterPlugin for HeadLinesFilter {
}
/// Returns the configuration options for this filter.
///
///
/// Defines the "count" parameter as required with no default.
///
///
/// # Returns
///
///
/// Vector of `FilterOption` describing parameters.
fn options(&self) -> Vec<FilterOption> {
vec![

18
src/filter_plugin/mod.rs
View File

@@ -101,7 +101,10 @@ pub trait FilterPlugin: Send {
/// // ... other methods
/// }
/// ```
fn filter(&mut self, mut reader: Box<dyn Read>, mut writer: Box<dyn Write>) -> Result<()>;
fn filter(&mut self, reader: &mut dyn Read, writer: &mut dyn Write) -> Result<()> {
let _ = std::io::copy(reader, writer)?;
Ok(())
}
/// Clones this plugin into a new boxed instance.
///
@@ -294,19 +297,16 @@ impl FilterChain {
for i in 0..plugins_len {
// Create a cursor for the current data
let mut cursor = std::io::Cursor::new(&current_data);
let input: Box<dyn Read> = Box::new(cursor);
let mut input = std::io::Cursor::new(std::mem::take(&mut current_data));
// For the last plugin, write directly to the output writer
if i == plugins_len - 1 {
let output: Box<dyn Write> = Box::new(writer);
self.plugins[i].filter(input, output)?;
self.plugins[i].filter(&mut input, writer)?;
} else {
// For intermediate plugins, write to a buffer
let mut output = Vec::new();
let output_ref: Box<dyn Write> = Box::new(&mut output);
self.plugins[i].filter(input, output_ref)?;
current_data = output;
let mut output_vec = Vec::new();
self.plugins[i].filter(&mut input, &mut output_vec)?;
current_data = output_vec;
}
}
Ok(())

12
src/filter_plugin/skip.rs
View File

@@ -32,13 +32,13 @@ impl FilterPlugin for SkipBytesFilter {
/// # Returns
///
/// Returns `Ok(())` on success, or an `io::Error` if reading or writing fails.
fn filter(&mut self, mut reader: Box<dyn Read>, mut writer: Box<dyn Write>) -> Result<()> {
fn filter(&mut self, reader: &mut dyn Read, writer: &mut dyn Write) -> Result<()> {
// Skip bytes in chunks
if self.remaining > 0 {
let mut buffer = vec![0; PIPESIZE];
while self.remaining > 0 {
let to_read = std::cmp::min(self.remaining, PIPESIZE);
let bytes_read = reader.as_mut().read(&mut buffer[..to_read])?;
let bytes_read = reader.read(&mut buffer[..to_read])?;
if bytes_read == 0 {
break;
}
@@ -47,7 +47,7 @@ impl FilterPlugin for SkipBytesFilter {
}
// Copy the remaining data using io::copy for efficiency
std::io::copy(reader.as_mut(), writer.as_mut())?;
std::io::copy(reader, writer)?;
Ok(())
}
@@ -107,14 +107,14 @@ impl FilterPlugin for SkipLinesFilter {
/// # Returns
///
/// Returns `Ok(())` on success, or an `io::Error` if reading or writing fails.
fn filter(&mut self, mut reader: Box<dyn Read>, mut writer: Box<dyn Write>) -> Result<()> {
let mut buf_reader = std::io::BufReader::new(reader.as_mut());
fn filter(&mut self, reader: &mut dyn Read, writer: &mut dyn Write) -> Result<()> {
let mut buf_reader = std::io::BufReader::new(reader);
for line in buf_reader.by_ref().lines() {
let line = line?;
if self.remaining > 0 {
self.remaining -= 1;
} else {
writeln!(writer.as_mut(), "{}", line)?;
writeln!(writer, "{}", line)?;
}
}
Ok(())

6
src/filter_plugin/strip_ansi.rs
View File

@@ -32,9 +32,9 @@ impl FilterPlugin for StripAnsiFilter {
/// # Returns
///
/// Returns `Ok(())` on success, or an `io::Error` if reading or writing fails.
fn filter(&mut self, mut reader: Box<dyn Read>, mut writer: Box<dyn Write>) -> Result<()> {
let mut ansi_writer = Writer::new(writer.as_mut());
std::io::copy(reader.as_mut(), &mut ansi_writer)?;
fn filter(&mut self, reader: &mut dyn Read, writer: &mut dyn Write) -> Result<()> {
let mut ansi_writer = Writer::new(writer);
std::io::copy(reader, &mut ansi_writer)?;
ansi_writer.flush()?;
Ok(())
}

12
src/filter_plugin/tail.rs
View File

@@ -35,10 +35,10 @@ impl FilterPlugin for TailBytesFilter {
/// # Returns
///
/// Returns `Ok(())` on success, or an `io::Error` if reading or writing fails.
fn filter(&mut self, mut reader: Box<dyn Read>, mut writer: Box<dyn Write>) -> Result<()> {
fn filter(&mut self, reader: &mut dyn Read, writer: &mut dyn Write) -> Result<()> {
let mut temp_buffer = vec![0; PIPESIZE];
loop {
let bytes_read = reader.as_mut().read(&mut temp_buffer)?;
let bytes_read = reader.read(&mut temp_buffer)?;
if bytes_read == 0 {
break;
}
@@ -54,7 +54,7 @@ impl FilterPlugin for TailBytesFilter {
// Write the buffered data at the end
let result: Vec<u8> = self.buffer.iter().cloned().collect();
writer.as_mut().write_all(&result)?;
writer.write_all(&result)?;
Ok(())
}
@@ -117,8 +117,8 @@ impl FilterPlugin for TailLinesFilter {
/// # Returns
///
/// Returns `Ok(())` on success, or an `io::Error` if reading or writing fails.
fn filter(&mut self, mut reader: Box<dyn Read>, mut writer: Box<dyn Write>) -> Result<()> {
let mut buf_reader = std::io::BufReader::new(reader.as_mut());
fn filter(&mut self, reader: &mut dyn Read, writer: &mut dyn Write) -> Result<()> {
let mut buf_reader = std::io::BufReader::new(reader);
for line in buf_reader.by_ref().lines() {
let line = line?;
if self.lines.len() == self.count {
@@ -129,7 +129,7 @@ impl FilterPlugin for TailLinesFilter {
// Write the buffered lines
for line in &self.lines {
writeln!(writer.as_mut(), "{}", line)?;
writeln!(writer, "{}", line)?;
}
Ok(())
}

2
src/meta_plugin/exec.rs
View File

@@ -127,7 +127,7 @@ impl MetaPluginExec {
.stderr(Stdio::piped());
match cmd.spawn() {
Ok(child) => {
Ok(mut child) => {
let stdin = child.stdin.take().unwrap();
self.writer = Some(Box::new(stdin));
self.process = Some(child);

2
src/meta_plugin/keep_pid.rs
View File

@@ -26,7 +26,7 @@ impl KeepPidMetaPlugin {
// Set default outputs
let default_outputs = &["keep_pid"];
base.initialize_plugin(default_outputs, _options, outputs);
base.initialize_plugin(default_outputs, &_options, &outputs);
KeepPidMetaPlugin {
is_finalized: false,

22
src/meta_plugin/magic_file.rs
View File

@@ -4,13 +4,14 @@ use magic::{Cookie, CookieFlags};
use std::process::{Command, Stdio};
use std::io::{self, Write};
use std::path::Path;
use log::debug;
use crate::meta_plugin::{MetaPlugin, MetaPluginType, BaseMetaPlugin, MetaPluginResponse, MetaData, process_metadata_outputs};
#[cfg(feature = "magic")]
#[derive(Debug)]
pub struct MagicFileMetaPlugin {
pub struct MagicFileMetaPluginImpl {
buffer: Vec<u8>,
max_buffer_size: usize,
is_finalized: bool,
@@ -19,11 +20,11 @@ pub struct MagicFileMetaPlugin {
}
#[cfg(feature = "magic")]
impl MagicFileMetaPlugin {
impl MagicFileMetaPluginImpl {
pub fn new(
options: Option<std::collections::HashMap<String, serde_yaml::Value>>,
outputs: Option<std::collections::HashMap<String, serde_yaml::Value>>,
) -> MagicFileMetaPlugin {
) -> MagicFileMetaPluginImpl {
let mut base = BaseMetaPlugin::new();
// Set default outputs
@@ -36,7 +37,7 @@ impl MagicFileMetaPlugin {
.and_then(|v| v.as_u64())
.unwrap_or(crate::common::PIPESIZE as u64) as usize;
MagicFileMetaPlugin {
MagicFileMetaPluginImpl {
buffer: Vec::new(),
max_buffer_size,
is_finalized: false,
@@ -68,7 +69,7 @@ impl MagicFileMetaPlugin {
let types_to_process = [
("mime_type", CookieFlags::MIME_TYPE),
("mime_encoding", CookieFlags::MIME_ENCODING),
("file_type", CookieFlags::NONE),
("file_type", CookieFlags::empty()),
];
for (name, flags) in types_to_process.iter() {
@@ -90,7 +91,7 @@ impl MagicFileMetaPlugin {
}
#[cfg(feature = "magic")]
impl MetaPlugin for MagicFileMetaPlugin {
impl MetaPlugin for MagicFileMetaPluginImpl {
fn is_finalized(&self) -> bool {
self.is_finalized
}
@@ -111,7 +112,7 @@ impl MetaPlugin for MagicFileMetaPlugin {
}
};
if let Err(e) = cookie.load(&[]) {
if let Err(e) = cookie.load(&[] as &[&Path]) {
debug!("META: MagicFile plugin: failed to load magic database: {}", e);
return MetaPluginResponse {
metadata: Vec::new(),
@@ -404,12 +405,10 @@ impl MetaPlugin for FallbackMagicFileMetaPlugin {
}
#[cfg(feature = "magic")]
use MagicFileMetaPlugin as MagicFileMetaPluginImpl;
pub use MagicFileMetaPluginImpl as MagicFileMetaPlugin;
#[cfg(not(feature = "magic"))]
use FallbackMagicFileMetaPlugin as MagicFileMetaPluginImpl;
pub use MagicFileMetaPluginImpl as MagicFileMetaPlugin;
pub use FallbackMagicFileMetaPlugin as MagicFileMetaPlugin;
use crate::meta_plugin::register_meta_plugin;
@@ -419,3 +418,4 @@ fn register_magic_file_plugin() {
Box::new(MagicFileMetaPlugin::new(options, outputs))
});
}

2
src/meta_plugin/read_rate.rs
View File

@@ -51,7 +51,7 @@ impl ReadRateMetaPlugin {
// Set default outputs
let default_outputs = &["read_rate"];
base.initialize_plugin(default_outputs, _options, outputs);
base.initialize_plugin(default_outputs, &_options, &outputs);
ReadRateMetaPlugin {
start_time: None,

2
src/meta_plugin/read_time.rs
View File

@@ -18,7 +18,7 @@ impl ReadTimeMetaPlugin {
// Set default outputs
let default_outputs = &["read_time"];
base.initialize_plugin(default_outputs, _options, outputs);
base.initialize_plugin(default_outputs, &_options, &outputs);
ReadTimeMetaPlugin {
start_time: None,

2
src/meta_plugin/shell.rs
View File

@@ -41,7 +41,7 @@ impl ShellMetaPlugin {
// Set default outputs
let default_outputs = &["shell"];
base.initialize_plugin(default_outputs, _options, outputs);
base.initialize_plugin(default_outputs, &_options, &outputs);
ShellMetaPlugin {
is_finalized: false,

2
src/meta_plugin/shell_pid.rs
View File

@@ -17,7 +17,7 @@ impl ShellPidMetaPlugin {
// Set default outputs
let default_outputs = &["shell_pid"];
base.initialize_plugin(default_outputs, options, outputs);
base.initialize_plugin(default_outputs, &options, &outputs);
ShellPidMetaPlugin {
is_finalized: false,

4
src/meta_plugin/text.rs
View File

@@ -43,8 +43,8 @@ impl TextMetaPlugin {
base.initialize_plugin(
&["text", "text_word_count", "text_line_count",
"text_line_max_len", "text_line_mean_len", "text_line_median_len"],
options,
outputs,
&options,
&outputs,
);
// Set disabled outputs to null based on options

4
src/meta_plugin/user.rs
View File

@@ -29,8 +29,8 @@ impl UserMetaPlugin {
// Initialize with helper function
base.initialize_plugin(
&["user_uid", "user_gid", "user_name", "user_group"],
options,
outputs,
&options,
&outputs,
);
UserMetaPlugin {