LibWeb: Implement transferable streams AOs

These AOs are common between transferable ReadableStream, WritableStream
and TransformStream objects.

Libraries/LibWeb/Streams/AbstractOperations.cpp

@@ -9,12 +9,27 @@
  */
 
 #include <LibJS/Runtime/ArrayBuffer.h>
+#include <LibJS/Runtime/NativeFunction.h>
 #include <LibJS/Runtime/TypedArray.h>
+#include <LibWeb/Bindings/ExceptionOrUtils.h>
+#include <LibWeb/DOM/IDLEventListener.h>
+#include <LibWeb/HTML/EventNames.h>
+#include <LibWeb/HTML/MessageEvent.h>
+#include <LibWeb/HTML/MessagePort.h>
 #include <LibWeb/HTML/StructuredSerialize.h>
+#include <LibWeb/HTML/StructuredSerializeOptions.h>
 #include <LibWeb/Streams/AbstractOperations.h>
 #include <LibWeb/Streams/QueuingStrategy.h>
+#include <LibWeb/Streams/ReadableStream.h>
+#include <LibWeb/Streams/ReadableStreamDefaultController.h>
+#include <LibWeb/Streams/ReadableStreamOperations.h>
+#include <LibWeb/Streams/WritableStream.h>
+#include <LibWeb/Streams/WritableStreamDefaultController.h>
+#include <LibWeb/Streams/WritableStreamOperations.h>
 #include <LibWeb/WebIDL/AbstractOperations.h>
 #include <LibWeb/WebIDL/Buffers.h>
+#include <LibWeb/WebIDL/CallbackType.h>
+#include <LibWeb/WebIDL/DOMException.h>
 
 namespace Web::Streams {
 
@@ -50,6 +65,359 @@ GC::Ref<SizeAlgorithm> extract_size_algorithm(JS::VM& vm, QueuingStrategy const&
     });
 }
 
