file_engine/operations/
copy.rs1use std::path::PathBuf;
2
3use tokio_util::sync::CancellationToken;
4
5use crate::error::Result;
6use crate::handle::Handle;
7use crate::planner::{BatchConfig, ErrorStrategy, OperationOutcome, SortOrder};
8use crate::profiler::DEFAULT_SMALL_FILE_THRESHOLD;
9use crate::progress::ProgressReporter;
10
11use super::default_concurrency;
12use super::pipeline::run_copy_pipeline;
13
14pub struct CopyBuilder {
15 source: PathBuf,
16 dest: PathBuf,
17 overwrite: bool,
18 preserve_permissions: bool,
21 allow_filesystem_integrity_risk: bool,
22 small_file_threshold: Option<u64>,
23 batch_config: BatchConfig,
24 concurrency: Option<usize>,
25}
26
27impl CopyBuilder {
28 pub(crate) fn new(source: impl Into<PathBuf>, dest: impl Into<PathBuf>) -> Self {
29 Self {
30 source: source.into(),
31 dest: dest.into(),
32 overwrite: false,
33 preserve_permissions: false,
34 allow_filesystem_integrity_risk: false,
35 small_file_threshold: None,
36 batch_config: BatchConfig::default(),
37 concurrency: None,
38 }
39 }
40
41 pub fn overwrite(mut self, overwrite: bool) -> Self {
42 self.overwrite = overwrite;
43 self
44 }
45
46 #[cfg(all(unix, feature = "permissions"))]
47 pub fn preserve_permissions(mut self, preserve: bool) -> Self {
48 self.preserve_permissions = preserve;
49 self
50 }
51
52 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.small_file_threshold.unwrap_or(DEFAULT_SMALL_FILE_THRESHOLD);
100 let cancel_for_task = cancel.clone();
101
102 let join_handle = tokio::spawn(async move {
103 run_copy_pipeline(
104 &self.source,
105 &self.dest,
106 self.overwrite,
107 self.preserve_permissions,
108 self.allow_filesystem_integrity_risk,
109 threshold,
110 &self.batch_config,
111 concurrency,
112 cancel_for_task,
113 reporter,
114 )
115 .await
116 });
117
118 Ok(Handle::new(join_handle, rx, cancel))
119 }
120}