You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by yu...@apache.org on 2026/05/24 15:27:39 UTC
(opendal) branch feat/bindings-go-layer created (now 0fcc32101)
This is an automated email from the ASF dual-hosted git repository.
yuchanns pushed a change to branch feat/bindings-go-layer
in repository https://gitbox.apache.org/repos/asf/opendal.git
at 0fcc32101 feat(bindings/go): support layers
This branch includes the following new commits:
new 175697e9f feat(bindings/c): support layers
new 0fcc32101 feat(bindings/go): support layers
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(opendal) 01/02: feat(bindings/c): support layers
Posted by yu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yuchanns pushed a commit to branch feat/bindings-go-layer
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 175697e9f6169531129c4d5e70ff4236de49525b
Author: Hanchin Hsieh <me...@yuchanns.xyz>
AuthorDate: Sun May 24 23:27:16 2026 +0800
feat(bindings/c): support layers
---
bindings/c/include/opendal.h | 54 +++++++++++
bindings/c/src/layer.rs | 166 +++++++++++++++++++++++++++++++++
bindings/c/src/lib.rs | 3 +
bindings/c/src/operator.rs | 90 ++++++++++++++----
bindings/c/tests/Makefile | 3 +-
bindings/c/tests/test_runner.cpp | 2 +
bindings/c/tests/test_suites_layer.cpp | 109 ++++++++++++++++++++++
7 files changed, 408 insertions(+), 19 deletions(-)
diff --git a/bindings/c/include/opendal.h b/bindings/c/include/opendal.h
index b845ab310..0b2c14269 100644
--- a/bindings/c/include/opendal.h
+++ b/bindings/c/include/opendal.h
@@ -188,6 +188,19 @@ typedef struct opendal_lister {
void *inner;
} opendal_lister;
+/**
+ * \brief The layers to apply when initializing an opendal_operator.
+ *
+ * \note This is also a heap-allocated struct, please free it after you use it.
+ */
+typedef struct opendal_operator_layers {
+ /**
+ * The pointer to the Vec<OperatorLayer> in the Rust code.
+ * Only touch this on judging whether it is NULL.
+ */
+ void *inner;
+} opendal_operator_layers;
+
/**
* \brief Carries all metadata associated with a **path**.
*
@@ -698,6 +711,33 @@ struct opendal_result_lister_next opendal_lister_next(struct opendal_lister *sel
*/
void opendal_lister_free(struct opendal_lister *ptr);
+/**
+ * \brief Construct a heap-allocated opendal_operator_layers.
+ */
+struct opendal_operator_layers *opendal_operator_layers_new(void);
+
+/**
+ * \brief Add a retry layer.
+ */
+void opendal_operator_layers_add_retry(struct opendal_operator_layers *self,
+ uint8_t jitter,
+ float factor,
+ uint64_t min_delay_ns,
+ uint64_t max_delay_ns,
+ uint64_t max_times);
+
+/**
+ * \brief Add a timeout layer.
+ */
+void opendal_operator_layers_add_timeout(struct opendal_operator_layers *self,
+ uint64_t timeout_ns,
+ uint64_t io_timeout_ns);
+
+/**
+ * \brief Free the allocated memory used by opendal_operator_layers.
+ */
+void opendal_operator_layers_free(struct opendal_operator_layers *ptr);
+
/**
* \brief Free the heap-allocated metadata used by opendal_metadata
*/
@@ -829,6 +869,20 @@ void opendal_operator_free(const struct opendal_operator *ptr);
struct opendal_result_operator_new opendal_operator_new(const char *scheme,
const struct opendal_operator_options *options);
+/**
+ * \brief Construct an operator based on scheme, options, and explicit layers.
+ *
+ * Unlike opendal_operator_new, this function will not add any default layer.
+ * Layers will be applied exactly as they were added to opendal_operator_layers.
+ *
+ * # Safety
+ *
+ * The only unsafe case is passing an invalid c string pointer to the scheme argument.
+ */
+struct opendal_result_operator_new opendal_operator_new_with_layers(const char *scheme,
+ const struct opendal_operator_options *options,
+ const struct opendal_operator_layers *layers);
+
/**
* \brief Blocking write raw bytes to `path`.
*
diff --git a/bindings/c/src/layer.rs b/bindings/c/src/layer.rs
new file mode 100644
index 000000000..6cc17dd64
--- /dev/null
+++ b/bindings/c/src/layer.rs
@@ -0,0 +1,166 @@
+// 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.
+
+use std::ffi::c_void;
+use std::time::Duration;
+
+use ::opendal as core;
+
+enum OperatorLayer {
+ Retry {
+ jitter: bool,
+ factor: f32,
+ min_delay: Duration,
+ max_delay: Duration,
+ max_times: u64,
+ },
+ Timeout {
+ timeout: Duration,
+ io_timeout: Duration,
+ },
+}
+
+impl OperatorLayer {
+ fn apply(&self, op: core::Operator) -> core::Operator {
+ match self {
+ Self::Retry {
+ jitter,
+ factor,
+ min_delay,
+ max_delay,
+ max_times,
+ } => {
+ let mut layer = core::layers::RetryLayer::new()
+ .with_factor(*factor)
+ .with_min_delay(*min_delay)
+ .with_max_delay(*max_delay)
+ .with_max_times(*max_times as usize);
+ if *jitter {
+ layer = layer.with_jitter();
+ }
+ op.layer(layer)
+ }
+ Self::Timeout {
+ timeout,
+ io_timeout,
+ } => op.layer(
+ core::layers::TimeoutLayer::new()
+ .with_timeout(*timeout)
+ .with_io_timeout(*io_timeout),
+ ),
+ }
+ }
+}
+
+/// \brief The layers to apply when initializing an opendal_operator.
+///
+/// \note This is also a heap-allocated struct, please free it after you use it.
+#[repr(C)]
+pub struct opendal_operator_layers {
+ /// The pointer to the Vec<OperatorLayer> in the Rust code.
+ /// Only touch this on judging whether it is NULL.
+ inner: *mut c_void,
+}
+
+impl opendal_operator_layers {
+ pub(crate) fn apply(&self, mut op: core::Operator) -> core::Operator {
+ for layer in self.deref() {
+ op = layer.apply(op);
+ }
+ op
+ }
+
+ fn deref(&self) -> &[OperatorLayer] {
+ // Safety: the inner should never be null once constructed.
+ unsafe { &*(self.inner as *mut Vec<OperatorLayer>) }
+ }
+
+ fn deref_mut(&mut self) -> &mut Vec<OperatorLayer> {
+ // Safety: the inner should never be null once constructed.
+ unsafe { &mut *(self.inner as *mut Vec<OperatorLayer>) }
+ }
+
+ /// \brief Construct a heap-allocated opendal_operator_layers.
+ #[no_mangle]
+ pub extern "C" fn opendal_operator_layers_new() -> *mut Self {
+ let layers: Vec<OperatorLayer> = Vec::new();
+ let layers = Self {
+ inner: Box::into_raw(Box::new(layers)) as _,
+ };
+ Box::into_raw(Box::new(layers))
+ }
+
+ /// \brief Add a retry layer.
+ pub fn add_retry(
+ &mut self,
+ jitter: u8,
+ factor: f32,
+ min_delay_ns: u64,
+ max_delay_ns: u64,
+ max_times: u64,
+ ) {
+ self.deref_mut().push(OperatorLayer::Retry {
+ jitter: jitter != 0,
+ factor,
+ min_delay: Duration::from_nanos(min_delay_ns),
+ max_delay: Duration::from_nanos(max_delay_ns),
+ max_times,
+ });
+ }
+
+ /// \brief Add a retry layer.
+ #[no_mangle]
+ pub unsafe extern "C" fn opendal_operator_layers_add_retry(
+ &mut self,
+ jitter: u8,
+ factor: f32,
+ min_delay_ns: u64,
+ max_delay_ns: u64,
+ max_times: u64,
+ ) {
+ self.add_retry(jitter, factor, min_delay_ns, max_delay_ns, max_times);
+ }
+
+ /// \brief Add a timeout layer.
+ pub fn add_timeout(&mut self, timeout_ns: u64, io_timeout_ns: u64) {
+ self.deref_mut().push(OperatorLayer::Timeout {
+ timeout: Duration::from_nanos(timeout_ns),
+ io_timeout: Duration::from_nanos(io_timeout_ns),
+ });
+ }
+
+ /// \brief Add a timeout layer.
+ #[no_mangle]
+ pub unsafe extern "C" fn opendal_operator_layers_add_timeout(
+ &mut self,
+ timeout_ns: u64,
+ io_timeout_ns: u64,
+ ) {
+ self.add_timeout(timeout_ns, io_timeout_ns);
+ }
+
+ /// \brief Free the allocated memory used by opendal_operator_layers.
+ #[no_mangle]
+ pub unsafe extern "C" fn opendal_operator_layers_free(ptr: *mut opendal_operator_layers) {
+ unsafe {
+ if !ptr.is_null() {
+ drop(Box::from_raw((*ptr).inner as *mut Vec<OperatorLayer>));
+ drop(Box::from_raw(ptr));
+ }
+ }
+ }
+}
diff --git a/bindings/c/src/lib.rs b/bindings/c/src/lib.rs
index 754df7b35..ede881e03 100644
--- a/bindings/c/src/lib.rs
+++ b/bindings/c/src/lib.rs
@@ -39,6 +39,9 @@ pub use error::opendal_error;
mod lister;
pub use lister::opendal_lister;
+mod layer;
+pub use layer::opendal_operator_layers;
+
mod metadata;
pub use metadata::opendal_metadata;
diff --git a/bindings/c/src/operator.rs b/bindings/c/src/operator.rs
index 428ce4439..bb6de2f5a 100644
--- a/bindings/c/src/operator.rs
+++ b/bindings/c/src/operator.rs
@@ -92,6 +92,25 @@ fn build_operator(
let op = core::Operator::via_iter(schema, map)?.layer(core::layers::RetryLayer::new());
+ build_blocking_operator(op)
+}
+
+fn build_operator_with_layers(
+ schema: &str,
+ map: HashMap<String, String>,
+ layers: *const opendal_operator_layers,
+) -> core::Result<core::blocking::Operator> {
+ core::init_default_registry();
+
+ let mut op = core::Operator::via_iter(schema, map)?;
+ if !layers.is_null() {
+ op = unsafe { (*layers).apply(op) };
+ }
+
+ build_blocking_operator(op)
+}
+
+fn build_blocking_operator(op: core::Operator) -> core::Result<core::blocking::Operator> {
let runtime =
tokio::runtime::Handle::try_current().unwrap_or_else(|_| RUNTIME.handle().clone());
let _guard = runtime.enter();
@@ -99,6 +118,33 @@ fn build_operator(
Ok(op)
}
+fn parse_operator_options(options: *const opendal_operator_options) -> HashMap<String, String> {
+ let mut map = HashMap::<String, String>::default();
+ if !options.is_null() {
+ unsafe {
+ for (k, v) in (*options).deref() {
+ map.insert(k.to_string(), v.to_string());
+ }
+ }
+ }
+ map
+}
+
+fn new_operator_result(op: core::Result<core::blocking::Operator>) -> opendal_result_operator_new {
+ match op {
+ Ok(op) => opendal_result_operator_new {
+ op: Box::into_raw(Box::new(opendal_operator {
+ inner: Box::into_raw(Box::new(op)) as _,
+ })),
+ error: std::ptr::null_mut(),
+ },
+ Err(e) => opendal_result_operator_new {
+ op: std::ptr::null_mut(),
+ error: opendal_error::new(e),
+ },
+ }
+}
+
/// \brief Construct an operator based on `scheme` and `options`
///
/// Uses an array of key-value pairs to initialize the operator based on provided `scheme`
@@ -146,25 +192,33 @@ pub unsafe extern "C" fn opendal_operator_new(
.to_str()
.expect("malformed scheme");
- let mut map = HashMap::<String, String>::default();
- if !options.is_null() {
- for (k, v) in (*options).deref() {
- map.insert(k.to_string(), v.to_string());
- }
- }
+ let map = parse_operator_options(options);
- match build_operator(scheme, map) {
- Ok(op) => opendal_result_operator_new {
- op: Box::into_raw(Box::new(opendal_operator {
- inner: Box::into_raw(Box::new(op)) as _,
- })),
- error: std::ptr::null_mut(),
- },
- Err(e) => opendal_result_operator_new {
- op: std::ptr::null_mut(),
- error: opendal_error::new(e),
- },
- }
+ new_operator_result(build_operator(scheme, map))
+}
+
+/// \brief Construct an operator based on scheme, options, and explicit layers.
+///
+/// Unlike opendal_operator_new, this function will not add any default layer.
+/// Layers will be applied exactly as they were added to opendal_operator_layers.
+///
+/// # Safety
+///
+/// The only unsafe case is passing an invalid c string pointer to the scheme argument.
+#[no_mangle]
+pub unsafe extern "C" fn opendal_operator_new_with_layers(
+ scheme: *const c_char,
+ options: *const opendal_operator_options,
+ layers: *const opendal_operator_layers,
+) -> opendal_result_operator_new {
+ assert!(!scheme.is_null());
+ let scheme = std::ffi::CStr::from_ptr(scheme)
+ .to_str()
+ .expect("malformed scheme");
+
+ let map = parse_operator_options(options);
+
+ new_operator_result(build_operator_with_layers(scheme, map, layers))
}
/// \brief Blocking write raw bytes to `path`.
diff --git a/bindings/c/tests/Makefile b/bindings/c/tests/Makefile
index 630070022..867a5287a 100644
--- a/bindings/c/tests/Makefile
+++ b/bindings/c/tests/Makefile
@@ -26,7 +26,7 @@ endif
# Source files
FRAMEWORK_SOURCES = test_framework.cpp
-SUITE_SOURCES = test_suites_basic.cpp test_suites_list.cpp test_suites_reader_writer.cpp test_suites_presign.cpp
+SUITE_SOURCES = test_suites_basic.cpp test_suites_list.cpp test_suites_reader_writer.cpp test_suites_presign.cpp test_suites_layer.cpp
RUNNER_SOURCES = test_runner.cpp
ALL_SOURCES = $(FRAMEWORK_SOURCES) $(SUITE_SOURCES) $(RUNNER_SOURCES)
@@ -105,4 +105,5 @@ test_framework.o: test_framework.h
test_suites_basic.o: test_framework.h
test_suites_list.o: test_framework.h
test_suites_reader_writer.o: test_framework.h
+test_suites_layer.o: test_framework.h
test_runner.o: test_framework.h
diff --git a/bindings/c/tests/test_runner.cpp b/bindings/c/tests/test_runner.cpp
index cf1bf2397..63a6c1ff1 100644
--- a/bindings/c/tests/test_runner.cpp
+++ b/bindings/c/tests/test_runner.cpp
@@ -24,6 +24,7 @@
// External test suite declarations
extern opendal_test_suite basic_suite;
extern opendal_test_suite list_suite;
+extern opendal_test_suite layer_suite;
extern opendal_test_suite presign_suite;
extern opendal_test_suite reader_writer_suite;
@@ -31,6 +32,7 @@ extern opendal_test_suite reader_writer_suite;
static opendal_test_suite* all_suites[] = {
&basic_suite,
&list_suite,
+ &layer_suite,
&presign_suite,
&reader_writer_suite,
};
diff --git a/bindings/c/tests/test_suites_layer.cpp b/bindings/c/tests/test_suites_layer.cpp
new file mode 100644
index 000000000..774170f74
--- /dev/null
+++ b/bindings/c/tests/test_suites_layer.cpp
@@ -0,0 +1,109 @@
+/*
+ * 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.
+ */
+
+#include "test_framework.h"
+
+void test_operator_with_timeout_layer(opendal_test_context* ctx)
+{
+ opendal_operator_layers* layers = opendal_operator_layers_new();
+ OPENDAL_ASSERT_NOT_NULL(layers, "Should create operator layers");
+
+ opendal_operator_layers_add_timeout(layers, 60ULL * 1000 * 1000 * 1000, 10ULL * 1000 * 1000 * 1000);
+
+ opendal_result_operator_new result = opendal_operator_new_with_layers(
+ ctx->config->scheme, ctx->config->options, layers);
+ opendal_operator_layers_free(layers);
+
+ OPENDAL_ASSERT_NO_ERROR(result.error, "Operator with timeout layer should be created");
+ OPENDAL_ASSERT_NOT_NULL(result.op, "Operator with timeout layer should not be null");
+
+ opendal_operator_info* info = opendal_operator_info_new(result.op);
+ OPENDAL_ASSERT_NOT_NULL(info, "Layered operator info should not be null");
+ opendal_operator_info_free(info);
+
+ const char* path = "test_timeout_layer.txt";
+ const char* content = "Hello, OpenDAL timeout layer!";
+ opendal_bytes data;
+ data.data = (uint8_t*)content;
+ data.len = strlen(content);
+ data.capacity = strlen(content);
+
+ opendal_error* error = opendal_operator_write(result.op, path, &data);
+ OPENDAL_ASSERT_NO_ERROR(error, "Write with timeout layer should succeed");
+
+ opendal_result_read read_result = opendal_operator_read(result.op, path);
+ OPENDAL_ASSERT_NO_ERROR(read_result.error, "Read with timeout layer should succeed");
+ OPENDAL_ASSERT_EQ(strlen(content), read_result.data.len, "Read data length should match");
+ OPENDAL_ASSERT(memcmp(content, read_result.data.data, read_result.data.len) == 0,
+ "Read content should match");
+
+ opendal_bytes_free(&read_result.data);
+ opendal_operator_delete(result.op, path);
+ opendal_operator_free(result.op);
+}
+
+void test_operator_with_retry_layer(opendal_test_context* ctx)
+{
+ opendal_operator_layers* layers = opendal_operator_layers_new();
+ OPENDAL_ASSERT_NOT_NULL(layers, "Should create operator layers");
+
+ opendal_operator_layers_add_retry(layers, 1, 2.0f, 1000ULL * 1000 * 1000, 60ULL * 1000 * 1000 * 1000, 3);
+
+ opendal_result_operator_new result = opendal_operator_new_with_layers(
+ ctx->config->scheme, ctx->config->options, layers);
+ opendal_operator_layers_free(layers);
+
+ OPENDAL_ASSERT_NO_ERROR(result.error, "Operator with retry layer should be created");
+ OPENDAL_ASSERT_NOT_NULL(result.op, "Operator with retry layer should not be null");
+
+ opendal_operator_info* info = opendal_operator_info_new(result.op);
+ OPENDAL_ASSERT_NOT_NULL(info, "Layered operator info should not be null");
+ opendal_operator_info_free(info);
+
+ const char* path = "test_retry_layer.txt";
+ const char* content = "Hello, OpenDAL retry layer!";
+ opendal_bytes data;
+ data.data = (uint8_t*)content;
+ data.len = strlen(content);
+ data.capacity = strlen(content);
+
+ opendal_error* error = opendal_operator_write(result.op, path, &data);
+ OPENDAL_ASSERT_NO_ERROR(error, "Write with retry layer should succeed");
+
+ opendal_result_read read_result = opendal_operator_read(result.op, path);
+ OPENDAL_ASSERT_NO_ERROR(read_result.error, "Read with retry layer should succeed");
+ OPENDAL_ASSERT_EQ(strlen(content), read_result.data.len, "Read data length should match");
+ OPENDAL_ASSERT(memcmp(content, read_result.data.data, read_result.data.len) == 0,
+ "Read content should match");
+
+ opendal_bytes_free(&read_result.data);
+ opendal_operator_delete(result.op, path);
+ opendal_operator_free(result.op);
+}
+
+opendal_test_case layer_tests[] = {
+ { "operator_with_retry_layer", test_operator_with_retry_layer, make_capability_read_write() },
+ { "operator_with_timeout_layer", test_operator_with_timeout_layer, make_capability_read_write() },
+};
+
+opendal_test_suite layer_suite = {
+ "Layer",
+ layer_tests,
+ sizeof(layer_tests) / sizeof(layer_tests[0])
+};
(opendal) 02/02: feat(bindings/go): support layers
Posted by yu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yuchanns pushed a commit to branch feat/bindings-go-layer
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 0fcc3210188eebce8f7dca5943c720c9a5233866
Author: Hanchin Hsieh <me...@yuchanns.xyz>
AuthorDate: Sun May 24 23:27:23 2026 +0800
feat(bindings/go): support layers
---
bindings/go/layer.go | 278 +++++++++++++++++++++++
bindings/go/opendal.go | 24 +-
bindings/go/operator.go | 19 +-
bindings/go/tests/behavior_tests/layer_test.go | 84 +++++++
bindings/go/tests/behavior_tests/opendal_test.go | 8 +-
5 files changed, 399 insertions(+), 14 deletions(-)
diff --git a/bindings/go/layer.go b/bindings/go/layer.go
new file mode 100644
index 000000000..4b7110a3b
--- /dev/null
+++ b/bindings/go/layer.go
@@ -0,0 +1,278 @@
+/*
+ * 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.
+ */
+
+package opendal
+
+import (
+ "context"
+ "errors"
+ "math"
+ "time"
+ "unsafe"
+
+ "github.com/jupiterrider/ffi"
+)
+
+// OperatorOption configures operator layers.
+type OperatorOption interface {
+ apply(*operatorConfig) error
+}
+
+type operatorConfig struct {
+ layers []operatorLayer
+}
+
+type operatorOptionFunc func(*operatorConfig) error
+
+func (f operatorOptionFunc) apply(config *operatorConfig) error {
+ return f(config)
+}
+
+type operatorLayer interface {
+ apply(context.Context, *operatorLayers) error
+}
+
+type timeoutLayer struct {
+ timeout time.Duration
+ ioTimeout time.Duration
+}
+
+type retryLayer struct {
+ jitter bool
+ factor float32
+ minDelay time.Duration
+ maxDelay time.Duration
+ maxTimes uint64
+}
+
+type retryConfig struct {
+ jitter bool
+ factor float32
+ minDelay time.Duration
+ maxDelay time.Duration
+ maxTimes uint64
+}
+
+// RetryOption configures the retry layer.
+type RetryOption interface {
+ applyRetry(*retryConfig) error
+}
+
+type retryOptionFunc func(*retryConfig) error
+
+func (f retryOptionFunc) applyRetry(config *retryConfig) error {
+ return f(config)
+}
+
+func defaultRetryConfig() retryConfig {
+ return retryConfig{
+ factor: 2,
+ minDelay: time.Second,
+ maxDelay: time.Minute,
+ maxTimes: 3,
+ }
+}
+
+// WithRetry adds a retry layer to the operator.
+//
+// Layers are applied in the order passed to NewOperator. When combining retry
+// and timeout, pass WithTimeout before WithRetry so each retry attempt has its
+// own timeout.
+func WithRetry(options ...RetryOption) OperatorOption {
+ return operatorOptionFunc(func(config *operatorConfig) error {
+ retry := defaultRetryConfig()
+ for _, option := range options {
+ if option == nil {
+ continue
+ }
+ if err := option.applyRetry(&retry); err != nil {
+ return err
+ }
+ }
+ if retry.maxDelay < retry.minDelay {
+ return errors.New("retry max delay must be greater than or equal to retry min delay")
+ }
+ config.layers = append(config.layers, retryLayer{
+ jitter: retry.jitter,
+ factor: retry.factor,
+ minDelay: retry.minDelay,
+ maxDelay: retry.maxDelay,
+ maxTimes: retry.maxTimes,
+ })
+ return nil
+ })
+}
+
+// RetryMaxTimes sets the retry layer max times.
+func RetryMaxTimes(maxTimes int) RetryOption {
+ return retryOptionFunc(func(config *retryConfig) error {
+ if maxTimes <= 0 {
+ return errors.New("retry max times must be positive")
+ }
+ config.maxTimes = uint64(maxTimes)
+ return nil
+ })
+}
+
+// RetryFactor sets the retry layer backoff factor.
+func RetryFactor(factor float64) RetryOption {
+ return retryOptionFunc(func(config *retryConfig) error {
+ if math.IsNaN(factor) || math.IsInf(factor, 0) || factor < 1 {
+ return errors.New("retry factor must be finite and greater than or equal to 1")
+ }
+ config.factor = float32(factor)
+ return nil
+ })
+}
+
+// RetryJitter enables jitter for the retry layer.
+func RetryJitter() RetryOption {
+ return retryOptionFunc(func(config *retryConfig) error {
+ config.jitter = true
+ return nil
+ })
+}
+
+// RetryMinDelay sets the retry layer min delay.
+func RetryMinDelay(delay time.Duration) RetryOption {
+ return retryOptionFunc(func(config *retryConfig) error {
+ if delay <= 0 {
+ return errors.New("retry min delay must be positive")
+ }
+ config.minDelay = delay
+ return nil
+ })
+}
+
+// RetryMaxDelay sets the retry layer max delay.
+func RetryMaxDelay(delay time.Duration) RetryOption {
+ return retryOptionFunc(func(config *retryConfig) error {
+ if delay <= 0 {
+ return errors.New("retry max delay must be positive")
+ }
+ config.maxDelay = delay
+ return nil
+ })
+}
+
+// WithTimeout adds a timeout layer to the operator.
+//
+// timeout controls non-IO operations, while ioTimeout controls read, write,
+// list, and streaming IO operations.
+func WithTimeout(timeout, ioTimeout time.Duration) OperatorOption {
+ return operatorOptionFunc(func(config *operatorConfig) error {
+ if timeout <= 0 {
+ return errors.New("timeout must be positive")
+ }
+ if ioTimeout <= 0 {
+ return errors.New("io timeout must be positive")
+ }
+ config.layers = append(config.layers, timeoutLayer{
+ timeout: timeout,
+ ioTimeout: ioTimeout,
+ })
+ return nil
+ })
+}
+
+func (l timeoutLayer) apply(ctx context.Context, layers *operatorLayers) error {
+ ffiOperatorLayersAddTimeout.symbol(ctx)(layers, uint64(l.timeout), uint64(l.ioTimeout))
+ return nil
+}
+
+func (l retryLayer) apply(ctx context.Context, layers *operatorLayers) error {
+ jitter := uint8(0)
+ if l.jitter {
+ jitter = 1
+ }
+ ffiOperatorLayersAddRetry.symbol(ctx)(
+ layers,
+ jitter,
+ l.factor,
+ uint64(l.minDelay),
+ uint64(l.maxDelay),
+ l.maxTimes,
+ )
+ return nil
+}
+
+type operatorLayers struct{}
+
+var ffiOperatorLayersNew = newFFI(ffiOpts{
+ sym: "opendal_operator_layers_new",
+ rType: &ffi.TypePointer,
+}, func(_ context.Context, ffiCall ffiCall) func() (layers *operatorLayers) {
+ return func() (layers *operatorLayers) {
+ ffiCall(unsafe.Pointer(&layers))
+ return
+ }
+})
+
+var ffiOperatorLayersAddRetry = newFFI(ffiOpts{
+ sym: "opendal_operator_layers_add_retry",
+ rType: &ffi.TypeVoid,
+ aTypes: []*ffi.Type{
+ &ffi.TypePointer,
+ &ffi.TypeUint8,
+ &ffi.TypeFloat,
+ &ffi.TypeUint64,
+ &ffi.TypeUint64,
+ &ffi.TypeUint64,
+ },
+}, func(_ context.Context, ffiCall ffiCall) func(layers *operatorLayers, jitter uint8, factor float32, minDelayNS, maxDelayNS, maxTimes uint64) {
+ return func(layers *operatorLayers, jitter uint8, factor float32, minDelayNS, maxDelayNS, maxTimes uint64) {
+ ffiCall(
+ nil,
+ unsafe.Pointer(&layers),
+ unsafe.Pointer(&jitter),
+ unsafe.Pointer(&factor),
+ unsafe.Pointer(&minDelayNS),
+ unsafe.Pointer(&maxDelayNS),
+ unsafe.Pointer(&maxTimes),
+ )
+ }
+})
+
+var ffiOperatorLayersAddTimeout = newFFI(ffiOpts{
+ sym: "opendal_operator_layers_add_timeout",
+ rType: &ffi.TypeVoid,
+ aTypes: []*ffi.Type{&ffi.TypePointer, &ffi.TypeUint64, &ffi.TypeUint64},
+}, func(_ context.Context, ffiCall ffiCall) func(layers *operatorLayers, timeoutNS, ioTimeoutNS uint64) {
+ return func(layers *operatorLayers, timeoutNS, ioTimeoutNS uint64) {
+ ffiCall(
+ nil,
+ unsafe.Pointer(&layers),
+ unsafe.Pointer(&timeoutNS),
+ unsafe.Pointer(&ioTimeoutNS),
+ )
+ }
+})
+
+var ffiOperatorLayersFree = newFFI(ffiOpts{
+ sym: "opendal_operator_layers_free",
+ rType: &ffi.TypeVoid,
+ aTypes: []*ffi.Type{&ffi.TypePointer},
+}, func(_ context.Context, ffiCall ffiCall) func(layers *operatorLayers) {
+ return func(layers *operatorLayers) {
+ ffiCall(
+ nil,
+ unsafe.Pointer(&layers),
+ )
+ }
+})
diff --git a/bindings/go/opendal.go b/bindings/go/opendal.go
index a1d72a5fc..d550faee7 100644
--- a/bindings/go/opendal.go
+++ b/bindings/go/opendal.go
@@ -165,7 +165,7 @@ type Operator struct {
// - error: An error if initialization fails, or nil if successful.
//
// Note: Remember to call Close() on the returned Operator when it's no longer needed.
-func NewOperator(scheme Scheme, opts OperatorOptions) (op *Operator, err error) {
+func NewOperator(scheme Scheme, opts OperatorOptions, operatorOptions ...OperatorOption) (op *Operator, err error) {
err = scheme.LoadOnce()
if err != nil {
return
@@ -195,7 +195,27 @@ func NewOperator(scheme Scheme, opts OperatorOptions) (op *Operator, err error)
}
}
- inner, err := ffiOperatorNew.symbol(ctx)(scheme, options)
+ config := &operatorConfig{}
+ for _, operatorOption := range operatorOptions {
+ if operatorOption == nil {
+ continue
+ }
+ err = operatorOption.apply(config)
+ if err != nil {
+ return
+ }
+ }
+
+ var inner *opendalOperator
+ layers := ffiOperatorLayersNew.symbol(ctx)()
+ defer ffiOperatorLayersFree.symbol(ctx)(layers)
+ for _, layer := range config.layers {
+ err = layer.apply(ctx, layers)
+ if err != nil {
+ return
+ }
+ }
+ inner, err = ffiOperatorNewWithLayers.symbol(ctx)(scheme, options, layers)
if err != nil {
return
}
diff --git a/bindings/go/operator.go b/bindings/go/operator.go
index ec5e65f09..792b0d953 100644
--- a/bindings/go/operator.go
+++ b/bindings/go/operator.go
@@ -98,16 +98,18 @@ func (op *Operator) Rename(src, dest string) error {
return ffiOperatorRename.symbol(op.ctx)(op.inner, src, dest)
}
-var ffiOperatorNew = newFFI(ffiOpts{
- sym: "opendal_operator_new",
+func normalizeScheme(scheme Scheme) (*byte, error) {
+ return BytePtrFromString(strings.ReplaceAll(scheme.Name(), "_", "-"))
+}
+
+var ffiOperatorNewWithLayers = newFFI(ffiOpts{
+ sym: "opendal_operator_new_with_layers",
rType: &typeResultOperatorNew,
- aTypes: []*ffi.Type{&ffi.TypePointer, &ffi.TypePointer},
-}, func(ctx context.Context, ffiCall ffiCall) func(scheme Scheme, opts *operatorOptions) (op *opendalOperator, err error) {
- return func(scheme Scheme, opts *operatorOptions) (op *opendalOperator, err error) {
- // This is a temporary fix; it can be removed once we fix the template generation code in opendal-go-services.
- normalizedSchemeName := strings.ReplaceAll(scheme.Name(), "_", "-")
+ aTypes: []*ffi.Type{&ffi.TypePointer, &ffi.TypePointer, &ffi.TypePointer},
+}, func(ctx context.Context, ffiCall ffiCall) func(scheme Scheme, opts *operatorOptions, layers *operatorLayers) (op *opendalOperator, err error) {
+ return func(scheme Scheme, opts *operatorOptions, layers *operatorLayers) (op *opendalOperator, err error) {
var byteName *byte
- byteName, err = BytePtrFromString(normalizedSchemeName)
+ byteName, err = normalizeScheme(scheme)
if err != nil {
return nil, err
}
@@ -116,6 +118,7 @@ var ffiOperatorNew = newFFI(ffiOpts{
unsafe.Pointer(&result),
unsafe.Pointer(&byteName),
unsafe.Pointer(&opts),
+ unsafe.Pointer(&layers),
)
if result.error != nil {
err = parseError(ctx, result.error)
diff --git a/bindings/go/tests/behavior_tests/layer_test.go b/bindings/go/tests/behavior_tests/layer_test.go
new file mode 100644
index 000000000..881f3f787
--- /dev/null
+++ b/bindings/go/tests/behavior_tests/layer_test.go
@@ -0,0 +1,84 @@
+/*
+ * 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.
+ */
+
+package opendal_test
+
+import (
+ "time"
+
+ "github.com/apache/opendal/bindings/go"
+ "github.com/google/uuid"
+ "github.com/stretchr/testify/require"
+)
+
+func testsLayer(cap *opendal.Capability) []behaviorTest {
+ if !cap.Read() || !cap.Write() {
+ return nil
+ }
+ return []behaviorTest{
+ testOperatorWithRetryLayer,
+ testOperatorWithTimeoutLayer,
+ }
+}
+
+func testOperatorWithRetryLayer(assert *require.Assertions, _ *opendal.Operator, _ *fixture) {
+ layeredOp, closeFunc, err := newOperator(opendal.WithRetry(
+ opendal.RetryMaxTimes(3),
+ opendal.RetryFactor(2),
+ opendal.RetryJitter(),
+ opendal.RetryMinDelay(time.Second),
+ opendal.RetryMaxDelay(time.Minute),
+ ))
+ assert.NoError(err)
+ defer closeFunc()
+
+ assert.NotNil(layeredOp.Info())
+
+ path := uuid.NewString()
+ content := []byte("Hello, OpenDAL retry layer!")
+
+ assert.NoError(layeredOp.Write(path, content))
+ defer func() {
+ _ = layeredOp.Delete(path)
+ }()
+
+ got, err := layeredOp.Read(path)
+ assert.NoError(err)
+ assert.Equal(content, got)
+}
+
+func testOperatorWithTimeoutLayer(assert *require.Assertions, _ *opendal.Operator, _ *fixture) {
+ layeredOp, closeFunc, err := newOperator(opendal.WithTimeout(time.Minute, 10*time.Second))
+ assert.NoError(err)
+ defer closeFunc()
+
+ assert.NotNil(layeredOp.Info())
+
+ path := uuid.NewString()
+ content := []byte("Hello, OpenDAL timeout layer!")
+
+ assert.NoError(layeredOp.Write(path, content))
+ defer func() {
+ _ = layeredOp.Delete(path)
+ }()
+
+ got, err := layeredOp.Read(path)
+ assert.NoError(err)
+ assert.Equal(content, got)
+}
diff --git a/bindings/go/tests/behavior_tests/opendal_test.go b/bindings/go/tests/behavior_tests/opendal_test.go
index 9b1295c84..d2b226992 100644
--- a/bindings/go/tests/behavior_tests/opendal_test.go
+++ b/bindings/go/tests/behavior_tests/opendal_test.go
@@ -44,7 +44,7 @@ func TestMain(m *testing.M) {
closeFunc func()
err error
)
- op, closeFunc, err = newOperator()
+ op, closeFunc, err = newOperator(opendal.WithRetry())
if err != nil {
panic(err)
}
@@ -66,6 +66,7 @@ func TestBehavior(t *testing.T) {
tests = append(tests, testsCreateDir(cap)...)
tests = append(tests, testsDelete(cap)...)
tests = append(tests, testsList(cap)...)
+ tests = append(tests, testsLayer(cap)...)
tests = append(tests, testsRead(cap)...)
tests = append(tests, testsPresign(cap)...)
tests = append(tests, testsRename(cap)...)
@@ -96,11 +97,10 @@ func TestBehavior(t *testing.T) {
}
}
-func newOperator() (op *opendal.Operator, closeFunc func(), err error) {
+func newOperator(operatorOptions ...opendal.OperatorOption) (op *opendal.Operator, closeFunc func(), err error) {
test := os.Getenv("OPENDAL_TEST")
var scheme opendal.Scheme
for _, s := range schemes {
- // This is a temporary fix; it can be removed once we fix the template generation code in opendal-go-services.
normalizedSchemeName := strings.ReplaceAll(test, "_", "-")
if s.Name() != test && s.Name() != normalizedSchemeName {
continue
@@ -133,7 +133,7 @@ func newOperator() (op *opendal.Operator, closeFunc func(), err error) {
opts[strings.ToLower(strings.TrimPrefix(key, prefix))] = value
}
- op, err = opendal.NewOperator(scheme, opts)
+ op, err = opendal.NewOperator(scheme, opts, operatorOptions...)
if err != nil {
err = fmt.Errorf("create operator must succeed: %s", err)
}