refactor!: move get to ObjectStoreExt

See apache#385 and apache#405.

Author: crepererum

src/azure/credential.rs

@@ -1073,7 +1073,7 @@ mod tests {
     use super::*;
     use crate::azure::MicrosoftAzureBuilder;
     use crate::client::mock_server::MockServer;
-    use crate::{ObjectStore, Path};
+    use crate::{ObjectStoreExt, Path};
 
     #[tokio::test]
     async fn test_managed_identity() {

src/buffered.rs

@@ -45,13 +45,14 @@ pub const DEFAULT_BUFFER_SIZE: usize = 1024 * 1024;
 /// very [high first-byte latencies], on the order of 100-200ms, and so avoiding unnecessary
 /// round-trips is critical to throughput.
 ///
-/// Systems looking to sequentially scan a file should instead consider using [`ObjectStore::get`],
+/// Systems looking to sequentially scan a file should instead consider using [`ObjectStoreExt::get`],
 /// or [`ObjectStore::get_opts`], or [`ObjectStore::get_range`] to read a particular range.
 ///
 /// Systems looking to read multiple ranges of a file should instead consider using
 /// [`ObjectStore::get_ranges`], which will optimise the vectored IO.
 ///
 /// [high first-byte latencies]: https://docs.aws.amazon.com/AmazonS3/latest/userguide/optimizing-performance.html
+/// [`ObjectStoreExt::get`]: crate::ObjectStoreExt::get
 pub struct BufReader {
     /// The object store to fetch data from
     store: Arc<dyn ObjectStore>,

src/client/get.rs

@@ -487,7 +487,7 @@ mod http_tests {
     use crate::client::{HttpError, HttpErrorKind, HttpResponseBody};
     use crate::http::HttpBuilder;
     use crate::path::Path;
-    use crate::{ClientOptions, ObjectStore, RetryConfig};
+    use crate::{ClientOptions, ObjectStoreExt, RetryConfig};
     use bytes::Bytes;
     use futures::FutureExt;
     use http::header::{CONTENT_LENGTH, CONTENT_RANGE, ETAG, RANGE};

src/lib.rs

@@ -204,15 +204,15 @@
 //!
 //! # Fetch objects
 //!
-//! Use the [`ObjectStore::get`] method to fetch the data bytes
+//! Use the [`ObjectStoreExt::get`] / [`ObjectStore::get_opts`] method to fetch the data bytes
 //! from remote storage or files in the local filesystem as a stream.
 //!
 //! ```ignore-wasm32
 //! # use futures::TryStreamExt;
 //! # use object_store::local::LocalFileSystem;
 //! # use std::sync::Arc;
 //! #  use bytes::Bytes;
-//! # use object_store::{path::Path, ObjectStore, GetResult};
+//! # use object_store::{path::Path, ObjectStore, ObjectStoreExt, GetResult};
 //! # fn get_object_store() -> Arc<dyn ObjectStore> {
 //! #   Arc::new(LocalFileSystem::new())
 //! # }
@@ -403,7 +403,7 @@
 //! ```
 //! # use std::collections::btree_map::Entry;
 //! # use std::collections::HashMap;
-//! # use object_store::{GetOptions, GetResult, ObjectStore, Result, Error};
+//! # use object_store::{GetOptions, GetResult, ObjectStore, ObjectStoreExt, Result, Error};
 //! # use std::sync::Arc;
 //! # use std::time::{Duration, Instant};
 //! # use bytes::Bytes;
@@ -468,7 +468,7 @@
 //! storage, without relying on a separate DBMS.
 //!
 //! ```
-//! # use object_store::{Error, ObjectStore, PutMode, UpdateVersion};
+//! # use object_store::{Error, ObjectStore, ObjectStoreExt, PutMode, UpdateVersion};
 //! # use std::sync::Arc;
 //! # use bytes::Bytes;
 //! # use tokio::io::AsyncWriteExt;
@@ -651,38 +651,6 @@ pub trait ObjectStore: std::fmt::Display + Send + Sync + Debug + 'static {
         opts: PutMultipartOptions,
     ) -> Result<Box<dyn MultipartUpload>>;
 
-    /// Return the bytes that are stored at the specified location.
-    ///
-    /// ## Example
-    ///
-    /// This example uses a basic local filesystem object store to get an object.
-    ///
-    /// ```ignore-wasm32
-    /// # use object_store::local::LocalFileSystem;
-    /// # use tempfile::tempdir;
-    /// # use object_store::{path::Path, ObjectStore, ObjectStoreExt};
-    /// async fn get_example() {
-    ///     let tmp = tempdir().unwrap();
-    ///     let store = LocalFileSystem::new_with_prefix(tmp.path()).unwrap();
-    ///     let location = Path::from("example.txt");
-    ///     let content = b"Hello, Object Store!";
-    ///
-    ///     // Put the object into the store
-    ///     store
-    ///         .put(&location, content.as_ref().into())
-    ///         .await
-    ///         .expect("Failed to put object");
-    ///
-    ///     // Get the object from the store
-    ///     let get_result = store.get(&location).await.expect("Failed to get object");
-    ///     let bytes = get_result.bytes().await.expect("Failed to read bytes");
-    ///     println!("Retrieved content: {}", String::from_utf8_lossy(&bytes));
-    /// }
-    /// ```
-    async fn get(&self, location: &Path) -> Result<GetResult> {
-        self.get_opts(location, GetOptions::default()).await
-    }
-
     /// Perform a get request with options
     ///
     /// ## Example
@@ -1115,10 +1083,6 @@ macro_rules! as_ref_impl {
                 self.as_ref().put_multipart_opts(location, opts).await
             }
 
-            async fn get(&self, location: &Path) -> Result<GetResult> {
-                self.as_ref().get(location).await
-            }
-
             async fn get_opts(&self, location: &Path, options: GetOptions) -> Result<GetResult> {
                 self.as_ref().get_opts(location, options).await
             }
@@ -1212,6 +1176,36 @@ pub trait ObjectStoreExt: ObjectStore {
         &self,
         location: &Path,
     ) -> impl Future<Output = Result<Box<dyn MultipartUpload>>>;
+
+    /// Return the bytes that are stored at the specified location.
+    ///
+    /// ## Example
+    ///
+    /// This example uses a basic local filesystem object store to get an object.
+    ///
+    /// ```ignore-wasm32
+    /// # use object_store::local::LocalFileSystem;
+    /// # use tempfile::tempdir;
+    /// # use object_store::{path::Path, ObjectStore, ObjectStoreExt};
+    /// async fn get_example() {
+    ///     let tmp = tempdir().unwrap();
+    ///     let store = LocalFileSystem::new_with_prefix(tmp.path()).unwrap();
+    ///     let location = Path::from("example.txt");
+    ///     let content = b"Hello, Object Store!";
+    ///
+    ///     // Put the object into the store
+    ///     store
+    ///         .put(&location, content.as_ref().into())
+    ///         .await
+    ///         .expect("Failed to put object");
+    ///
+    ///     // Get the object from the store
+    ///     let get_result = store.get(&location).await.expect("Failed to get object");
+    ///     let bytes = get_result.bytes().await.expect("Failed to read bytes");
+    ///     println!("Retrieved content: {}", String::from_utf8_lossy(&bytes));
+    /// }
+    /// ```
+    fn get(&self, location: &Path) -> impl Future<Output = Result<GetResult>>;
 }
 
 impl<T> ObjectStoreExt for T
@@ -1227,6 +1221,10 @@ where
         self.put_multipart_opts(location, PutMultipartOptions::default())
             .await
     }
+
+    async fn get(&self, location: &Path) -> Result<GetResult> {
+        self.get_opts(location, GetOptions::default()).await
+    }
 }
 
 /// Result of a list call that includes objects, prefixes (directories) and a

