file_engine/operations/
copy.rs

1use std::path::PathBuf;
2use std::time::Instant;
3
4use tokio_util::sync::CancellationToken;
5
6use crate::error::Result;
7use crate::handle::Handle;
8use crate::planner::{BatchConfig, ErrorStrategy, OperationOutcome, SortOrder};
9use crate::profiler::DEFAULT_SMALL_FILE_THRESHOLD;
10use crate::progress::ProgressReporter;
11
12use super::default_concurrency;
13use super::pipeline::run_copy_pipeline;
14
15pub struct CopyBuilder {
16    source: PathBuf,
17    dest: PathBuf,
18    overwrite: bool,
19    /// Unconditional field even though only `.preserve_permissions()`
20    /// (Unix-only) can ever set it.
21    preserve_permissions: bool,
22    allow_filesystem_integrity_risk: bool,
23    small_file_threshold: Option<u64>,
24    batch_config: BatchConfig,
25    concurrency: Option<usize>,
26}
27
28impl CopyBuilder {
29    pub(crate) fn new(source: impl Into<PathBuf>, dest: impl Into<PathBuf>) -> Self {
30        Self {
31            source: source.into(),
32            dest: dest.into(),
33            overwrite: false,
34            preserve_permissions: false,
35            allow_filesystem_integrity_risk: false,
36            small_file_threshold: None,
37            batch_config: BatchConfig::default(),
38            concurrency: None,
39        }
40    }
41
42    pub fn overwrite(mut self, overwrite: bool) -> Self {
43        self.overwrite = overwrite;
44        self
45    }
46
47    #[cfg(all(unix, feature = "permissions"))]
48    pub fn preserve_permissions(mut self, preserve: bool) -> Self {
49        self.preserve_permissions = preserve;
50        self
51    }
52
53    /// Proceed even when the destination filesystem has a known
54    /// write-integrity risk on this platform (currently: exFAT on
55    /// macOS) — without this, `.start()`'s `Handle` resolves to
56    /// `Err(Error::FilesystemIntegrityRisk)` before any data is written.
57    /// See `docs/guide/filesystem-safety.md`.
58    pub fn allow_filesystem_integrity_risk(mut self, allow: bool) -> Self {
59        self.allow_filesystem_integrity_risk = allow;
60        self
61    }
62
63    pub fn small_file_threshold(mut self, bytes: u64) -> Self {
64        self.small_file_threshold = Some(bytes);
65        self
66    }
67
68    pub fn max_bytes_per_batch(mut self, bytes: u64) -> Self {
69        self.batch_config.max_bytes_per_batch = bytes;
70        self
71    }
72
73    pub fn max_files_per_batch(mut self, n: usize) -> Self {
74        self.batch_config.max_files_per_batch = Some(n);
75        self
76    }
77
78    pub fn batch_sort_order(mut self, order: SortOrder) -> Self {
79        self.batch_config.sort_order = order;
80        self
81    }
82
83    pub fn on_error(mut self, strategy: ErrorStrategy) -> Self {
84        self.batch_config.error_strategy = strategy;
85        self
86    }
87
88    pub fn batch_concurrency(mut self, n: usize) -> Self {
89        self.concurrency = Some(n);
90        self
91    }
92
93    pub fn start(self) -> Result<Handle<OperationOutcome>> {
94        let cancel = CancellationToken::new();
95        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
96        let reporter = ProgressReporter::new(tx);
97
98        let concurrency = self.concurrency.unwrap_or_else(default_concurrency);
99        let threshold = self
100            .small_file_threshold
101            .unwrap_or(DEFAULT_SMALL_FILE_THRESHOLD);
102        let cancel_for_task = cancel.clone();
103
104        let join_handle = tokio::spawn(async move {
105            let started = Instant::now();
106            let mut outcome = run_copy_pipeline(
107                &self.source,
108                &self.dest,
109                self.overwrite,
110                self.preserve_permissions,
111                self.allow_filesystem_integrity_risk,
112                threshold,
113                &self.batch_config,
114                concurrency,
115                cancel_for_task,
116                reporter,
117            )
118            .await?;
119            outcome.duration = started.elapsed();
120            Ok(outcome)
121        });
122
123        Ok(Handle::new(join_handle, rx, cancel))
124    }
125}
file_engine/operations/copy.rs

file_engine/operations/
copy.rs
