quic: For streams allow ReadableStream as body

doc/api/quic.md

@@ -475,7 +475,7 @@ added: v23.8.0
 -->
 
 * `options` {Object}
-  * `body` {ArrayBuffer | ArrayBufferView | Blob}
+  * `body` {ArrayBuffer | ArrayBufferView | Blob | ReadableStream<Uint8Array>}
   * `sendOrder` {number}
 * Returns: {Promise} for a {quic.QuicStream}
 
@@ -489,7 +489,7 @@ added: v23.8.0
 -->
 
 * `options` {Object}
-  * `body` {ArrayBuffer | ArrayBufferView | Blob}
+  * `body` {ArrayBuffer | ArrayBufferView | Blob | ReadableStream<Uint8Array>}
   * `sendOrder` {number}
 * Returns: {Promise} for a {quic.QuicStream}
 
@@ -820,7 +820,7 @@ The callback to invoke when the stream is reset. Read/write.
 added: v23.8.0
 -->
 
-* Type: {ReadableStream}
+* Type: {ReadableStream | undefined}
 
 ### `stream.session`
 
@@ -832,6 +832,16 @@ added: v23.8.0
 
 The session that created this stream. Read only.
 
+### `stream.setOutbound(outbound)`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* `outbound` {ArrayBuffer|SharedArrayBuffer|ArrayBufferView|Blob|ReadableStream}
+
+Set the outbound datasource.
+
 ### `stream.stats`
 
 <!-- YAML

lib/internal/blob.js

@@ -475,6 +475,14 @@ function createBlobReaderStream(reader) {
           if (buffer !== undefined && buffer.byteLength !== 0) {
             c.enqueue(new Uint8Array(buffer));
           }
+          if (status === 3) {
+            // status wait!
+            // the source will call us (this function!) again !
+            // we should not schedule another readnext,
+            // as we end up in a busy loop, calling pull over and
+            // over and pilling up next callbacks in dataqueue
+            return;
+          }
           // We keep reading until we either reach EOS, some error, or we
           // hit the flow rate of the stream (c.desiredSize).
           // We use set immediate here because we have to allow the event

lib/internal/quic/quic.js

@@ -57,6 +57,10 @@ const {
   CLOSECONTEXT_START_FAILURE: kCloseContextStartFailure,
 } = internalBinding('quic');
 
+const {
+  DataQueueFeeder,
+} = internalBinding('dataqueuefeeder');
+
 const {
   isArrayBuffer,
   isArrayBufferView,
@@ -114,6 +118,10 @@ const {
   buildNgHeaderString,
 } = require('internal/http2/util');
 
+const {
+  isReadableStream,
+} = require('internal/webstreams/readablestream');
+
 const kEmptyObject = { __proto__: null };
 
 const {
@@ -546,6 +554,37 @@ setCallbacks({
 function validateBody(body) {
   // TODO(@jasnell): Support streaming sources
   if (body === undefined) return body;
+  if (isReadableStream(body)) {
+    const feeder = new DataQueueFeeder();
+    const reader = body.getReader();
+
+    const feeding = async () => {
+      await feeder.ready();
+      let cont = true;
+
+      while (cont) {
+        let read;
+        try {
+          read = await reader.read();
+        } catch (error) {
+          feeder.error(error);
+        }
+        const { value, done } = read;
+        try {
+          cont = await feeder.submit(value, done);
+        } catch (error) {
+          reader.cancel(error.toString());
+          break;
+        }
+      }
+      if (!cont) {
+        reader.releaseLock();
+      }
+
+    };
+    feeding();
+    return feeder;
+  }
   // Transfer ArrayBuffers...
   if (isArrayBuffer(body)) {
     return ArrayBufferPrototypeTransfer(body);
@@ -578,6 +617,7 @@ function validateBody(body) {
     'ArrayBuffer',
     'ArrayBufferView',
     'Blob',
+    'ReadableStream',
   ], body);
 }
 
@@ -802,10 +842,10 @@ class QuicStream {
   /**
    * Sets the outbound data source for the stream. This can only be called
    * once and must be called before any data will be sent. The body can be
-   * an ArrayBuffer, a TypedArray or DataView, or a Blob. If the stream
-   * is destroyed or already has an outbound data source, an error will
-   * be thrown.
-   * @param {ArrayBuffer|SharedArrayBuffer|ArrayBufferView|Blob} outbound
+   * an ArrayBuffer, a TypedArray or DataView, or a Blob, a ReadableStream.
+   * If the stream is destroyed or already has an outbound data source,
+   * an error will be thrown.
+   * @param {ArrayBuffer|SharedArrayBuffer|ArrayBufferView|Blob|ReadableStream} outbound
    */
   setOutbound(outbound) {
     QuicStream.#assertIsQuicStream(this);