LibWeb: Add and use an ad-hoc ReadableStreamDefaultReader::read_all_chunks AO

Userland/Libraries/LibWeb/Streams/AbstractOperations.cpp

@@ -304,22 +304,23 @@ JS::NonnullGCPtr<WebIDL::Promise> readable_stream_pipe_to(ReadableStream& source
     //       JavaScript-modifiable reader, writer, and stream APIs (i.e. methods on the appropriate prototypes) must not
     //       be used. Instead, the streams must be manipulated directly.
 
-    // FIXME: Currently a naive implementation that uses ReadableStreamDefaultReader::read_all_bytes() to read all chunks
+    // FIXME: Currently a naive implementation that uses ReadableStreamDefaultReader::read_all_chunks() to read all chunks
     //        from the source and then through the callback success_steps writes those chunks to the destination.
-    auto success_steps = [promise, &realm, writer](Vector<ByteBuffer> const& bytes) {
-        for (auto byte_buffer : bytes) {
-            auto buffer = JS::ArrayBuffer::create(realm, move(byte_buffer));
-            auto inner_promise = writable_stream_default_writer_write(writer, JS::Value { buffer });
-            WebIDL::resolve_promise(realm, inner_promise, JS::js_undefined());
-        }
+    auto chunk_steps = [&realm, writer](ByteBuffer chunk) {
+        auto buffer = JS::ArrayBuffer::create(realm, move(chunk));
+        auto promise = writable_stream_default_writer_write(writer, JS::Value { buffer });
+        WebIDL::resolve_promise(realm, promise, JS::js_undefined());
+    };
+
+    auto success_steps = [promise, &realm](ByteBuffer) {
         WebIDL::resolve_promise(realm, promise, JS::js_undefined());
     };
 
     auto failure_steps = [promise, &realm](JS::Value error) {
         WebIDL::reject_promise(realm, promise, error);
     };
 
-    reader->read_all_bytes(move(success_steps), move(failure_steps));
+    reader->read_all_chunks(move(chunk_steps), move(success_steps), move(failure_steps));
 
     // 16. Return promise.
     return promise;

Userland/Libraries/LibWeb/Streams/ReadableStreamDefaultReader.cpp

@@ -65,12 +65,13 @@ void ReadableStreamDefaultReader::visit_edges(Cell::Visitor& visitor)
 }
 
 // https://streams.spec.whatwg.org/#read-loop
-ReadLoopReadRequest::ReadLoopReadRequest(JS::VM& vm, JS::Realm& realm, ReadableStreamDefaultReader& reader, SuccessSteps success_steps, FailureSteps failure_steps)
+ReadLoopReadRequest::ReadLoopReadRequest(JS::VM& vm, JS::Realm& realm, ReadableStreamDefaultReader& reader, SuccessSteps success_steps, FailureSteps failure_steps, ChunkSteps chunk_steps)
     : m_vm(vm)
     , m_realm(realm)
     , m_reader(reader)
     , m_success_steps(move(success_steps))
     , m_failure_steps(move(failure_steps))
+    , m_chunk_steps(move(chunk_steps))
 {
 }
 
@@ -87,7 +88,12 @@ void ReadLoopReadRequest::on_chunk(JS::Value chunk)
     auto const& buffer = array.viewed_array_buffer()->buffer();
 
     // 2. Append the bytes represented by chunk to bytes.
-    m_byte_chunks.append(buffer);
+    m_bytes.append(buffer);
+
+    if (m_chunk_steps) {
+        // FIXME: Can we move the buffer out of the `chunk`? Unclear if that is safe.
+        m_chunk_steps(MUST(ByteBuffer::copy(buffer)));
+    }
 
     // FIXME: As the spec suggests, implement this non-recursively - instead of directly. It is not too big of a deal currently
     //        as we enqueue the entire blob buffer in one go, meaning that we only recurse a single time. Once we begin queuing
@@ -101,7 +107,7 @@ void ReadLoopReadRequest::on_chunk(JS::Value chunk)
 void ReadLoopReadRequest::on_close()
 {
     // 1. Call successSteps with bytes.
-    m_success_steps(m_byte_chunks);
+    m_success_steps(move(m_bytes));
 }
 
 // error steps, given e
@@ -195,6 +201,22 @@ void ReadableStreamDefaultReader::read_all_bytes(ReadLoopReadRequest::SuccessSte
     readable_stream_default_reader_read(*this, read_request);
 }
 
+void ReadableStreamDefaultReader::read_all_chunks(ReadLoopReadRequest::ChunkSteps chunk_steps, ReadLoopReadRequest::SuccessSteps success_steps, ReadLoopReadRequest::FailureSteps failure_steps)
+{
+    // AD-HOC: Some spec steps direct us to "read all chunks" from a stream, but there isn't an AO defined to do that.
+    //         We implement those steps by using the "read all bytes" definition, with a custom callback to receive
+    //         each chunk that is read.
+    auto& realm = this->realm();
+    auto& vm = realm.vm();
+
+    // 1. Let readRequest be a new read request with the following items:
+    //    NOTE: items and steps in ReadLoopReadRequest.
+    auto read_request = heap().allocate_without_realm<ReadLoopReadRequest>(vm, realm, *this, move(success_steps), move(failure_steps), move(chunk_steps));
+
+    // 2. Perform ! ReadableStreamDefaultReaderRead(this, readRequest).
+    readable_stream_default_reader_read(*this, read_request);
+}
+
 // FIXME: This function is a promise-based wrapper around "read all bytes". The spec changed this function to not use promises
 //        in https://github.com/whatwg/streams/commit/f894acdd417926a2121710803cef593e15127964 - however, it seems that the
 //        FileAPI blob specification has not been updated to match, see: https://github.com/w3c/FileAPI/issues/187.
@@ -204,12 +226,8 @@ JS::NonnullGCPtr<WebIDL::Promise> ReadableStreamDefaultReader::read_all_bytes_de
 
     auto promise = WebIDL::create_promise(realm);
 
-    auto success_steps = [promise, &realm](Vector<ByteBuffer> const& byte_chunks) {
-        ByteBuffer concatenated_byte_chunks;
-        for (auto const& chunk : byte_chunks)
-            concatenated_byte_chunks.append(chunk);
-
-        auto buffer = JS::ArrayBuffer::create(realm, move(concatenated_byte_chunks));
+    auto success_steps = [promise, &realm](ByteBuffer bytes) {
+        auto buffer = JS::ArrayBuffer::create(realm, move(bytes));
         WebIDL::resolve_promise(realm, promise, buffer);
     };
 

Userland/Libraries/LibWeb/Streams/ReadableStreamDefaultReader.h

@@ -37,12 +37,15 @@ class ReadLoopReadRequest final : public ReadRequest {
 
 public:
     // successSteps, which is an algorithm accepting a byte sequence
-    using SuccessSteps = JS::SafeFunction<void(Vector<ByteBuffer> const&)>;
+    using SuccessSteps = JS::SafeFunction<void(ByteBuffer)>;
 
     // failureSteps, which is an algorithm accepting a JavaScript value
     using FailureSteps = JS::SafeFunction<void(JS::Value error)>;
 
-    ReadLoopReadRequest(JS::VM& vm, JS::Realm& realm, ReadableStreamDefaultReader& reader, SuccessSteps success_steps, FailureSteps failure_steps);
+    // AD-HOC: callback triggered on every chunk received from the stream.
+    using ChunkSteps = JS::SafeFunction<void(ByteBuffer)>;
+
+    ReadLoopReadRequest(JS::VM& vm, JS::Realm& realm, ReadableStreamDefaultReader& reader, SuccessSteps success_steps, FailureSteps failure_steps, ChunkSteps chunk_steps = {});
 
     virtual void on_chunk(JS::Value chunk) override;
 
@@ -56,9 +59,10 @@ class ReadLoopReadRequest final : public ReadRequest {
     JS::VM& m_vm;
     JS::NonnullGCPtr<JS::Realm> m_realm;
     JS::NonnullGCPtr<ReadableStreamDefaultReader> m_reader;
-    Vector<ByteBuffer> m_byte_chunks;
+    ByteBuffer m_bytes;
     SuccessSteps m_success_steps;
     FailureSteps m_failure_steps;
+    ChunkSteps m_chunk_steps;
 };
 
 // https://streams.spec.whatwg.org/#readablestreamdefaultreader
@@ -76,6 +80,7 @@ class ReadableStreamDefaultReader final
     JS::NonnullGCPtr<JS::Promise> read();
 
     void read_all_bytes(ReadLoopReadRequest::SuccessSteps, ReadLoopReadRequest::FailureSteps);
+    void read_all_chunks(ReadLoopReadRequest::ChunkSteps, ReadLoopReadRequest::SuccessSteps, ReadLoopReadRequest::FailureSteps);
     JS::NonnullGCPtr<WebIDL::Promise> read_all_bytes_deprecated();
 
     void release_lock();