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 2023/05/24 10:43:10 UTC
[incubator-opendal] 01/01: refactor(core): Implement RFC-2299 for BlockingOperator::write_with
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch blocking-stat-with
in repository https://gitbox.apache.org/repos/asf/incubator-opendal.git
commit 008dba5412d92d62954c4032cbed53ac8bb0fac8
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Wed May 24 18:42:36 2023 +0800
refactor(core): Implement RFC-2299 for BlockingOperator::write_with
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/types/mod.rs | 1 +
core/src/types/operator/blocking_operator.rs | 51 +++++++-------
core/src/types/operator/mod.rs | 1 +
core/src/types/operator/operator_functions.rs | 98 +++++++++++++++++++++++++++
core/src/types/operator/operator_futures.rs | 2 +-
5 files changed, 127 insertions(+), 26 deletions(-)
diff --git a/core/src/types/mod.rs b/core/src/types/mod.rs
index ce9b51f2..bb202abf 100644
--- a/core/src/types/mod.rs
+++ b/core/src/types/mod.rs
@@ -41,6 +41,7 @@ pub use list::BlockingLister;
pub use list::Lister;
mod operator;
+pub use operator::operator_functions;
pub use operator::operator_futures;
pub use operator::BlockingOperator;
pub use operator::Operator;
diff --git a/core/src/types/operator/blocking_operator.rs b/core/src/types/operator/blocking_operator.rs
index d58750bc..ce2478f2 100644
--- a/core/src/types/operator/blocking_operator.rs
+++ b/core/src/types/operator/blocking_operator.rs
@@ -25,6 +25,8 @@ use crate::ops::*;
use crate::raw::*;
use crate::*;
+use super::operator_functions::*;
+
/// BlockingOperator is the entry for all public blocking APIs.
///
/// Read [`concepts`][docs::concepts] for know more about [`Operator`].
@@ -448,12 +450,7 @@ impl BlockingOperator {
/// # }
/// ```
pub fn write(&self, path: &str, bs: impl Into<Bytes>) -> Result<()> {
- let bs = bs.into();
- self.write_with(
- path,
- OpWrite::new().with_content_length(bs.len() as u64),
- bs,
- )
+ self.write_with(path, bs).call()
}
/// Copy a file from `from` to `to`.
@@ -583,35 +580,39 @@ impl BlockingOperator {
/// # use opendal::Result;
/// # use opendal::BlockingOperator;
/// use bytes::Bytes;
- /// use opendal::ops::OpWrite;
///
/// # async fn test(op: BlockingOperator) -> Result<()> {
/// let bs = b"hello, world!".to_vec();
- /// let ow = OpWrite::new().with_content_type("text/plain");
- /// let _ = op.write_with("hello.txt", ow, bs)?;
+ /// let _ = op.write_with("hello.txt", bs).content_type("text/plain").call()?;
/// # Ok(())
/// # }
/// ```
- pub fn write_with(&self, path: &str, args: OpWrite, bs: impl Into<Bytes>) -> Result<()> {
+ pub fn write_with(&self, path: &str, bs: impl Into<Bytes>) -> FunctionWrite {
let path = normalize_path(path);
- if !validate_path(&path, EntryMode::FILE) {
- return Err(
- Error::new(ErrorKind::IsADirectory, "write path is a directory")
- .with_operation("BlockingOperator::write_with")
- .with_context("service", self.info().scheme().into_static())
- .with_context("path", &path),
- );
- }
-
let bs = bs.into();
- let (_, mut w) = self
- .inner()
- .blocking_write(&path, args.with_content_length(bs.len() as u64))?;
- w.write(bs)?;
- w.close()?;
- Ok(())
+ FunctionWrite(OperatorFunction::new(
+ self.inner().clone(),
+ path,
+ (OpWrite::default().with_content_length(bs.len() as u64), bs),
+ |inner, path, (args, bs)| {
+ if !validate_path(&path, EntryMode::FILE) {
+ return Err(
+ Error::new(ErrorKind::IsADirectory, "write path is a directory")
+ .with_operation("BlockingOperator::write_with")
+ .with_context("service", inner.info().scheme().into_static())
+ .with_context("path", &path),
+ );
+ }
+
+ let (_, mut w) = inner.blocking_write(&path, args)?;
+ w.write(bs)?;
+ w.close()?;
+
+ Ok(())
+ },
+ ))
}
/// Write multiple bytes into given path.
diff --git a/core/src/types/operator/mod.rs b/core/src/types/operator/mod.rs
index 6689a679..879f1130 100644
--- a/core/src/types/operator/mod.rs
+++ b/core/src/types/operator/mod.rs
@@ -30,4 +30,5 @@ pub use builder::OperatorBuilder;
mod metadata;
pub use metadata::OperatorInfo;
+pub mod operator_functions;
pub mod operator_futures;
diff --git a/core/src/types/operator/operator_functions.rs b/core/src/types/operator/operator_functions.rs
new file mode 100644
index 00000000..48a4a58f
--- /dev/null
+++ b/core/src/types/operator/operator_functions.rs
@@ -0,0 +1,98 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Functions provides the functions generated by [`BlockingOperator`]
+//!
+//! By using functions, users can add more options for operation.
+
+use bytes::Bytes;
+
+use crate::ops::*;
+use crate::raw::*;
+use crate::*;
+
+/// OperatorFunction is the function generated by [`BlockingOperator`].
+///
+/// The function will consume all the input to generate a result.
+pub(crate) struct OperatorFunction<T, R> {
+ inner: FusedAccessor,
+ path: String,
+ args: T,
+ f: fn(FusedAccessor, String, T) -> Result<R>,
+}
+
+impl<T, R> OperatorFunction<T, R> {
+ pub fn new(
+ inner: FusedAccessor,
+ path: String,
+ args: T,
+ f: fn(FusedAccessor, String, T) -> Result<R>,
+ ) -> Self {
+ Self {
+ inner,
+ path,
+ args,
+ f,
+ }
+ }
+
+ fn map_args(self, f: impl FnOnce(T) -> T) -> Self {
+ Self {
+ inner: self.inner,
+ path: self.path,
+ args: f(self.args),
+ f: self.f,
+ }
+ }
+
+ fn call(self) -> Result<R> {
+ (self.f)(self.inner, self.path, self.args)
+ }
+}
+
+/// Function that generated by [`BlockingOperator::write_with`].
+///
+/// Users can add more options by public functions provided by this struct.
+pub struct FunctionWrite(
+ /// The args for FunctionWrite is a bit special because we also
+ /// need to move the bytes input this function.
+ pub(crate) OperatorFunction<(OpWrite, Bytes), ()>,
+);
+
+impl FunctionWrite {
+ /// Set the content length for this operation.
+ pub fn content_length(mut self, v: u64) -> Self {
+ self.0 = self
+ .0
+ .map_args(|(args, bs)| (args.with_content_length(v), bs));
+ self
+ }
+
+ /// Set the content type for this operation.
+ pub fn content_type(mut self, v: &str) -> Self {
+ self.0 = self
+ .0
+ .map_args(|(args, bs)| (args.with_content_type(v), bs));
+ self
+ }
+
+ /// Call the function to comsume all the input and generate a
+ /// result.
+ pub fn call(self) -> Result<()> {
+ self.0.call()
+ }
+}
diff --git a/core/src/types/operator/operator_futures.rs b/core/src/types/operator/operator_futures.rs
index a95b1efd..cbfaa321 100644
--- a/core/src/types/operator/operator_futures.rs
+++ b/core/src/types/operator/operator_futures.rs
@@ -33,7 +33,7 @@ use crate::*;
/// OperatorFuture is the future generated by [`Operator`].
///
-/// The future will consume all the input to generate a future.
+/// The future will consume all the input to generate a result.
pub(crate) enum OperatorFuture<T, F> {
/// Idle state, waiting for the future to be polled
Idle(