Buffer through

core/shared/src/main/scala/fs2/Stream.scala

@@ -346,6 +346,25 @@ final class Stream[+F[_], +O] private[fs2] (private[fs2] val underlying: Pull[F,
     go(Nil, false, this).stream
   }
 
+  // A builder for queues.
+  trait MakeQueue[F[_]] {
+    def create[A]: F[Queue[F, A]]
+  }
+
+  /** Buffers the chunks of this stream through a queue that is created by the supplied builder,
+    * and returns a new stream that consumes the chunks from that queue. The new stream terminates
+    * after both this stream terminates and all chunks are emitted. Errors from this stream are
+    * propagated to the new stream.
+    */
+  def bufferThrough[F2[x] >: F[x]: Concurrent, O2 >: O](
+      mkQueue: MakeQueue[F2]
+  ): Stream[F2, O2] =
+    Stream.eval(mkQueue.create[Option[Chunk[O2]]]).flatMap { queue =>
+      Stream
+        .fromQueueNoneTerminatedChunk(queue)
+        .concurrently(this.enqueueNoneTerminatedChunks(queue))
+    }
+
   /** Emits only elements that are distinct from their immediate predecessors,
     * using natural equality for comparison.
     *