LibWeb: Mostly implement ReadableByteStreamController.[[PullSteps]]

Author: https://github.com/mattco98
Commit: https://github.com/SerenityOS/serenity/commit/51abecc8bc
Pull-request: https://github.com/SerenityOS/serenity/pull/18297
Reviewed-by: https://github.com/linusg

Userland/Libraries/LibWeb/Streams/AbstractOperations.cpp

@@ -765,6 +765,34 @@ void readable_byte_stream_controller_clear_pending_pull_intos(ReadableByteStream
     controller.pending_pull_intos().clear();
 }
 
+// https://streams.spec.whatwg.org/#abstract-opdef-readablebytestreamcontrollerfillreadrequestfromqueue
+WebIDL::ExceptionOr<void> readable_byte_stream_controller_fill_read_request_from_queue(ReadableByteStreamController& controller, NonnullRefPtr<ReadRequest> read_request)
+{
+    auto& vm = controller.vm();
+    auto& realm = controller.realm();
+
+    // 1. Assert: controller.[[queueTotalSize]] > 0.
+    VERIFY(controller.queue_total_size() > 0);
+
+    // 2. Let entry be controller.[[queue]][0].
+    // 3. Remove entry from controller.[[queue]].
+    auto entry = controller.queue().take_first();
+
+    // 4. Set controller.[[queueTotalSize]] to controller.[[queueTotalSize]] − entry’s byte length.
+    controller.set_queue_total_size(controller.queue_total_size() - entry.byte_length);
+
+    // 5. Perform ! ReadableByteStreamControllerHandleQueueDrain(controller).
+    readable_byte_stream_controller_handle_queue_drain(controller);
+
+    // 6. Let view be ! Construct(%Uint8Array%, « entry’s buffer, entry’s byte offset, entry’s byte length »).
+    auto view = MUST_OR_THROW_OOM(JS::construct(vm, *realm.intrinsics().uint8_array_constructor(), entry.buffer, JS::Value(entry.byte_offset), JS::Value(entry.byte_length)));
+
+    // 7. Perform readRequest’s chunk steps, given view.
+    read_request->on_chunk(view);
+
+    return {};
+}
+
 // https://streams.spec.whatwg.org/#readable-byte-stream-controller-get-desired-size
 Optional<double> readable_byte_stream_controller_get_desired_size(ReadableByteStreamController const& controller)
 {
@@ -783,6 +811,26 @@ Optional<double> readable_byte_stream_controller_get_desired_size(ReadableByteSt
     return controller.strategy_hwm() - controller.queue_total_size();
 }
 
+// https://streams.spec.whatwg.org/#readable-byte-stream-controller-handle-queue-drain
+void readable_byte_stream_controller_handle_queue_drain(ReadableByteStreamController& controller)
+{
+    // 1. Assert: controller.[[stream]].[[state]] is "readable".
+    VERIFY(controller.stream()->state() == ReadableStream::State::Readable);
+
+    // 2. If controller.[[queueTotalSize]] is 0 and controller.[[closeRequested]] is true,
+    if (controller.queue_total_size() == 0 && controller.close_requested()) {
+        // 1. Perform ! ReadableByteStreamControllerClearAlgorithms(controller).
+        readable_byte_stream_controller_clear_algorithms(controller);
+
+        // 2. Perform ! ReadableStreamClose(controller.[[stream]]).
+        readable_stream_close(*controller.stream());
+    }
+    // 3. Otherwise,
+    else {
+        // FIXME: 1. Perform ! ReadableByteStreamControllerCallPullIfNeeded(controller).
+    }
+}
+
 // https://streams.spec.whatwg.org/#readable-byte-stream-controller-invalidate-byob-request
 void readable_byte_stream_controller_invalidate_byob_request(ReadableByteStreamController& controller)
 {

Userland/Libraries/LibWeb/Streams/AbstractOperations.h

@@ -55,7 +55,9 @@ WebIDL::ExceptionOr<void> set_up_readable_stream_default_controller_from_underly
 
 void readable_byte_stream_controller_clear_algorithms(ReadableByteStreamController&);
 void readable_byte_stream_controller_clear_pending_pull_intos(ReadableByteStreamController&);
+WebIDL::ExceptionOr<void> readable_byte_stream_controller_fill_read_request_from_queue(ReadableByteStreamController&, NonnullRefPtr<ReadRequest>);
 Optional<double> readable_byte_stream_controller_get_desired_size(ReadableByteStreamController const&);
+void readable_byte_stream_controller_handle_queue_drain(ReadableByteStreamController&);
 void readable_byte_stream_controller_invalidate_byob_request(ReadableByteStreamController&);
 
 WebIDL::ExceptionOr<JS::NonnullGCPtr<WritableStreamDefaultWriter>> acquire_writable_stream_default_writer(WritableStream&);

Userland/Libraries/LibWeb/Streams/ReadableByteStreamController.cpp

@@ -7,6 +7,7 @@
 #include <LibWeb/Streams/ReadableByteStreamController.h>
 #include <LibWeb/Streams/ReadableStream.h>
 #include <LibWeb/Streams/ReadableStreamBYOBRequest.h>
+#include <LibWeb/Streams/ReadableStreamDefaultReader.h>
 
 namespace Web::Streams {
 
@@ -41,6 +42,70 @@ WebIDL::ExceptionOr<JS::GCPtr<WebIDL::Promise>> ReadableByteStreamController::ca
     return result;
 }
 
+// https://streams.spec.whatwg.org/#rbs-controller-private-pull
+WebIDL::ExceptionOr<void> ReadableByteStreamController::pull_steps(NonnullRefPtr<ReadRequest> read_request)
+{
+    auto& vm = this->vm();
+    auto& realm = this->realm();
+
+    // 1. Let stream be this.[[stream]].
+
+    // 2. Assert: ! ReadableStreamHasDefaultReader(stream) is true.
+    VERIFY(readable_stream_has_default_reader(*m_stream));
+
+    // 3. If this.[[queueTotalSize]] > 0,
+    if (m_queue_total_size > 0) {
+        // 1. Assert: ! ReadableStreamGetNumReadRequests(stream) is 0.
+        VERIFY(readable_stream_get_num_read_requests(*m_stream) == 0);
+
+        // 2. Perform ! ReadableByteStreamControllerFillReadRequestFromQueue(this, readRequest).
+        TRY(readable_byte_stream_controller_fill_read_request_from_queue(*this, read_request));
+
+        // 3. Return.
+        return {};
+    }
+
+    // 4. Let autoAllocateChunkSize be this.[[autoAllocateChunkSize]].
+
+    // 5. If autoAllocateChunkSize is not undefined,
+    if (m_auto_allocate_chunk_size.has_value()) {
+        // 1. Let buffer be Construct(%ArrayBuffer%, « autoAllocateChunkSize »).
+        auto buffer = JS::ArrayBuffer::create(realm, *m_auto_allocate_chunk_size);
+
+        // 2. If buffer is an abrupt completion,
+        if (buffer.is_throw_completion()) {
+            // 1. Perform readRequest’s error steps, given buffer.[[Value]].
+            read_request->on_error(*buffer.throw_completion().value());
+
+            // 2. Return.
+            return {};
+        }
+
+        // 3. Let pullIntoDescriptor be a new pull-into descriptor with buffer buffer.[[Value]], buffer byte length autoAllocateChunkSize, byte offset 0,
+        //    byte length autoAllocateChunkSize, bytes filled 0, element size 1, view constructor %Uint8Array%, and reader type "default".
+        PullIntoDescriptor pull_into_descriptor {
+            .buffer = buffer.release_value(),
+            .buffer_byte_length = *m_auto_allocate_chunk_size,
+            .byte_offset = 0,
+            .byte_length = *m_auto_allocate_chunk_size,
+            .bytes_filled = 0,
+            .element_size = 1,
+            .view_constructor = *realm.intrinsics().uint8_array_constructor(),
+            .reader_type = ReaderType::Default,
+        };
+
+        // 4. Append pullIntoDescriptor to this.[[pendingPullIntos]].
+        TRY_OR_THROW_OOM(vm, m_pending_pull_intos.try_append(move(pull_into_descriptor)));
+    }
+
+    // 6. Perform ! ReadableStreamAddReadRequest(stream, readRequest).
+    readable_stream_add_read_request(*m_stream, read_request);
+
+    // FIXME: 7. Perform ! ReadableByteStreamControllerCallPullIfNeeded(this).
+
+    return {};
+}
+
 void ReadableByteStreamController::visit_edges(Cell::Visitor& visitor)
 {
     Base::visit_edges(visitor);

Userland/Libraries/LibWeb/Streams/ReadableByteStreamController.h

@@ -117,6 +117,7 @@ public:
     void set_stream(JS::GCPtr<ReadableStream> stream) { m_stream = stream; }
 
     WebIDL::ExceptionOr<JS::GCPtr<WebIDL::Promise>> cancel_steps(JS::Value reason);
+    WebIDL::ExceptionOr<void> pull_steps(NonnullRefPtr<ReadRequest>);
 
 private:
     explicit ReadableByteStreamController(JS::Realm&);