+struct PromiseHolder : public JS::Cell {
+    GC_CELL(PromiseHolder, JS::Cell);
+    GC_DECLARE_ALLOCATOR(PromiseHolder);
+
+    explicit PromiseHolder(GC::Ptr<WebIDL::Promise> promise)
+        : promise(promise)
+    {
+    }
+
+    virtual void visit_edges(JS::Cell::Visitor& visitor) override
+    {
+        Base::visit_edges(visitor);
+        visitor.visit(promise);
+    }
+
+    GC::Ptr<WebIDL::Promise> promise;
+};
+
+GC_DEFINE_ALLOCATOR(PromiseHolder);
+
+static void add_message_event_listener(JS::Realm& realm, HTML::MessagePort& port, FlyString const& name, Function<void(JS::VM&, HTML::MessageEvent const&)> handler)
+{
+    auto behavior = [handler = GC::create_function(realm.heap(), move(handler))](JS::VM& vm) {
+        auto event = vm.argument(0);
+        VERIFY(event.is_object());
+
+        auto& message_event = as<HTML::MessageEvent>(event.as_object());
+        handler->function()(vm, message_event);
+
+        return JS::js_undefined();
+    };
+
+    auto function = JS::NativeFunction::create(realm, move(behavior), 1, FlyString {}, &realm);
+    auto callback = realm.heap().allocate<WebIDL::CallbackType>(function, realm);
+    auto listener = DOM::IDLEventListener::create(realm, callback);
+
+    port.add_event_listener_without_options(name, listener);
+}
+
+// https://streams.spec.whatwg.org/#abstract-opdef-crossrealmtransformsenderror
+void cross_realm_transform_send_error(JS::Realm& realm, HTML::MessagePort& port, JS::Value error)
+{
+    // 1. Perform PackAndPostMessage(port, "error", error), discarding the result.
+    (void)pack_and_post_message(realm, port, "error"sv, error);
+}
+
+// https://streams.spec.whatwg.org/#abstract-opdef-packandpostmessagehandlingerror
+WebIDL::ExceptionOr<void> pack_and_post_message(JS::Realm& realm, HTML::MessagePort& port, StringView type, JS::Value value)
+{
+    auto& vm = realm.vm();
+
+    // 1. Let message be OrdinaryObjectCreate(null).
+    auto message = JS::Object::create(realm, nullptr);
+
+    // 2. Perform ! CreateDataProperty(message, "type", type).
+    MUST(message->create_data_property(vm.names.type, JS::PrimitiveString::create(vm, type)));
+
+    // 3. Perform ! CreateDataProperty(message, "value", value).
+    MUST(message->create_data_property(vm.names.value, value));
+
+    // 4. Let targetPort be the port with which port is entangled, if any; otherwise let it be null.
+    auto target_port = port.entangled_port();
+
+    // 5. Let options be «[ "transfer" → « » ]».
+    HTML::StructuredSerializeOptions options { .transfer = {} };
+
+    // 6. Run the message port post message steps providing targetPort, message, and options.
+    return port.message_port_post_message_steps(target_port, message, options);
+}
+
+// https://streams.spec.whatwg.org/#abstract-opdef-packandpostmessagehandlingerror
+WebIDL::ExceptionOr<void> pack_and_post_message_handling_error(JS::Realm& realm, HTML::MessagePort& port, StringView type, JS::Value value)
+{
+    // 1. Let result be PackAndPostMessage(port, type, value).
+    auto result = pack_and_post_message(realm, port, type, value);
+
+    // 2. If result is an abrupt completion,
+    if (result.is_exception()) {
+        auto error = Bindings::exception_to_throw_completion(realm.vm(), result.release_error());
+
+        // 1. Perform ! CrossRealmTransformSendError(port, result.[[Value]]).
+        cross_realm_transform_send_error(realm, port, error.value());
+    }
+
+    // 3. Return result as a completion record.
+    return result;
+}
+
+// https://streams.spec.whatwg.org/#abstract-opdef-setupcrossrealmtransformreadable
+void set_up_cross_realm_transform_readable(JS::Realm& realm, ReadableStream& stream, HTML::MessagePort& port)
+{
+    // 1. Perform ! InitializeReadableStream(stream).
+    initialize_readable_stream(stream);
+
+    // 2. Let controller be a new ReadableStreamDefaultController.
+    auto controller = realm.create<ReadableStreamDefaultController>(realm);
+
+    // 3. Add a handler for port’s message event with the following steps:
+    add_message_event_listener(realm, port, HTML::EventNames::message,
+        [&port, controller](JS::VM& vm, HTML::MessageEvent const& message) {
+            // 1. Let data be the data of the message.
+            auto data = message.data();
+
+            // 2. Assert: data is an Object.
+            VERIFY(data.is_object());
+
+            // 3. Let type be ! Get(data, "type").
+            auto type = MUST(data.get(vm, vm.names.type));
+
+            // 4. Let value be ! Get(data, "value").
+            auto value = MUST(data.get(vm, vm.names.value));
+
+            // 5. Assert: type is a String.
+            auto type_string = type.as_string().utf8_string_view();
+
+            // 6. If type is "chunk",
+            if (type_string == "chunk"sv) {
+                // 1. Perform ! ReadableStreamDefaultControllerEnqueue(controller, value).
+                MUST(readable_stream_default_controller_enqueue(controller, value));
+            }
+            // 7. Otherwise, if type is "close",
+            else if (type_string == "close"sv) {
+                // 1. Perform ! ReadableStreamDefaultControllerClose(controller).
+                readable_stream_default_controller_close(controller);
+
+                // 2. Disentangle port.
+                port.disentangle();
+            }
+            // 8. Otherwise, if type is "error",
+            else if (type_string == "error"sv) {
+                // 1. Perform ! ReadableStreamDefaultControllerError(controller, value).
+                readable_stream_default_controller_error(controller, value);
+
+                // 2. Disentangle port.
+                port.disentangle();
+            }
+        });
+
+    // 4. Add a handler for port’s messageerror event with the following steps:
+    add_message_event_listener(realm, port, HTML::EventNames::messageerror,
+        [&realm, &port, controller](JS::VM&, HTML::MessageEvent const&) {
+            // 1. Let error be a new "DataCloneError" DOMException.
+            auto error = WebIDL::DataCloneError::create(realm, "Unable to transfer stream"_string);
+
+            // 2. Perform ! CrossRealmTransformSendError(port, error).
+            cross_realm_transform_send_error(realm, port, error);
+
+            // 3. Perform ! ReadableStreamDefaultControllerError(controller, error).
+            readable_stream_default_controller_error(controller, error);
+
+            // 4. Disentangle port.
+            port.disentangle();
+        });
+
+    // FIXME: 5. Enable port’s port message queue.
+
+    // 6. Let startAlgorithm be an algorithm that returns undefined.
+    auto start_algorithm = GC::create_function(realm.heap(), []() -> WebIDL::ExceptionOr<JS::Value> {
+        return JS::js_undefined();
+    });
+
+    // 7. Let pullAlgorithm be the following steps:
+    auto pull_algorithm = GC::create_function(realm.heap(), [&realm, &port]() -> GC::Ref<WebIDL::Promise> {
+        // 1. Perform ! PackAndPostMessage(port, "pull", undefined).
+        MUST(pack_and_post_message(realm, port, "pull"sv, JS::js_undefined()));
+
+        // 2. Return a promise resolved with undefined.
+        return WebIDL::create_resolved_promise(realm, JS::js_undefined());
+    });
+
+    // 8. Let cancelAlgorithm be the following steps, taking a reason argument:
+    auto cancel_algorithm = GC::create_function(realm.heap(), [&realm, &port](JS::Value reason) -> GC::Ref<WebIDL::Promise> {
+        // 1. Let result be PackAndPostMessageHandlingError(port, "error", reason).
+        auto result = pack_and_post_message_handling_error(realm, port, "error"sv, reason);
+
+        // 2. Disentangle port.
+        port.disentangle();
+
+        // 3. If result is an abrupt completion, return a promise rejected with result.[[Value]].
+        if (result.is_error())
+            return WebIDL::create_rejected_promise_from_exception(realm, result.release_error());
+
+        // 4. Otherwise, return a promise resolved with undefined.
+        return WebIDL::create_resolved_promise(realm, JS::js_undefined());
+    });
+
+    // 9. Let sizeAlgorithm be an algorithm that returns 1.
+    auto size_algorithm = GC::create_function(realm.heap(), [](JS::Value) -> JS::Completion {
+        return JS::Value { 1 };
+    });
+
+    // 10. Perform ! SetUpReadableStreamDefaultController(stream, controller, startAlgorithm, pullAlgorithm, cancelAlgorithm, 0, sizeAlgorithm).
+    MUST(set_up_readable_stream_default_controller(stream, controller, start_algorithm, pull_algorithm, cancel_algorithm, 0, size_algorithm));
+}
+
+// https://streams.spec.whatwg.org/#abstract-opdef-setupcrossrealmtransformwritable
+void set_up_cross_realm_transform_writable(JS::Realm& realm, WritableStream& stream, HTML::MessagePort& port)
+{
+    // 1. Perform ! InitializeWritableStream(stream).
+    initialize_writable_stream(stream);
+
+    // 2. Let controller be a new WritableStreamDefaultController.
+    auto controller = realm.create<WritableStreamDefaultController>(realm);
+
+    // 3. Let backpressurePromise be a new promise.
+    auto backpressure_promise = realm.heap().allocate<PromiseHolder>(WebIDL::create_promise(realm));
+
+    // 4. Add a handler for port’s message event with the following steps:
+    add_message_event_listener(realm, port, HTML::EventNames::message,
+        [&realm, controller, backpressure_promise](JS::VM& vm, HTML::MessageEvent const& message) {
+            // 1. Let data be the data of the message.
+            auto data = message.data();
+
+            // 2. Assert: data is an Object.
+            VERIFY(data.is_object());
+
+            // 3. Let type be ! Get(data, "type").
+            auto type = MUST(data.get(vm, vm.names.type));
+
+            // 4. Let value be ! Get(data, "value").
+            auto value = MUST(data.get(vm, vm.names.value));
+
+            // 5. Assert: type is a String.
+            auto type_string = type.as_string().utf8_string_view();
+
+            // 6. If type is "pull",
+            if (type_string == "pull"sv) {
+                // 1. If backpressurePromise is not undefined,
+                if (backpressure_promise->promise) {
+                    // 1. Resolve backpressurePromise with undefined.
+                    WebIDL::resolve_promise(realm, *backpressure_promise->promise, JS::js_undefined());
+
+                    // 2. Set backpressurePromise to undefined.
+                    backpressure_promise->promise = nullptr;
+                }
+            }
+            // 7. Otherwise, if type is "error",
+            else if (type_string == "error"sv) {
+                // 1. Perform ! WritableStreamDefaultControllerErrorIfNeeded(controller, value).
+                writable_stream_default_controller_error_if_needed(controller, value);
+
+                // 2. If backpressurePromise is not undefined,
+                if (backpressure_promise->promise) {
+                    // 1. Resolve backpressurePromise with undefined.
+                    WebIDL::resolve_promise(realm, *backpressure_promise->promise, JS::js_undefined());
+
+                    // 2. Set backpressurePromise to undefined.
+                    backpressure_promise->promise = nullptr;
+                }
+            }
+        });
+
+    // 5. Add a handler for port’s messageerror event with the following steps:
+    add_message_event_listener(realm, port, HTML::EventNames::messageerror,
+        [&realm, &port, controller](JS::VM&, HTML::MessageEvent const&) {
+            // 1. Let error be a new "DataCloneError" DOMException
+            auto error = WebIDL::DataCloneError::create(realm, "Unable to transfer stream"_string);
+
+            // 2. Perform ! CrossRealmTransformSendError(port, error).
+            cross_realm_transform_send_error(realm, port, error);
+
+            // 3. Perform ! WritableStreamDefaultControllerErrorIfNeeded(controller, error).
+            writable_stream_default_controller_error_if_needed(controller, error);
+
+            // 4. Disentangle port.
+            port.disentangle();
+        });
+
+    // FIXME: 6. Enable port’s port message queue.
+
+    // 7. Let startAlgorithm be an algorithm that returns undefined.
+    auto start_algorithm = GC::create_function(realm.heap(), []() -> WebIDL::ExceptionOr<JS::Value> {
+        return JS::js_undefined();
+    });
+
+    // 8. Let writeAlgorithm be the following steps, taking a chunk argument:
+    auto write_algorithm = GC::create_function(realm.heap(), [&realm, &port, backpressure_promise](JS::Value chunk) -> GC::Ref<WebIDL::Promise> {
+        // 1. If backpressurePromise is undefined, set backpressurePromise to a promise resolved with undefined.
+        if (!backpressure_promise->promise)
+            backpressure_promise->promise = WebIDL::create_resolved_promise(realm, JS::js_undefined());
+
+        // FIXME: The steps below ("Return a promise rejected/resolved with ...") seem to indicate we should be creating
+        //        a promise on-the-fly. But in order for the error from PackAndPostMessageHandlingError to be propagated
+        //        back to the original ReadableStream, we must actually fulfill the promise created from reacting to the
+        //        backpressure promise. This is explicitly tested by WPT.
+        auto reaction_promise = realm.heap().allocate<PromiseHolder>(nullptr);
+
+        // 2. Return the result of reacting to backpressurePromise with the following fulfillment steps:
+        reaction_promise->promise = WebIDL::upon_fulfillment(*backpressure_promise->promise,
+            GC::create_function(realm.heap(), [&realm, &port, backpressure_promise, reaction_promise, chunk](JS::Value) -> WebIDL::ExceptionOr<JS::Value> {
+                // 1. Set backpressurePromise to a new promise.
+                backpressure_promise->promise = WebIDL::create_promise(realm);
+
+                // 2. Let result be PackAndPostMessageHandlingError(port, "chunk", chunk).
+                auto result = pack_and_post_message_handling_error(realm, port, "chunk"sv, chunk);
+
+                // 3. If result is an abrupt completion,
+                if (result.is_error()) {
+                    // 1. Disentangle port.
+                    port.disentangle();
+
+                    // 2. Return a promise rejected with result.[[Value]].
+                    auto error = Bindings::exception_to_throw_completion(realm.vm(), result.release_error());
+                    WebIDL::reject_promise(realm, *reaction_promise->promise, error.value());
+                }
+                // 4. Otherwise, return a promise resolved with undefined.
+                else {
+                    WebIDL::resolve_promise(realm, *reaction_promise->promise, JS::js_undefined());
+                }
+
+                return reaction_promise->promise;
+            }));
+
+        return *reaction_promise->promise;
+    });
+
+    // 9. Let closeAlgorithm be the folowing steps:
+    auto close_algorithm = GC::create_function(realm.heap(), [&realm, &port]() -> GC::Ref<WebIDL::Promise> {
+        // 1. Perform ! PackAndPostMessage(port, "close", undefined).
+        MUST(pack_and_post_message(realm, port, "close"sv, JS::js_undefined()));
+
+        // 2. Disentangle port.
+        port.disentangle();
+
+        // 3. Return a promise resolved with undefined.
+        return WebIDL::create_resolved_promise(realm, JS::js_undefined());
+    });
+
+    // 10. Let abortAlgorithm be the following steps, taking a reason argument:
+    auto abort_algorithm = GC::create_function(realm.heap(), [&realm, &port](JS::Value reason) -> GC::Ref<WebIDL::Promise> {
+        // 1. Let result be PackAndPostMessageHandlingError(port, "error", reason).
+        auto result = pack_and_post_message_handling_error(realm, port, "error"sv, reason);
+
+        // 2. Disentangle port.
+        port.disentangle();
+
+        // 3. If result is an abrupt completion, return a promise rejected with result.[[Value]].
+        if (result.is_error())
+            return WebIDL::create_rejected_promise_from_exception(realm, result.release_error());
+
+        // 4. Otherwise, return a promise resolved with undefined.
+        return WebIDL::create_resolved_promise(realm, JS::js_undefined());
+    });
+
+    // 11. Let sizeAlgorithm be an algorithm that returns 1.
+    auto size_algorithm = GC::create_function(realm.heap(), [](JS::Value) -> JS::Completion {
+        return JS::Value { 1 };
+    });
+
+    // 12. Perform ! SetUpWritableStreamDefaultController(stream, controller, startAlgorithm, writeAlgorithm, closeAlgorithm, abortAlgorithm, 1, sizeAlgorithm).
+    MUST(set_up_writable_stream_default_controller(stream, controller, start_algorithm, write_algorithm, close_algorithm, abort_algorithm, 1, size_algorithm));
+}
+
 // https://streams.spec.whatwg.org/#can-transfer-array-buffer
 bool can_transfer_array_buffer(JS::ArrayBuffer const& array_buffer)
 {