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/06/17 10:40:30 UTC
(opendal) 01/01: refactor(object_store): Polish implementation details of object_store
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch object_store
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 432764380954fa619708cf21c6af43e2a45abafe
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Mon Jun 17 18:40:00 2024 +0800
refactor(object_store): Polish implementation details of object_store
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
integrations/object_store/src/lib.rs | 540 +++------------------
integrations/object_store/src/{lib.rs => store.rs} | 210 ++++----
.../object_store/src/{send_wrapper.rs => utils.rs} | 131 +++--
3 files changed, 255 insertions(+), 626 deletions(-)
diff --git a/integrations/object_store/src/lib.rs b/integrations/object_store/src/lib.rs
index 6c18102e9e..df72863ae9 100644
--- a/integrations/object_store/src/lib.rs
+++ b/integrations/object_store/src/lib.rs
@@ -15,373 +15,60 @@
// specific language governing permissions and limitations
// under the License.
-mod send_wrapper;
-
-use std::future::IntoFuture;
-use std::ops::Range;
-
-use async_trait::async_trait;
-use bytes::Bytes;
-use futures::stream::BoxStream;
-use futures::FutureExt;
-use futures::StreamExt;
-use futures::TryStreamExt;
-use object_store::path::Path;
-use object_store::GetOptions;
-use object_store::GetResult;
-use object_store::GetResultPayload;
-use object_store::ListResult;
-use object_store::MultipartUpload;
-use object_store::ObjectMeta;
-use object_store::ObjectStore;
-use object_store::PutMultipartOpts;
-use object_store::PutOptions;
-use object_store::PutPayload;
-use object_store::PutResult;
-use object_store::Result;
-use opendal::Entry;
-use opendal::Metadata;
-use opendal::Metakey;
-use opendal::Operator;
-use send_wrapper::IntoSendFuture;
-use send_wrapper::IntoSendStream;
-
-#[derive(Debug)]
-pub struct OpendalStore {
- inner: Operator,
-}
-
-impl OpendalStore {
- /// Create OpendalStore by given Operator.
- pub fn new(op: Operator) -> Self {
- Self { inner: op }
- }
-}
-
-impl std::fmt::Display for OpendalStore {
- fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
- write!(f, "OpenDAL({:?})", self.inner)
- }
-}
-
-#[async_trait]
-impl ObjectStore for OpendalStore {
- async fn put(&self, location: &Path, bytes: PutPayload) -> Result<PutResult> {
- let bytes: Bytes = bytes.into();
- self.inner
- .write(location.as_ref(), bytes)
- .into_send()
- .await
- .map_err(|err| format_object_store_error(err, location.as_ref()))?;
- Ok(PutResult {
- e_tag: None,
- version: None,
- })
- }
-
- async fn put_opts(
- &self,
- _location: &Path,
- _bytes: PutPayload,
- _opts: PutOptions,
- ) -> Result<PutResult> {
- Err(object_store::Error::NotSupported {
- source: Box::new(opendal::Error::new(
- opendal::ErrorKind::Unsupported,
- "put_opts is not implemented so far",
- )),
- })
- }
-
- async fn put_multipart(&self, _location: &Path) -> Result<Box<dyn MultipartUpload>> {
- Err(object_store::Error::NotSupported {
- source: Box::new(opendal::Error::new(
- opendal::ErrorKind::Unsupported,
- "put_multipart is not implemented so far",
- )),
- })
- }
-
- async fn put_multipart_opts(
- &self,
- _location: &Path,
- _opts: PutMultipartOpts,
- ) -> Result<Box<dyn MultipartUpload>> {
- Err(object_store::Error::NotSupported {
- source: Box::new(opendal::Error::new(
- opendal::ErrorKind::Unsupported,
- "put_multipart_opts is not implemented so far",
- )),
- })
- }
-
- async fn get(&self, location: &Path) -> Result<GetResult> {
- let meta = self
- .inner
- .stat(location.as_ref())
- .into_send()
- .await
- .map_err(|err| format_object_store_error(err, location.as_ref()))?;
-
- let meta = ObjectMeta {
- location: location.clone(),
- last_modified: meta.last_modified().unwrap_or_default(),
- size: meta.content_length() as usize,
- e_tag: meta.etag().map(|x| x.to_string()),
- version: meta.version().map(|x| x.to_string()),
- };
- let r = self
- .inner
- .reader(location.as_ref())
- .into_send()
- .await
- .map_err(|err| format_object_store_error(err, location.as_ref()))?;
-
- let stream = r
- .into_bytes_stream(0..meta.size as u64)
- .await
- .map_err(|err| object_store::Error::Generic {
- store: "IoError",
- source: Box::new(err),
- })?
- .into_send()
- .map_err(|err| object_store::Error::Generic {
- store: "IoError",
- source: Box::new(err),
- });
-
- Ok(GetResult {
- payload: GetResultPayload::Stream(Box::pin(stream)),
- range: 0..meta.size,
- meta,
- attributes: Default::default(),
- })
- }
-
- async fn get_opts(&self, _location: &Path, _options: GetOptions) -> Result<GetResult> {
- Err(object_store::Error::NotSupported {
- source: Box::new(opendal::Error::new(
- opendal::ErrorKind::Unsupported,
- "get_opts is not implemented so far",
- )),
- })
- }
-
- async fn get_range(&self, location: &Path, range: Range<usize>) -> Result<Bytes> {
- let bs = self
- .inner
- .read_with(location.as_ref())
- .range(range.start as u64..range.end as u64)
- .into_future()
- .into_send()
- .await
- .map_err(|err| format_object_store_error(err, location.as_ref()))?;
-
- Ok(bs.to_bytes())
- }
-
- async fn head(&self, location: &Path) -> Result<ObjectMeta> {
- let meta = self
- .inner
- .stat(location.as_ref())
- .into_send()
- .await
- .map_err(|err| format_object_store_error(err, location.as_ref()))?;
-
- Ok(ObjectMeta {
- location: location.clone(),
- last_modified: meta.last_modified().unwrap_or_default(),
- size: meta.content_length() as usize,
- e_tag: meta.etag().map(|x| x.to_string()),
- version: meta.version().map(|x| x.to_string()),
- })
- }
-
- async fn delete(&self, location: &Path) -> Result<()> {
- self.inner
- .delete(location.as_ref())
- .into_send()
- .await
- .map_err(|err| format_object_store_error(err, location.as_ref()))?;
-
- Ok(())
- }
-
- fn list(&self, prefix: Option<&Path>) -> BoxStream<'_, Result<ObjectMeta>> {
- // object_store `Path` always removes trailing slash
- // need to add it back
- let path = prefix.map_or("".into(), |x| format!("{}/", x));
-
- let fut = async move {
- let stream = self
- .inner
- .lister_with(&path)
- .metakey(Metakey::ContentLength | Metakey::LastModified)
- .recursive(true)
- .await
- .map_err(|err| format_object_store_error(err, &path))?;
-
- let stream = stream.then(|res| async {
- let entry = res.map_err(|err| format_object_store_error(err, ""))?;
- let meta = entry.metadata();
-
- Ok(format_object_meta(entry.path(), meta))
- });
- Ok::<_, object_store::Error>(stream)
- };
-
- fut.into_stream().try_flatten().into_send().boxed()
- }
-
- fn list_with_offset(
- &self,
- prefix: Option<&Path>,
- offset: &Path,
- ) -> BoxStream<'_, Result<ObjectMeta>> {
- let path = prefix.map_or("".into(), |x| format!("{}/", x));
- let offset = offset.clone();
-
- let fut = async move {
- let fut = if self.inner.info().full_capability().list_with_start_after {
- self.inner
- .lister_with(&path)
- .start_after(offset.as_ref())
- .metakey(Metakey::ContentLength | Metakey::LastModified)
- .recursive(true)
- .into_future()
- .into_send()
- .await
- .map_err(|err| format_object_store_error(err, &path))?
- .then(try_format_object_meta)
- .into_send()
- .boxed()
- } else {
- self.inner
- .lister_with(&path)
- .metakey(Metakey::ContentLength | Metakey::LastModified)
- .recursive(true)
- .into_future()
- .into_send()
- .await
- .map_err(|err| format_object_store_error(err, &path))?
- .try_filter(move |entry| futures::future::ready(entry.path() > offset.as_ref()))
- .then(try_format_object_meta)
- .into_send()
- .boxed()
- };
- Ok::<_, object_store::Error>(fut)
- };
-
- fut.into_stream().into_send().try_flatten().boxed()
- }
-
- async fn list_with_delimiter(&self, prefix: Option<&Path>) -> Result<ListResult> {
- let path = prefix.map_or("".into(), |x| format!("{}/", x));
- let mut stream = self
- .inner
- .lister_with(&path)
- .metakey(Metakey::Mode | Metakey::ContentLength | Metakey::LastModified)
- .into_future()
- .into_send()
- .await
- .map_err(|err| format_object_store_error(err, &path))?
- .into_send();
-
- let mut common_prefixes = Vec::new();
- let mut objects = Vec::new();
-
- while let Some(res) = stream.next().into_send().await {
- let entry = res.map_err(|err| format_object_store_error(err, ""))?;
- let meta = entry.metadata();
-
- if meta.is_dir() {
- common_prefixes.push(entry.path().into());
- } else {
- objects.push(format_object_meta(entry.path(), meta));
- }
- }
-
- Ok(ListResult {
- common_prefixes,
- objects,
- })
- }
-
- async fn copy(&self, _from: &Path, _to: &Path) -> Result<()> {
- Err(object_store::Error::NotSupported {
- source: Box::new(opendal::Error::new(
- opendal::ErrorKind::Unsupported,
- "copy is not implemented so far",
- )),
- })
- }
-
- async fn rename(&self, _from: &Path, _to: &Path) -> Result<()> {
- Err(object_store::Error::NotSupported {
- source: Box::new(opendal::Error::new(
- opendal::ErrorKind::Unsupported,
- "rename is not implemented so far",
- )),
- })
- }
-
- async fn copy_if_not_exists(&self, _from: &Path, _to: &Path) -> Result<()> {
- Err(object_store::Error::NotSupported {
- source: Box::new(opendal::Error::new(
- opendal::ErrorKind::Unsupported,
- "copy_if_not_exists is not implemented so far",
- )),
- })
- }
-}
-
-fn format_object_store_error(err: opendal::Error, path: &str) -> object_store::Error {
- use opendal::ErrorKind;
- match err.kind() {
- ErrorKind::NotFound => object_store::Error::NotFound {
- path: path.to_string(),
- source: Box::new(err),
- },
- ErrorKind::Unsupported => object_store::Error::NotSupported {
- source: Box::new(err),
- },
- ErrorKind::AlreadyExists => object_store::Error::AlreadyExists {
- path: path.to_string(),
- source: Box::new(err),
- },
- kind => object_store::Error::Generic {
- store: kind.into_static(),
- source: Box::new(err),
- },
- }
-}
-
-fn format_object_meta(path: &str, meta: &Metadata) -> ObjectMeta {
- let version = match meta.metakey().contains(Metakey::Version) {
- true => meta.version().map(|x| x.to_string()),
- false => None,
- };
-
- let e_tag = match meta.metakey().contains(Metakey::Etag) {
- true => meta.etag().map(|x| x.to_string()),
- false => None,
- };
-
- ObjectMeta {
- location: path.into(),
- last_modified: meta.last_modified().unwrap_or_default(),
- size: meta.content_length() as usize,
- e_tag,
- version,
- }
-}
-
-async fn try_format_object_meta(res: Result<Entry, opendal::Error>) -> Result<ObjectMeta> {
- let entry = res.map_err(|err| format_object_store_error(err, ""))?;
- let meta = entry.metadata();
-
- Ok(format_object_meta(entry.path(), meta))
-}
+//! object_store_opendal is an object store implementation using opendal.
+//!
+//! This crate can help you to access 30 more storage services with the same API.
+//!
+//! ```no_run
+//! use std::sync::Arc;
+//!
+//! use bytes::Bytes;
+//! use object_store::path::Path;
+//! use object_store::ObjectStore;
+//! use object_store_opendal::OpendalStore;
+//! use opendal::services::S3;
+//! use opendal::{Builder, Operator};
+//!
+//! #[tokio::main]
+//! async fn main() {
+//! let builder = S3::from_map(
+//! vec![
+//! ("access_key".to_string(), "my_access_key".to_string()),
+//! ("secret_key".to_string(), "my_secret_key".to_string()),
+//! ("endpoint".to_string(), "my_endpoint".to_string()),
+//! ("region".to_string(), "my_region".to_string()),
+//! ]
+//! .into_iter()
+//! .collect(),
+//! );
+//!
+//! // Create a new operator
+//! let operator = Operator::new(builder).unwrap().finish();
+//!
+//! // Create a new object store
+//! let object_store = Arc::new(OpendalStore::new(operator));
+//!
+//! let path = Path::from("data/nested/test.txt");
+//! let bytes = Bytes::from_static(b"hello, world! I am nested.");
+//!
+//! object_store.put(&path, bytes.clone().into()).await.unwrap();
+//!
+//! let content = object_store
+//! .get(&path)
+//! .await
+//! .unwrap()
+//! .bytes()
+//! .await
+//! .unwrap();
+//!
+//! assert_eq!(content, bytes);
+//! }
+//! ```
+
+mod store;
+pub use store::OpendalStore;
+
+mod utils;
// Make sure `send_wrapper` works as expected
#[cfg(all(feature = "send_wrapper", target_arch = "wasm32"))]
@@ -407,122 +94,3 @@ mod assert_send {
assert_send(store.list_with_delimiter(None));
}
}
-
-#[cfg(test)]
-mod tests {
- use std::sync::Arc;
-
- use object_store::path::Path;
- use object_store::ObjectStore;
- use opendal::services;
-
- use super::*;
-
- async fn create_test_object_store() -> Arc<dyn ObjectStore> {
- let op = Operator::new(services::Memory::default()).unwrap().finish();
- let object_store = Arc::new(OpendalStore::new(op));
-
- let path: Path = "data/test.txt".into();
- let bytes = Bytes::from_static(b"hello, world!");
- object_store.put(&path, bytes.into()).await.unwrap();
-
- let path: Path = "data/nested/test.txt".into();
- let bytes = Bytes::from_static(b"hello, world! I am nested.");
- object_store.put(&path, bytes.into()).await.unwrap();
-
- object_store
- }
-
- #[tokio::test]
- async fn test_basic() {
- let op = Operator::new(services::Memory::default()).unwrap().finish();
- let object_store: Arc<dyn ObjectStore> = Arc::new(OpendalStore::new(op));
-
- // Retrieve a specific file
- let path: Path = "data/test.txt".into();
-
- let bytes = Bytes::from_static(b"hello, world!");
- object_store.put(&path, bytes.clone().into()).await.unwrap();
-
- let meta = object_store.head(&path).await.unwrap();
-
- assert_eq!(meta.size, 13);
-
- assert_eq!(
- object_store
- .get(&path)
- .await
- .unwrap()
- .bytes()
- .await
- .unwrap(),
- bytes
- );
- }
-
- #[tokio::test]
- async fn test_list() {
- let object_store = create_test_object_store().await;
- let path: Path = "data/".into();
- let results = object_store.list(Some(&path)).collect::<Vec<_>>().await;
- assert_eq!(results.len(), 2);
- let mut locations = results
- .iter()
- .map(|x| x.as_ref().unwrap().location.as_ref())
- .collect::<Vec<_>>();
-
- let expected_files = vec![
- (
- "data/nested/test.txt",
- Bytes::from_static(b"hello, world! I am nested."),
- ),
- ("data/test.txt", Bytes::from_static(b"hello, world!")),
- ];
-
- let expected_locations = expected_files.iter().map(|x| x.0).collect::<Vec<&str>>();
-
- locations.sort();
- assert_eq!(locations, expected_locations);
-
- for (location, bytes) in expected_files {
- let path: Path = location.into();
- assert_eq!(
- object_store
- .get(&path)
- .await
- .unwrap()
- .bytes()
- .await
- .unwrap(),
- bytes
- );
- }
- }
-
- #[tokio::test]
- async fn test_list_with_delimiter() {
- let object_store = create_test_object_store().await;
- let path: Path = "data/".into();
- let result = object_store.list_with_delimiter(Some(&path)).await.unwrap();
- assert_eq!(result.objects.len(), 1);
- assert_eq!(result.common_prefixes.len(), 1);
- assert_eq!(result.objects[0].location.as_ref(), "data/test.txt");
- assert_eq!(result.common_prefixes[0].as_ref(), "data/nested");
- }
-
- #[tokio::test]
- async fn test_list_with_offset() {
- let object_store = create_test_object_store().await;
- let path: Path = "data/".into();
- let offset: Path = "data/nested/test.txt".into();
- let result = object_store
- .list_with_offset(Some(&path), &offset)
- .collect::<Vec<_>>()
- .await;
- assert_eq!(result.len(), 1);
- assert_eq!(
- result[0].as_ref().unwrap().location.as_ref(),
- "data/test.txt"
- );
- }
-}
diff --git a/integrations/object_store/src/lib.rs b/integrations/object_store/src/store.rs
similarity index 78%
copy from integrations/object_store/src/lib.rs
copy to integrations/object_store/src/store.rs
index 6c18102e9e..cbfc1b9845 100644
--- a/integrations/object_store/src/lib.rs
+++ b/integrations/object_store/src/store.rs
@@ -15,11 +15,11 @@
// specific language governing permissions and limitations
// under the License.
-mod send_wrapper;
-
+use std::fmt::{Debug, Display, Formatter};
use std::future::IntoFuture;
use std::ops::Range;
+use crate::utils::*;
use async_trait::async_trait;
use bytes::Bytes;
use futures::stream::BoxStream;
@@ -38,15 +38,60 @@ use object_store::PutMultipartOpts;
use object_store::PutOptions;
use object_store::PutPayload;
use object_store::PutResult;
-use object_store::Result;
-use opendal::Entry;
-use opendal::Metadata;
-use opendal::Metakey;
use opendal::Operator;
-use send_wrapper::IntoSendFuture;
-use send_wrapper::IntoSendStream;
-
-#[derive(Debug)]
+use opendal::{Buffer, Metakey};
+
+/// OpendalStore implements ObjectStore trait by using opendal.
+///
+/// This allows users to use opendal as an object store without extra cost.
+///
+/// Visit [`opendal::services`] for more information about supported services.
+///
+/// ```no_run
+/// use std::sync::Arc;
+///
+/// use bytes::Bytes;
+/// use object_store::path::Path;
+/// use object_store::ObjectStore;
+/// use object_store_opendal::OpendalStore;
+/// use opendal::services::S3;
+/// use opendal::{Builder, Operator};
+///
+/// #[tokio::main]
+/// async fn main() {
+/// let builder = S3::from_map(
+/// vec![
+/// ("access_key".to_string(), "my_access_key".to_string()),
+/// ("secret_key".to_string(), "my_secret_key".to_string()),
+/// ("endpoint".to_string(), "my_endpoint".to_string()),
+/// ("region".to_string(), "my_region".to_string()),
+/// ]
+/// .into_iter()
+/// .collect(),
+/// );
+///
+/// // Create a new operator
+/// let operator = Operator::new(builder).unwrap().finish();
+///
+/// // Create a new object store
+/// let object_store = Arc::new(OpendalStore::new(operator));
+///
+/// let path = Path::from("data/nested/test.txt");
+/// let bytes = Bytes::from_static(b"hello, world! I am nested.");
+///
+/// object_store.put(&path, bytes.clone().into()).await.unwrap();
+///
+/// let content = object_store
+/// .get(&path)
+/// .await
+/// .unwrap()
+/// .bytes()
+/// .await
+/// .unwrap();
+///
+/// assert_eq!(content, bytes);
+/// }
+/// ```
pub struct OpendalStore {
inner: Operator,
}
@@ -58,18 +103,42 @@ impl OpendalStore {
}
}
-impl std::fmt::Display for OpendalStore {
- fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
- write!(f, "OpenDAL({:?})", self.inner)
+impl Debug for OpendalStore {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let info = self.inner.info();
+ f.debug_struct("OpendalStore")
+ .field("scheme", &info.scheme())
+ .field("name", &info.name())
+ .field("root", &info.root())
+ .field("capability", &info.full_capability())
+ .finish()
+ }
+}
+
+impl Display for OpendalStore {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let info = self.inner.info();
+ write!(
+ f,
+ "Opendal({}, bucket={}, root={})",
+ info.scheme(),
+ info.name(),
+ info.root()
+ )
+ }
+}
+
+impl From<Operator> for OpendalStore {
+ fn from(value: Operator) -> Self {
+ Self::new(value)
}
}
#[async_trait]
impl ObjectStore for OpendalStore {
- async fn put(&self, location: &Path, bytes: PutPayload) -> Result<PutResult> {
- let bytes: Bytes = bytes.into();
+ async fn put(&self, location: &Path, bytes: PutPayload) -> object_store::Result<PutResult> {
self.inner
- .write(location.as_ref(), bytes)
+ .write(location.as_ref(), Buffer::from_iter(bytes.into_iter()))
.into_send()
.await
.map_err(|err| format_object_store_error(err, location.as_ref()))?;
@@ -84,7 +153,7 @@ impl ObjectStore for OpendalStore {
_location: &Path,
_bytes: PutPayload,
_opts: PutOptions,
- ) -> Result<PutResult> {
+ ) -> object_store::Result<PutResult> {
Err(object_store::Error::NotSupported {
source: Box::new(opendal::Error::new(
opendal::ErrorKind::Unsupported,
@@ -93,7 +162,10 @@ impl ObjectStore for OpendalStore {
})
}
- async fn put_multipart(&self, _location: &Path) -> Result<Box<dyn MultipartUpload>> {
+ async fn put_multipart(
+ &self,
+ _location: &Path,
+ ) -> object_store::Result<Box<dyn MultipartUpload>> {
Err(object_store::Error::NotSupported {
source: Box::new(opendal::Error::new(
opendal::ErrorKind::Unsupported,
@@ -106,7 +178,7 @@ impl ObjectStore for OpendalStore {
&self,
_location: &Path,
_opts: PutMultipartOpts,
- ) -> Result<Box<dyn MultipartUpload>> {
+ ) -> object_store::Result<Box<dyn MultipartUpload>> {
Err(object_store::Error::NotSupported {
source: Box::new(opendal::Error::new(
opendal::ErrorKind::Unsupported,
@@ -115,7 +187,7 @@ impl ObjectStore for OpendalStore {
})
}
- async fn get(&self, location: &Path) -> Result<GetResult> {
+ async fn get(&self, location: &Path) -> object_store::Result<GetResult> {
let meta = self
.inner
.stat(location.as_ref())
@@ -130,6 +202,7 @@ impl ObjectStore for OpendalStore {
e_tag: meta.etag().map(|x| x.to_string()),
version: meta.version().map(|x| x.to_string()),
};
+
let r = self
.inner
.reader(location.as_ref())
@@ -158,7 +231,11 @@ impl ObjectStore for OpendalStore {
})
}
- async fn get_opts(&self, _location: &Path, _options: GetOptions) -> Result<GetResult> {
+ async fn get_opts(
+ &self,
+ _location: &Path,
+ _options: GetOptions,
+ ) -> object_store::Result<GetResult> {
Err(object_store::Error::NotSupported {
source: Box::new(opendal::Error::new(
opendal::ErrorKind::Unsupported,
@@ -167,7 +244,7 @@ impl ObjectStore for OpendalStore {
})
}
- async fn get_range(&self, location: &Path, range: Range<usize>) -> Result<Bytes> {
+ async fn get_range(&self, location: &Path, range: Range<usize>) -> object_store::Result<Bytes> {
let bs = self
.inner
.read_with(location.as_ref())
@@ -180,7 +257,7 @@ impl ObjectStore for OpendalStore {
Ok(bs.to_bytes())
}
- async fn head(&self, location: &Path) -> Result<ObjectMeta> {
+ async fn head(&self, location: &Path) -> object_store::Result<ObjectMeta> {
let meta = self
.inner
.stat(location.as_ref())
@@ -197,7 +274,7 @@ impl ObjectStore for OpendalStore {
})
}
- async fn delete(&self, location: &Path) -> Result<()> {
+ async fn delete(&self, location: &Path) -> object_store::Result<()> {
self.inner
.delete(location.as_ref())
.into_send()
@@ -207,7 +284,7 @@ impl ObjectStore for OpendalStore {
Ok(())
}
- fn list(&self, prefix: Option<&Path>) -> BoxStream<'_, Result<ObjectMeta>> {
+ fn list(&self, prefix: Option<&Path>) -> BoxStream<'_, object_store::Result<ObjectMeta>> {
// object_store `Path` always removes trailing slash
// need to add it back
let path = prefix.map_or("".into(), |x| format!("{}/", x));
@@ -237,7 +314,7 @@ impl ObjectStore for OpendalStore {
&self,
prefix: Option<&Path>,
offset: &Path,
- ) -> BoxStream<'_, Result<ObjectMeta>> {
+ ) -> BoxStream<'_, object_store::Result<ObjectMeta>> {
let path = prefix.map_or("".into(), |x| format!("{}/", x));
let offset = offset.clone();
@@ -275,7 +352,7 @@ impl ObjectStore for OpendalStore {
fut.into_stream().into_send().try_flatten().boxed()
}
- async fn list_with_delimiter(&self, prefix: Option<&Path>) -> Result<ListResult> {
+ async fn list_with_delimiter(&self, prefix: Option<&Path>) -> object_store::Result<ListResult> {
let path = prefix.map_or("".into(), |x| format!("{}/", x));
let mut stream = self
.inner
@@ -307,7 +384,7 @@ impl ObjectStore for OpendalStore {
})
}
- async fn copy(&self, _from: &Path, _to: &Path) -> Result<()> {
+ async fn copy(&self, _from: &Path, _to: &Path) -> object_store::Result<()> {
Err(object_store::Error::NotSupported {
source: Box::new(opendal::Error::new(
opendal::ErrorKind::Unsupported,
@@ -316,7 +393,7 @@ impl ObjectStore for OpendalStore {
})
}
- async fn rename(&self, _from: &Path, _to: &Path) -> Result<()> {
+ async fn rename(&self, _from: &Path, _to: &Path) -> object_store::Result<()> {
Err(object_store::Error::NotSupported {
source: Box::new(opendal::Error::new(
opendal::ErrorKind::Unsupported,
@@ -325,7 +402,7 @@ impl ObjectStore for OpendalStore {
})
}
- async fn copy_if_not_exists(&self, _from: &Path, _to: &Path) -> Result<()> {
+ async fn copy_if_not_exists(&self, _from: &Path, _to: &Path) -> object_store::Result<()> {
Err(object_store::Error::NotSupported {
source: Box::new(opendal::Error::new(
opendal::ErrorKind::Unsupported,
@@ -335,79 +412,6 @@ impl ObjectStore for OpendalStore {
}
}
-fn format_object_store_error(err: opendal::Error, path: &str) -> object_store::Error {
- use opendal::ErrorKind;
- match err.kind() {
- ErrorKind::NotFound => object_store::Error::NotFound {
- path: path.to_string(),
- source: Box::new(err),
- },
- ErrorKind::Unsupported => object_store::Error::NotSupported {
- source: Box::new(err),
- },
- ErrorKind::AlreadyExists => object_store::Error::AlreadyExists {
- path: path.to_string(),
- source: Box::new(err),
- },
- kind => object_store::Error::Generic {
- store: kind.into_static(),
- source: Box::new(err),
- },
- }
-}
-
-fn format_object_meta(path: &str, meta: &Metadata) -> ObjectMeta {
- let version = match meta.metakey().contains(Metakey::Version) {
- true => meta.version().map(|x| x.to_string()),
- false => None,
- };
-
- let e_tag = match meta.metakey().contains(Metakey::Etag) {
- true => meta.etag().map(|x| x.to_string()),
- false => None,
- };
-
- ObjectMeta {
- location: path.into(),
- last_modified: meta.last_modified().unwrap_or_default(),
- size: meta.content_length() as usize,
- e_tag,
- version,
- }
-}
-
-async fn try_format_object_meta(res: Result<Entry, opendal::Error>) -> Result<ObjectMeta> {
- let entry = res.map_err(|err| format_object_store_error(err, ""))?;
- let meta = entry.metadata();
-
- Ok(format_object_meta(entry.path(), meta))
-}
-
-// Make sure `send_wrapper` works as expected
-#[cfg(all(feature = "send_wrapper", target_arch = "wasm32"))]
-mod assert_send {
- use object_store::ObjectStore;
-
- #[allow(dead_code)]
- fn assert_send<T: Send>(_: T) {}
-
- #[allow(dead_code)]
- fn assertion() {
- let op = super::Operator::new(opendal::services::Memory::default())
- .unwrap()
- .finish();
- let store = super::OpendalStore::new(op);
- assert_send(store.put(&"test".into(), bytes::Bytes::new()));
- assert_send(store.get(&"test".into()));
- assert_send(store.get_range(&"test".into(), 0..1));
- assert_send(store.head(&"test".into()));
- assert_send(store.delete(&"test".into()));
- assert_send(store.list(None));
- assert_send(store.list_with_offset(None, &"test".into()));
- assert_send(store.list_with_delimiter(None));
- }
-}
-
#[cfg(test)]
mod tests {
use std::sync::Arc;
diff --git a/integrations/object_store/src/send_wrapper.rs b/integrations/object_store/src/utils.rs
similarity index 53%
rename from integrations/object_store/src/send_wrapper.rs
rename to integrations/object_store/src/utils.rs
index 82b4e89854..c4c593797e 100644
--- a/integrations/object_store/src/send_wrapper.rs
+++ b/integrations/object_store/src/utils.rs
@@ -15,14 +15,103 @@
// specific language governing permissions and limitations
// under the License.
-//! Conditionally add the `Send` marker trait for the wrapped type.
-//! Only take effect when the `send_wrapper` feature is enabled.
-
use futures::Stream;
+use object_store::ObjectMeta;
+use opendal::{Entry, Metadata, Metakey};
+
+/// Conditionally add the `Send` marker trait for the wrapped type.
+/// Only take effect when the `send_wrapper` feature is enabled.
#[cfg(not(feature = "send_wrapper"))]
-pub use noop_wrapper::NoopWrapper as SendWrapper;
+use noop_wrapper::NoopWrapper as SendWrapper;
#[cfg(feature = "send_wrapper")]
-pub use send_wrapper::SendWrapper;
+use send_wrapper::SendWrapper;
+
+/// Format `opendal::Error` to `object_store::Error`.
+pub fn format_object_store_error(err: opendal::Error, path: &str) -> object_store::Error {
+ use opendal::ErrorKind;
+ match err.kind() {
+ ErrorKind::NotFound => object_store::Error::NotFound {
+ path: path.to_string(),
+ source: Box::new(err),
+ },
+ ErrorKind::Unsupported => object_store::Error::NotSupported {
+ source: Box::new(err),
+ },
+ ErrorKind::AlreadyExists => object_store::Error::AlreadyExists {
+ path: path.to_string(),
+ source: Box::new(err),
+ },
+ kind => object_store::Error::Generic {
+ store: kind.into_static(),
+ source: Box::new(err),
+ },
+ }
+}
+
+/// Format `opendal::Metadata` to `object_store::ObjectMeta`.
+pub fn format_object_meta(path: &str, meta: &Metadata) -> ObjectMeta {
+ let version = match meta.metakey().contains(Metakey::Version) {
+ true => meta.version().map(|x| x.to_string()),
+ false => None,
+ };
+
+ let e_tag = match meta.metakey().contains(Metakey::Etag) {
+ true => meta.etag().map(|x| x.to_string()),
+ false => None,
+ };
+
+ ObjectMeta {
+ location: path.into(),
+ last_modified: meta.last_modified().unwrap_or_default(),
+ size: meta.content_length() as usize,
+ e_tag,
+ version,
+ }
+}
+
+/// Try to format `opendal::Entry` to `object_store::ObjectMeta`.
+pub async fn try_format_object_meta(
+ res: object_store::Result<Entry, opendal::Error>,
+) -> object_store::Result<ObjectMeta> {
+ let entry = res.map_err(|err| format_object_store_error(err, ""))?;
+ let meta = entry.metadata();
+
+ Ok(format_object_meta(entry.path(), meta))
+}
+
+/// Make given future `Send`.
+pub trait IntoSendFuture {
+ type Output;
+
+ fn into_send(self) -> Self::Output;
+}
+
+impl<T> IntoSendFuture for T
+where
+ T: futures::Future,
+{
+ type Output = SendWrapper<T>;
+ fn into_send(self) -> Self::Output {
+ SendWrapper::new(self)
+ }
+}
+
+/// Make given Stream `Send`.
+pub trait IntoSendStream {
+ type Output;
+
+ fn into_send(self) -> Self::Output;
+}
+
+impl<T> IntoSendStream for T
+where
+ T: Stream,
+{
+ type Output = SendWrapper<T>;
+ fn into_send(self) -> Self::Output {
+ SendWrapper::new(self)
+ }
+}
#[cfg(not(feature = "send_wrapper"))]
mod noop_wrapper {
@@ -70,35 +159,3 @@ mod noop_wrapper {
}
}
}
-
-pub trait IntoSendFuture {
- type Output;
-
- fn into_send(self) -> Self::Output;
-}
-
-impl<T> IntoSendFuture for T
-where
- T: futures::Future,
-{
- type Output = SendWrapper<T>;
- fn into_send(self) -> Self::Output {
- SendWrapper::new(self)
- }
-}
-
-pub trait IntoSendStream {
- type Output;
-
- fn into_send(self) -> Self::Output;
-}
-
-impl<T> IntoSendStream for T
-where
- T: Stream,
-{
- type Output = SendWrapper<T>;
- fn into_send(self) -> Self::Output {
- SendWrapper::new(self)
- }
-}