refactor(raw/oio): Reorganize to allow adding more features

core/src/layers/complete.rs

@@ -25,13 +25,13 @@ use std::task::Poll;
 use async_trait::async_trait;
 use bytes::Bytes;
 
-use crate::raw::oio::into_reader::RangeReader;
-use crate::raw::oio::to_flat_pager;
-use crate::raw::oio::to_hierarchy_pager;
+use crate::raw::oio::into_flat_page;
+use crate::raw::oio::into_hierarchy_page;
+use crate::raw::oio::ByRangeSeekableReader;
 use crate::raw::oio::Entry;
-use crate::raw::oio::IntoStreamableReader;
-use crate::raw::oio::ToFlatPager;
-use crate::raw::oio::ToHierarchyPager;
+use crate::raw::oio::FlatPager;
+use crate::raw::oio::HierarchyPager;
+use crate::raw::oio::StreamableReader;
 use crate::raw::*;
 use crate::*;
 
@@ -172,7 +172,7 @@ impl<A: Accessor> CompleteReaderAccessor<A> {
         match (seekable, streamable) {
             (true, true) => Ok((rp, CompleteReader::AlreadyComplete(r))),
             (true, false) => {
-                let r = oio::into_streamable_reader(r, 256 * 1024);
+                let r = oio::into_streamable_read(r, 256 * 1024);
                 Ok((rp, CompleteReader::NeedStreamable(r)))
             }
             _ => {
@@ -193,12 +193,12 @@ impl<A: Accessor> CompleteReaderAccessor<A> {
                         (offset, size)
                     }
                 };
-                let r = oio::into_reader::by_range(self.inner.clone(), path, r, offset, size);
+                let r = oio::into_seekable_read_by_range(self.inner.clone(), path, r, offset, size);
 
                 if streamable {
                     Ok((rp, CompleteReader::NeedSeekable(r)))
                 } else {
-                    let r = oio::into_streamable_reader(r, 256 * 1024);
+                    let r = oio::into_streamable_read(r, 256 * 1024);
                     Ok((rp, CompleteReader::NeedBoth(r)))
                 }
             }
@@ -223,7 +223,7 @@ impl<A: Accessor> CompleteReaderAccessor<A> {
         match (seekable, streamable) {
             (true, true) => Ok((rp, CompleteReader::AlreadyComplete(r))),
             (true, false) => {
-                let r = oio::into_streamable_reader(r, 256 * 1024);
+                let r = oio::into_streamable_read(r, 256 * 1024);
                 Ok((rp, CompleteReader::NeedStreamable(r)))
             }
             (false, _) => Err(Error::new(
@@ -254,7 +254,7 @@ impl<A: Accessor> CompleteReaderAccessor<A> {
                 let (rp, p) = self.inner.list(path, args).await?;
                 Ok((rp, CompletePager::AlreadyComplete(p)))
             } else {
-                let p = to_flat_pager(
+                let p = into_flat_page(
                     self.inner.clone(),
                     path,
                     args.with_delimiter("/").limit().unwrap_or(1000),
@@ -269,7 +269,7 @@ impl<A: Accessor> CompleteReaderAccessor<A> {
                 Ok((rp, CompletePager::AlreadyComplete(p)))
             } else {
                 let (_, p) = self.inner.list(path, args.with_delimiter("")).await?;
-                let p = to_hierarchy_pager(p, path);
+                let p = into_hierarchy_page(p, path);
                 Ok((RpList::default(), CompletePager::NeedHierarchy(p)))
             };
         }
@@ -303,7 +303,7 @@ impl<A: Accessor> CompleteReaderAccessor<A> {
                 let (rp, p) = self.inner.blocking_list(path, args)?;
                 Ok((rp, CompletePager::AlreadyComplete(p)))
             } else {
-                let p = to_flat_pager(
+                let p = into_flat_page(
                     self.inner.clone(),
                     path,
                     args.with_delimiter("/").limit().unwrap_or(1000),
@@ -318,8 +318,8 @@ impl<A: Accessor> CompleteReaderAccessor<A> {
                 Ok((rp, CompletePager::AlreadyComplete(p)))
             } else {
                 let (_, p) = self.inner.blocking_list(path, args.with_delimiter(""))?;
-                let p: ToHierarchyPager<<A as Accessor>::BlockingPager> =
-                    to_hierarchy_pager(p, path);
+                let p: HierarchyPager<<A as Accessor>::BlockingPager> =
+                    into_hierarchy_page(p, path);
                 Ok((RpList::default(), CompletePager::NeedHierarchy(p)))
             };
         }
@@ -532,9 +532,9 @@ impl<A: Accessor> LayeredAccessor for CompleteReaderAccessor<A> {
 
 pub enum CompleteReader<A: Accessor, R> {
     AlreadyComplete(R),
-    NeedSeekable(RangeReader<A>),
-    NeedStreamable(IntoStreamableReader<R>),
-    NeedBoth(IntoStreamableReader<RangeReader<A>>),
+    NeedSeekable(ByRangeSeekableReader<A>),
+    NeedStreamable(StreamableReader<R>),
+    NeedBoth(StreamableReader<ByRangeSeekableReader<A>>),
 }
 
 impl<A, R> oio::Read for CompleteReader<A, R>
@@ -614,8 +614,8 @@ where
 
 pub enum CompletePager<A: Accessor, P> {
     AlreadyComplete(P),
-    NeedFlat(ToFlatPager<Arc<A>, P>),
-    NeedHierarchy(ToHierarchyPager<P>),
+    NeedFlat(FlatPager<Arc<A>, P>),
+    NeedHierarchy(HierarchyPager<P>),
 }
 
 #[async_trait]

core/src/raw/http_util/client.rs

@@ -27,7 +27,7 @@ use http::Response;
 use super::body::IncomingAsyncBody;
 use super::parse_content_length;
 use super::AsyncBody;
-use crate::raw::oio::into_stream;
+use crate::raw::*;
 use crate::Error;
 use crate::ErrorKind;
 use crate::Result;
@@ -150,10 +150,7 @@ impl HttpClient {
                 .set_source(err)
         });
 
-        let body = IncomingAsyncBody::new(
-            Box::new(into_stream::from_futures_stream(stream)),
-            content_length,
-        );
+        let body = IncomingAsyncBody::new(Box::new(oio::into_stream(stream)), content_length);
 
         let resp = hr.body(body).expect("response must build succeed");
 

core/src/raw/http_util/multipart.rs

@@ -38,7 +38,7 @@ use http::Version;
 use super::new_request_build_error;
 use super::AsyncBody;
 use super::IncomingAsyncBody;
-use crate::raw::oio::into_stream;
+use crate::raw::oio;
 use crate::*;
 
 /// Multipart is a builder for multipart/form-data.
@@ -320,7 +320,7 @@ impl MixedPart {
         let bs: Bytes = self.content;
         let length = bs.len();
         let body = IncomingAsyncBody::new(
-            Box::new(into_stream::from_futures_stream(stream::iter(vec![Ok(bs)]))),
+            Box::new(oio::into_stream(stream::iter(vec![Ok(bs)]))),
             Some(length as u64),
         );
 

core/src/raw/mod.rs

@@ -27,28 +27,16 @@
 //!   them whenever possible.
 
 mod accessor;
-pub use accessor::Accessor;
-pub use accessor::AccessorInfo;
-pub use accessor::FusedAccessor;
+pub use accessor::*;
 
 mod layer;
-pub use layer::Layer;
-pub use layer::LayeredAccessor;
-
-pub mod oio;
+pub use layer::*;
 
 mod path;
-pub use path::build_abs_path;
-pub use path::build_rel_path;
-pub use path::build_rooted_abs_path;
-pub use path::get_basename;
-pub use path::get_parent;
-pub use path::normalize_path;
-pub use path::normalize_root;
-pub use path::validate_path;
+pub use path::*;
 
 mod operation;
-pub use operation::Operation;
+pub use operation::*;
 
 mod version;
 pub use version::VERSION;
@@ -70,3 +58,4 @@ pub use chrono_util::*;
 
 // Expose as a pub mod to avoid confusing.
 pub mod adapters;
+pub mod oio;

core/src/raw/oio/into_stream/mod.rs → core/src/raw/oio/append/mod.rs

@@ -15,11 +15,5 @@
 // specific language governing permissions and limitations
 // under the License.
 
-//! into_stream will provide different implementations to convert into
-//! [`oio::Stream`][crate::raw::oio::Stream]
-
-mod from_futures_stream;
-pub use from_futures_stream::from_futures_stream;
-
-mod from_futures_reader;
-pub use from_futures_reader::from_futures_reader;
+mod api;
+pub use api::*;