mod backseeking;
mod buffered;
mod chunked; mod stdin;
pub mod strategies;
use std::io::{self, SeekFrom, Write};
use std::path::{Path, PathBuf};
pub use backseeking::*;
pub use buffered::*;
use bytes::BytesMut;
pub use chunked::*;
use miette::{IntoDiagnostic, Result};
use owo_colors::OwoColorize;
pub use stdin::*;
pub use strategies::*;
use crate::defaults::io::*;
use crate::defaults::processing::*;
use crate::errors::{FileError, TaleError, find_similar_files};
use crate::multiplexed::watcher::{MultiFileWatcher, WatchEvent, WatcherConfig};
use crate::{config, process_line};
async fn wait_for_file_creation(target_path: &Path) -> Result<()> {
use std::time::Duration;
eprintln!("Watching for '{}'…", target_path.display().yellow().bold());
let parent_dir = target_path.parent().ok_or_else(|| {
TaleError::from(Box::new(FileError::NotFound {
path: target_path.to_path_buf(),
similar_files: vec!["Parent directory not found".to_string()],
}))
})?;
if !parent_dir.exists() {
return Err(TaleError::from(Box::new(FileError::NotFound {
path: parent_dir.to_path_buf(),
similar_files: vec!["Parent directory must exist for file watching".to_string()],
}))
.into());
}
let target_filename = target_path.file_name().and_then(|n| n.to_str()).ok_or_else(|| {
miette::Report::from(TaleError::from(Box::new(FileError::NotFound {
path: target_path.to_path_buf(),
similar_files: vec!["Invalid filename".to_string()],
})))
})?;
let mut watcher = MultiFileWatcher::new(WatcherConfig::default());
watcher.add_files(vec![parent_dir]).await?;
let mut event_receiver = watcher.watch().await?;
let mut elapsed_seconds = 0;
let mut last_message_time = std::time::Instant::now();
loop {
match tokio::time::timeout(Duration::from_secs(5), event_receiver.recv()).await {
Ok(Some(event)) => {
match event {
WatchEvent::FileCreated(created_path) => {
if let Some(created_filename) = created_path.file_name().and_then(|n| n.to_str())
&& created_filename == target_filename
{
eprintln!("+++> '{}' created; tailing", target_filename.yellow().bold());
return Ok(());
}
}
WatchEvent::Error(_err) => {
}
_ => {
}
}
}
Ok(None) => {
return Err(TaleError::from(Box::new(FileError::NotFound {
path: target_path.to_path_buf(),
similar_files: vec!["File watcher stopped unexpectedly".to_string()],
}))
.into());
}
Err(_) => {
elapsed_seconds += 5;
if last_message_time.elapsed() >= Duration::from_secs(30) {
eprintln!(
"Still watching for '{}' ({}s elapsed)...",
target_path.display().yellow().bold(),
elapsed_seconds.bright_magenta()
);
last_message_time = std::time::Instant::now();
}
if target_path.exists() {
eprintln!("+++> '{}' created; tailing", target_filename.yellow().bold());
return Ok(());
}
}
}
}
}
pub async fn handle_file(fpath: &Path) -> Result<()> {
let sticky = config::sticky();
if !fpath.exists() {
if sticky {
wait_for_file_creation(fpath).await?;
if !fpath.is_file() {
return Err(TaleError::from(Box::new(FileError::not_a_file_with_type(fpath.to_path_buf()))).into());
}
} else {
let similar_files = find_similar_files(fpath);
return Err(TaleError::from(Box::new(FileError::not_found_with_suggestions(
fpath.to_path_buf(),
similar_files,
)))
.into());
}
}
if !fpath.is_file() {
return Err(TaleError::from(Box::new(FileError::not_a_file_with_type(fpath.to_path_buf()))).into());
}
let mut processor = create_file_processor(fpath, None).map_err(|e| enhance_error_context(e, fpath))?;
if let FileProcessorType::BackSeeking(mut backseeker) = processor {
return backseeker.tail();
}
let offset = config::offset();
let offset_unit = config::offset_unit();
match (offset.is_positive(), offset_unit) {
(true, config::OffsetUnit::Lines) if offset > 0 => {
processor.skip_lines(offset as u64)?;
let mut buffer = BytesMut::with_capacity(OUTPUT_BUFFER_CAPACITY);
let mut outlock = io::stdout().lock();
processor.process_lines(|line| {
process_line(line, &mut buffer, &mut outlock)
.map_err(|e| TaleError::from(std::io::Error::other(e.to_string())))
})?;
outlock.flush().into_diagnostic()?;
}
_ => {
let mut buffer = BytesMut::with_capacity(OUTPUT_BUFFER_CAPACITY);
let mut outlock = io::stdout().lock();
processor.process_lines(|line| {
process_line(line, &mut buffer, &mut outlock)
.map_err(|e| TaleError::from(std::io::Error::other(e.to_string())))
})?;
outlock.flush().into_diagnostic()?;
}
}
Ok(())
}
fn enhance_error_context(error: TaleError, path: &Path) -> TaleError {
match error {
TaleError::Io(io_error) => {
let crate::errors::IoError::OperationFailed { source, .. } = io_error.as_ref();
if source.kind() == std::io::ErrorKind::PermissionDenied {
let suggestion = if cfg!(unix) {
Some(format!("Try: chmod +r {}", path.display()))
} else {
Some("Check file permissions in Properties".to_string())
};
Box::new(FileError::permission_denied_with_suggestion(
path.to_path_buf(),
suggestion,
))
.into()
} else {
TaleError::Io(io_error)
}
}
other => other,
}
}
pub fn create_file_processor<P: AsRef<Path>>(
path: P,
file_size_hint: Option<u64>,
) -> Result<FileProcessorType<'static>, TaleError> {
let path = path.as_ref();
let _strategy = if cfg!(debug_assertions) && config::conservative() {
Strategy::Static(StaticStrategy::conservative())
} else {
Strategy::default()
};
let file_size = file_size_hint.unwrap_or_else(|| std::fs::metadata(path).map(|m| m.len()).unwrap_or(0));
let offset = config::offset();
let offset_unit = config::offset_unit();
let large_offset = offset.abs() > LARGE_OFFSET_THRESHOLD as i64;
let use_chunked = !config::disable_chunked()
&& (config::force_chunked()
|| (file_size > CHUNKED_WITH_OFFSET_FILE_SIZE && large_offset)
|| file_size > ALWAYS_CHUNKED_FILE_SIZE);
if offset < 0 || matches!(offset_unit, config::OffsetUnit::Bytes | config::OffsetUnit::Blocks) {
let processor = BackSeekingProcessor::new(PathBuf::from(path));
return Ok(FileProcessorType::BackSeeking(processor));
}
if use_chunked {
let reader = ChunkedFileReader::with_optimal_config(path)?;
return Ok(FileProcessorType::Chunked(Box::new(reader)));
}
let reader = BufferedFileProcessor::new(path)?;
Ok(FileProcessorType::Buffered(reader))
}
pub trait FileProcessor {
fn process_lines<F>(&mut self, line_processor: F) -> Result<(), TaleError>
where
F: FnMut(&str) -> Result<(), TaleError>;
fn skip_lines(&mut self, count: u64) -> Result<(), TaleError>;
fn file_size(&self) -> u64;
fn seek(&mut self, pos: SeekFrom) -> Result<u64, TaleError>;
fn position(&self) -> u64;
}
pub enum FileProcessorType<'a> {
Buffered(BufferedFileProcessor),
Chunked(Box<ChunkedFileReader>),
BackSeeking(BackSeekingProcessor<'a>),
}
impl<'a> FileProcessor for FileProcessorType<'a> {
fn process_lines<F>(&mut self, line_processor: F) -> Result<(), TaleError>
where
F: FnMut(&str) -> Result<(), TaleError>,
{
match self {
FileProcessorType::Buffered(processor) => processor.process_lines(line_processor),
FileProcessorType::Chunked(processor) => processor.process_lines(line_processor),
FileProcessorType::BackSeeking(processor) => processor.process_lines(line_processor),
}
}
fn skip_lines(&mut self, count: u64) -> Result<(), TaleError> {
match self {
FileProcessorType::Buffered(processor) => processor.skip_lines(count),
FileProcessorType::Chunked(processor) => processor.skip_lines(count),
FileProcessorType::BackSeeking(processor) => processor.skip_lines(count),
}
}
fn file_size(&self) -> u64 {
match self {
FileProcessorType::Buffered(processor) => processor.file_size(),
FileProcessorType::Chunked(processor) => processor.file_size(),
FileProcessorType::BackSeeking(processor) => processor.file_size(),
}
}
fn seek(&mut self, pos: SeekFrom) -> Result<u64, TaleError> {
match self {
FileProcessorType::Buffered(processor) => processor.seek(pos),
FileProcessorType::Chunked(processor) => processor.seek(pos),
FileProcessorType::BackSeeking(processor) => processor.seek(pos),
}
}
fn position(&self) -> u64 {
match self {
FileProcessorType::Buffered(processor) => processor.position(),
FileProcessorType::Chunked(processor) => processor.position(),
FileProcessorType::BackSeeking(processor) => processor.position(),
}
}
}
#[cfg(test)]
mod tests {
use std::io::Write;
use tempfile::NamedTempFile;
use super::*;
use crate::config::ConfigOpts;
use crate::tests::TestLogPattern;
fn create_test_file(content: &str) -> NamedTempFile {
let mut file = NamedTempFile::new().expect("Failed to create temp file");
file.write_all(content.as_bytes()).expect("Failed to write test data");
file.flush().expect("Failed to flush test file");
file
}
#[test]
fn chonk_size_optimizer() {
assert_eq!(optimal_chunk_size(500_000, None), 8_192);
assert_eq!(optimal_chunk_size(50_000_000, None), 131_072);
assert_eq!(optimal_chunk_size(500_000_000, None), 524_288);
}
#[test]
fn processor_selection() {
let testfp = crate::tests::create_test_file(120_000, TestLogPattern::Canonical);
crate::config::with_config(
ConfigOpts {
offset: -20,
offset_unit: config::OffsetUnit::Lines,
force_chunked: false,
disable_chunked: false,
..ConfigOpts::default()
},
|| {
let result = create_file_processor(&testfp, Some(1_000_000_000))
.expect("should create processor for negative offset");
assert!(
matches!(result, FileProcessorType::BackSeeking(_)),
"Negative offset should use Simple processor"
);
},
);
crate::config::with_config(
ConfigOpts {
offset: 100,
offset_unit: config::OffsetUnit::Bytes,
force_chunked: false,
disable_chunked: false,
..ConfigOpts::default()
},
|| {
let result = create_file_processor(&testfp, Some(1_000_000_000))
.expect("should create processor for byte offset");
assert!(
matches!(result, FileProcessorType::BackSeeking(_)),
"Byte offset should use Simple processor"
);
},
);
crate::config::with_config(
ConfigOpts {
offset: 100,
offset_unit: config::OffsetUnit::Blocks,
force_chunked: false,
disable_chunked: false,
..ConfigOpts::default()
},
|| {
let result = create_file_processor(&testfp, Some(1_000_000_000))
.expect("should create processor for block offset");
assert!(
matches!(result, FileProcessorType::BackSeeking(_)),
"Block offset should use Simple processor"
);
},
);
crate::config::with_config(
ConfigOpts {
offset: 100,
offset_unit: config::OffsetUnit::Lines,
force_chunked: true,
disable_chunked: false,
..ConfigOpts::default()
},
|| {
let result = create_file_processor(&testfp, Some(100_000_000))
.expect("should create processor for force_chunked");
assert!(
matches!(result, FileProcessorType::Chunked(_)),
"force_chunked should use Chunked processor"
);
},
);
crate::config::with_config(
ConfigOpts {
offset: 20_000, offset_unit: config::OffsetUnit::Lines,
force_chunked: false,
disable_chunked: true,
..ConfigOpts::default()
},
|| {
let result = create_file_processor(&testfp, Some(200_000_000))
.expect("should create processor for disable_chunked");
assert!(
matches!(result, FileProcessorType::Buffered(_)),
"disable_chunked should prevent Chunked processor"
);
},
);
crate::config::with_config(
ConfigOpts {
offset: 100,
offset_unit: config::OffsetUnit::Lines,
force_chunked: false,
disable_chunked: false,
..ConfigOpts::default()
},
|| {
let result = create_file_processor(&testfp, Some(1_500_000_000)) .expect("should create processor for large file");
assert!(
matches!(result, FileProcessorType::Chunked(_)),
"Large file (>1GB) should use Chunked processor"
);
},
);
crate::config::with_config(
ConfigOpts {
offset: 20_000, offset_unit: config::OffsetUnit::Lines,
force_chunked: false,
disable_chunked: false,
..ConfigOpts::default()
},
|| {
let result = create_file_processor(&testfp, Some(150_000_000)) .expect("should create processor for large file + large offset");
assert!(
matches!(result, FileProcessorType::Chunked(_)),
"Large file (>100MB) + large offset (>10k) should use Chunked processor"
);
},
);
crate::config::with_config(
ConfigOpts {
offset: 100, offset_unit: config::OffsetUnit::Lines,
force_chunked: false,
disable_chunked: false,
..ConfigOpts::default()
},
|| {
let result = create_file_processor(&testfp, Some(10_000_000)) .expect("should create processor for small file + small offset");
assert!(
matches!(result, FileProcessorType::Buffered(_)),
"Small file + small offset should use Buffered processor"
);
},
);
}
#[test]
fn can_create_chonker() {
let data = b"line1\nline2\nline3\n".to_vec();
let chunk = FileChunk::new(data.clone(), 0, data.len() as u64);
assert_eq!(chunk.size(), data.len());
assert!(!chunk.is_empty());
assert!(chunk.starts_at_line_boundary);
assert!(chunk.ends_at_line_boundary);
}
#[test]
fn can_chunkread_small_files() -> Result<(), TaleError> {
let test_data = "line1\nline2\nline3\n";
let temp_file = create_test_file(test_data);
let config = ChunkConfig {
overlap_size: 2,
low_memory_mode: true,
};
let strategy = StaticStrategy {
chunk_size: 8, config: config.clone(),
};
let mut reader = ChunkedFileReader::with_strategy(temp_file.path(), Strategy::Static(strategy))?;
assert_eq!(reader.file_size(), test_data.len() as u64);
assert_eq!(reader.position(), 0);
assert!(!reader.is_at_end());
let mut all_content = String::new();
while let Some(chunk) = reader.read_chunk()? {
let chunk_str = std::str::from_utf8(&chunk.data).expect("we expected a valid utf8 string in this test");
all_content.push_str(chunk_str);
}
assert_eq!(all_content, test_data);
assert!(reader.is_at_end());
Ok(())
}
#[test]
fn line_boundary_handling() {
let data = b"line1\nline2\npartial".to_vec();
let data_len = data.len();
let mut chunk = FileChunk::new(data, 0, data_len as u64);
assert!(!chunk.ends_at_line_boundary);
let remainder = chunk.split_at_last_line();
assert!(remainder.is_some());
assert_eq!(
remainder.expect("we expected some remainder after the end of the line"),
b"partial"
);
assert_eq!(chunk.data, b"line1\nline2\n");
assert!(chunk.ends_at_line_boundary);
}
#[test]
fn chunk_iterator_works() {
let data = b"line1\nline2\nline3".to_vec();
let data_len = data.len();
let chunk = FileChunk::new(data, 0, data_len as u64);
let lines: Vec<&str> = chunk.lines().collect();
assert_eq!(lines, vec!["line1", "line2", "line3"]);
}
#[test]
fn buffer_thing_works() -> Result<(), TaleError> {
let test_data = "line1\nline2\nline3\n";
let temp_file = create_test_file(test_data);
let mut processor = BufferedFileProcessor::new(temp_file.path())?;
assert_eq!(processor.file_size(), test_data.len() as u64);
assert_eq!(processor.position(), 0);
let mut lines = Vec::new();
processor.process_lines(|line| {
lines.push(line.to_string());
Ok(())
})?;
assert_eq!(lines, vec!["line1", "line2", "line3"]);
Ok(())
}
#[test]
fn abstract_processor_impl_factory_noun() -> Result<(), TaleError> {
let test_data = "line1\nline2\nline3\n";
let temp_file = create_test_file(test_data);
config::with_config(ConfigOpts::default(), || {
let mut processor = create_file_processor(temp_file.path(), None).expect("should create processor");
assert_eq!(processor.file_size(), test_data.len() as u64);
let mut lines = Vec::new();
processor
.process_lines(|line| {
lines.push(line.to_string());
Ok(())
})
.expect("should process lines");
assert_eq!(lines, vec!["line1", "line2", "line3"]);
});
Ok(())
}
#[test]
fn can_skip_chunked_lines() -> Result<(), TaleError> {
let test_data = "line1\nline2\nline3\nline4\nline5\nline6\nline7\nline8\nline9\nline10\n";
let temp_file = create_test_file(test_data);
let config = ChunkConfig {
overlap_size: 2,
low_memory_mode: true,
};
let strategy = StaticStrategy {
chunk_size: 15, config: config.clone(),
};
let mut reader = ChunkedFileReader::with_strategy(temp_file.path(), Strategy::Static(strategy))?;
reader.skip_lines(3)?;
let mut remaining_lines = Vec::new();
reader.process_lines(|line| {
remaining_lines.push(line.to_string());
Ok(())
})?;
assert_eq!(
remaining_lines,
vec!["line4", "line5", "line6", "line7", "line8", "line9", "line10"]
);
Ok(())
}
#[test]
fn chunked_skip_lines_partial_chunk() -> Result<(), TaleError> {
let test_data = "a\nb\nc\nd\ne\nf\ng\nh\ni\nj\n";
let temp_file = create_test_file(test_data);
let config = ChunkConfig {
overlap_size: 1,
low_memory_mode: true,
};
let strategy = StaticStrategy {
chunk_size: 8, config: config.clone(),
};
let mut reader = ChunkedFileReader::with_strategy(temp_file.path(), Strategy::Static(strategy))?;
reader.skip_lines(5)?;
let mut next_lines = Vec::new();
reader.process_lines(|line| {
next_lines.push(line.to_string());
if next_lines.len() >= 2 {
return Ok(()); }
Ok(())
})?;
assert!(next_lines.len() >= 2);
assert_eq!(next_lines[0], "f");
assert_eq!(next_lines[1], "g");
Ok(())
}
}