[ChunksTimeout] Consumes the stream and return the buffered items immediately

Summary:
When the underlying stream is an exclusive reference (&mut stream), and we need to drop the `ChunksTimout` stream without losing the buffered items.

Author: muhamadazmy

tokio-stream/src/stream_ext/chunks_timeout.rs

@@ -33,6 +33,11 @@ impl<S: Stream> ChunksTimeout<S> {
             cap: max_size,
         }
     }
+
+    /// Consumes the stream and immediately return the buffered items.
+    pub fn remaining(self) -> Vec<S::Item> {
+        self.items
+    }
 }
 
 impl<S: Stream> Stream for ChunksTimeout<S> {