this repo has no description
feat: add progress bar to gzip
+273
-137
+23
-39
src/bzip2.rs
+23
-39
src/bzip2.rs
···
1
1
use crate::{
2
-
progress::{progress_bar, ProgressArgs},
2
+
progress::{copy_with_progress, ProgressArgs},
3
3
utils::*,
4
4
};
5
5
use bzip2::write::{BzDecoder, BzEncoder};
···
84
84
CmprssOutput::Pipe(pipe) => Box::new(pipe) as Box<dyn Write + Send>,
85
85
};
86
86
let mut encoder = BzEncoder::new(output_stream, Compression::new(self.level));
87
-
let mut bar = progress_bar(file_size, self.progress_args.progress, &output);
88
-
if let Some(progress) = &mut bar {
89
-
// Copy the input to the output in chunks so that we can update the progress bar
90
-
let mut buffer = vec![0; self.progress_args.chunk_size.size_in_bytes];
91
-
loop {
92
-
let bytes_read = input_stream.read(&mut buffer)?;
93
-
if bytes_read == 0 {
94
-
break;
95
-
}
96
-
encoder.write_all(&buffer[..bytes_read])?;
97
-
progress.update_input(encoder.total_in());
98
-
progress.update_output(encoder.total_out());
99
-
}
100
-
encoder.flush()?;
101
-
progress.update_output(encoder.total_out());
102
-
progress.finish();
103
-
} else {
104
-
io::copy(&mut input_stream, &mut encoder)?;
105
-
}
87
+
88
+
// Use the custom output function to handle progress bar updates
89
+
copy_with_progress(
90
+
&mut input_stream,
91
+
&mut encoder,
92
+
self.progress_args.chunk_size.size_in_bytes,
93
+
file_size,
94
+
self.progress_args.progress,
95
+
&output,
96
+
)?;
97
+
106
98
Ok(())
107
99
}
108
100
···
128
120
CmprssOutput::Pipe(pipe) => Box::new(pipe) as Box<dyn Write + Send>,
129
121
};
130
122
let mut decoder = BzDecoder::new(output_stream);
131
-
let mut bar = progress_bar(file_size, self.progress_args.progress, &output);
132
-
if let Some(progress) = &mut bar {
133
-
// Copy the input to the output in chunks so that we can update the progress bar
134
-
let mut buffer = vec![0; self.progress_args.chunk_size.size_in_bytes];
135
-
loop {
136
-
let bytes_read = input_stream.read(&mut buffer)?;
137
-
if bytes_read == 0 {
138
-
break;
139
-
}
140
-
decoder.write_all(&buffer[..bytes_read])?;
141
-
progress.update_input(decoder.total_in());
142
-
progress.update_output(decoder.total_out());
143
-
}
144
-
decoder.flush()?;
145
-
progress.update_output(decoder.total_out());
146
-
progress.finish();
147
-
} else {
148
-
io::copy(&mut input_stream, &mut decoder)?;
149
-
}
123
+
124
+
// Use the custom output function to handle progress bar updates
125
+
copy_with_progress(
126
+
&mut input_stream,
127
+
&mut decoder,
128
+
self.progress_args.chunk_size.size_in_bytes,
129
+
file_size,
130
+
self.progress_args.progress,
131
+
&output,
132
+
)?;
133
+
150
134
Ok(())
151
135
}
152
136
}
+58
-17
src/gzip.rs
+58
-17
src/gzip.rs
···
1
+
use crate::progress::{copy_with_progress, ProgressArgs};
1
2
use crate::utils::*;
2
3
use clap::Args;
3
4
use flate2::write::GzEncoder;
4
5
use flate2::{read::GzDecoder, Compression};
5
6
use std::fs::File;
6
-
use std::io::{self, Read, Write};
7
+
use std::io::{self, BufReader, BufWriter, Read, Write};
7
8
8
9
#[derive(Args, Debug)]
9
10
pub struct GzipArgs {
···
12
13
13
14
#[clap(flatten)]
14
15
pub level_args: LevelArgs,
16
+
17
+
#[clap(flatten)]
18
+
pub progress_args: ProgressArgs,
15
19
}
16
20
17
21
pub struct Gzip {
18
22
pub compression_level: u32,
23
+
pub progress_args: ProgressArgs,
19
24
}
20
25
21
26
impl Default for Gzip {
22
27
fn default() -> Self {
23
28
Gzip {
24
29
compression_level: 6,
30
+
progress_args: ProgressArgs::default(),
25
31
}
26
32
}
27
33
}
···
30
36
pub fn new(args: &GzipArgs) -> Gzip {
31
37
Gzip {
32
38
compression_level: args.level_args.level.level,
39
+
progress_args: args.progress_args,
33
40
}
34
41
}
35
42
}
···
72
79
}
73
80
}
74
81
}
75
-
let mut input_stream = match input {
82
+
let mut file_size = None;
83
+
let mut input_stream: Box<dyn Read + Send> = match input {
76
84
CmprssInput::Path(paths) => {
77
85
if paths.len() > 1 {
78
-
return cmprss_error("only 1 file can be compressed at a time");
86
+
return Err(io::Error::new(
87
+
io::ErrorKind::InvalidInput,
88
+
"Multiple input files not supported for gzip",
89
+
));
79
90
}
80
-
Box::new(File::open(paths[0].as_path())?)
91
+
let path = &paths[0];
92
+
file_size = Some(std::fs::metadata(path)?.len());
93
+
Box::new(BufReader::new(File::open(path)?))
81
94
}
82
-
CmprssInput::Pipe(pipe) => Box::new(pipe) as Box<dyn Read + Send>,
95
+
CmprssInput::Pipe(stdin) => Box::new(BufReader::new(stdin)),
83
96
};
84
-
let output_stream = match output {
85
-
CmprssOutput::Path(path) => Box::new(File::create(path)?),
86
-
CmprssOutput::Pipe(pipe) => Box::new(pipe) as Box<dyn Write + Send>,
97
+
98
+
let output_stream: Box<dyn Write + Send> = match &output {
99
+
CmprssOutput::Path(path) => Box::new(BufWriter::new(File::create(path)?)),
100
+
CmprssOutput::Pipe(stdout) => Box::new(BufWriter::new(stdout)),
87
101
};
88
102
89
103
let mut encoder = GzEncoder::new(output_stream, Compression::new(self.compression_level));
90
-
std::io::copy(&mut input_stream, &mut encoder)?;
104
+
105
+
// Use the custom output function to handle progress bar updates with CountingWriter
106
+
copy_with_progress(
107
+
&mut input_stream,
108
+
&mut encoder,
109
+
self.progress_args.chunk_size.size_in_bytes,
110
+
file_size,
111
+
self.progress_args.progress,
112
+
&output,
113
+
)?;
114
+
91
115
encoder.finish()?;
92
116
Ok(())
93
117
}
94
118
95
119
/// Extract a gzip archive
96
120
fn extract(&self, input: CmprssInput, output: CmprssOutput) -> Result<(), io::Error> {
97
-
let input_stream = match input {
121
+
let mut file_size = None;
122
+
let input_stream: Box<dyn Read + Send> = match input {
98
123
CmprssInput::Path(paths) => {
99
124
if paths.len() > 1 {
100
-
return cmprss_error("only 1 file can be extracted at a time");
125
+
return Err(io::Error::new(
126
+
io::ErrorKind::InvalidInput,
127
+
"Multiple input files not supported for gzip extraction",
128
+
));
101
129
}
102
-
Box::new(File::open(paths[0].as_path())?)
130
+
let path = &paths[0];
131
+
file_size = Some(std::fs::metadata(path)?.len());
132
+
Box::new(BufReader::new(File::open(path)?))
103
133
}
104
-
CmprssInput::Pipe(pipe) => Box::new(pipe) as Box<dyn Read + Send>,
134
+
CmprssInput::Pipe(stdin) => Box::new(BufReader::new(stdin)),
105
135
};
106
-
let mut output_stream = match output {
107
-
CmprssOutput::Path(path) => Box::new(File::create(path)?),
108
-
CmprssOutput::Pipe(pipe) => Box::new(pipe) as Box<dyn Write + Send>,
136
+
137
+
let mut output_stream: Box<dyn Write + Send> = match &output {
138
+
CmprssOutput::Path(path) => Box::new(BufWriter::new(File::create(path)?)),
139
+
CmprssOutput::Pipe(stdout) => Box::new(BufWriter::new(stdout)),
109
140
};
110
141
111
142
let mut decoder = GzDecoder::new(input_stream);
112
-
std::io::copy(&mut decoder, &mut output_stream)?;
143
+
144
+
// Use the utility function to handle progress bar updates
145
+
copy_with_progress(
146
+
&mut decoder,
147
+
&mut output_stream,
148
+
self.progress_args.chunk_size.size_in_bytes,
149
+
file_size,
150
+
self.progress_args.progress,
151
+
&output,
152
+
)?;
153
+
113
154
Ok(())
114
155
}
115
156
}
+169
-42
src/progress.rs
+169
-42
src/progress.rs
···
1
1
use crate::utils::CmprssOutput;
2
2
use clap::Args;
3
3
use indicatif::{HumanBytes, ProgressBar};
4
+
use std::io::{self, Read, Write};
4
5
use std::str::FromStr;
6
+
use std::time::Duration;
7
+
use std::time::Instant;
5
8
6
9
#[derive(clap::ValueEnum, Clone, Copy, Debug, Default)]
7
10
pub enum ProgressDisplay {
···
70
73
pub chunk_size: ChunkSize,
71
74
}
72
75
73
-
/// Progress bar for the compress process
74
-
pub struct Progress {
75
-
/// The progress bar
76
-
bar: ProgressBar,
77
-
/// The number of bytes read from the input
78
-
input_read: u64,
79
-
/// The number of bytes written to the output
80
-
output_written: u64,
81
-
}
82
-
83
-
/// Create a progress bar if necessary
84
-
pub fn progress_bar(
76
+
/// Create a progress bar if necessary based on settings
77
+
pub fn create_progress_bar(
85
78
input_size: Option<u64>,
86
79
progress: ProgressDisplay,
87
80
output: &CmprssOutput,
88
-
) -> Option<Progress> {
81
+
) -> Option<ProgressBar> {
89
82
match (progress, output) {
90
83
(ProgressDisplay::Auto, CmprssOutput::Pipe(_)) => None,
91
84
(ProgressDisplay::Off, _) => None,
92
-
(_, _) => Some(Progress::new(input_size)),
85
+
(_, _) => {
86
+
let bar = match input_size {
87
+
Some(size) => ProgressBar::new(size),
88
+
None => ProgressBar::new_spinner(),
89
+
};
90
+
bar.set_style(
91
+
indicatif::ProgressStyle::default_bar()
92
+
.template("{spinner:.green} [{elapsed_precise}] ({eta}) [{bar:40.cyan/blue}] {bytes}/{total_bytes} => {msg}").unwrap()
93
+
.progress_chars("#>-"),
94
+
);
95
+
bar.enable_steady_tick(Duration::from_millis(100));
96
+
Some(bar)
97
+
}
93
98
}
94
99
}
95
100
96
-
impl Progress {
97
-
/// Create a new progress bar
98
-
/// Draws to stderr by default
99
-
pub fn new(input_size: Option<u64>) -> Self {
100
-
let bar = match input_size {
101
-
Some(size) => ProgressBar::new(size),
102
-
None => ProgressBar::new_spinner(),
103
-
};
104
-
bar.set_style(
105
-
indicatif::ProgressStyle::default_bar()
106
-
.template("{spinner:.green} [{elapsed_precise}] ({eta}) [{bar:40.cyan/blue}] {bytes}/{total_bytes} => {msg}").unwrap()
107
-
.progress_chars("#>-"),
108
-
);
109
-
Progress {
101
+
/// A reader that tracks progress of bytes read
102
+
pub struct ProgressReader<R> {
103
+
inner: R,
104
+
bar: Option<ProgressBar>,
105
+
total_read: u64,
106
+
last_update: Instant,
107
+
bytes_since_update: u64,
108
+
bytes_per_update: u64,
109
+
}
110
+
111
+
impl<R: Read> ProgressReader<R> {
112
+
pub fn new(inner: R, bar: Option<ProgressBar>) -> Self {
113
+
ProgressReader {
114
+
inner,
115
+
bar,
116
+
total_read: 0,
117
+
last_update: Instant::now(),
118
+
bytes_since_update: 0,
119
+
bytes_per_update: 8192, // Start with 8KB, will adjust dynamically
120
+
}
121
+
}
122
+
123
+
/// Updates the progress bar if enough bytes have been read since the last update.
124
+
/// Dynamically adjusts the update frequency to target ~100ms between updates by
125
+
/// tracking the elapsed time and adjusting bytes_per_update accordingly.
126
+
fn maybe_update_progress(&mut self, bytes_read: u64) {
127
+
if let Some(ref bar) = self.bar {
128
+
self.bytes_since_update += bytes_read;
129
+
130
+
if self.bytes_since_update >= self.bytes_per_update {
131
+
let now = Instant::now();
132
+
let elapsed = now.duration_since(self.last_update);
133
+
134
+
// Update the progress
135
+
bar.set_position(self.total_read);
136
+
137
+
// Adjust bytes_per_update to target ~100ms between updates
138
+
if elapsed < Duration::from_millis(50) {
139
+
self.bytes_per_update *= 2;
140
+
} else if elapsed > Duration::from_millis(150) {
141
+
self.bytes_per_update = (self.bytes_per_update / 2).max(1024);
142
+
}
143
+
144
+
self.last_update = now;
145
+
self.bytes_since_update = 0;
146
+
}
147
+
}
148
+
}
149
+
}
150
+
151
+
impl<R: Read> Read for ProgressReader<R> {
152
+
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
153
+
let bytes_read = self.inner.read(buf)?;
154
+
if bytes_read > 0 {
155
+
self.total_read += bytes_read as u64;
156
+
self.maybe_update_progress(bytes_read as u64);
157
+
}
158
+
Ok(bytes_read)
159
+
}
160
+
}
161
+
162
+
/// A writer that tracks progress of bytes written
163
+
pub struct ProgressWriter<W> {
164
+
inner: W,
165
+
bar: Option<ProgressBar>,
166
+
total_written: u64,
167
+
last_update: Instant,
168
+
bytes_since_update: u64,
169
+
bytes_per_update: u64,
170
+
}
171
+
172
+
impl<W: Write> ProgressWriter<W> {
173
+
pub fn new(inner: W, bar: Option<ProgressBar>) -> Self {
174
+
ProgressWriter {
175
+
inner,
110
176
bar,
111
-
input_read: 0,
112
-
output_written: 0,
177
+
total_written: 0,
178
+
last_update: Instant::now(),
179
+
bytes_since_update: 0,
180
+
bytes_per_update: 8192, // Start with 8KB, will adjust dynamically
181
+
}
182
+
}
183
+
184
+
pub fn finish(self) {
185
+
if let Some(bar) = self.bar {
186
+
bar.finish();
187
+
}
188
+
}
189
+
190
+
/// Updates the progress bar if enough bytes have been written since the last update.
191
+
/// Dynamically adjusts the update frequency to target ~100ms between updates by
192
+
/// tracking the elapsed time and adjusting bytes_per_update accordingly.
193
+
fn maybe_update_progress(&mut self, bytes_written: u64) {
194
+
if let Some(ref bar) = self.bar {
195
+
self.bytes_since_update += bytes_written;
196
+
197
+
if self.bytes_since_update >= self.bytes_per_update {
198
+
let now = Instant::now();
199
+
let elapsed = now.duration_since(self.last_update);
200
+
201
+
// Update the progress
202
+
bar.set_message(HumanBytes(self.total_written).to_string());
203
+
204
+
// Adjust bytes_per_update to target ~100ms between updates
205
+
if elapsed < Duration::from_millis(50) {
206
+
self.bytes_per_update *= 2;
207
+
} else if elapsed > Duration::from_millis(150) {
208
+
self.bytes_per_update = (self.bytes_per_update / 2).max(1024);
209
+
}
210
+
211
+
self.last_update = now;
212
+
self.bytes_since_update = 0;
213
+
}
113
214
}
114
215
}
216
+
}
115
217
116
-
/// Update the progress bar with the number of bytes read from the input
117
-
pub fn update_input(&mut self, bytes_read: u64) {
118
-
self.input_read = bytes_read;
119
-
self.bar.set_position(self.input_read);
218
+
impl<W: Write> Write for ProgressWriter<W> {
219
+
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
220
+
let bytes_written = self.inner.write(buf)?;
221
+
if bytes_written > 0 {
222
+
self.total_written += bytes_written as u64;
223
+
self.maybe_update_progress(bytes_written as u64);
224
+
}
225
+
Ok(bytes_written)
120
226
}
121
227
122
-
/// Update the progress bar with the number of bytes written to the output
123
-
pub fn update_output(&mut self, bytes_written: u64) {
124
-
self.output_written = bytes_written;
125
-
self.bar
126
-
.set_message(HumanBytes(self.output_written).to_string());
228
+
fn flush(&mut self) -> io::Result<()> {
229
+
self.inner.flush()
127
230
}
231
+
}
128
232
129
-
/// Finish the progress bar
130
-
pub fn finish(&self) {
131
-
self.bar.finish();
233
+
/// Process data with progress bar updates
234
+
pub fn copy_with_progress<R: Read, W: Write>(
235
+
reader: R,
236
+
writer: W,
237
+
chunk_size: usize,
238
+
input_size: Option<u64>,
239
+
progress_display: ProgressDisplay,
240
+
output: &CmprssOutput,
241
+
) -> io::Result<()> {
242
+
// Create the progress bar if needed
243
+
let progress_bar = create_progress_bar(input_size, progress_display, output);
244
+
245
+
// Create reader and writer with progress tracking
246
+
let mut reader = ProgressReader::new(reader, progress_bar.clone());
247
+
let mut writer = ProgressWriter::new(writer, progress_bar);
248
+
249
+
let mut buffer = vec![0; chunk_size];
250
+
loop {
251
+
let bytes_read = reader.read(&mut buffer)?;
252
+
if bytes_read == 0 {
253
+
break;
254
+
}
255
+
writer.write_all(&buffer[..bytes_read])?;
132
256
}
257
+
writer.flush()?;
258
+
writer.finish();
259
+
Ok(())
133
260
}
134
261
135
262
#[cfg(test)]
+23
-39
src/xz.rs
+23
-39
src/xz.rs
···
1
1
use crate::{
2
-
progress::{progress_bar, ProgressArgs},
2
+
progress::{copy_with_progress, ProgressArgs},
3
3
utils::*,
4
4
};
5
5
use clap::Args;
···
76
76
CmprssOutput::Pipe(pipe) => Box::new(pipe) as Box<dyn Write + Send>,
77
77
};
78
78
let mut encoder = XzEncoder::new(output_stream, self.level);
79
-
let mut bar = progress_bar(file_size, self.progress_args.progress, &output);
80
-
if let Some(progress) = &mut bar {
81
-
// Copy the input to the output in chunks so that we can update the progress bar
82
-
let mut buffer = vec![0; self.progress_args.chunk_size.size_in_bytes];
83
-
loop {
84
-
let bytes_read = input_stream.read(&mut buffer)?;
85
-
if bytes_read == 0 {
86
-
break;
87
-
}
88
-
encoder.write_all(&buffer[..bytes_read])?;
89
-
progress.update_input(encoder.total_in());
90
-
progress.update_output(encoder.total_out());
91
-
}
92
-
encoder.flush()?;
93
-
progress.update_output(encoder.total_out());
94
-
progress.finish();
95
-
} else {
96
-
io::copy(&mut input_stream, &mut encoder)?;
97
-
}
79
+
80
+
// Use the custom output function to handle progress bar updates
81
+
copy_with_progress(
82
+
&mut input_stream,
83
+
&mut encoder,
84
+
self.progress_args.chunk_size.size_in_bytes,
85
+
file_size,
86
+
self.progress_args.progress,
87
+
&output,
88
+
)?;
89
+
98
90
Ok(())
99
91
}
100
92
···
119
111
CmprssOutput::Pipe(pipe) => Box::new(pipe) as Box<dyn Write + Send>,
120
112
};
121
113
let mut decoder = XzDecoder::new(output_stream);
122
-
let mut bar = progress_bar(file_size, self.progress_args.progress, &output);
123
-
if let Some(progress) = &mut bar {
124
-
// Copy the input to the output in chunks so that we can update the progress bar
125
-
let mut buffer = vec![0; self.progress_args.chunk_size.size_in_bytes];
126
-
loop {
127
-
let bytes_read = input_stream.read(&mut buffer)?;
128
-
if bytes_read == 0 {
129
-
break;
130
-
}
131
-
decoder.write_all(&buffer[..bytes_read])?;
132
-
progress.update_input(decoder.total_in());
133
-
progress.update_output(decoder.total_out());
134
-
}
135
-
decoder.flush()?;
136
-
progress.update_output(decoder.total_out());
137
-
progress.finish();
138
-
} else {
139
-
io::copy(&mut input_stream, &mut decoder)?;
140
-
}
114
+
115
+
// Use the custom output function to handle progress bar updates
116
+
copy_with_progress(
117
+
&mut input_stream,
118
+
&mut decoder,
119
+
self.progress_args.chunk_size.size_in_bytes,
120
+
file_size,
121
+
self.progress_args.progress,
122
+
&output,
123
+
)?;
124
+
141
125
Ok(())
142
126
}
143
127
}