Revert "impl fn collect(mut self) -> Buffer"

This reverts commit 482ff97.

core/src/raw/adapters/typed_kv/backend.rs

@@ -53,7 +53,7 @@ where
 }
 
 #[cfg_attr(not(target_arch = "wasm32"), async_trait)]
-#[cfg_attr(target_arch = "wasm32", async_trait(? Send))]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
 impl<S: Adapter> Accessor for Backend<S> {
     type Reader = Buffer;
     type BlockingReader = Buffer;
@@ -258,7 +258,11 @@ impl<S> KvWriter<S> {
     }
 
     fn build(&mut self) -> Value {
-        let value = self.buf.take().map(QueueBuf::collect).unwrap_or_default();
+        let value = self
+            .buf
+            .take()
+            .map(QueueBuf::into_buffer)
+            .unwrap_or_default();
 
         let mut metadata = Metadata::new(EntryMode::FILE);
         metadata.set_content_length(value.len() as u64);

core/src/raw/oio/buf/queue_buf.rs

@@ -16,6 +16,7 @@
 // under the License.
 
 use crate::*;
+use bytes::Buf;
 use std::collections::VecDeque;
 
 /// QueueBuf is a queue of [`Buffer`].
@@ -56,7 +57,7 @@ impl QueueBuf {
         self.len() == 0
     }
 
-    /// Collect the buffers from the queue into a new [`Buffer`].
+    /// Build a new [`Buffer`] from the queue.
     ///
     /// If the queue is empty, it will return an empty buffer. Otherwise, it will iterate over all
     /// buffers and collect them into a new buffer.
@@ -67,7 +68,28 @@ impl QueueBuf {
     /// most of them should be acceptable since we can expect the item length of buffers are slower
     /// than 4k.
     #[inline]
-    pub fn collect(mut self) -> Buffer {
+    pub fn collect(&self) -> Buffer {
+        if self.0.is_empty() {
+            Buffer::new()
+        } else if self.0.len() == 1 {
+            self.0.clone().pop_front().unwrap()
+        } else {
+            self.0.clone().into_iter().flatten().collect()
+        }
+    }
+
+    /// Convert to a new [`Buffer`] from the queue.
+    ///
+    /// If the queue is empty, it will return an empty buffer. Otherwise, it will iterate over all
+    /// buffers and collect them into a new buffer.
+    ///
+    /// # Notes
+    ///
+    /// There are allocation overheads when collecting multiple buffers into a new buffer. But
+    /// most of them should be acceptable since we can expect the item length of buffers are slower
+    /// than 4k.
+    #[inline]
+    pub fn into_buffer(mut self) -> Buffer {
         if self.0.is_empty() {
             Buffer::new()
         } else if self.0.len() == 1 {
@@ -77,6 +99,24 @@ impl QueueBuf {
         }
     }
 
+    /// Advance the buffer queue by `cnt` bytes.
+    #[inline]
+    pub fn advance(&mut self, cnt: usize) {
+        assert!(cnt <= self.len(), "cannot advance past {cnt} bytes");
+
+        let mut new_cnt = cnt;
+        while new_cnt > 0 {
+            let buf = self.0.front_mut().expect("buffer must be valid");
+            if new_cnt < buf.remaining() {
+                buf.advance(new_cnt);
+                break;
+            } else {
+                new_cnt -= buf.remaining();
+                self.0.pop_front();
+            }
+        }
+    }
+
     /// Clear the buffer queue.
     #[inline]
     pub fn clear(&mut self) {

core/src/raw/oio/write/exact_buf_write.rs

@@ -47,8 +47,8 @@ impl<W: oio::Write> ExactBufWriter<W> {
 impl<W: oio::Write> oio::Write for ExactBufWriter<W> {
     async fn write(&mut self, mut bs: Buffer) -> Result<usize> {
         if self.buffer.len() >= self.buffer_size {
-            let buffer = std::mem::take(&mut self.buffer);
-            self.inner.write(buffer.collect()).await?;
+            let written = self.inner.write(self.buffer.collect()).await?;
+            self.buffer.advance(written);
         }
 
         let remaining = self.buffer_size - self.buffer.len();
@@ -59,8 +59,15 @@ impl<W: oio::Write> oio::Write for ExactBufWriter<W> {
     }
 
     async fn close(&mut self) -> Result<()> {
-        let buffer = std::mem::take(&mut self.buffer);
-        self.inner.write(buffer.collect()).await?;
+        loop {
+            if self.buffer.is_empty() {
+                break;
+            }
+
+            let written = self.inner.write(self.buffer.collect()).await?;
+            self.buffer.advance(written);
+        }
+
         self.inner.close().await
     }