Compare commits
4 Commits
8b7d76017f
...
v0.3.0
| Author | SHA1 | Date | |
|---|---|---|---|
| 3bac1243a1 | |||
| d196662ac6 | |||
| 44d917c264 | |||
| 46e980fe76 |
25
Cargo.lock
generated
25
Cargo.lock
generated
@@ -154,12 +154,24 @@ version = "0.8.7"
|
|||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b"
|
checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b"
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "equivalent"
|
||||||
|
version = "1.0.2"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f"
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "find-msvc-tools"
|
name = "find-msvc-tools"
|
||||||
version = "0.1.9"
|
version = "0.1.9"
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582"
|
checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582"
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "hashbrown"
|
||||||
|
version = "0.17.0"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "4f467dd6dccf739c208452f8014c75c18bb8301b050ad1cfb27153803edb0f51"
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "heck"
|
name = "heck"
|
||||||
version = "0.5.0"
|
version = "0.5.0"
|
||||||
@@ -190,6 +202,16 @@ dependencies = [
|
|||||||
"cc",
|
"cc",
|
||||||
]
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "indexmap"
|
||||||
|
version = "2.14.0"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9"
|
||||||
|
dependencies = [
|
||||||
|
"equivalent",
|
||||||
|
"hashbrown",
|
||||||
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "is_terminal_polyfill"
|
name = "is_terminal_polyfill"
|
||||||
version = "1.70.2"
|
version = "1.70.2"
|
||||||
@@ -413,8 +435,9 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "zfsbackup"
|
name = "zfsbackup"
|
||||||
version = "0.2.3"
|
version = "0.3.0"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"chrono",
|
"chrono",
|
||||||
"clap",
|
"clap",
|
||||||
|
"indexmap",
|
||||||
]
|
]
|
||||||
|
|||||||
@@ -1,9 +1,10 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "zfsbackup"
|
name = "zfsbackup"
|
||||||
version = "0.2.3"
|
version = "0.3.0"
|
||||||
edition = "2024"
|
edition = "2024"
|
||||||
publish = false
|
publish = false
|
||||||
|
|
||||||
[dependencies]
|
[dependencies]
|
||||||
chrono = "0.4.44"
|
chrono = "0.4.44"
|
||||||
clap = { version = "4.6.1", features = ["derive"] }
|
clap = { version = "4.6.1", features = ["derive"] }
|
||||||
|
indexmap = "2.14.0"
|
||||||
|
|||||||
143
src/job.rs
143
src/job.rs
@@ -2,11 +2,13 @@ use crate::command;
|
|||||||
use crate::progress::BackupEvent;
|
use crate::progress::BackupEvent;
|
||||||
use crate::progress::filter;
|
use crate::progress::filter;
|
||||||
use chrono::{Local, NaiveDateTime};
|
use chrono::{Local, NaiveDateTime};
|
||||||
|
use indexmap::set::IndexSet;
|
||||||
use std::collections::HashSet;
|
use std::collections::HashSet;
|
||||||
|
use std::fmt;
|
||||||
use std::sync::mpsc::Sender;
|
use std::sync::mpsc::Sender;
|
||||||
|
|
||||||
pub struct Job {
|
pub struct Job {
|
||||||
datasets: HashSet<String>,
|
datasets: IndexSet<String>,
|
||||||
target: String,
|
target: String,
|
||||||
source_zfs_command: Vec<String>,
|
source_zfs_command: Vec<String>,
|
||||||
target_zfs_command: Vec<String>,
|
target_zfs_command: Vec<String>,
|
||||||
@@ -14,26 +16,25 @@ pub struct Job {
|
|||||||
sender: Option<Sender<BackupEvent>>,
|
sender: Option<Sender<BackupEvent>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug)]
|
#[derive(Copy, Clone)]
|
||||||
enum JobSide {
|
enum JobSide {
|
||||||
Source,
|
Source,
|
||||||
Destination,
|
Target,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug)]
|
|
||||||
struct Snapshot {
|
struct Snapshot {
|
||||||
snapshot: String,
|
dataset: String,
|
||||||
|
snapshot_id: String,
|
||||||
side: JobSide,
|
side: JobSide,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Job {
|
impl fmt::Display for Snapshot {
|
||||||
pub fn dump(&self) {
|
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||||
println!("Datasets: {:?}", self.datasets);
|
write!(f, "{}@{}", self.dataset, self.snapshot_id)
|
||||||
println!("Target: {}", self.target);
|
}
|
||||||
println!("Source ZFS Command: {:?}", self.source_zfs_command);
|
|
||||||
println!("Target ZFS Command: {:?}", self.target_zfs_command);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
impl Job {
|
||||||
fn send_event(&self, event: BackupEvent) {
|
fn send_event(&self, event: BackupEvent) {
|
||||||
if let Some(sender) = &self.sender {
|
if let Some(sender) = &self.sender {
|
||||||
sender.send(event).ok();
|
sender.send(event).ok();
|
||||||
@@ -43,16 +44,21 @@ impl Job {
|
|||||||
fn get_side_command(&self, side: JobSide) -> Vec<&str> {
|
fn get_side_command(&self, side: JobSide) -> Vec<&str> {
|
||||||
match side {
|
match side {
|
||||||
JobSide::Source => self.source_zfs_command.iter().map(|s| s.as_str()).collect(),
|
JobSide::Source => self.source_zfs_command.iter().map(|s| s.as_str()).collect(),
|
||||||
JobSide::Destination => self.target_zfs_command.iter().map(|s| s.as_str()).collect(),
|
JobSide::Target => self.target_zfs_command.iter().map(|s| s.as_str()).collect(),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
fn create_snapshot(&self, dataset: &str, side: JobSide) -> Result<String, String> {
|
fn create_snapshot(&self, dataset: &str, side: JobSide) -> Result<Snapshot, String> {
|
||||||
let snapshot = format!("{}@{}", dataset, Local::now().format("%Y-%m-%dT%H:%M:%S"));
|
let snapshot = Snapshot {
|
||||||
|
dataset: dataset.to_string(),
|
||||||
|
snapshot_id: Local::now().format("%Y-%m-%dT%H:%M:%S").to_string(),
|
||||||
|
side,
|
||||||
|
};
|
||||||
|
let name = snapshot.to_string();
|
||||||
let mut cmd = self.get_side_command(side);
|
let mut cmd = self.get_side_command(side);
|
||||||
cmd.extend(["snapshot", &snapshot]);
|
cmd.extend(["snapshot", &name]);
|
||||||
command::exec_command(&cmd)?;
|
command::exec_command(&cmd)?;
|
||||||
self.send_event(BackupEvent::SnapshotCreated(snapshot.clone()));
|
self.send_event(BackupEvent::SnapshotCreated(name.clone()));
|
||||||
Ok(snapshot)
|
Ok(snapshot)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -77,6 +83,7 @@ impl Job {
|
|||||||
self.send_event(BackupEvent::Estimate(size));
|
self.send_event(BackupEvent::Estimate(size));
|
||||||
Ok(size)
|
Ok(size)
|
||||||
}
|
}
|
||||||
|
|
||||||
fn send_receive(
|
fn send_receive(
|
||||||
&self,
|
&self,
|
||||||
source: &str,
|
source: &str,
|
||||||
@@ -91,10 +98,13 @@ impl Job {
|
|||||||
}
|
}
|
||||||
send_cmd.push(source);
|
send_cmd.push(source);
|
||||||
|
|
||||||
let mut receive_cmd = self.get_side_command(JobSide::Destination);
|
let mut receive_cmd = self.get_side_command(JobSide::Target);
|
||||||
receive_cmd.extend(["receive", "-F", dest]);
|
receive_cmd.extend(["receive", "-F", dest]);
|
||||||
let filters = match self.sender.as_ref() {
|
let filters: Vec<Box<dyn command::Filter>> = match self.sender.as_ref() {
|
||||||
Some(sender) => vec![filter::CountingReaderBuilder::build(sender.clone(), total)],
|
Some(sender) => vec![Box::new(filter::ByteCountFilter::new(
|
||||||
|
sender.clone(),
|
||||||
|
total,
|
||||||
|
))],
|
||||||
None => vec![],
|
None => vec![],
|
||||||
};
|
};
|
||||||
command::exec_piped_commands(&send_cmd, &receive_cmd, filters)?;
|
command::exec_piped_commands(&send_cmd, &receive_cmd, filters)?;
|
||||||
@@ -102,7 +112,7 @@ impl Job {
|
|||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
fn list_snapshots(&self, source: &str, side: JobSide) -> Result<Vec<String>, String> {
|
fn list_snapshot_ids(&self, source: &str, side: JobSide) -> Result<Vec<String>, String> {
|
||||||
let mut cmd = self.get_side_command(side);
|
let mut cmd = self.get_side_command(side);
|
||||||
cmd.extend(["list", "-H", "-o", "name", "-t", "snapshot", source]);
|
cmd.extend(["list", "-H", "-o", "name", "-t", "snapshot", source]);
|
||||||
let output = command::exec_command(&cmd)?;
|
let output = command::exec_command(&cmd)?;
|
||||||
@@ -125,16 +135,16 @@ impl Job {
|
|||||||
Ok(result)
|
Ok(result)
|
||||||
}
|
}
|
||||||
|
|
||||||
fn find_latest_matching_snapshot(&self, source: &str, dest: &str) -> Result<String, String> {
|
fn find_latest_matching_snapshot_id(&self, source: &str, dest: &str) -> Result<String, String> {
|
||||||
self.find_matching_snapshots(source, dest)?
|
self.find_matching_snapshot_ids(source, dest)?
|
||||||
.into_iter()
|
.into_iter()
|
||||||
.next()
|
.next()
|
||||||
.ok_or(String::from("no matching snapshots"))
|
.ok_or(String::from("no matching snapshots"))
|
||||||
}
|
}
|
||||||
|
|
||||||
fn find_matching_snapshots(&self, source: &str, dest: &str) -> Result<Vec<String>, String> {
|
fn find_matching_snapshot_ids(&self, source: &str, dest: &str) -> Result<Vec<String>, String> {
|
||||||
let source_snapshots = self.list_snapshots(source, JobSide::Source)?;
|
let source_snapshots = self.list_snapshot_ids(source, JobSide::Source)?;
|
||||||
let mut dest_snapshots = self.list_snapshots(dest, JobSide::Destination)?;
|
let mut dest_snapshots = self.list_snapshot_ids(dest, JobSide::Target)?;
|
||||||
dest_snapshots.sort_by(|a, b| b.cmp(a));
|
dest_snapshots.sort_by(|a, b| b.cmp(a));
|
||||||
|
|
||||||
let source_set: HashSet<String> = source_snapshots.into_iter().collect();
|
let source_set: HashSet<String> = source_snapshots.into_iter().collect();
|
||||||
@@ -152,9 +162,9 @@ impl Job {
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn find_old_snapshots(&self, source: &str, dest: &str) -> Result<Vec<Snapshot>, String> {
|
fn find_old_snapshots(&self, source: &str, dest: &str) -> Result<Vec<Snapshot>, String> {
|
||||||
let source_snapshots = self.list_snapshots(source, JobSide::Source)?;
|
let source_snapshots = self.list_snapshot_ids(source, JobSide::Source)?;
|
||||||
let dest_snapshots = self.list_snapshots(dest, JobSide::Destination)?;
|
let dest_snapshots = self.list_snapshot_ids(dest, JobSide::Target)?;
|
||||||
let matching_snapshots = self.find_matching_snapshots(source, dest)?;
|
let matching_snapshots = self.find_matching_snapshot_ids(source, dest)?;
|
||||||
if matching_snapshots.is_empty() {
|
if matching_snapshots.is_empty() {
|
||||||
return Err(String::from("no matching snapshots found"));
|
return Err(String::from("no matching snapshots found"));
|
||||||
}
|
}
|
||||||
@@ -163,7 +173,8 @@ impl Job {
|
|||||||
.into_iter()
|
.into_iter()
|
||||||
.filter(|s| !retain.contains(s))
|
.filter(|s| !retain.contains(s))
|
||||||
.map(|s| Snapshot {
|
.map(|s| Snapshot {
|
||||||
snapshot: format!("{}@{}", source, s),
|
dataset: source.to_string(),
|
||||||
|
snapshot_id: s,
|
||||||
side: JobSide::Source,
|
side: JobSide::Source,
|
||||||
})
|
})
|
||||||
.chain(
|
.chain(
|
||||||
@@ -171,22 +182,22 @@ impl Job {
|
|||||||
.into_iter()
|
.into_iter()
|
||||||
.filter(|s| !retain.contains(s))
|
.filter(|s| !retain.contains(s))
|
||||||
.map(|s| Snapshot {
|
.map(|s| Snapshot {
|
||||||
snapshot: format!("{}@{}", dest, s),
|
dataset: dest.to_string(),
|
||||||
side: JobSide::Destination,
|
snapshot_id: s,
|
||||||
|
side: JobSide::Target,
|
||||||
}),
|
}),
|
||||||
)
|
)
|
||||||
.collect();
|
.collect();
|
||||||
Ok(result)
|
Ok(result)
|
||||||
}
|
}
|
||||||
|
|
||||||
fn delete_snapshot(&self, snapshot: Snapshot) -> Result<String, String> {
|
fn delete_snapshot(&self, snapshot: Snapshot) -> Result<(), String> {
|
||||||
let mut cmd = self.get_side_command(snapshot.side);
|
let mut cmd = self.get_side_command(snapshot.side);
|
||||||
cmd.extend(["destroy", &snapshot.snapshot]);
|
let name = snapshot.to_string();
|
||||||
let res = command::exec_command(&cmd);
|
cmd.extend(["destroy", &name]);
|
||||||
if res.is_ok() {
|
command::exec_command(&cmd)?;
|
||||||
self.send_event(BackupEvent::SnapshotDeleted(snapshot.snapshot.clone()));
|
self.send_event(BackupEvent::SnapshotDeleted(name.clone()));
|
||||||
}
|
Ok(())
|
||||||
res
|
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn run(&self, execute: bool) -> Result<(), String> {
|
pub fn run(&self, execute: bool) -> Result<(), String> {
|
||||||
@@ -200,7 +211,7 @@ impl Job {
|
|||||||
// TODO: This will assume the destination doesn't exist if the
|
// TODO: This will assume the destination doesn't exist if the
|
||||||
// command fails for any reason.
|
// command fails for any reason.
|
||||||
let dest = format!("{}/{}", self.target, source);
|
let dest = format!("{}/{}", self.target, source);
|
||||||
cmd = self.get_side_command(JobSide::Destination);
|
cmd = self.get_side_command(JobSide::Target);
|
||||||
cmd.extend(["list", "-H", "-o", "name", &dest]);
|
cmd.extend(["list", "-H", "-o", "name", &dest]);
|
||||||
let dest_exists = command::exec_command(&cmd).is_ok();
|
let dest_exists = command::exec_command(&cmd).is_ok();
|
||||||
|
|
||||||
@@ -215,20 +226,19 @@ impl Job {
|
|||||||
let inc_snapshot = format!(
|
let inc_snapshot = format!(
|
||||||
"{}@{}",
|
"{}@{}",
|
||||||
source,
|
source,
|
||||||
self.find_latest_matching_snapshot(source, &dest)?
|
self.find_latest_matching_snapshot_id(source, &dest)?
|
||||||
);
|
);
|
||||||
|
|
||||||
let snapshot = self.create_snapshot(source, JobSide::Source)?;
|
let snapshot = self.create_snapshot(source, JobSide::Source)?;
|
||||||
let total = self.estimate(&snapshot, Some(&inc_snapshot)).ok();
|
let total = self
|
||||||
|
.estimate(&snapshot.to_string(), Some(&inc_snapshot))
|
||||||
|
.ok();
|
||||||
if !execute {
|
if !execute {
|
||||||
self.send_event(BackupEvent::DryrunCompleted(source.clone()));
|
self.send_event(BackupEvent::DryrunCompleted(source.clone()));
|
||||||
self.delete_snapshot(Snapshot {
|
self.delete_snapshot(snapshot)?;
|
||||||
snapshot: snapshot.clone(),
|
continue;
|
||||||
side: JobSide::Source,
|
|
||||||
})?;
|
|
||||||
return Ok(());
|
|
||||||
}
|
}
|
||||||
self.send_receive(&snapshot, &dest, Some(&inc_snapshot), total)?;
|
self.send_receive(&snapshot.to_string(), &dest, Some(&inc_snapshot), total)?;
|
||||||
} else {
|
} else {
|
||||||
self.send_event(BackupEvent::StartingFullBackup {
|
self.send_event(BackupEvent::StartingFullBackup {
|
||||||
source: source.clone(),
|
source: source.clone(),
|
||||||
@@ -237,16 +247,13 @@ impl Job {
|
|||||||
total: self.datasets.len(),
|
total: self.datasets.len(),
|
||||||
});
|
});
|
||||||
let snapshot = self.create_snapshot(source, JobSide::Source)?;
|
let snapshot = self.create_snapshot(source, JobSide::Source)?;
|
||||||
let total = self.estimate(&snapshot, None).ok();
|
let total = self.estimate(&snapshot.to_string(), None).ok();
|
||||||
if !execute {
|
if !execute {
|
||||||
self.send_event(BackupEvent::DryrunCompleted(source.clone()));
|
self.send_event(BackupEvent::DryrunCompleted(source.clone()));
|
||||||
self.delete_snapshot(Snapshot {
|
self.delete_snapshot(snapshot)?;
|
||||||
snapshot: snapshot.clone(),
|
continue;
|
||||||
side: JobSide::Source,
|
|
||||||
})?;
|
|
||||||
return Ok(());
|
|
||||||
}
|
}
|
||||||
self.send_receive(&snapshot, &dest, None, total)?;
|
self.send_receive(&snapshot.to_string(), &dest, None, total)?;
|
||||||
}
|
}
|
||||||
|
|
||||||
// Clean up snapshots
|
// Clean up snapshots
|
||||||
@@ -278,24 +285,23 @@ fn parse_command(commandstr: &str) -> Vec<String> {
|
|||||||
|
|
||||||
for c in commandstr.chars() {
|
for c in commandstr.chars() {
|
||||||
match c {
|
match c {
|
||||||
'\'' => {
|
'\'' if !in_double_quote => {
|
||||||
if !in_double_quote {
|
|
||||||
in_single_quote = !in_single_quote;
|
in_single_quote = !in_single_quote;
|
||||||
} else {
|
}
|
||||||
|
'\'' => {
|
||||||
arg.push(c);
|
arg.push(c);
|
||||||
}
|
}
|
||||||
|
'"' if !in_single_quote => {
|
||||||
|
in_double_quote = !in_double_quote;
|
||||||
}
|
}
|
||||||
'"' => {
|
'"' => {
|
||||||
if !in_single_quote {
|
|
||||||
in_double_quote = !in_double_quote;
|
|
||||||
} else {
|
|
||||||
arg.push(c);
|
arg.push(c);
|
||||||
}
|
}
|
||||||
|
' ' if in_single_quote || in_double_quote => {
|
||||||
|
arg.push(c);
|
||||||
}
|
}
|
||||||
' ' => {
|
' ' => {
|
||||||
if in_single_quote || in_double_quote {
|
if !arg.is_empty() {
|
||||||
arg.push(c);
|
|
||||||
} else if !arg.is_empty() {
|
|
||||||
command.push(arg.clone());
|
command.push(arg.clone());
|
||||||
arg.clear();
|
arg.clear();
|
||||||
}
|
}
|
||||||
@@ -356,7 +362,7 @@ impl JobBuilder {
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub fn build(self) -> Result<Job, String> {
|
pub fn build(self) -> Result<Job, String> {
|
||||||
let mut datasets: HashSet<String> = HashSet::new();
|
let mut datasets: IndexSet<String> = IndexSet::new();
|
||||||
for source in &self.sources {
|
for source in &self.sources {
|
||||||
let recurse = source.ends_with("/...");
|
let recurse = source.ends_with("/...");
|
||||||
let source = source.trim_end_matches("/...");
|
let source = source.trim_end_matches("/...");
|
||||||
@@ -376,7 +382,7 @@ impl JobBuilder {
|
|||||||
return Err(String::from("no matching source datasets found"));
|
return Err(String::from("no matching source datasets found"));
|
||||||
}
|
}
|
||||||
Ok(Job {
|
Ok(Job {
|
||||||
datasets: datasets,
|
datasets,
|
||||||
target: self.target,
|
target: self.target,
|
||||||
source_zfs_command: self.source_zfs_command,
|
source_zfs_command: self.source_zfs_command,
|
||||||
target_zfs_command: self.target_zfs_command,
|
target_zfs_command: self.target_zfs_command,
|
||||||
@@ -386,13 +392,6 @@ impl JobBuilder {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
struct JobConfig {
|
|
||||||
sources: Vec<String>,
|
|
||||||
target: String,
|
|
||||||
source_zfs_command: Option<String>,
|
|
||||||
target_zfs_command: Option<String>,
|
|
||||||
}
|
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
|
|||||||
@@ -4,7 +4,7 @@ use std::io::{self, Read};
|
|||||||
use std::sync::mpsc::Sender;
|
use std::sync::mpsc::Sender;
|
||||||
use std::time::{Duration, Instant};
|
use std::time::{Duration, Instant};
|
||||||
|
|
||||||
struct CountingReader<R: Read> {
|
struct ByteCountReader<R: Read> {
|
||||||
inner: R,
|
inner: R,
|
||||||
sender: Sender<BackupEvent>,
|
sender: Sender<BackupEvent>,
|
||||||
bytes: u64,
|
bytes: u64,
|
||||||
@@ -12,7 +12,7 @@ struct CountingReader<R: Read> {
|
|||||||
total: Option<u64>,
|
total: Option<u64>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<R: Read> Read for CountingReader<R> {
|
impl<R: Read> Read for ByteCountReader<R> {
|
||||||
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
|
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
|
||||||
let n = self.inner.read(buf)?;
|
let n = self.inner.read(buf)?;
|
||||||
self.bytes += n as u64;
|
self.bytes += n as u64;
|
||||||
@@ -29,31 +29,31 @@ impl<R: Read> Read for CountingReader<R> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<R: Read> CountingReader<R> {
|
impl<R: Read> ByteCountReader<R> {
|
||||||
fn new(inner: R, sender: Sender<BackupEvent>, total: Option<u64>) -> Self {
|
fn new(inner: R, sender: Sender<BackupEvent>, total: Option<u64>) -> Self {
|
||||||
Self {
|
Self {
|
||||||
inner,
|
inner,
|
||||||
sender,
|
sender,
|
||||||
total,
|
|
||||||
bytes: 0,
|
bytes: 0,
|
||||||
last_send: Instant::now() - Duration::from_secs(1),
|
last_send: Instant::now() - Duration::from_secs(1),
|
||||||
|
total,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pub struct CountingReaderBuilder {
|
pub(crate) struct ByteCountFilter {
|
||||||
sender: Sender<BackupEvent>,
|
sender: Sender<BackupEvent>,
|
||||||
total: Option<u64>,
|
total: Option<u64>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Filter for CountingReaderBuilder {
|
impl ByteCountFilter {
|
||||||
fn filter(&self, reader: Box<dyn Read>) -> Box<dyn Read> {
|
pub(crate) fn new(sender: Sender<BackupEvent>, total: Option<u64>) -> Self {
|
||||||
Box::new(CountingReader::new(reader, self.sender.clone(), self.total))
|
Self { sender, total }
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl CountingReaderBuilder {
|
impl Filter for ByteCountFilter {
|
||||||
pub fn build(sender: Sender<BackupEvent>, total: Option<u64>) -> Box<dyn Filter> {
|
fn filter(&self, inner: Box<dyn Read>) -> Box<dyn Read> {
|
||||||
Box::new(Self { sender, total })
|
Box::new(ByteCountReader::new(inner, self.sender.clone(), self.total))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user