Buffered, BufferUnordered: Assert maximum size of buffer is non-zero

Author: taiki-e

futures-util/src/stream/stream/buffer_unordered.rs

@@ -46,6 +46,7 @@ where
         St: Stream,
         St::Item: Future,
     {
+        assert!(n > 0);
         Self {
             stream: super::Fuse::new(stream),
             in_progress_queue: FuturesUnordered::new(),

futures-util/src/stream/stream/buffered.rs

@@ -44,6 +44,7 @@ where
     St::Item: Future,
 {
     pub(super) fn new(stream: St, n: usize) -> Self {
+        assert!(n > 0);
         Self {
             stream: super::Fuse::new(stream),
             in_progress_queue: FuturesOrdered::new(),

futures-util/src/stream/stream/mod.rs

@@ -1142,6 +1142,10 @@ pub trait StreamExt: Stream {
     ///
     /// This method is only available when the `std` or `alloc` feature of this
     /// library is activated, and it is activated by default.
+    ///
+    /// # Panics
+    ///
+    /// This method will panic if `n` is zero.
     #[cfg_attr(feature = "cfg-target-has-atomic", cfg(target_has_atomic = "ptr"))]
     #[cfg(feature = "alloc")]
     fn buffered(self, n: usize) -> Buffered<Self>
@@ -1165,6 +1169,10 @@ pub trait StreamExt: Stream {
     /// This method is only available when the `std` or `alloc` feature of this
     /// library is activated, and it is activated by default.
     ///
+    /// # Panics
+    ///
+    /// This method will panic if `n` is zero.
+    ///
     /// # Examples
     ///
     /// ```

futures/tests/stream.rs

@@ -1,4 +1,4 @@
-use futures::channel::mpsc;
+use futures::{channel::mpsc, future::BoxFuture};
 use futures::executor::block_on;
 use futures::future::{self, Future};
 use futures::sink::SinkExt;
@@ -118,6 +118,22 @@ fn take_until() {
     });
 }
 
+#[test]
+#[should_panic]
+fn buffered_panic_on_cap_zero() {
+    let (_, rx1) = mpsc::channel::<BoxFuture<()>>(1);
+
+    let _ = rx1.buffered(0);
+}
+
+#[test]
+#[should_panic]
+fn buffer_unordered_panic_on_cap_zero() {
+    let (_, rx1) = mpsc::channel::<BoxFuture<()>>(1);
+
+    let _ = rx1.buffer_unordered(0);
+}
+
 #[test]
 #[should_panic]
 fn chunks_panic_on_cap_zero() {