src/limit.rs

@@ -93,12 +93,6 @@ impl<T: ObjectStore> ObjectStore for LimitStore<T> {
         }))
     }
 
-    async fn get(&self, location: &Path) -> Result<GetResult> {
-        let permit = Arc::clone(&self.semaphore).acquire_owned().await.unwrap();
-        let r = self.inner.get(location).await?;
-        Ok(permit_get_result(r, permit))
-    }
-
     async fn get_opts(&self, location: &Path, options: GetOptions) -> Result<GetResult> {
         let permit = Arc::clone(&self.semaphore).acquire_owned().await.unwrap();
         let r = self.inner.get_opts(location, options).await?;

src/local.rs

@@ -1756,7 +1756,7 @@ mod unix_test {
     use tempfile::TempDir;
 
     use crate::local::LocalFileSystem;
-    use crate::{ObjectStore, Path};
+    use crate::{ObjectStore, ObjectStoreExt, Path};
 
     #[tokio::test]
     async fn test_fifo() {

src/parse.rs

@@ -401,7 +401,7 @@ mod tests {
     #[tokio::test]
     #[cfg(all(feature = "http", not(target_arch = "wasm32")))]
     async fn test_url_http() {
-        use crate::client::mock_server::MockServer;
+        use crate::{ObjectStoreExt, client::mock_server::MockServer};
         use http::{Response, header::USER_AGENT};
 
         let server = MockServer::new().await;

src/prefix.rs

@@ -113,11 +113,6 @@ impl<T: ObjectStore> ObjectStore for PrefixStore<T> {
         self.inner.put_multipart_opts(&full_path, opts).await
     }
 
-    async fn get(&self, location: &Path) -> Result<GetResult> {
-        let full_path = self.full_path(location);
-        self.inner.get(&full_path).await
-    }
-
     async fn get_range(&self, location: &Path, range: Range<u64>) -> Result<Bytes> {
         let full_path = self.full_path(location);
         self.inner.get_range(&full_path, range).await

src/throttle.rs

@@ -40,7 +40,7 @@ pub struct ThrottleConfig {
     /// the operation.
     pub wait_delete_per_call: Duration,
 
-    /// Sleep duration for every byte received during [`get`](ThrottledStore::get).
+    /// Sleep duration for every byte received during [`get_opts`](ThrottledStore::get_opts).
     ///
     /// Sleeping is performed after the underlying store returned and only for successful gets. The
     /// sleep duration is additive to [`wait_get_per_call`](Self::wait_get_per_call).
@@ -50,7 +50,7 @@ pub struct ThrottleConfig {
     /// resulting sleep time will be partial as well.
     pub wait_get_per_byte: Duration,
 
-    /// Sleep duration for every call to [`get`](ThrottledStore::get).
+    /// Sleep duration for every call to [`get_opts`](ThrottledStore::get_opts).
     ///
     /// Sleeping is done before the underlying store is called and independently of the success of
     /// the operation. The sleep duration is additive to
@@ -170,16 +170,6 @@ impl<T: ObjectStore> ObjectStore for ThrottledStore<T> {
         }))
     }
 
-    async fn get(&self, location: &Path) -> Result<GetResult> {
-        sleep(self.config().wait_get_per_call).await;
-
-        // need to copy to avoid moving / referencing `self`
-        let wait_get_per_byte = self.config().wait_get_per_byte;
-
-        let result = self.inner.get(location).await?;
-        Ok(throttle_get(result, wait_get_per_byte))
-    }
-
     async fn get_opts(&self, location: &Path, options: GetOptions) -> Result<GetResult> {
         sleep(self.config().wait_get_per_call).await;