opendal/services/ftp/
backend.rs1use std::fmt::Debug;
19use std::fmt::Formatter;
20use std::str;
21use std::str::FromStr;
22use std::sync::Arc;
23
24use super::core::FtpCore;
25use http::Uri;
26use log::debug;
27use services::ftp::core::Manager;
28use suppaftp::list::File;
29use suppaftp::types::Response;
30use suppaftp::FtpError;
31use suppaftp::Status;
32use tokio::sync::OnceCell;
33use uuid::Uuid;
34
35use super::delete::FtpDeleter;
36use super::err::parse_error;
37use super::lister::FtpLister;
38use super::reader::FtpReader;
39use super::writer::FtpWriter;
40use crate::raw::*;
41use crate::services::FtpConfig;
42use crate::*;
43
44impl Configurator for FtpConfig {
45 type Builder = FtpBuilder;
46 fn into_builder(self) -> Self::Builder {
47 FtpBuilder { config: self }
48 }
49}
50
51#[doc = include_str!("docs.md")]
53#[derive(Default)]
54pub struct FtpBuilder {
55 config: FtpConfig,
56}
57
58impl Debug for FtpBuilder {
59 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
60 f.debug_struct("FtpBuilder")
61 .field("config", &self.config)
62 .finish()
63 }
64}
65
66impl FtpBuilder {
67 pub fn endpoint(mut self, endpoint: &str) -> Self {
69 self.config.endpoint = if endpoint.is_empty() {
70 None
71 } else {
72 Some(endpoint.to_string())
73 };
74
75 self
76 }
77
78 pub fn root(mut self, root: &str) -> Self {
80 self.config.root = if root.is_empty() {
81 None
82 } else {
83 Some(root.to_string())
84 };
85
86 self
87 }
88
89 pub fn user(mut self, user: &str) -> Self {
91 self.config.user = if user.is_empty() {
92 None
93 } else {
94 Some(user.to_string())
95 };
96
97 self
98 }
99
100 pub fn password(mut self, password: &str) -> Self {
102 self.config.password = if password.is_empty() {
103 None
104 } else {
105 Some(password.to_string())
106 };
107
108 self
109 }
110}
111
112impl Builder for FtpBuilder {
113 const SCHEME: Scheme = Scheme::Ftp;
114 type Config = FtpConfig;
115
116 fn build(self) -> Result<impl Access> {
117 debug!("ftp backend build started: {:?}", &self);
118 let endpoint = match &self.config.endpoint {
119 None => return Err(Error::new(ErrorKind::ConfigInvalid, "endpoint is empty")),
120 Some(v) => v,
121 };
122
123 let endpoint_uri = match endpoint.parse::<Uri>() {
124 Err(e) => {
125 return Err(Error::new(ErrorKind::ConfigInvalid, "endpoint is invalid")
126 .with_context("endpoint", endpoint)
127 .set_source(e));
128 }
129 Ok(uri) => uri,
130 };
131
132 let host = endpoint_uri.host().unwrap_or("127.0.0.1");
133 let port = endpoint_uri.port_u16().unwrap_or(21);
134
135 let endpoint = format!("{host}:{port}");
136
137 let enable_secure = match endpoint_uri.scheme_str() {
138 Some("ftp") => false,
139 Some("ftps") | None => true,
142
143 Some(s) => {
144 return Err(Error::new(
145 ErrorKind::ConfigInvalid,
146 "endpoint is unsupported or invalid",
147 )
148 .with_context("endpoint", s));
149 }
150 };
151
152 let root = normalize_root(&self.config.root.unwrap_or_default());
153
154 let user = match &self.config.user {
155 None => "".to_string(),
156 Some(v) => v.clone(),
157 };
158
159 let password = match &self.config.password {
160 None => "".to_string(),
161 Some(v) => v.clone(),
162 };
163
164 let accessor_info = AccessorInfo::default();
165 accessor_info
166 .set_scheme(Scheme::Ftp)
167 .set_root(&root)
168 .set_native_capability(Capability {
169 stat: true,
170 stat_has_content_length: true,
171 stat_has_last_modified: true,
172
173 read: true,
174
175 write: true,
176 write_can_multi: true,
177 write_can_append: true,
178
179 delete: true,
180 create_dir: true,
181
182 list: true,
183 list_has_content_length: true,
184 list_has_last_modified: true,
185
186 shared: true,
187
188 ..Default::default()
189 });
190 let manager = Manager {
191 endpoint: endpoint.clone(),
192 root: root.clone(),
193 user: user.clone(),
194 password: password.clone(),
195 enable_secure,
196 };
197 let core = Arc::new(FtpCore {
198 info: accessor_info.into(),
199 manager,
200 pool: OnceCell::new(),
201 });
202
203 Ok(FtpBackend { core })
204 }
205}
206
207#[derive(Clone)]
209pub struct FtpBackend {
210 core: Arc<FtpCore>,
211}
212
213impl Debug for FtpBackend {
214 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
215 f.debug_struct("Backend").finish()
216 }
217}
218
219impl Access for FtpBackend {
220 type Reader = FtpReader;
221 type Writer = FtpWriter;
222 type Lister = FtpLister;
223 type Deleter = oio::OneShotDeleter<FtpDeleter>;
224 type BlockingReader = ();
225 type BlockingWriter = ();
226 type BlockingLister = ();
227 type BlockingDeleter = ();
228
229 fn info(&self) -> Arc<AccessorInfo> {
230 self.core.info.clone()
231 }
232
233 async fn create_dir(&self, path: &str, _: OpCreateDir) -> Result<RpCreateDir> {
234 let mut ftp_stream = self.core.ftp_connect(Operation::CreateDir).await?;
235
236 let paths: Vec<&str> = path.split_inclusive('/').collect();
237
238 let mut curr_path = String::new();
239
240 for path in paths {
241 curr_path.push_str(path);
242 match ftp_stream.mkdir(&curr_path).await {
243 Err(FtpError::UnexpectedResponse(Response {
245 status: Status::FileUnavailable,
246 ..
247 }))
248 | Ok(()) => (),
249 Err(e) => {
250 return Err(parse_error(e));
251 }
252 }
253 }
254
255 Ok(RpCreateDir::default())
256 }
257
258 async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
259 let file = self.ftp_stat(path).await?;
260
261 let mode = if file.is_file() {
262 EntryMode::FILE
263 } else if file.is_directory() {
264 EntryMode::DIR
265 } else {
266 EntryMode::Unknown
267 };
268
269 let mut meta = Metadata::new(mode);
270 meta.set_content_length(file.size() as u64);
271 meta.set_last_modified(file.modified().into());
272
273 Ok(RpStat::new(meta))
274 }
275
276 async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
277 let ftp_stream = self.core.ftp_connect(Operation::Read).await?;
278
279 let reader = FtpReader::new(ftp_stream, path.to_string(), args).await?;
280 Ok((RpRead::new(), reader))
281 }
282
283 async fn write(&self, path: &str, op: OpWrite) -> Result<(RpWrite, Self::Writer)> {
284 let parent = get_parent(path);
286 let paths: Vec<&str> = parent.split('/').collect();
287
288 let mut ftp_stream = self.core.ftp_connect(Operation::Write).await?;
290 let mut curr_path = String::new();
291
292 for path in paths {
293 if path.is_empty() {
294 continue;
295 }
296 curr_path.push_str(path);
297 curr_path.push('/');
298 match ftp_stream.mkdir(&curr_path).await {
299 Err(FtpError::UnexpectedResponse(Response {
301 status: Status::FileUnavailable,
302 ..
303 }))
304 | Ok(()) => (),
305 Err(e) => {
306 return Err(parse_error(e));
307 }
308 }
309 }
310
311 let tmp_path = if op.append() {
312 None
313 } else {
314 let uuid = Uuid::new_v4().to_string();
315 Some(format!("{}.{}", path, uuid))
316 };
317
318 let w = FtpWriter::new(ftp_stream, path.to_string(), tmp_path);
319
320 Ok((RpWrite::new(), w))
321 }
322
323 async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
324 Ok((
325 RpDelete::default(),
326 oio::OneShotDeleter::new(FtpDeleter::new(self.core.clone())),
327 ))
328 }
329
330 async fn list(&self, path: &str, _: OpList) -> Result<(RpList, Self::Lister)> {
331 let mut ftp_stream = self.core.ftp_connect(Operation::List).await?;
332
333 let pathname = if path == "/" { None } else { Some(path) };
334 let files = ftp_stream.list(pathname).await.map_err(parse_error)?;
335
336 Ok((
337 RpList::default(),
338 FtpLister::new(if path == "/" { "" } else { path }, files),
339 ))
340 }
341}
342
343impl FtpBackend {
344 pub async fn ftp_stat(&self, path: &str) -> Result<File> {
345 let mut ftp_stream = self.core.ftp_connect(Operation::Stat).await?;
346
347 let (parent, basename) = (get_parent(path), get_basename(path));
348
349 let pathname = if parent == "/" { None } else { Some(parent) };
350
351 let resp = ftp_stream.list(pathname).await.map_err(parse_error)?;
352
353 let mut files = resp
355 .into_iter()
356 .filter_map(|file| File::from_str(file.as_str()).ok())
357 .filter(|f| f.name() == basename.trim_end_matches('/'))
358 .collect::<Vec<File>>();
359
360 if files.is_empty() {
361 Err(Error::new(
362 ErrorKind::NotFound,
363 "file is not found during list",
364 ))
365 } else {
366 Ok(files.remove(0))
367 }
368 }
369}
370
371#[cfg(test)]
372mod build_test {
373 use super::FtpBuilder;
374 use crate::*;
375
376 #[test]
377 fn test_build() {
378 let b = FtpBuilder::default()
380 .endpoint("ftps://ftp_server.local")
381 .build();
382 assert!(b.is_ok());
383
384 let b = FtpBuilder::default()
386 .endpoint("ftp://ftp_server.local:1234")
387 .build();
388 assert!(b.is_ok());
389
390 let b = FtpBuilder::default()
392 .endpoint("ftp_server.local:8765")
393 .build();
394 assert!(b.is_ok());
395
396 let b = FtpBuilder::default()
398 .endpoint("invalidscheme://ftp_server.local:8765")
399 .build();
400 assert!(b.is_err());
401 let e = b.unwrap_err();
402 assert_eq!(e.kind(), ErrorKind::ConfigInvalid);
403 }
404}