You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2024/04/10 16:20:45 UTC
(opendal) branch main updated: feat(services/ftp): Support multiple write (#4425)
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/opendal.git
The following commit(s) were added to refs/heads/main by this push:
new 060960c171 feat(services/ftp): Support multiple write (#4425)
060960c171 is described below
commit 060960c1715ae763983da084935c6b1f23224999
Author: x³u³ <xx...@outlook.com>
AuthorDate: Thu Apr 11 00:20:40 2024 +0800
feat(services/ftp): Support multiple write (#4425)
* feat(services/ftp): Support multiple write
* feat(services/ftp): Support append mode and store data_stream inside FtpWriter
* feat(services/ftp): FtpWriter adapts oio::Write
---
core/src/services/ftp/backend.rs | 22 +++++++--
core/src/services/ftp/writer.rs | 101 +++++++++++++++++++++++++++++----------
2 files changed, 93 insertions(+), 30 deletions(-)
diff --git a/core/src/services/ftp/backend.rs b/core/src/services/ftp/backend.rs
index 962fd8cd8e..b9c0ca34c5 100644
--- a/core/src/services/ftp/backend.rs
+++ b/core/src/services/ftp/backend.rs
@@ -37,12 +37,12 @@ use suppaftp::FtpError;
use suppaftp::ImplAsyncFtpStream;
use suppaftp::Status;
use tokio::sync::OnceCell;
+use uuid::Uuid;
use super::err::parse_error;
use super::lister::FtpLister;
use super::reader::FtpReader;
use super::writer::FtpWriter;
-use super::writer::FtpWriters;
use crate::raw::*;
use crate::*;
@@ -285,7 +285,7 @@ impl Debug for FtpBackend {
#[async_trait]
impl Accessor for FtpBackend {
type Reader = FtpReader;
- type Writer = FtpWriters;
+ type Writer = FtpWriter;
type Lister = FtpLister;
type BlockingReader = ();
type BlockingWriter = ();
@@ -301,6 +301,8 @@ impl Accessor for FtpBackend {
read: true,
write: true,
+ write_can_multi: true,
+ write_can_append: true,
delete: true,
create_dir: true,
@@ -360,7 +362,7 @@ impl Accessor for FtpBackend {
Ok((RpRead::new(), FtpReader::new(self.clone(), path, args)))
}
- async fn write(&self, path: &str, _: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+ async fn write(&self, path: &str, op: OpWrite) -> Result<(RpWrite, Self::Writer)> {
// Ensure the parent dir exists.
let parent = get_parent(path);
let paths: Vec<&str> = parent.split('/').collect();
@@ -370,7 +372,11 @@ impl Accessor for FtpBackend {
let mut curr_path = String::new();
for path in paths {
+ if path.is_empty() {
+ continue;
+ }
curr_path.push_str(path);
+ curr_path.push('/');
match ftp_stream.mkdir(&curr_path).await {
// Do nothing if status is FileUnavailable or OK(()) is return.
Err(FtpError::UnexpectedResponse(Response {
@@ -384,8 +390,14 @@ impl Accessor for FtpBackend {
}
}
- let w = FtpWriter::new(self.clone(), path.to_string());
- let w = oio::OneShotWriter::new(w);
+ let tmp_path = if op.append() {
+ None
+ } else {
+ let uuid = Uuid::new_v4().to_string();
+ Some(format!("{}.{}", path, uuid))
+ };
+
+ let w = FtpWriter::new(ftp_stream, path.to_string(), tmp_path);
Ok((RpWrite::new(), w))
}
diff --git a/core/src/services/ftp/writer.rs b/core/src/services/ftp/writer.rs
index fa9f959e2b..c206d157c7 100644
--- a/core/src/services/ftp/writer.rs
+++ b/core/src/services/ftp/writer.rs
@@ -15,19 +15,24 @@
// specific language governing permissions and limitations
// under the License.
-use bytes::Bytes;
-use futures::AsyncWriteExt;
+use bb8::PooledConnection;
+use bytes::Buf;
+use futures::{AsyncRead, AsyncWrite, AsyncWriteExt};
-use super::backend::FtpBackend;
use crate::raw::*;
use crate::services::ftp::err::parse_error;
use crate::*;
-pub type FtpWriters = oio::OneShotWriter<FtpWriter>;
+use super::backend::Manager;
+
+trait DataStream: AsyncRead + AsyncWrite {}
+impl<T> DataStream for T where T: AsyncRead + AsyncWrite {}
pub struct FtpWriter {
- backend: FtpBackend,
- path: String,
+ target_path: String,
+ tmp_path: Option<String>,
+ ftp_stream: PooledConnection<'static, Manager>,
+ data_stream: Option<Box<dyn DataStream + Sync + Send + Unpin + 'static>>,
}
/// # TODO
@@ -36,31 +41,77 @@ pub struct FtpWriter {
///
/// After we can use data stream, we should return it directly.
impl FtpWriter {
- pub fn new(backend: FtpBackend, path: String) -> Self {
- FtpWriter { backend, path }
+ pub fn new(
+ ftp_stream: PooledConnection<'static, Manager>,
+ target_path: String,
+ tmp_path: Option<String>,
+ ) -> Self {
+ FtpWriter {
+ target_path,
+ tmp_path,
+ ftp_stream,
+ data_stream: None,
+ }
}
}
-/// # Safety
-///
-/// We will only take `&mut Self` reference for FtpWriter.
-unsafe impl Sync for FtpWriter {}
+impl oio::Write for FtpWriter {
+ async fn write(&mut self, bs: Buffer) -> Result<usize> {
+ let path = if let Some(tmp_path) = &self.tmp_path {
+ tmp_path
+ } else {
+ &self.target_path
+ };
-impl oio::OneShotWrite for FtpWriter {
- async fn write_once(&self, bs: Bytes) -> Result<()> {
- let mut ftp_stream = self.backend.ftp_connect(Operation::Write).await?;
- let mut data_stream = ftp_stream
- .append_with_stream(&self.path)
- .await
- .map_err(parse_error)?;
- data_stream.write_all(&bs).await.map_err(|err| {
- Error::new(ErrorKind::Unexpected, "copy from ftp stream").set_source(err)
- })?;
+ if self.data_stream.is_none() {
+ self.data_stream = Some(Box::new(
+ self.ftp_stream
+ .append_with_stream(path)
+ .await
+ .map_err(parse_error)?,
+ ));
+ }
- ftp_stream
- .finalize_put_stream(data_stream)
+ let size = self
+ .data_stream
+ .as_mut()
+ .unwrap()
+ .write(bs.chunk())
.await
- .map_err(parse_error)?;
+ .map_err(|err| {
+ Error::new(ErrorKind::Unexpected, "copy from ftp stream").set_source(err)
+ })?;
+
+ Ok(size)
+ }
+
+ async fn close(&mut self) -> Result<()> {
+ let data_stream = self.data_stream.take();
+ if let Some(mut data_stream) = data_stream {
+ data_stream.flush().await.map_err(|err| {
+ Error::new(ErrorKind::Unexpected, "flush data stream failed").set_source(err)
+ })?;
+
+ self.ftp_stream
+ .finalize_put_stream(data_stream)
+ .await
+ .map_err(parse_error)?;
+
+ if let Some(tmp_path) = &self.tmp_path {
+ self.ftp_stream
+ .rename(tmp_path, &self.target_path)
+ .await
+ .map_err(parse_error)?;
+ }
+ }
+
Ok(())
}
+
+ async fn abort(&mut self) -> Result<()> {
+ Err(Error::new(
+ ErrorKind::Unsupported,
+ "FtpWriter doesn't support abort",
+ ))
+ }
}