LibWeb: Implement ReadableStream.pipeTo()

Author: https://github.com/kennethmyhra
Commit: d3b2cd8ab6
Pull-request: https://github.com/SerenityOS/serenity/pull/23869
Reviewed-by: https://github.com/shannonbooth ✅

Userland/Libraries/LibWeb/Streams/ReadableStream.cpp

@@ -1,12 +1,14 @@
 /*
  * Copyright (c) 2022, Linus Groh <linusg@serenityos.org>
  * Copyright (c) 2023-2024, Shannon Booth <shannon@serenityos.org>
+ * Copyright (c) 2024, Kenneth Myhra <kennethmyhra@serenityos.org>
  *
  * SPDX-License-Identifier: BSD-2-Clause
  */
 
 #include <LibJS/Runtime/PromiseCapability.h>
 #include <LibWeb/Bindings/Intrinsics.h>
+#include <LibWeb/DOM/AbortSignal.h>
 #include <LibWeb/Streams/AbstractOperations.h>
 #include <LibWeb/Streams/ReadableByteStreamController.h>
 #include <LibWeb/Streams/ReadableStream.h>
@@ -108,6 +110,31 @@ WebIDL::ExceptionOr<ReadableStreamReader> ReadableStream::get_reader(ReadableStr
     return ReadableStreamReader { TRY(acquire_readable_stream_byob_reader(*this)) };
 }
 
+WebIDL::ExceptionOr<JS::NonnullGCPtr<JS::Object>> ReadableStream::pipe_to(WritableStream& destination, StreamPipeOptions const& options)
+{
+    auto& realm = this->realm();
+
+    // 1. If ! IsReadableStreamLocked(this) is true, return a promise rejected with a TypeError exception.
+    if (is_readable_stream_locked(*this)) {
+        auto promise = WebIDL::create_promise(realm);
+        WebIDL::reject_promise(realm, promise, JS::TypeError::create(realm, "Failed to execute 'pipeTo' on 'ReadableStream': Cannot pipe a locked stream"sv));
+        return promise->promise();
+    }
+
+    // 2. If ! IsWritableStreamLocked(destination) is true, return a promise rejected with a TypeError exception.
+    if (is_writable_stream_locked(destination)) {
+        auto promise = WebIDL::create_promise(realm);
+        WebIDL::reject_promise(realm, promise, JS::TypeError::create(realm, "Failed to execute 'pipeTo' on 'ReadableStream':  Cannot pipe to a locked stream"sv));
+        return promise->promise();
+    }
+
+    // 3. Let signal be options["signal"] if it exists, or undefined otherwise.
+    auto signal = options.signal.has_value() ? JS::Value(options.signal.value().ptr()) : JS::js_undefined();
+
+    // 4. Return ! ReadableStreamPipeTo(this, destination, options["preventClose"], options["preventAbort"], options["preventCancel"], signal).
+    return MUST(readable_stream_pipe_to(*this, destination, options.prevent_close, options.prevent_abort, options.prevent_cancel, signal))->promise();
+}
+
 // https://streams.spec.whatwg.org/#readablestream-tee
 WebIDL::ExceptionOr<ReadableStreamPair> ReadableStream::tee()
 {

Userland/Libraries/LibWeb/Streams/ReadableStream.h

@@ -1,5 +1,6 @@
 /*
  * Copyright (c) 2022, Linus Groh <linusg@serenityos.org>
+ * Copyright (c) 2024, Kenneth Myhra <kennethmyhra@serenityos.org>
  *
  * SPDX-License-Identifier: BSD-2-Clause
  */
@@ -26,6 +27,13 @@ struct ReadableStreamGetReaderOptions {
     Optional<Bindings::ReadableStreamReaderMode> mode;
 };
 
+struct StreamPipeOptions {
+    bool prevent_close { false };
+    bool prevent_abort { false };
+    bool prevent_cancel { false };
+    Optional<JS::NonnullGCPtr<DOM::AbortSignal>> signal;
+};
+
 struct ReadableStreamPair {
     // Define a couple container-like methods so this type may be used as the return type of the IDL `tee` implementation.
     size_t size() const { return 2; }
@@ -62,6 +70,7 @@ public:
     bool locked() const;
     WebIDL::ExceptionOr<JS::GCPtr<JS::Object>> cancel(JS::Value reason);
     WebIDL::ExceptionOr<ReadableStreamReader> get_reader(ReadableStreamGetReaderOptions const& = {});
+    WebIDL::ExceptionOr<JS::NonnullGCPtr<JS::Object>> pipe_to(WritableStream& destination, StreamPipeOptions const& = {});
     WebIDL::ExceptionOr<ReadableStreamPair> tee();
 
     Optional<ReadableStreamController>& controller() { return m_controller; }

Userland/Libraries/LibWeb/Streams/ReadableStream.idl

@@ -1,6 +1,15 @@
+#import <DOM/AbortSignal.idl>
 #import <Streams/QueuingStrategy.idl>
 #import <Streams/ReadableStreamBYOBReader.idl>
 #import <Streams/ReadableStreamDefaultReader.idl>
+#import <Streams/WritableStream.idl>
+
+dictionary StreamPipeOptions {
+    boolean preventClose = false;
+    boolean preventAbort = false;
+    boolean preventCancel = false;
+    AbortSignal signal;
+};
 
 // https://streams.spec.whatwg.org/#enumdef-readablestreamreadermode
 enum ReadableStreamReaderMode { "byob" };
@@ -22,7 +31,7 @@ interface ReadableStream {
     Promise<undefined> cancel(optional any reason);
     ReadableStreamReader getReader(optional ReadableStreamGetReaderOptions options = {});
     // FIXME: ReadableStream pipeThrough(ReadableWritablePair transform, optional StreamPipeOptions options = {});
-    // FIXME: Promise<undefined> pipeTo(WritableStream destination, optional StreamPipeOptions options = {});
+    Promise<undefined> pipeTo(WritableStream destination, optional StreamPipeOptions options = {});
     sequence<ReadableStream> tee();
 
     // FIXME: async iterable<any>(optional ReadableStreamIteratorOptions options = {});