tale_ndjson/multiplexed/
mod.rs

1//! Weaving together output from more than one file.
2
3use std::io::{self, Write};
4use std::path::PathBuf;
5use std::time::Duration;
6
7use bytes::{Buf, BytesMut};
8use miette::{IntoDiagnostic, Result};
9
10mod batch;
11mod file_state;
12pub mod watcher;
13
14use crate::errors::TaleError;
15use crate::logpatterns::*;
16use crate::{config, process_line};
17
18/// Handle multi-file static mode (read all files once, no following)
19pub fn handle_static(paths: Vec<PathBuf>) -> Result<()> {
20    use file_state::FileStateManager;
21
22    let mut file_manager = FileStateManager::new();
23
24    // Add all files to the manager
25    for path in &paths {
26        file_manager.add_file(path)?;
27    }
28
29    // Read all lines from all files (for static reading)
30    let all_lines = file_manager.read_all_lines()?;
31
32    // Collect all lines, parse and convert to SourcedLine immediately
33    let mut all_sourced_lines: Vec<SourcedLine<'_>> = Vec::new();
34
35    all_lines.iter().for_each(|(file_path, lines)| {
36        lines.iter().enumerate().for_each(|(line_num, line_content)| {
37            let parsed: Printable<'_> = {
38                match serde_json::from_str::<Printable<'_>>(line_content) {
39                    Ok(printable) => printable,
40                    Err(_) => {
41                        // If parsing as Printable fails, it's either invalid JSON or doesn't match any
42                        // variant Use the Text fallback in either case
43                        Printable::Text(line_content.to_owned())
44                    }
45                }
46            };
47            let sourced = SourcedLine::new(parsed, file_path.clone(), line_num);
48            all_sourced_lines.push(sourced);
49        });
50    });
51    // Sort using the stable multi-file sorting function
52    all_sourced_lines.sort_by_key(|line| line.sort_key());
53
54    // Output the sorted lines using existing process_line function
55    let mut outlock = io::stdout().lock();
56    let mut buffer = BytesMut::with_capacity(2048);
57
58    for wrapped in all_sourced_lines {
59        wrapped.write(&mut buffer);
60        outlock.write_all(buffer.chunk()).into_diagnostic()?;
61        outlock.write_all(&[0x0a; 1]).into_diagnostic()?; // blank line
62        buffer.clear();
63    }
64
65    outlock.flush().into_diagnostic()?;
66    Ok(())
67}
68
69/// Handle multi-file tailing mode (watch for changes and follow)
70pub async fn handle_tailing(paths: Vec<PathBuf>) -> Result<()> {
71    use batch::{BatchConfig, BatchedLine, batched_with_config};
72    use watcher::{WatchEvent, create_watcher};
73
74    // Create the file watcher
75    let mut watcher = create_watcher();
76
77    // Add files to watch
78    watcher.add_files(paths).await?;
79
80    // Create batch processor with configuration from CLI
81    let batch_config = BatchConfig {
82        batch_window: Duration::from_millis(config::batch_window_ms()),
83        max_batch_size: 1000,
84        _max_buffer_memory: 10 * 1024 * 1024,
85    };
86    let mut batch_processor = batched_with_config(batch_config);
87
88    // Start the batch processor
89    let (line_sender, mut batch_receiver) = batch_processor.start().await?;
90
91    // Start watching files
92    let mut watch_events = watcher.watch().await?;
93
94    // Set up output
95    let mut outlock = io::stdout().lock();
96    let mut buffer = BytesMut::with_capacity(2048);
97
98    // Main coordination loop
99    loop {
100        tokio::select! {
101            // Handle file system events
102            watch_event = watch_events.recv() => {
103                match watch_event {
104                    Some(WatchEvent::FileModified(path)) => {
105                        // File was modified, read new lines
106                        if let Some(state) = watcher.file_manager_mut().get_state_mut(&path)
107                            && let Ok(_changed) = state.refresh()
108                            && let Ok(new_lines) = state.read_new_lines() {
109                            // Send lines to batch processor
110                            for (line_num, line) in new_lines.into_iter().enumerate() {
111                                let batched_line = BatchedLine::new(
112                                    line,
113                                    path.clone(),
114                                    line_num as u64
115                                );
116                                match line_sender.send(batched_line) {
117                                    Ok(v) => v,
118                                    Err(_) => return Err(TaleError::BatchedLineSender.into())
119                                }
120                            }
121                        }
122                    }
123                    Some(WatchEvent::Error(err)) => {
124                        eprintln!("Watch error: {err}");
125                    }
126                    Some(_) => {
127                        // Other events (create, delete) - could handle these in future
128                    }
129                    None => {
130                        // Watcher stopped
131                        break;
132                    }
133                }
134            }
135
136            // Handle sorted batches from batch processor
137            batch = batch_receiver.recv() => {
138                match batch {
139                    Some(sorted_lines) => {
140                        // Output the sorted batch
141                        for batched_line in sorted_lines {
142                            process_line(&batched_line.content, &mut buffer, &mut outlock)?;
143                        }
144                        outlock.flush().into_diagnostic()?;
145                    }
146                    None => {
147                        // Batch processor stopped
148                        break;
149                    }
150                }
151            }
152        }
153    }
154
155    Ok(())
156}
tale_ndjson/multiplexed/mod.rs

tale_ndjson/multiplexed/
mod.rs
