Fixed ftp write

Cargo.lock

@@ -147,7 +147,7 @@ checksum = "00b0228411908ca8685dba7fc2cdd70ec9990a6e753e89b6ac91a84c40fbaf4b"
 [[package]]
 name = "ftp"
 version = "3.0.1"
-source = "git+https://github.com/ChristianVisintin/rust-ftp#36fb333a99719266d445762fa2892ef7f627bc70"
+source = "git+https://github.com/ChristianVisintin/rust-ftp#cfe594f6e6acd6137087847357bf902207d4e477"
 dependencies = [
  "chrono",
  "lazy_static",

src/filetransfer/ftp_transfer.rs

@@ -34,6 +34,7 @@ use crate::utils::lstime_to_systime;
 
 // Includes
 use ftp::native_tls::TlsConnector;
+use ftp::status::{CLOSING_DATA_CONNECTION, REQUESTED_FILE_ACTION_OK};
 use ftp::FtpStream;
 use regex::Regex;
 use std::io::{Read, Write};
@@ -46,6 +47,7 @@ use std::time::SystemTime;
 pub struct FtpFileTransfer {
     stream: Option<FtpStream>,
     ftps: bool,
+    pending_write_response: bool,
 }
 
 impl FtpFileTransfer {
@@ -56,9 +58,29 @@ impl FtpFileTransfer {
         FtpFileTransfer {
             stream: None,
             ftps: ftps,
+            pending_write_response: false,
         }
     }
 
+    /// ### handle_pending_operations
+    /// 
+    /// Handle pending operations, such as closing write command after writing a file
+    fn handle_pending_operations(&mut self) -> Result<(), FileTransferError> {
+        if self.pending_write_response {
+            // Send write response ok
+            if let Some(stream) = &mut self.stream {
+                match stream.read_response_in(&[
+                    CLOSING_DATA_CONNECTION,
+                    REQUESTED_FILE_ACTION_OK,
+                ]) {
+                    Ok(_) => self.pending_write_response = false,
+                    Err(err) => return Err(FileTransferError::new_ex(FileTransferErrorType::ProtocolError, format!("Could not close write: {}", err)))
+                }
+            }
+        }
+        Ok(())
+    }
+
     /// ### parse_list_line
     ///
     /// Parse a line of LIST command output and instantiates an FsEntry from it
@@ -310,6 +332,10 @@ impl FileTransfer for FtpFileTransfer {
     /// Print working directory
 
     fn pwd(&mut self) -> Result<PathBuf, FileTransferError> {
+        // Close pending operations
+        if let Err(err) = self.handle_pending_operations() {
+            return Err(err);
+        }
         match &mut self.stream {
             Some(stream) => match stream.pwd() {
                 Ok(path) => Ok(PathBuf::from(path.as_str())),
@@ -329,6 +355,10 @@ impl FileTransfer for FtpFileTransfer {
     /// Change working directory
 
     fn change_dir(&mut self, dir: &Path) -> Result<PathBuf, FileTransferError> {
+        // Close pending operations
+        if let Err(err) = self.handle_pending_operations() {
+            return Err(err);
+        }
         match &mut self.stream {
             Some(stream) => match stream.cwd(&dir.to_string_lossy()) {
                 Ok(_) => Ok(PathBuf::from(dir)),
@@ -348,6 +378,10 @@ impl FileTransfer for FtpFileTransfer {
     /// List directory entries
 
     fn list_dir(&mut self, path: &Path) -> Result<Vec<FsEntry>, FileTransferError> {
+        // Close pending operations
+        if let Err(err) = self.handle_pending_operations() {
+            return Err(err);
+        }
         match &mut self.stream {
             Some(stream) => match stream.list(Some(&path.to_string_lossy())) {
                 Ok(entries) => {
@@ -376,6 +410,10 @@ impl FileTransfer for FtpFileTransfer {
     ///
     /// Make directory
     fn mkdir(&mut self, dir: &Path) -> Result<(), FileTransferError> {
+        // Close pending operations
+        if let Err(err) = self.handle_pending_operations() {
+            return Err(err);
+        }
         match &mut self.stream {
             Some(stream) => match stream.mkdir(&dir.to_string_lossy()) {
                 Ok(_) => Ok(()),
@@ -394,6 +432,10 @@ impl FileTransfer for FtpFileTransfer {
     ///
     /// Remove a file or a directory
     fn remove(&mut self, fsentry: &FsEntry) -> Result<(), FileTransferError> {
+        // Close pending operations
+        if let Err(err) = self.handle_pending_operations() {
+            return Err(err);
+        }
         if self.stream.is_none() {
             return Err(FileTransferError::new(
                 FileTransferErrorType::UninitializedSession,
@@ -446,6 +488,10 @@ impl FileTransfer for FtpFileTransfer {
     ///
     /// Rename file or a directory
     fn rename(&mut self, file: &FsEntry, dst: &Path) -> Result<(), FileTransferError> {
+        // Close pending operations
+        if let Err(err) = self.handle_pending_operations() {
+            return Err(err);
+        }
         match &mut self.stream {
             Some(stream) => {
                 // Get name
@@ -498,9 +544,17 @@ impl FileTransfer for FtpFileTransfer {
     /// Data contains the file data
     /// Returns file and its size
     fn send_file(&mut self, file_name: &Path) -> Result<Box<dyn Write>, FileTransferError> {
+        // Close pending operations
+        if let Err(err) = self.handle_pending_operations() {
+            return Err(err);
+        }
         match &mut self.stream {
             Some(stream) => match stream.put_with_stream(&file_name.to_string_lossy()) {
-                Ok(writer) => Ok(Box::new(writer)),
+                Ok(writer) => {
+                    // Set pending write response
+                    self.pending_write_response = true;
+                    Ok(Box::new(writer))
+                }    
                 Err(err) => Err(FileTransferError::new_ex(
                     FileTransferErrorType::FileCreateDenied,
                     format!("{}", err),
@@ -517,6 +571,10 @@ impl FileTransfer for FtpFileTransfer {
     /// Receive file from remote with provided name
     /// Returns file and its size
     fn recv_file(&mut self, file_name: &Path) -> Result<Box<dyn Read>, FileTransferError> {
+        // Close pending operations
+        if let Err(err) = self.handle_pending_operations() {
+            return Err(err);
+        }
         match &mut self.stream {
             Some(stream) => match stream.get(&file_name.to_string_lossy()) {
                 Ok(reader) => Ok(Box::new(reader)),