quic: for streams allow ReadableStream as body

After the change the body parameter for all
created QuicStreams can take a ReadableStream
of Uint8Array, Arraybuffers etc..

We introduce a DataQueueFeeder class, that may be
also used for other related mechanisms.

A locally created unidirectional Stream can not
have a reader. Therefore the interface is changed
to handle this case. (undercovered when writing
 the tests).

Furthermore, a ResumeStream must be added After
AddStream in QuicSession, as the ResumeStream
beforehand triggered with set_outbound is a no-op,
as Stream was not added to Session beforehand.

Fixes: #60234

Author: martenrichter

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`
 

lib/internal/quic/quic.js

@@ -34,6 +34,7 @@ const {
   Endpoint: Endpoint_,
   Http3Application: Http3,
   setCallbacks,
+  DataQueueFeeder,
 
   // The constants to be exposed to end users for various options.
   CC_ALGO_RENO_STR: CC_ALGO_RENO,
@@ -114,6 +115,10 @@ const {
   buildNgHeaderString,
 } = require('internal/http2/util');
 
+const {
+  isReadableStream,
+} = require('internal/webstreams/readablestream');
+
 const kEmptyObject = { __proto__: null };
 
 const {
@@ -546,6 +551,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 +614,7 @@ function validateBody(body) {
     'ArrayBuffer',
     'ArrayBufferView',
     'Blob',
+    'ReadableStream',
   ], body);
 }
 

src/dataqueue/queue.cc

@@ -18,6 +18,8 @@
 #include <memory>
 #include <vector>
 
+#include "../quic/streams.h"
+
 namespace node {
 
 using v8::ArrayBufferView;
@@ -1062,9 +1064,81 @@ class FdEntry final : public EntryImpl {
   friend class ReaderImpl;
 };
 
+}  // namespace
+// ============================================================================
+
+class FeederEntry final : public EntryImpl {
+ public:
+  FeederEntry(DataQueueFeeder* feeder) : feeder_(feeder) {
+  }
+
+  static std::unique_ptr<FeederEntry> Create(DataQueueFeeder* feeder) {
+    return std::make_unique<FeederEntry>(feeder);
+  }
+
+  std::shared_ptr<DataQueue::Reader> get_reader() override {
+    return ReaderImpl::Create(this);
+  }
+
+  std::unique_ptr<Entry> slice(
+      uint64_t start, std::optional<uint64_t> end = std::nullopt) override {
+      // we are not idempotent
+      return std::unique_ptr<Entry>(nullptr);
+  }
+
+  std::optional<uint64_t> size() const override {
+    return std::optional<uint64_t>();
+  }
+
+  bool is_idempotent() const override { return false; }
+
+  SET_NO_MEMORY_INFO()
+  SET_MEMORY_INFO_NAME(FeederEntry)
+  SET_SELF_SIZE(FeederEntry)
+
+ private:
+  DataQueueFeeder* feeder_;
+
+  class ReaderImpl final : public DataQueue::Reader,
+                           public std::enable_shared_from_this<ReaderImpl> {
+   public:
+    static std::shared_ptr<ReaderImpl> Create(FeederEntry* entry) {
+      return std::make_shared<ReaderImpl>(entry);
+    };
+
+    explicit ReaderImpl(FeederEntry* entry) : entry_(entry) {
+    }
+
+    ~ReaderImpl() {
+      entry_->feeder_->DrainAndClose();
+    }
+
+    int Pull(Next next,
+             int options,
+             DataQueue::Vec* data,
+             size_t count,
+             size_t max_count_hint = bob::kMaxCountHint) override {
+      if (entry_->feeder_->Done()) {
+        std::move(next)(bob::STATUS_EOS, nullptr, 0, [](uint64_t) {});
+        return bob::STATUS_EOS;
+      }
+      entry_->feeder_->addPendingPull(
+        DataQueueFeeder::PendingPull(std::move(next)));
+      entry_->feeder_->tryWakePulls();
+      return bob::STATUS_WAIT;
+    }
+
+    SET_NO_MEMORY_INFO()
+    SET_MEMORY_INFO_NAME(FeederEntry::Reader)
+    SET_SELF_SIZE(ReaderImpl)
+
+   private:
+    FeederEntry* entry_;
+  };
+};
+
 // ============================================================================
 
-}  // namespace
 
 std::shared_ptr<DataQueue> DataQueue::CreateIdempotent(
     std::vector<std::unique_ptr<Entry>> list) {
@@ -1138,6 +1212,11 @@ std::unique_ptr<DataQueue::Entry> DataQueue::CreateFdEntry(Environment* env,
   return FdEntry::Create(env, path);
 }
 
+std::unique_ptr<DataQueue::Entry> DataQueue::CreateFeederEntry(
+    DataQueueFeeder* feeder) {
+  return FeederEntry::Create(feeder);
+}
+
 void DataQueue::Initialize(Environment* env, v8::Local<v8::Object> target) {
   // Nothing to do here currently.
 }

src/dataqueue/queue.h

@@ -16,6 +16,8 @@
 #include <vector>
 
 namespace node {
+    using v8::Local;
+    using v8::Value;
 
 // Represents a sequenced collection of data sources that can be
 // consumed as a single logical stream of data. Sources can be
@@ -124,6 +126,8 @@ namespace node {
 // For non-idempotent DataQueues, only a single reader is ever allowed for
 // the DataQueue, and the data can only ever be read once.
 
+class DataQueueFeeder;
+
 class DataQueue : public MemoryRetainer {
  public:
   struct Vec {
@@ -224,6 +228,8 @@ class DataQueue : public MemoryRetainer {
   static std::unique_ptr<Entry> CreateFdEntry(Environment* env,
                                               v8::Local<v8::Value> path);
 
+  static std::unique_ptr<Entry> CreateFeederEntry(DataQueueFeeder* feeder);
+
   // Creates a Reader for the given queue. If the queue is idempotent,
   // any number of readers can be created, all of which are guaranteed
   // to provide the same data. Otherwise, only a single reader is

src/quic/bindingdata.h

@@ -15,7 +15,9 @@
 #include <unordered_map>
 #include "defs.h"
 
-namespace node::quic {
+namespace node {
+class DataQueueFeeder;
+namespace quic {
 
 class Endpoint;
 class Packet;
@@ -24,6 +26,7 @@ class Packet;
 
 // The FunctionTemplates the BindingData will store for us.
 #define QUIC_CONSTRUCTORS(V)                                                   \
+  V(dataqueuefeeder)                                                           \
   V(endpoint)                                                                  \
   V(http3application)                                                          \
   V(logstream)                                                                 \
@@ -68,6 +71,7 @@ class Packet;
   V(ciphers, "ciphers")                                                        \
   V(crl, "crl")                                                                \
   V(cubic, "cubic")                                                            \
+  V(dataqueuefeeder, "DataQueueFeeder")                                        \
   V(disable_stateless_reset, "disableStatelessReset")                          \
   V(enable_connect_protocol, "enableConnectProtocol")                          \
   V(enable_datagrams, "enableDatagrams")                                       \
@@ -264,6 +268,7 @@ struct CallbackScope final : public CallbackScopeBase {
   explicit CallbackScope(T* ptr) : CallbackScopeBase(ptr->env()), ref(ptr) {}
 };
 
-}  // namespace node::quic
+}  // namespace quic
+}  // namespace node
 
 #endif  // defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS

src/quic/quic.cc

@@ -27,6 +27,7 @@ void CreatePerIsolateProperties(IsolateData* isolate_data,
   Endpoint::InitPerIsolate(isolate_data, target);
   Session::InitPerIsolate(isolate_data, target);
   Stream::InitPerIsolate(isolate_data, target);
+  DataQueueFeeder::InitPerIsolate(isolate_data, target);
 }
 
 void CreatePerContextProperties(Local<Object> target,
@@ -38,13 +39,15 @@ void CreatePerContextProperties(Local<Object> target,
   Endpoint::InitPerContext(realm, target);
   Session::InitPerContext(realm, target);
   Stream::InitPerContext(realm, target);
+  DataQueueFeeder::InitPerContext(realm, target);
 }
 
 void RegisterExternalReferences(ExternalReferenceRegistry* registry) {
   BindingData::RegisterExternalReferences(registry);
   Endpoint::RegisterExternalReferences(registry);
   Session::RegisterExternalReferences(registry);
   Stream::RegisterExternalReferences(registry);
+  DataQueueFeeder::RegisterExternalReferences(registry);
 }
 
 }  // namespace quic

src/quic/session.cc

@@ -1940,6 +1940,8 @@ BaseObjectPtr<Stream> Session::CreateStream(
   if (auto stream = Stream::Create(this, id, std::move(data_source)))
       [[likely]] {
     AddStream(stream, option);
+    ResumeStream(id);  // ok, we need to resume, as the Resume before fails
+    // as the stream was not added yet
     return stream;
   }
   return {};