Fix path problem for DirStream.

Fix create() function, it would now create corresponding
directories if not exist.

Author: ArberSephirotheca

src/services/ftp/backend.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use log::info;
 use std::collections::HashMap;
 use std::fmt::Debug;
 use std::fmt::Formatter;
@@ -26,10 +27,9 @@ use anyhow::anyhow;
 use async_trait::async_trait;
 use futures::io::copy;
 use futures::AsyncReadExt;
-use log::info;
 use suppaftp::async_native_tls::TlsConnector;
 use suppaftp::list::File;
-use suppaftp::types::FileType;
+use suppaftp::types::{FileType, Response};
 use suppaftp::FtpError;
 use suppaftp::FtpStream;
 use suppaftp::Status;
@@ -237,47 +237,52 @@ impl Accessor for Backend {
     }
 
     async fn create(&self, args: &OpCreate) -> Result<()> {
-        let path = args.path();
-
         let mut ftp_stream = self.ftp_connect(Operation::Create).await?;
 
-        if args.mode() == ObjectMode::FILE {
-            ftp_stream
-                .put_file(&path, &mut "".as_bytes())
-                .await
-                .map_err(|e| {
-                    other(ObjectError::new(
-                        Operation::Create,
-                        path,
-                        anyhow!("put request: {e:?}"),
-                    ))
-                })?;
-
-            ftp_stream
-                .quit()
-                .await
-                .map_err(|e| new_request_quit_err(e, Operation::Create, path))?;
-
-            return Ok(());
+        let paths: Vec<&str> = args.path().split_inclusive('/').collect();
+
+        let mut curr_path = String::new();
+
+        for path in paths {
+            curr_path.push_str(path);
+            // try to create directory
+            if curr_path.ends_with('/') {
+                match ftp_stream.mkdir(&curr_path).await {
+                    // Do nothing if status is FileUnavailable or OK(()) is return.
+                    Err(FtpError::UnexpectedResponse(Response {
+                        status: Status::FileUnavailable,
+                        ..
+                    }))
+                    | Ok(()) => (),
+                    Err(e) => {
+                        return Err(other(ObjectError::new(
+                            Operation::Create,
+                            args.path(),
+                            anyhow!("mkdir request: {e:?}"),
+                        )));
+                    }
+                }
+            } else {
+                // else, create file
+                ftp_stream
+                    .put_file(&curr_path, &mut "".as_bytes())
+                    .await
+                    .map_err(|e| {
+                        other(ObjectError::new(
+                            Operation::Create,
+                            path,
+                            anyhow!("put request: {e:?}"),
+                        ))
+                    })?;
+            }
         }
 
-        if args.mode() == ObjectMode::DIR {
-            ftp_stream.mkdir(&path).await.map_err(|e| {
-                other(ObjectError::new(
-                    Operation::Create,
-                    path,
-                    anyhow!("mkdir request: {e:?}"),
-                ))
-            })?;
-
-            ftp_stream
-                .quit()
-                .await
-                .map_err(|e| new_request_quit_err(e, Operation::Create, path))?;
-            return Ok(());
-        }
+        ftp_stream
+            .quit()
+            .await
+            .map_err(|e| new_request_quit_err(e, Operation::Create, args.path()))?;
 
-        unreachable!()
+        return Ok(());
     }
 
     async fn read(&self, args: &OpRead) -> Result<BytesReader> {
@@ -319,6 +324,8 @@ impl Accessor for Backend {
     }
 
     async fn write(&self, args: &OpWrite, r: BytesReader) -> Result<u64> {
+        // TODO: create before write
+
         let path = args.path();
 
         let mut ftp_stream = self.ftp_connect(Operation::Write).await?;

src/services/ftp/dir_stream.rs

@@ -90,16 +90,18 @@ impl futures::Stream for DirStream {
                 Poll::Ready(Some(Err(parse_io_error(e, Operation::List, &self.path))))
             }
             Some(Ok(de)) => {
+                let path = self.path.to_string() + de.name();
+
                 let d = if de.is_file() {
-                    DirEntry::new(self.backend.clone(), ObjectMode::FILE, de.name())
+                    DirEntry::new(self.backend.clone(), ObjectMode::FILE, &path)
                 } else if de.is_directory() {
-                    let mut dir_path = de.name().to_string();
-                    if !dir_path.ends_with('/') {
-                        dir_path.push('/');
-                    }
-                    DirEntry::new(self.backend.clone(), ObjectMode::DIR, dir_path.as_str())
+                    DirEntry::new(
+                        self.backend.clone(),
+                        ObjectMode::DIR,
+                        &format!("{}/", &path),
+                    )
                 } else {
-                    DirEntry::new(self.backend.clone(), ObjectMode::Unknown, de.name())
+                    DirEntry::new(self.backend.clone(), ObjectMode::Unknown, &path)
                 };
 
                 debug!(