fix(core): Read chunk should respect users input

Signed-off-by: Xuanwo <[email protected]>

core/src/raw/ops.rs

@@ -392,15 +392,15 @@ pub struct OpReader {
     /// The concurrent requests that reader can send.
     concurrent: usize,
     /// The chunk size of each request.
-    chunk: usize,
+    chunk: Option<usize>,
 }
 
 impl Default for OpReader {
     fn default() -> Self {
         Self {
             range: BytesRange::default(),
             concurrent: 1,
-            chunk: 0,
+            chunk: None,
         }
     }
 }
@@ -435,12 +435,12 @@ impl OpReader {
 
     /// Set the chunk of the option
     pub fn with_chunk(mut self, chunk: usize) -> Self {
-        self.chunk = chunk;
+        self.chunk = Some(chunk.max(1));
         self
     }
 
     /// Get chunk from option
-    pub fn chunk(&self) -> usize {
+    pub fn chunk(&self) -> Option<usize> {
         self.chunk
     }
 }

core/src/types/reader.rs

@@ -139,15 +139,15 @@ pub mod into_stream {
 
     pub struct ReadFutureIterator {
         r: oio::Reader,
-        chunk: usize,
+        chunk: Option<usize>,
 
         offset: u64,
         end: Option<u64>,
         finished: Arc<AtomicBool>,
     }
 
     impl ReadFutureIterator {
-        pub fn new(r: oio::Reader, chunk: usize, range: impl RangeBounds<u64>) -> Self {
+        pub fn new(r: oio::Reader, chunk: Option<usize>, range: impl RangeBounds<u64>) -> Self {
             let start = match range.start_bound().cloned() {
                 Bound::Included(start) => start,
                 Bound::Excluded(start) => start + 1,
@@ -181,13 +181,12 @@ pub mod into_stream {
             }
 
             let offset = self.offset;
+            // TODO: replace with services preferred chunk size.
+            let chunk = self.chunk.unwrap_or(4 * 1024 * 1024);
             let limit = self
                 .end
-                .map(|end| (end - self.offset) as usize)
-                // TODO: replace with services preferred chunk size.
-                .unwrap_or(4 * 1024 * 1024)
-                // Align with reader's chunk size.
-                .max(self.chunk);
+                .map(|end| ((end - self.offset) as usize).min(chunk))
+                .unwrap_or(chunk);
             let finished = self.finished.clone();
             let r = self.r.clone();