mirrors/ladybird
0
0
Fork 0
mirror of https://github.com/LadybirdBrowser/ladybird.git synced 2025-07-28 11:49:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10

LibWeb: Move ReadableStream AOs into their own file

Browse source 
The main streams AO file has gotten very large, and is a bit difficult
to navigate. In an effort to improve DX, this migrates ReadableStream
AOs to their own file. And the helper classes used for the tee and pipe-
to operations are also in their own files.
This commit is contained in:
Timothy Flynn 2025-04-17 15:47:53 -04:00 • committed by Tim Flynn
commit a9ddd427cb
Notes: github-actions[bot] 2025-04-18 10:56:51 +00:00
34 changed files with 4145 additions and 3968 deletions
Download patch file Download diff file Expand all files Collapse all files

3
Libraries/LibWeb/CMakeLists.txt
View file

@ -749,6 +749,9 @@ set(SOURCES
Streams/ReadableStreamDefaultController.cpp Streams/ReadableStreamDefaultController.cpp
Streams/ReadableStreamDefaultReader.cpp Streams/ReadableStreamDefaultReader.cpp
Streams/ReadableStreamGenericReader.cpp Streams/ReadableStreamGenericReader.cpp
Streams/ReadableStreamOperations.cpp
Streams/ReadableStreamPipeTo.cpp
Streams/ReadableStreamTee.cpp
Streams/Transformer.cpp Streams/Transformer.cpp
Streams/TransformStream.cpp Streams/TransformStream.cpp
Streams/TransformStreamDefaultController.cpp Streams/TransformStreamDefaultController.cpp

1
Libraries/LibWeb/Compression/DecompressionStream.cpp
View file

@ -15,7 +15,6 @@
#include <LibWeb/Bindings/ExceptionOrUtils.h> #include <LibWeb/Bindings/ExceptionOrUtils.h>
#include <LibWeb/Bindings/Intrinsics.h> #include <LibWeb/Bindings/Intrinsics.h>
#include <LibWeb/Compression/DecompressionStream.h> #include <LibWeb/Compression/DecompressionStream.h>
#include <LibWeb/Streams/AbstractOperations.h>
#include <LibWeb/Streams/TransformStream.h> #include <LibWeb/Streams/TransformStream.h>
#include <LibWeb/WebIDL/AbstractOperations.h> #include <LibWeb/WebIDL/AbstractOperations.h>

2
Libraries/LibWeb/Fetch/BodyInit.cpp
View file

@ -14,7 +14,7 @@
#include <LibWeb/HTML/FormControlInfrastructure.h> #include <LibWeb/HTML/FormControlInfrastructure.h>
#include <LibWeb/HTML/Scripting/TemporaryExecutionContext.h> #include <LibWeb/HTML/Scripting/TemporaryExecutionContext.h>
#include <LibWeb/Platform/EventLoopPlugin.h> #include <LibWeb/Platform/EventLoopPlugin.h>
#include <LibWeb/Streams/AbstractOperations.h> #include <LibWeb/Streams/ReadableStream.h>
#include <LibWeb/WebIDL/AbstractOperations.h> #include <LibWeb/WebIDL/AbstractOperations.h>
#include <LibWeb/WebIDL/Buffers.h> #include <LibWeb/WebIDL/Buffers.h>
#include <LibWeb/WebIDL/ExceptionOr.h> #include <LibWeb/WebIDL/ExceptionOr.h>

2
Libraries/LibWeb/Fetch/FetchMethod.cpp
View file

@ -19,7 +19,7 @@
#include <LibWeb/Fetch/Request.h> #include <LibWeb/Fetch/Request.h>
#include <LibWeb/Fetch/Response.h> #include <LibWeb/Fetch/Response.h>
#include <LibWeb/HTML/Scripting/TemporaryExecutionContext.h> #include <LibWeb/HTML/Scripting/TemporaryExecutionContext.h>
#include <LibWeb/Streams/AbstractOperations.h> #include <LibWeb/Streams/ReadableStreamOperations.h>
#include <LibWeb/WebIDL/ExceptionOr.h> #include <LibWeb/WebIDL/ExceptionOr.h>
#include <LibWeb/WebIDL/Promise.h> #include <LibWeb/WebIDL/Promise.h>

2
Libraries/LibWeb/Fetch/Fetching/FetchedDataReceiver.cpp
View file

@ -11,7 +11,7 @@
#include <LibWeb/Fetch/Infrastructure/Task.h> #include <LibWeb/Fetch/Infrastructure/Task.h>
#include <LibWeb/HTML/Scripting/ExceptionReporter.h> #include <LibWeb/HTML/Scripting/ExceptionReporter.h>
#include <LibWeb/HTML/Scripting/TemporaryExecutionContext.h> #include <LibWeb/HTML/Scripting/TemporaryExecutionContext.h>
#include <LibWeb/Streams/AbstractOperations.h> #include <LibWeb/Streams/ReadableStream.h>
#include <LibWeb/WebIDL/Promise.h> #include <LibWeb/WebIDL/Promise.h>
namespace Web::Fetch::Fetching { namespace Web::Fetch::Fetching {

1
Libraries/LibWeb/Fetch/Fetching/Fetching.cpp
View file

@ -57,6 +57,7 @@
#include <LibWeb/ResourceTiming/PerformanceResourceTiming.h> #include <LibWeb/ResourceTiming/PerformanceResourceTiming.h>
#include <LibWeb/SRI/SRI.h> #include <LibWeb/SRI/SRI.h>
#include <LibWeb/SecureContexts/AbstractOperations.h> #include <LibWeb/SecureContexts/AbstractOperations.h>
#include <LibWeb/Streams/AbstractOperations.h>
#include <LibWeb/Streams/TransformStream.h> #include <LibWeb/Streams/TransformStream.h>
#include <LibWeb/Streams/TransformStreamDefaultController.h> #include <LibWeb/Streams/TransformStreamDefaultController.h>
#include <LibWeb/Streams/Transformer.h> #include <LibWeb/Streams/Transformer.h>

6
Libraries/LibWeb/Fetch/Infrastructure/HTTP/Bodies.cpp
View file

@ -13,7 +13,7 @@
#include <LibWeb/Fetch/Infrastructure/IncrementalReadLoopReadRequest.h> #include <LibWeb/Fetch/Infrastructure/IncrementalReadLoopReadRequest.h>
#include <LibWeb/Fetch/Infrastructure/Task.h> #include <LibWeb/Fetch/Infrastructure/Task.h>
#include <LibWeb/HTML/Scripting/TemporaryExecutionContext.h> #include <LibWeb/HTML/Scripting/TemporaryExecutionContext.h>
#include <LibWeb/Streams/AbstractOperations.h> #include <LibWeb/Streams/ReadableStream.h>
namespace Web::Fetch::Infrastructure { namespace Web::Fetch::Infrastructure {
@ -30,12 +30,12 @@ GC::Ref<Body> Body::create(JS::VM& vm, GC::Ref<Streams::ReadableStream> stream,
} }
Body::Body(GC::Ref<Streams::ReadableStream> stream) Body::Body(GC::Ref<Streams::ReadableStream> stream)
: m_stream(move(stream)) : m_stream(stream)
{ {
} }
Body::Body(GC::Ref<Streams::ReadableStream> stream, SourceType source, Optional<u64> length) Body::Body(GC::Ref<Streams::ReadableStream> stream, SourceType source, Optional<u64> length)
: m_stream(move(stream)) : m_stream(stream)
, m_source(move(source)) , m_source(move(source))
, m_length(move(length)) , m_length(move(length))
{ {

6
Libraries/LibWeb/FileAPI/Blob.cpp
View file

@ -19,8 +19,8 @@
#include <LibWeb/HTML/StructuredSerialize.h> #include <LibWeb/HTML/StructuredSerialize.h>
#include <LibWeb/Infra/Strings.h> #include <LibWeb/Infra/Strings.h>
#include <LibWeb/MimeSniff/MimeType.h> #include <LibWeb/MimeSniff/MimeType.h>
#include <LibWeb/Streams/AbstractOperations.h>
#include <LibWeb/Streams/ReadableStreamDefaultReader.h> #include <LibWeb/Streams/ReadableStreamDefaultReader.h>
#include <LibWeb/Streams/ReadableStreamOperations.h>
#include <LibWeb/WebIDL/AbstractOperations.h> #include <LibWeb/WebIDL/AbstractOperations.h>
#include <LibWeb/WebIDL/Buffers.h> #include <LibWeb/WebIDL/Buffers.h>
@ -350,14 +350,14 @@ GC::Ref<Streams::ReadableStream> Blob::get_stream()
}); });
if (maybe_error.is_error()) { if (maybe_error.is_error()) {
readable_stream_error(*stream, maybe_error.release_error().value()); Streams::readable_stream_error(*stream, maybe_error.release_error().value());
return; return;
} }
// FIXME: Spec bug: https://github.com/w3c/FileAPI/issues/206 // FIXME: Spec bug: https://github.com/w3c/FileAPI/issues/206
// //
// We need to close the stream so that the stream will finish reading. // We need to close the stream so that the stream will finish reading.
readable_stream_close(*stream); Streams::readable_stream_close(*stream);
})); }));
} }
} }

1
Libraries/LibWeb/FileAPI/FileReader.cpp
View file

@ -25,7 +25,6 @@
#include <LibWeb/HTML/Scripting/TemporaryExecutionContext.h> #include <LibWeb/HTML/Scripting/TemporaryExecutionContext.h>
#include <LibWeb/MimeSniff/MimeType.h> #include <LibWeb/MimeSniff/MimeType.h>
#include <LibWeb/Platform/EventLoopPlugin.h> #include <LibWeb/Platform/EventLoopPlugin.h>
#include <LibWeb/Streams/AbstractOperations.h>
#include <LibWeb/Streams/ReadableStream.h> #include <LibWeb/Streams/ReadableStream.h>
#include <LibWeb/Streams/ReadableStreamDefaultReader.h> #include <LibWeb/Streams/ReadableStreamDefaultReader.h>
#include <LibWeb/WebIDL/DOMException.h> #include <LibWeb/WebIDL/DOMException.h>

2
Libraries/LibWeb/FileAPI/FileReaderSync.cpp
View file

@ -12,7 +12,7 @@
#include <LibWeb/FileAPI/Blob.h> #include <LibWeb/FileAPI/Blob.h>
#include <LibWeb/FileAPI/FileReaderSync.h> #include <LibWeb/FileAPI/FileReaderSync.h>
#include <LibWeb/HTML/EventLoop/EventLoop.h> #include <LibWeb/HTML/EventLoop/EventLoop.h>
#include <LibWeb/Streams/AbstractOperations.h> #include <LibWeb/Streams/ReadableStream.h>
#include <LibWeb/Streams/ReadableStreamDefaultReader.h> #include <LibWeb/Streams/ReadableStreamDefaultReader.h>
#include <LibWeb/WebIDL/ExceptionOr.h> #include <LibWeb/WebIDL/ExceptionOr.h>

3861
Libraries/LibWeb/Streams/AbstractOperations.cpp
View file

File diff suppressed because it is too large Load diff

88
Libraries/LibWeb/Streams/AbstractOperations.h
View file

@ -3,6 +3,7 @@
* Copyright (c) 2023, Matthew Olsson <mattco@serenityos.org> * Copyright (c) 2023, Matthew Olsson <mattco@serenityos.org>
* Copyright (c) 2023-2025, Shannon Booth <shannon@serenityos.org> * Copyright (c) 2023-2025, Shannon Booth <shannon@serenityos.org>
* Copyright (c) 2023-2024, Kenneth Myhra <kennethmyhra@serenityos.org> * Copyright (c) 2023-2024, Kenneth Myhra <kennethmyhra@serenityos.org>
* Copyright (c) 2025, Tim Flynn <trflynn89@ladybird.org>
* *
* SPDX-License-Identifier: BSD-2-Clause * SPDX-License-Identifier: BSD-2-Clause
*/ */
@ -12,104 +13,17 @@
#include <LibGC/Ptr.h> #include <LibGC/Ptr.h>
#include <LibWeb/Forward.h> #include <LibWeb/Forward.h>
#include <LibWeb/Streams/Algorithms.h> #include <LibWeb/Streams/Algorithms.h>
#include <LibWeb/Streams/ReadableStream.h>
#include <LibWeb/WebIDL/CallbackType.h> #include <LibWeb/WebIDL/CallbackType.h>
#include <LibWeb/WebIDL/ExceptionOr.h> #include <LibWeb/WebIDL/ExceptionOr.h>
#include <LibWeb/WebIDL/Promise.h>
#include <LibWeb/WebIDL/Types.h>
namespace Web::Streams { namespace Web::Streams {
WebIDL::ExceptionOr<GC::Ref<ReadableStreamDefaultReader>> acquire_readable_stream_default_reader(ReadableStream&);
WebIDL::ExceptionOr<GC::Ref<ReadableStreamBYOBReader>> acquire_readable_stream_byob_reader(ReadableStream&);
bool is_readable_stream_locked(ReadableStream const&);
GC::Ref<SizeAlgorithm> extract_size_algorithm(JS::VM&, QueuingStrategy const&); GC::Ref<SizeAlgorithm> extract_size_algorithm(JS::VM&, QueuingStrategy const&);
WebIDL::ExceptionOr<double> extract_high_water_mark(QueuingStrategy const&, double default_hwm); WebIDL::ExceptionOr<double> extract_high_water_mark(QueuingStrategy const&, double default_hwm);
void readable_stream_close(ReadableStream&);
void readable_stream_error(ReadableStream&, JS::Value error);
WebIDL::ExceptionOr<GC::Ref<ReadableStream>> readable_stream_from_iterable(JS::VM& vm, JS::Value async_iterable);
void readable_stream_add_read_request(ReadableStream&, GC::Ref<ReadRequest>);
void readable_stream_add_read_into_request(ReadableStream&, GC::Ref<ReadIntoRequest>);
GC::Ref<WebIDL::Promise> readable_stream_cancel(ReadableStream&, JS::Value reason);
void readable_stream_fulfill_read_into_request(ReadableStream&, JS::Value chunk, bool done);
void readable_stream_fulfill_read_request(ReadableStream&, JS::Value chunk, bool done);
size_t readable_stream_get_num_read_into_requests(ReadableStream const&);
size_t readable_stream_get_num_read_requests(ReadableStream const&);
bool readable_stream_has_byob_reader(ReadableStream const&);
bool readable_stream_has_default_reader(ReadableStream const&);
GC::Ref<WebIDL::Promise> readable_stream_pipe_to(ReadableStream& source, WritableStream& dest, bool prevent_close, bool prevent_abort, bool prevent_cancel, GC::Ptr<DOM::AbortSignal> signal = {});
WebIDL::ExceptionOr<ReadableStreamPair> readable_stream_tee(JS::Realm&, ReadableStream&, bool clone_for_branch2);
WebIDL::ExceptionOr<ReadableStreamPair> readable_stream_default_tee(JS::Realm& realm, ReadableStream& stream, bool clone_for_branch2);
WebIDL::ExceptionOr<ReadableStreamPair> readable_byte_stream_tee(JS::Realm& realm, ReadableStream& stream);
GC::Ref<WebIDL::Promise> readable_stream_reader_generic_cancel(ReadableStreamGenericReaderMixin&, JS::Value reason);
void readable_stream_reader_generic_initialize(ReadableStreamReader, ReadableStream&);
void readable_stream_reader_generic_release(ReadableStreamGenericReaderMixin&);
void readable_stream_default_reader_error_read_requests(ReadableStreamDefaultReader&, JS::Value error);
void readable_stream_byob_reader_error_read_into_requests(ReadableStreamBYOBReader&, JS::Value error);
JS::Value readable_byte_stream_controller_convert_pull_into_descriptor(JS::Realm&, PullIntoDescriptor const&);
void readable_byte_stream_controller_pull_into(ReadableByteStreamController&, WebIDL::ArrayBufferView&, u64 min, ReadIntoRequest&);
void readable_stream_byob_reader_read(ReadableStreamBYOBReader&, WebIDL::ArrayBufferView&, u64 min, ReadIntoRequest&);
void readable_byte_stream_controller_fill_head_pull_into_descriptor(ReadableByteStreamController const&, u64 size, PullIntoDescriptor&);
void readable_stream_default_reader_read(ReadableStreamDefaultReader&, ReadRequest&);
void readable_stream_default_reader_release(ReadableStreamDefaultReader&);
void readable_stream_byob_reader_release(ReadableStreamBYOBReader&);
WebIDL::ExceptionOr<void> set_up_readable_stream_default_reader(ReadableStreamDefaultReader&, ReadableStream&);
WebIDL::ExceptionOr<void> set_up_readable_stream_byob_reader(ReadableStreamBYOBReader&, ReadableStream&);
void readable_stream_default_controller_close(ReadableStreamDefaultController&);
bool readable_stream_default_controller_has_backpressure(ReadableStreamDefaultController&);
WebIDL::ExceptionOr<void> readable_stream_default_controller_enqueue(ReadableStreamDefaultController&, JS::Value chunk);
void readable_stream_default_controller_call_pull_if_needed(ReadableStreamDefaultController&);
bool readable_stream_default_controller_should_call_pull(ReadableStreamDefaultController&);
void readable_stream_default_controller_clear_algorithms(ReadableStreamDefaultController&);
void readable_stream_default_controller_error(ReadableStreamDefaultController&, JS::Value error);
Optional<double> readable_stream_default_controller_get_desired_size(ReadableStreamDefaultController&);
bool readable_stream_default_controller_can_close_or_enqueue(ReadableStreamDefaultController&);
WebIDL::ExceptionOr<void> set_up_readable_stream_default_controller(ReadableStream&, ReadableStreamDefaultController&, GC::Ref<StartAlgorithm>, GC::Ref<PullAlgorithm>, GC::Ref<CancelAlgorithm>, double high_water_mark, GC::Ref<SizeAlgorithm>);
WebIDL::ExceptionOr<void> set_up_readable_stream_default_controller_from_underlying_source(ReadableStream&, JS::Value underlying_source_value, UnderlyingSource, double high_water_mark, GC::Ref<SizeAlgorithm>);
WebIDL::ExceptionOr<void> set_up_readable_byte_stream_controller(ReadableStream&, ReadableByteStreamController&, GC::Ref<StartAlgorithm>, GC::Ref<PullAlgorithm>, GC::Ref<CancelAlgorithm>, double high_water_mark, JS::Value auto_allocate_chunk_size);
WebIDL::ExceptionOr<void> set_up_readable_byte_stream_controller_from_underlying_source(ReadableStream&, JS::Value underlying_source, UnderlyingSource const& underlying_source_dict, double high_water_mark);
GC::Ptr<ReadableStreamBYOBRequest> readable_byte_stream_controller_get_byob_request(GC::Ref<ReadableByteStreamController>);
WebIDL::ExceptionOr<void> readable_byte_stream_controller_respond_in_readable_state(ReadableByteStreamController&, u64 bytes_written, PullIntoDescriptor&);
void readable_byte_stream_controller_respond_in_closed_state(ReadableByteStreamController&, PullIntoDescriptor&);
WebIDL::ExceptionOr<void> readable_byte_stream_controller_respond_internal(ReadableByteStreamController&, u64 bytes_written);
WebIDL::ExceptionOr<void> readable_byte_stream_controller_respond(ReadableByteStreamController&, u64 bytes_written);
WebIDL::ExceptionOr<void> readable_byte_stream_controller_respond_with_new_view(JS::Realm&, ReadableByteStreamController&, WebIDL::ArrayBufferView&);
WebIDL::ExceptionOr<void> readable_byte_stream_controller_enqueue(ReadableByteStreamController& controller, JS::Value chunk);
WebIDL::ExceptionOr<GC::Ref<JS::ArrayBuffer>> transfer_array_buffer(JS::Realm& realm, JS::ArrayBuffer& buffer); WebIDL::ExceptionOr<GC::Ref<JS::ArrayBuffer>> transfer_array_buffer(JS::Realm& realm, JS::ArrayBuffer& buffer);
WebIDL::ExceptionOr<void> readable_byte_stream_controller_enqueue_detached_pull_into_to_queue(ReadableByteStreamController& controller, PullIntoDescriptor& pull_into_descriptor);
void readable_byte_stream_controller_commit_pull_into_descriptor(ReadableStream&, PullIntoDescriptor const&);
void readable_byte_stream_controller_process_read_requests_using_queue(ReadableByteStreamController& controller);
[[nodiscard]] SinglyLinkedList<GC::Root<PullIntoDescriptor>> readable_byte_stream_controller_process_pull_into_descriptors_using_queue(ReadableByteStreamController&);
void readable_byte_stream_controller_enqueue_chunk_to_queue(ReadableByteStreamController& controller, GC::Ref<JS::ArrayBuffer> buffer, u32 byte_offset, u32 byte_length);
WebIDL::ExceptionOr<void> readable_byte_stream_controller_enqueue_cloned_chunk_to_queue(ReadableByteStreamController& controller, JS::ArrayBuffer& buffer, u64 byte_offset, u64 byte_length);
GC::Ref<PullIntoDescriptor> readable_byte_stream_controller_shift_pending_pull_into(ReadableByteStreamController& controller);
void readable_byte_stream_controller_call_pull_if_needed(ReadableByteStreamController&);
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_close(ReadableByteStreamController&);
void readable_byte_stream_controller_error(ReadableByteStreamController&, JS::Value error);
void readable_byte_stream_controller_fill_read_request_from_queue(ReadableByteStreamController&, GC::Ref<ReadRequest>);
bool readable_byte_stream_controller_fill_pull_into_descriptor_from_queue(ReadableByteStreamController&, PullIntoDescriptor&);
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&);
bool readable_byte_stream_controller_should_call_pull(ReadableByteStreamController const&);
WebIDL::ExceptionOr<GC::Ref<ReadableStream>> create_readable_stream(JS::Realm& realm, GC::Ref<StartAlgorithm> start_algorithm, GC::Ref<PullAlgorithm> pull_algorithm, GC::Ref<CancelAlgorithm> cancel_algorithm, Optional<double> high_water_mark = {}, GC::Ptr<SizeAlgorithm> size_algorithm = {});
WebIDL::ExceptionOr<GC::Ref<ReadableStream>> create_readable_byte_stream(JS::Realm& realm, GC::Ref<StartAlgorithm> start_algorithm, GC::Ref<PullAlgorithm> pull_algorithm, GC::Ref<CancelAlgorithm> cancel_algorithm);
WebIDL::ExceptionOr<GC::Ref<WritableStream>> create_writable_stream(JS::Realm& realm, GC::Ref<StartAlgorithm> start_algorithm, GC::Ref<WriteAlgorithm> write_algorithm, GC::Ref<CloseAlgorithm> close_algorithm, GC::Ref<AbortAlgorithm> abort_algorithm, double high_water_mark, GC::Ref<SizeAlgorithm> size_algorithm); WebIDL::ExceptionOr<GC::Ref<WritableStream>> create_writable_stream(JS::Realm& realm, GC::Ref<StartAlgorithm> start_algorithm, GC::Ref<WriteAlgorithm> write_algorithm, GC::Ref<CloseAlgorithm> close_algorithm, GC::Ref<AbortAlgorithm> abort_algorithm, double high_water_mark, GC::Ref<SizeAlgorithm> size_algorithm);
void initialize_readable_stream(ReadableStream&);
void initialize_writable_stream(WritableStream&); void initialize_writable_stream(WritableStream&);
WebIDL::ExceptionOr<GC::Ref<WritableStreamDefaultWriter>> acquire_writable_stream_default_writer(WritableStream&); WebIDL::ExceptionOr<GC::Ref<WritableStreamDefaultWriter>> acquire_writable_stream_default_writer(WritableStream&);

1
Libraries/LibWeb/Streams/ReadableByteStreamController.cpp
View file

@ -13,6 +13,7 @@
#include <LibWeb/Streams/ReadableStream.h> #include <LibWeb/Streams/ReadableStream.h>
#include <LibWeb/Streams/ReadableStreamBYOBRequest.h> #include <LibWeb/Streams/ReadableStreamBYOBRequest.h>
#include <LibWeb/Streams/ReadableStreamDefaultReader.h> #include <LibWeb/Streams/ReadableStreamDefaultReader.h>
#include <LibWeb/Streams/ReadableStreamOperations.h>
#include <LibWeb/WebIDL/Buffers.h> #include <LibWeb/WebIDL/Buffers.h>
namespace Web::Streams { namespace Web::Streams {

2
Libraries/LibWeb/Streams/ReadableByteStreamController.h
View file

@ -10,7 +10,7 @@
#include <AK/Optional.h> #include <AK/Optional.h>
#include <AK/SinglyLinkedList.h> #include <AK/SinglyLinkedList.h>
#include <LibWeb/Bindings/PlatformObject.h> #include <LibWeb/Bindings/PlatformObject.h>
#include <LibWeb/Streams/AbstractOperations.h> #include <LibWeb/Streams/Algorithms.h>
namespace Web::Streams { namespace Web::Streams {

1
Libraries/LibWeb/Streams/ReadableStream.cpp
View file

@ -18,6 +18,7 @@
#include <LibWeb/Streams/ReadableStreamBYOBRequest.h> #include <LibWeb/Streams/ReadableStreamBYOBRequest.h>
#include <LibWeb/Streams/ReadableStreamDefaultController.h> #include <LibWeb/Streams/ReadableStreamDefaultController.h>
#include <LibWeb/Streams/ReadableStreamDefaultReader.h> #include <LibWeb/Streams/ReadableStreamDefaultReader.h>
#include <LibWeb/Streams/ReadableStreamOperations.h>
#include <LibWeb/Streams/TransformStream.h> #include <LibWeb/Streams/TransformStream.h>
#include <LibWeb/Streams/UnderlyingSource.h> #include <LibWeb/Streams/UnderlyingSource.h>
#include <LibWeb/Streams/WritableStream.h> #include <LibWeb/Streams/WritableStream.h>

2
Libraries/LibWeb/Streams/ReadableStreamAsyncIterator.cpp
View file

@ -6,10 +6,10 @@
#include <LibWeb/Bindings/Intrinsics.h> #include <LibWeb/Bindings/Intrinsics.h>
#include <LibWeb/Bindings/ReadableStreamAsyncIteratorPrototype.h> #include <LibWeb/Bindings/ReadableStreamAsyncIteratorPrototype.h>
#include <LibWeb/Streams/AbstractOperations.h>
#include <LibWeb/Streams/ReadableStream.h> #include <LibWeb/Streams/ReadableStream.h>
#include <LibWeb/Streams/ReadableStreamAsyncIterator.h> #include <LibWeb/Streams/ReadableStreamAsyncIterator.h>
#include <LibWeb/Streams/ReadableStreamDefaultReader.h> #include <LibWeb/Streams/ReadableStreamDefaultReader.h>
#include <LibWeb/Streams/ReadableStreamOperations.h>
namespace Web::Bindings { namespace Web::Bindings {

2
Libraries/LibWeb/Streams/ReadableStreamBYOBReader.cpp
View file

@ -9,9 +9,9 @@
#include <LibJS/Runtime/TypedArray.h> #include <LibJS/Runtime/TypedArray.h>
#include <LibWeb/Bindings/Intrinsics.h> #include <LibWeb/Bindings/Intrinsics.h>
#include <LibWeb/Bindings/ReadableStreamBYOBReaderPrototype.h> #include <LibWeb/Bindings/ReadableStreamBYOBReaderPrototype.h>
#include <LibWeb/Streams/AbstractOperations.h>
#include <LibWeb/Streams/ReadableStream.h> #include <LibWeb/Streams/ReadableStream.h>
#include <LibWeb/Streams/ReadableStreamBYOBReader.h> #include <LibWeb/Streams/ReadableStreamBYOBReader.h>
#include <LibWeb/Streams/ReadableStreamOperations.h>
#include <LibWeb/WebIDL/Buffers.h> #include <LibWeb/WebIDL/Buffers.h>
#include <LibWeb/WebIDL/ExceptionOr.h> #include <LibWeb/WebIDL/ExceptionOr.h>

1
Libraries/LibWeb/Streams/ReadableStreamBYOBRequest.cpp
View file

@ -10,6 +10,7 @@
#include <LibWeb/Bindings/ReadableStreamBYOBRequestPrototype.h> #include <LibWeb/Bindings/ReadableStreamBYOBRequestPrototype.h>
#include <LibWeb/Streams/ReadableByteStreamController.h> #include <LibWeb/Streams/ReadableByteStreamController.h>
#include <LibWeb/Streams/ReadableStreamBYOBRequest.h> #include <LibWeb/Streams/ReadableStreamBYOBRequest.h>
#include <LibWeb/Streams/ReadableStreamOperations.h>
#include <LibWeb/WebIDL/Buffers.h> #include <LibWeb/WebIDL/Buffers.h>
namespace Web::Streams { namespace Web::Streams {

1
Libraries/LibWeb/Streams/ReadableStreamDefaultController.cpp
View file

@ -10,6 +10,7 @@
#include <LibWeb/Streams/ReadableStream.h> #include <LibWeb/Streams/ReadableStream.h>
#include <LibWeb/Streams/ReadableStreamDefaultController.h> #include <LibWeb/Streams/ReadableStreamDefaultController.h>
#include <LibWeb/Streams/ReadableStreamDefaultReader.h> #include <LibWeb/Streams/ReadableStreamDefaultReader.h>
#include <LibWeb/Streams/ReadableStreamOperations.h>
#include <LibWeb/WebIDL/ExceptionOr.h> #include <LibWeb/WebIDL/ExceptionOr.h>
#include <LibWeb/WebIDL/Promise.h> #include <LibWeb/WebIDL/Promise.h>

2
Libraries/LibWeb/Streams/ReadableStreamDefaultReader.cpp
View file

@ -16,9 +16,9 @@
#include <LibWeb/Bindings/Intrinsics.h> #include <LibWeb/Bindings/Intrinsics.h>
#include <LibWeb/Bindings/ReadableStreamDefaultReaderPrototype.h> #include <LibWeb/Bindings/ReadableStreamDefaultReaderPrototype.h>
#include <LibWeb/Fetch/Infrastructure/IncrementalReadLoopReadRequest.h> #include <LibWeb/Fetch/Infrastructure/IncrementalReadLoopReadRequest.h>
#include <LibWeb/Streams/AbstractOperations.h>
#include <LibWeb/Streams/ReadableStream.h> #include <LibWeb/Streams/ReadableStream.h>
#include <LibWeb/Streams/ReadableStreamDefaultReader.h> #include <LibWeb/Streams/ReadableStreamDefaultReader.h>
#include <LibWeb/Streams/ReadableStreamOperations.h>
#include <LibWeb/WebIDL/ExceptionOr.h> #include <LibWeb/WebIDL/ExceptionOr.h>
#include <LibWeb/WebIDL/Promise.h> #include <LibWeb/WebIDL/Promise.h>

4
Libraries/LibWeb/Streams/ReadableStreamDefaultReader.h
View file

@ -20,7 +20,9 @@ struct ReadableStreamReadResult {
bool done; bool done;
}; };
namespace Detail {
class ReadableStreamPipeTo; class ReadableStreamPipeTo;
}
class ReadRequest : public JS::Cell { class ReadRequest : public JS::Cell {
GC_CELL(ReadRequest, JS::Cell); GC_CELL(ReadRequest, JS::Cell);
@ -99,7 +101,7 @@ public:
SinglyLinkedList<GC::Ref<ReadRequest>>& read_requests() { return m_read_requests; } SinglyLinkedList<GC::Ref<ReadRequest>>& read_requests() { return m_read_requests; }
void set_readable_stream_pipe_to_operation(Badge<ReadableStreamPipeTo>, GC::Ptr<JS::Cell> readable_stream_pipe_to_operation) { m_readable_stream_pipe_to_operation = readable_stream_pipe_to_operation; } void set_readable_stream_pipe_to_operation(Badge<Detail::ReadableStreamPipeTo>, GC::Ptr<JS::Cell> readable_stream_pipe_to_operation) { m_readable_stream_pipe_to_operation = readable_stream_pipe_to_operation; }
private: private:
explicit ReadableStreamDefaultReader(JS::Realm&); explicit ReadableStreamDefaultReader(JS::Realm&);

2
Libraries/LibWeb/Streams/ReadableStreamGenericReader.cpp
View file

@ -6,9 +6,9 @@
#include <LibJS/Runtime/Promise.h> #include <LibJS/Runtime/Promise.h>
#include <LibJS/Runtime/PromiseCapability.h> #include <LibJS/Runtime/PromiseCapability.h>
#include <LibWeb/Streams/AbstractOperations.h>
#include <LibWeb/Streams/ReadableStream.h> #include <LibWeb/Streams/ReadableStream.h>
#include <LibWeb/Streams/ReadableStreamGenericReader.h> #include <LibWeb/Streams/ReadableStreamGenericReader.h>
#include <LibWeb/Streams/ReadableStreamOperations.h>
#include <LibWeb/WebIDL/Promise.h> #include <LibWeb/WebIDL/Promise.h>
namespace Web::Streams { namespace Web::Streams {

2883
Libraries/LibWeb/Streams/ReadableStreamOperations.cpp Normal file
View file

File diff suppressed because it is too large Load diff

110
Libraries/LibWeb/Streams/ReadableStreamOperations.h Normal file
View file

@ -0,0 +1,110 @@
/*
* Copyright (c) 2022, Linus Groh <linusg@serenityos.org>
* Copyright (c) 2023, Matthew Olsson <mattco@serenityos.org>
* Copyright (c) 2023-2025, Shannon Booth <shannon@serenityos.org>
* Copyright (c) 2023-2024, Kenneth Myhra <kennethmyhra@serenityos.org>
* Copyright (c) 2025, Tim Flynn <trflynn89@ladybird.org>
*
* SPDX-License-Identifier: BSD-2-Clause
*/
#pragma once
#include <LibGC/Ptr.h>
#include <LibWeb/Forward.h>
#include <LibWeb/Streams/Algorithms.h>
#include <LibWeb/Streams/ReadableStream.h>
#include <LibWeb/WebIDL/ExceptionOr.h>
namespace Web::Streams {
// 4.9.1. Working with readable streams, https://streams.spec.whatwg.org/#rs-abstract-ops
WebIDL::ExceptionOr<GC::Ref<ReadableStreamBYOBReader>> acquire_readable_stream_byob_reader(ReadableStream&);
WebIDL::ExceptionOr<GC::Ref<ReadableStreamDefaultReader>> acquire_readable_stream_default_reader(ReadableStream&);
WebIDL::ExceptionOr<GC::Ref<ReadableStream>> create_readable_stream(JS::Realm& realm, GC::Ref<StartAlgorithm> start_algorithm, GC::Ref<PullAlgorithm> pull_algorithm, GC::Ref<CancelAlgorithm> cancel_algorithm, Optional<double> high_water_mark = {}, GC::Ptr<SizeAlgorithm> size_algorithm = {});
WebIDL::ExceptionOr<GC::Ref<ReadableStream>> create_readable_byte_stream(JS::Realm& realm, GC::Ref<StartAlgorithm> start_algorithm, GC::Ref<PullAlgorithm> pull_algorithm, GC::Ref<CancelAlgorithm> cancel_algorithm);
void initialize_readable_stream(ReadableStream&);
bool is_readable_stream_locked(ReadableStream const&);
WebIDL::ExceptionOr<GC::Ref<ReadableStream>> readable_stream_from_iterable(JS::VM& vm, JS::Value async_iterable);
GC::Ref<WebIDL::Promise> readable_stream_pipe_to(ReadableStream& source, WritableStream& dest, bool prevent_close, bool prevent_abort, bool prevent_cancel, GC::Ptr<DOM::AbortSignal> signal = {});
WebIDL::ExceptionOr<ReadableStreamPair> readable_stream_tee(JS::Realm&, ReadableStream&, bool clone_for_branch2);
WebIDL::ExceptionOr<ReadableStreamPair> readable_stream_default_tee(JS::Realm& realm, ReadableStream& stream, bool clone_for_branch2);
WebIDL::ExceptionOr<ReadableStreamPair> readable_byte_stream_tee(JS::Realm& realm, ReadableStream& stream);
// 4.9.2. Interfacing with controllers, https://streams.spec.whatwg.org/#rs-abstract-ops-used-by-controllers
void readable_stream_add_read_into_request(ReadableStream&, GC::Ref<ReadIntoRequest>);
void readable_stream_add_read_request(ReadableStream&, GC::Ref<ReadRequest>);
GC::Ref<WebIDL::Promise> readable_stream_cancel(ReadableStream&, JS::Value reason);
void readable_stream_close(ReadableStream&);
void readable_stream_error(ReadableStream&, JS::Value error);
void readable_stream_fulfill_read_into_request(ReadableStream&, JS::Value chunk, bool done);
void readable_stream_fulfill_read_request(ReadableStream&, JS::Value chunk, bool done);
size_t readable_stream_get_num_read_into_requests(ReadableStream const&);
size_t readable_stream_get_num_read_requests(ReadableStream const&);
bool readable_stream_has_byob_reader(ReadableStream const&);
bool readable_stream_has_default_reader(ReadableStream const&);
// 4.9.3. Readers, https://streams.spec.whatwg.org/#rs-reader-abstract-ops
GC::Ref<WebIDL::Promise> readable_stream_reader_generic_cancel(ReadableStreamGenericReaderMixin&, JS::Value reason);
void readable_stream_reader_generic_initialize(ReadableStreamReader const&, ReadableStream&);
void readable_stream_reader_generic_release(ReadableStreamGenericReaderMixin&);
void readable_stream_byob_reader_error_read_into_requests(ReadableStreamBYOBReader&, JS::Value error);
void readable_stream_byob_reader_read(ReadableStreamBYOBReader&, WebIDL::ArrayBufferView&, u64 min, ReadIntoRequest&);
void readable_stream_byob_reader_release(ReadableStreamBYOBReader&);
void readable_stream_default_reader_error_read_requests(ReadableStreamDefaultReader&, JS::Value error);
void readable_stream_default_reader_read(ReadableStreamDefaultReader&, ReadRequest&);
void readable_stream_default_reader_release(ReadableStreamDefaultReader&);
WebIDL::ExceptionOr<void> set_up_readable_stream_byob_reader(ReadableStreamBYOBReader&, ReadableStream&);
WebIDL::ExceptionOr<void> set_up_readable_stream_default_reader(ReadableStreamDefaultReader&, ReadableStream&);
// 4.9.4. Default controllers, https://streams.spec.whatwg.org/#rs-default-controller-abstract-ops
void readable_stream_default_controller_call_pull_if_needed(ReadableStreamDefaultController&);
bool readable_stream_default_controller_should_call_pull(ReadableStreamDefaultController&);
void readable_stream_default_controller_clear_algorithms(ReadableStreamDefaultController&);
void readable_stream_default_controller_close(ReadableStreamDefaultController&);
WebIDL::ExceptionOr<void> readable_stream_default_controller_enqueue(ReadableStreamDefaultController&, JS::Value chunk);
void readable_stream_default_controller_error(ReadableStreamDefaultController&, JS::Value error);
Optional<double> readable_stream_default_controller_get_desired_size(ReadableStreamDefaultController&);
bool readable_stream_default_controller_has_backpressure(ReadableStreamDefaultController&);
bool readable_stream_default_controller_can_close_or_enqueue(ReadableStreamDefaultController&);
WebIDL::ExceptionOr<void> set_up_readable_stream_default_controller(ReadableStream&, ReadableStreamDefaultController&, GC::Ref<StartAlgorithm>, GC::Ref<PullAlgorithm>, GC::Ref<CancelAlgorithm>, double high_water_mark, GC::Ref<SizeAlgorithm>);
WebIDL::ExceptionOr<void> set_up_readable_stream_default_controller_from_underlying_source(ReadableStream&, JS::Value underlying_source_value, UnderlyingSource, double high_water_mark, GC::Ref<SizeAlgorithm>);
// 4.9.5. Byte stream controllers, https://streams.spec.whatwg.org/#rbs-controller-abstract-ops
void readable_byte_stream_controller_call_pull_if_needed(ReadableByteStreamController&);
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_close(ReadableByteStreamController&);
void readable_byte_stream_controller_commit_pull_into_descriptor(ReadableStream&, PullIntoDescriptor const&);
JS::Value readable_byte_stream_controller_convert_pull_into_descriptor(JS::Realm&, PullIntoDescriptor const&);
WebIDL::ExceptionOr<void> readable_byte_stream_controller_enqueue(ReadableByteStreamController& controller, JS::Value chunk);
void readable_byte_stream_controller_enqueue_chunk_to_queue(ReadableByteStreamController& controller, GC::Ref<JS::ArrayBuffer> buffer, u32 byte_offset, u32 byte_length);
WebIDL::ExceptionOr<void> readable_byte_stream_controller_enqueue_cloned_chunk_to_queue(ReadableByteStreamController& controller, JS::ArrayBuffer& buffer, u64 byte_offset, u64 byte_length);
WebIDL::ExceptionOr<void> readable_byte_stream_controller_enqueue_detached_pull_into_to_queue(ReadableByteStreamController& controller, PullIntoDescriptor& pull_into_descriptor);
void readable_byte_stream_controller_error(ReadableByteStreamController&, JS::Value error);
void readable_byte_stream_controller_fill_head_pull_into_descriptor(ReadableByteStreamController const&, u64 size, PullIntoDescriptor&);
bool readable_byte_stream_controller_fill_pull_into_descriptor_from_queue(ReadableByteStreamController&, PullIntoDescriptor&);
void readable_byte_stream_controller_fill_read_request_from_queue(ReadableByteStreamController&, ReadRequest&);
GC::Ptr<ReadableStreamBYOBRequest> readable_byte_stream_controller_get_byob_request(ReadableByteStreamController&);
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&);
[[nodiscard]] SinglyLinkedList<GC::Root<PullIntoDescriptor>> readable_byte_stream_controller_process_pull_into_descriptors_using_queue(ReadableByteStreamController&);
void readable_byte_stream_controller_process_read_requests_using_queue(ReadableByteStreamController& controller);
void readable_byte_stream_controller_pull_into(ReadableByteStreamController&, WebIDL::ArrayBufferView&, u64 min, ReadIntoRequest&);
WebIDL::ExceptionOr<void> readable_byte_stream_controller_respond(ReadableByteStreamController&, u64 bytes_written);
void readable_byte_stream_controller_respond_in_closed_state(ReadableByteStreamController&, PullIntoDescriptor&);
WebIDL::ExceptionOr<void> readable_byte_stream_controller_respond_in_readable_state(ReadableByteStreamController&, u64 bytes_written, PullIntoDescriptor&);
WebIDL::ExceptionOr<void> readable_byte_stream_controller_respond_internal(ReadableByteStreamController&, u64 bytes_written);
WebIDL::ExceptionOr<void> readable_byte_stream_controller_respond_with_new_view(JS::Realm&, ReadableByteStreamController&, WebIDL::ArrayBufferView&);
GC::Ref<PullIntoDescriptor> readable_byte_stream_controller_shift_pending_pull_into(ReadableByteStreamController& controller);
bool readable_byte_stream_controller_should_call_pull(ReadableByteStreamController const&);
WebIDL::ExceptionOr<void> set_up_readable_byte_stream_controller(ReadableStream&, ReadableByteStreamController&, GC::Ref<StartAlgorithm>, GC::Ref<PullAlgorithm>, GC::Ref<CancelAlgorithm>, double high_water_mark, JS::Value auto_allocate_chunk_size);
WebIDL::ExceptionOr<void> set_up_readable_byte_stream_controller_from_underlying_source(ReadableStream&, JS::Value underlying_source, UnderlyingSource const& underlying_source_dict, double high_water_mark);
}

411
Libraries/LibWeb/Streams/ReadableStreamPipeTo.cpp Normal file
View file

@ -0,0 +1,411 @@
/*
* Copyright (c) 2025, Tim Flynn <trflynn89@ladybird.org>
*
* SPDX-License-Identifier: BSD-2-Clause
*/
#include <LibWeb/HTML/EventLoop/EventLoop.h>
#include <LibWeb/HTML/Scripting/TemporaryExecutionContext.h>
#include <LibWeb/Streams/AbstractOperations.h>
#include <LibWeb/Streams/ReadableStreamDefaultReader.h>
#include <LibWeb/Streams/ReadableStreamOperations.h>
#include <LibWeb/Streams/ReadableStreamPipeTo.h>
#include <LibWeb/Streams/WritableStream.h>
#include <LibWeb/Streams/WritableStreamDefaultWriter.h>
#include <LibWeb/WebIDL/Promise.h>
namespace Web::Streams::Detail {
class ReadableStreamPipeToReadRequest final : public ReadRequest {
GC_CELL(ReadableStreamPipeToReadRequest, ReadRequest);
GC_DECLARE_ALLOCATOR(ReadableStreamPipeToReadRequest);
using OnChunk = GC::Ref<GC::Function<void(JS::Value)>>;
using OnComplete = GC::Ref<GC::Function<void()>>;
// This has a return value just for compatibility with WebIDL::react_to_promise.
using OnError = GC::Ref<GC::Function<WebIDL::ExceptionOr<JS::Value>(JS::Value)>>;
public:
virtual void on_chunk(JS::Value chunk) override
{
m_on_chunk->function()(chunk);
}
virtual void on_close() override
{
m_on_complete->function()();
}
virtual void on_error(JS::Value error) override
{
MUST(m_on_error->function()(error));
}
private:
ReadableStreamPipeToReadRequest(OnChunk on_chunk, OnComplete on_complete, OnError on_error)
: m_on_chunk(on_chunk)
, m_on_complete(on_complete)
, m_on_error(on_error)
{
}
virtual void visit_edges(Visitor& visitor) override
{
Base::visit_edges(visitor);
visitor.visit(m_on_chunk);
visitor.visit(m_on_complete);
visitor.visit(m_on_error);
}
OnChunk m_on_chunk;
OnComplete m_on_complete;
OnError m_on_error;
};
GC_DEFINE_ALLOCATOR(ReadableStreamPipeTo);
GC_DEFINE_ALLOCATOR(ReadableStreamPipeToReadRequest);
// https://streams.spec.whatwg.org/#ref-for-in-parallel
ReadableStreamPipeTo::ReadableStreamPipeTo(
GC::Ref<JS::Realm> realm,
GC::Ref<WebIDL::Promise> promise,
GC::Ref<ReadableStream> source,
GC::Ref<WritableStream> destination,
GC::Ref<ReadableStreamDefaultReader> reader,
GC::Ref<WritableStreamDefaultWriter> writer,
bool prevent_close,
bool prevent_abort,
bool prevent_cancel)
: m_realm(realm)
, m_promise(promise)
, m_source(source)
, m_destination(destination)
, m_reader(reader)
, m_writer(writer)
, m_prevent_close(prevent_close)
, m_prevent_abort(prevent_abort)
, m_prevent_cancel(prevent_cancel)
{
m_reader->set_readable_stream_pipe_to_operation({}, this);
}
void ReadableStreamPipeTo::visit_edges(Cell::Visitor& visitor)
{
Base::visit_edges(visitor);
visitor.visit(m_realm);
visitor.visit(m_promise);
visitor.visit(m_source);
visitor.visit(m_destination);
visitor.visit(m_reader);
visitor.visit(m_writer);
visitor.visit(m_signal);
visitor.visit(m_pending_writes);
visitor.visit(m_unwritten_chunks);
}
void ReadableStreamPipeTo::process()
{
if (check_for_error_and_close_states())
return;
auto ready_promise = m_writer->ready();
if (ready_promise && WebIDL::is_promise_fulfilled(*ready_promise)) {
read_chunk();
return;
}
auto when_ready = GC::create_function(m_realm->heap(), [this](JS::Value) -> WebIDL::ExceptionOr<JS::Value> {
read_chunk();
return JS::js_undefined();
});
auto shutdown = GC::create_function(heap(), [this](JS::Value) -> WebIDL::ExceptionOr<JS::Value> {
check_for_error_and_close_states();
return JS::js_undefined();
});
if (ready_promise)
WebIDL::react_to_promise(*ready_promise, when_ready, shutdown);
if (auto promise = m_reader->closed())
WebIDL::react_to_promise(*promise, shutdown, shutdown);
}
void ReadableStreamPipeTo::set_abort_signal(GC::Ref<DOM::AbortSignal> signal, DOM::AbortSignal::AbortSignal::AbortAlgorithmID signal_id)
{
m_signal = signal;
m_signal_id = signal_id;
}
// https://streams.spec.whatwg.org/#rs-pipeTo-shutdown-with-action
void ReadableStreamPipeTo::shutdown_with_action(GC::Ref<GC::Function<GC::Ref<WebIDL::Promise>()>> action, Optional<JS::Value> original_error)
{
// 1. If shuttingDown is true, abort these substeps.
if (m_shutting_down)
return;
// 2. Set shuttingDown to true.
m_shutting_down = true;
auto on_pending_writes_complete = [this, action, original_error = move(original_error)]() mutable {
HTML::TemporaryExecutionContext execution_context { m_realm, HTML::TemporaryExecutionContext::CallbacksEnabled::Yes };
// 4. Let p be the result of performing action.
auto promise = action->function()();
WebIDL::react_to_promise(promise,
// 5. Upon fulfillment of p, finalize, passing along originalError if it was given.
GC::create_function(heap(), [this, original_error = move(original_error)](JS::Value) mutable -> WebIDL::ExceptionOr<JS::Value> {
finish(move(original_error));
return JS::js_undefined();
}),
// 6. Upon rejection of p with reason newError, finalize with newError.
GC::create_function(heap(), [this](JS::Value new_error) -> WebIDL::ExceptionOr<JS::Value> {
finish(new_error);
return JS::js_undefined();
}));
};
// 3. If dest.[[state]] is "writable" and ! WritableStreamCloseQueuedOrInFlight(dest) is false,
if (m_destination->state() == WritableStream::State::Writable && !writable_stream_close_queued_or_in_flight(m_destination)) {
// 1. If any chunks have been read but not yet written, write them to dest.
write_unwritten_chunks();
// 2. Wait until every chunk that has been read has been written (i.e. the corresponding promises have settled).
wait_for_pending_writes_to_complete(move(on_pending_writes_complete));
} else {
on_pending_writes_complete();
}
}
// https://streams.spec.whatwg.org/#rs-pipeTo-shutdown
void ReadableStreamPipeTo::shutdown(Optional<JS::Value> error)
{
// 1. If shuttingDown is true, abort these substeps.
if (m_shutting_down)
return;
// 2. Set shuttingDown to true.
m_shutting_down = true;
auto on_pending_writes_complete = [this, error = move(error)]() mutable {
HTML::TemporaryExecutionContext execution_context { m_realm, HTML::TemporaryExecutionContext::CallbacksEnabled::Yes };
// 4. Finalize, passing along error if it was given.
finish(move(error));
};
// 3. If dest.[[state]] is "writable" and ! WritableStreamCloseQueuedOrInFlight(dest) is false,
if (m_destination->state() == WritableStream::State::Writable && !writable_stream_close_queued_or_in_flight(m_destination)) {
// 1. If any chunks have been read but not yet written, write them to dest.
write_unwritten_chunks();
// 2. Wait until every chunk that has been read has been written (i.e. the corresponding promises have settled).
wait_for_pending_writes_to_complete(move(on_pending_writes_complete));
} else {
on_pending_writes_complete();
}
}
void ReadableStreamPipeTo::read_chunk()
{
// Shutdown must stop activity: if shuttingDown becomes true, the user agent must not initiate further reads from
// reader, and must only perform writes of already-read chunks, as described below. In particular, the user agent
// must check the below conditions before performing any reads or writes, since they might lead to immediate shutdown.
if (check_for_error_and_close_states())
return;
auto on_chunk = GC::create_function(heap(), [this](JS::Value chunk) {
m_unwritten_chunks.append(chunk);
if (check_for_error_and_close_states())
return;
HTML::queue_a_microtask(nullptr, GC::create_function(m_realm->heap(), [this]() {
HTML::TemporaryExecutionContext execution_context { m_realm, HTML::TemporaryExecutionContext::CallbacksEnabled::Yes };
write_chunk();
process();
}));
});
auto on_complete = GC::create_function(heap(), [this]() {
if (!check_for_error_and_close_states())
finish();
});
auto shutdown = GC::create_function(heap(), [this](JS::Value) -> WebIDL::ExceptionOr<JS::Value> {
check_for_error_and_close_states();
return JS::js_undefined();
});
auto read_request = heap().allocate<ReadableStreamPipeToReadRequest>(on_chunk, on_complete, shutdown);
readable_stream_default_reader_read(m_reader, read_request);
if (auto promise = m_writer->closed())
WebIDL::react_to_promise(*promise, shutdown, shutdown);
}
void ReadableStreamPipeTo::write_chunk()
{
// Shutdown must stop activity: if shuttingDown becomes true, the user agent must not initiate further reads from
// reader, and must only perform writes of already-read chunks, as described below. In particular, the user agent
// must check the below conditions before performing any reads or writes, since they might lead to immediate shutdown.
if (!m_shutting_down && check_for_error_and_close_states())
return;
auto promise = writable_stream_default_writer_write(m_writer, m_unwritten_chunks.take_first());
WebIDL::mark_promise_as_handled(promise);
m_pending_writes.append(promise);
}
void ReadableStreamPipeTo::write_unwritten_chunks()
{
while (!m_unwritten_chunks.is_empty())
write_chunk();
}
void ReadableStreamPipeTo::wait_for_pending_writes_to_complete(Function<void()> on_complete)
{
auto handler = GC::create_function(heap(), [this, on_complete = move(on_complete)]() {
m_pending_writes.clear();
on_complete();
});
auto success_steps = [handler](Vector<JS::Value> const&) { handler->function()(); };
auto failure_steps = [handler](JS::Value) { handler->function()(); };
WebIDL::wait_for_all(m_realm, m_pending_writes, move(success_steps), move(failure_steps));
}
// https://streams.spec.whatwg.org/#rs-pipeTo-finalize
// We call this `finish` instead of `finalize` to avoid conflicts with GC::Cell::finalize.
void ReadableStreamPipeTo::finish(Optional<JS::Value> error)
{
// 1. Perform ! WritableStreamDefaultWriterRelease(writer).
writable_stream_default_writer_release(m_writer);
// 2. If reader implements ReadableStreamBYOBReader, perform ! ReadableStreamBYOBReaderRelease(reader).
// 3. Otherwise, perform ! ReadableStreamDefaultReaderRelease(reader).
readable_stream_default_reader_release(m_reader);
// 4. If signal is not undefined, remove abortAlgorithm from signal.
if (m_signal)
m_signal->remove_abort_algorithm(m_signal_id);
// 5. If error was given, reject promise with error.
if (error.has_value()) {
WebIDL::reject_promise(m_realm, m_promise, *error);
}
// 6. Otherwise, resolve promise with undefined.
else {
WebIDL::resolve_promise(m_realm, m_promise, JS::js_undefined());
}
m_reader->set_readable_stream_pipe_to_operation({}, nullptr);
}
bool ReadableStreamPipeTo::check_for_error_and_close_states()
{
// Error and close states must be propagated: the following conditions must be applied in order.
return m_shutting_down
|| check_for_forward_errors()
|| check_for_backward_errors()
|| check_for_forward_close()
|| check_for_backward_close();
}
bool ReadableStreamPipeTo::check_for_forward_errors()
{
// 1. Errors must be propagated forward: if source.[[state]] is or becomes "errored", then
if (m_source->state() == ReadableStream::State::Errored) {
// 1. If preventAbort is false, shutdown with an action of ! WritableStreamAbort(dest, source.[[storedError]])
// and with source.[[storedError]].
if (!m_prevent_abort) {
auto action = GC::create_function(heap(), [this]() {
return writable_stream_abort(m_destination, m_source->stored_error());
});
shutdown_with_action(action, m_source->stored_error());
}
// 2. Otherwise, shutdown with source.[[storedError]].
else {
shutdown(m_source->stored_error());
}
}
return m_shutting_down;
}
bool ReadableStreamPipeTo::check_for_backward_errors()
{
// 2. Errors must be propagated backward: if dest.[[state]] is or becomes "errored", then
if (m_destination->state() == WritableStream::State::Errored) {
// 1. If preventCancel is false, shutdown with an action of ! ReadableStreamCancel(source, dest.[[storedError]])
// and with dest.[[storedError]].
if (!m_prevent_cancel) {
auto action = GC::create_function(heap(), [this]() {
return readable_stream_cancel(m_source, m_destination->stored_error());
});
shutdown_with_action(action, m_destination->stored_error());
}
// 2. Otherwise, shutdown with dest.[[storedError]].
else {
shutdown(m_destination->stored_error());
}
}
return m_shutting_down;
}
bool ReadableStreamPipeTo::check_for_forward_close()
{
// 3. Closing must be propagated forward: if source.[[state]] is or becomes "closed", then
if (m_source->state() == ReadableStream::State::Closed) {
// 1. If preventClose is false, shutdown with an action of ! WritableStreamDefaultWriterCloseWithErrorPropagation(writer).
if (!m_prevent_close) {
auto action = GC::create_function(heap(), [this]() {
return writable_stream_default_writer_close_with_error_propagation(m_writer);
});
shutdown_with_action(action);
}
// 2. Otherwise, shutdown.
else {
shutdown();
}
}
return m_shutting_down;
}
bool ReadableStreamPipeTo::check_for_backward_close()
{
// 4. Closing must be propagated backward: if ! WritableStreamCloseQueuedOrInFlight(dest) is true or dest.[[state]] is "closed", then
if (writable_stream_close_queued_or_in_flight(m_destination) || m_destination->state() == WritableStream::State::Closed) {
// 1. Assert: no chunks have been read or written.
// 2. Let destClosed be a new TypeError.
auto destination_closed = JS::TypeError::create(m_realm, "Destination stream was closed during piping operation"sv);
// 3. If preventCancel is false, shutdown with an action of ! ReadableStreamCancel(source, destClosed) and with destClosed.
if (!m_prevent_cancel) {
auto action = GC::create_function(heap(), [this, destination_closed]() {
return readable_stream_cancel(m_source, destination_closed);
});
shutdown_with_action(action, destination_closed);
}
// 4. Otherwise, shutdown with destClosed.
else {
shutdown(destination_closed);
}
}
return m_shutting_down;
}
}

81
Libraries/LibWeb/Streams/ReadableStreamPipeTo.h Normal file
View file

@ -0,0 +1,81 @@
/*
* Copyright (c) 2025, Tim Flynn <trflynn89@ladybird.org>
*
* SPDX-License-Identifier: BSD-2-Clause
*/
#pragma once
#include <AK/Vector.h>
#include <LibGC/Ptr.h>
#include <LibJS/Forward.h>
#include <LibJS/Heap/Cell.h>
#include <LibWeb/DOM/AbortSignal.h>
#include <LibWeb/Forward.h>
namespace Web::Streams::Detail {
// https://streams.spec.whatwg.org/#ref-for-in-parallel
class ReadableStreamPipeTo final : public JS::Cell {
GC_CELL(ReadableStreamPipeTo, JS::Cell);
GC_DECLARE_ALLOCATOR(ReadableStreamPipeTo);
public:
void process();
void set_abort_signal(GC::Ref<DOM::AbortSignal>, DOM::AbortSignal::AbortSignal::AbortAlgorithmID);
void shutdown_with_action(GC::Ref<GC::Function<GC::Ref<WebIDL::Promise>()>> action, Optional<JS::Value> original_error = {});
void shutdown(Optional<JS::Value> error = {});
private:
ReadableStreamPipeTo(
GC::Ref<JS::Realm>,
GC::Ref<WebIDL::Promise>,
GC::Ref<ReadableStream> source,
GC::Ref<WritableStream> destination,
GC::Ref<ReadableStreamDefaultReader> reader,
GC::Ref<WritableStreamDefaultWriter> writer,
bool prevent_close,
bool prevent_abort,
bool prevent_cancel);
virtual void visit_edges(Cell::Visitor& visitor) override;
void read_chunk();
void write_chunk();
void write_unwritten_chunks();
void wait_for_pending_writes_to_complete(Function<void()> on_complete);
void finish(Optional<JS::Value> error = {});
bool check_for_error_and_close_states();
bool check_for_forward_errors();
bool check_for_backward_errors();
bool check_for_forward_close();
bool check_for_backward_close();
GC::Ref<JS::Realm> m_realm;
GC::Ref<WebIDL::Promise> m_promise;
GC::Ref<ReadableStream> m_source;
GC::Ref<WritableStream> m_destination;
GC::Ref<ReadableStreamDefaultReader> m_reader;
GC::Ref<WritableStreamDefaultWriter> m_writer;
GC::Ptr<DOM::AbortSignal> m_signal;
DOM::AbortSignal::AbortAlgorithmID m_signal_id { 0 };
Vector<GC::Ref<WebIDL::Promise>> m_pending_writes;
Vector<JS::Value, 1> m_unwritten_chunks;
bool m_prevent_close { false };
bool m_prevent_abort { false };
bool m_prevent_cancel { false };
bool m_shutting_down { false };
};
}

488
Libraries/LibWeb/Streams/ReadableStreamTee.cpp Normal file
View file

@ -0,0 +1,488 @@
/*
* Copyright (c) 2025, Tim Flynn <trflynn89@ladybird.org>
*
* SPDX-License-Identifier: BSD-2-Clause
*/
#include <LibWeb/Bindings/ExceptionOrUtils.h>
#include <LibWeb/HTML/EventLoop/EventLoop.h>
#include <LibWeb/HTML/Scripting/TemporaryExecutionContext.h>
#include <LibWeb/Streams/AbstractOperations.h>
#include <LibWeb/Streams/ReadableByteStreamController.h>
#include <LibWeb/Streams/ReadableStreamDefaultController.h>
#include <LibWeb/Streams/ReadableStreamOperations.h>
#include <LibWeb/Streams/ReadableStreamTee.h>
#include <LibWeb/WebIDL/Buffers.h>
#include <LibWeb/WebIDL/Promise.h>
namespace Web::Streams::Detail {
GC_DEFINE_ALLOCATOR(ReadableStreamTeeParams);
GC_DEFINE_ALLOCATOR(ReadableStreamTeeReadRequest);
void ReadableStreamTeeParams::visit_edges(Visitor& visitor)
{
Base::visit_edges(visitor);
visitor.visit(reason1);
visitor.visit(reason2);
visitor.visit(branch1);
visitor.visit(branch2);
visitor.visit(pull_algorithm);
}
// https://streams.spec.whatwg.org/#ref-for-read-request③
ReadableStreamTeeReadRequest::ReadableStreamTeeReadRequest(
JS::Realm& realm,
GC::Ref<ReadableStream> stream,
GC::Ref<ReadableStreamTeeParams> params,
GC::Ref<WebIDL::Promise> cancel_promise,
bool clone_for_branch2)
: m_realm(realm)
, m_stream(stream)
, m_params(params)
, m_cancel_promise(cancel_promise)
, m_clone_for_branch2(clone_for_branch2)
{
}
void ReadableStreamTeeReadRequest::visit_edges(Visitor& visitor)
{
Base::visit_edges(visitor);
visitor.visit(m_realm);
visitor.visit(m_stream);
visitor.visit(m_params);
visitor.visit(m_cancel_promise);
}
// https://streams.spec.whatwg.org/#ref-for-read-request-chunk-steps③
void ReadableStreamTeeReadRequest::on_chunk(JS::Value chunk)
{
// 1. Queue a microtask to perform the following steps:
HTML::queue_a_microtask(nullptr, GC::create_function(m_realm->heap(), [this, chunk]() {
HTML::TemporaryExecutionContext execution_context { m_realm, HTML::TemporaryExecutionContext::CallbacksEnabled::Yes };
auto controller1 = m_params->branch1->controller()->get<GC::Ref<ReadableStreamDefaultController>>();
auto controller2 = m_params->branch2->controller()->get<GC::Ref<ReadableStreamDefaultController>>();
// 1. Set readAgain to false.
m_params->read_again = false;
// 2. Let chunk1 and chunk2 be chunk.
auto chunk1 = chunk;
auto chunk2 = chunk;
// 3. If canceled2 is false and cloneForBranch2 is true,
if (!m_params->canceled2 && m_clone_for_branch2) {
// 1. Let cloneResult be StructuredClone(chunk2).
auto clone_result = structured_clone(m_realm, chunk2);
// 2. If cloneResult is an abrupt completion,
if (clone_result.is_exception()) {
auto completion = Bindings::exception_to_throw_completion(m_realm->vm(), clone_result.release_error());
// 1. Perform ! ReadableStreamDefaultControllerError(branch1.[[controller]], cloneResult.[[Value]]).
readable_stream_default_controller_error(controller1, completion.value());
// 2. Perform ! ReadableStreamDefaultControllerError(branch2.[[controller]], cloneResult.[[Value]]).
readable_stream_default_controller_error(controller2, completion.value());
// 3. Resolve cancelPromise with ! ReadableStreamCancel(stream, cloneResult.[[Value]]).
auto cancel_result = readable_stream_cancel(m_stream, completion.value());
// Note: We need to manually convert the result to an ECMAScript value here, by extracting its [[Promise]] slot.
WebIDL::resolve_promise(m_realm, m_cancel_promise, cancel_result->promise());
// 4. Return.
return;
}
// 3. Otherwise, set chunk2 to cloneResult.[[Value]].
chunk2 = clone_result.release_value();
}
// 4. If canceled1 is false, perform ! ReadableStreamDefaultControllerEnqueue(branch1.[[controller]], chunk1).
if (!m_params->canceled1) {
MUST(readable_stream_default_controller_enqueue(controller1, chunk1));
}
// 5. If canceled2 is false, perform ! ReadableStreamDefaultControllerEnqueue(branch2.[[controller]], chunk2).
if (!m_params->canceled2) {
MUST(readable_stream_default_controller_enqueue(controller2, chunk2));
}
// 6. Set reading to false.
m_params->reading = false;
// 7. If readAgain is true, perform pullAlgorithm.
if (m_params->read_again) {
m_params->pull_algorithm->function()();
}
}));
// NOTE: The microtask delay here is necessary because it takes at least a microtask to detect errors, when we
// use reader.[[closedPromise]] below. We want errors in stream to error both branches immediately, so we
// cannot let successful synchronously-available reads happen ahead of asynchronously-available errors.
}
// https://streams.spec.whatwg.org/#ref-for-read-request-close-steps②
void ReadableStreamTeeReadRequest::on_close()
{
auto controller1 = m_params->branch1->controller()->get<GC::Ref<ReadableStreamDefaultController>>();
auto controller2 = m_params->branch2->controller()->get<GC::Ref<ReadableStreamDefaultController>>();
// 1. Set reading to false.
m_params->reading = false;
// 2. If canceled1 is false, perform ! ReadableStreamDefaultControllerClose(branch1.[[controller]]).
if (!m_params->canceled1) {
readable_stream_default_controller_close(controller1);
}
// 3. If canceled2 is false, perform ! ReadableStreamDefaultControllerClose(branch2.[[controller]]).
if (!m_params->canceled2) {
readable_stream_default_controller_close(controller2);
}
// 4. If canceled1 is false or canceled2 is false, resolve cancelPromise with undefined.
if (!m_params->canceled1 || !m_params->canceled2) {
WebIDL::resolve_promise(m_realm, m_cancel_promise, JS::js_undefined());
}
}
// https://streams.spec.whatwg.org/#ref-for-read-request-error-steps③
void ReadableStreamTeeReadRequest::on_error(JS::Value)
{
// 1. Set reading to false.
m_params->reading = false;
}
GC_DEFINE_ALLOCATOR(ReadableByteStreamTeeParams);
GC_DEFINE_ALLOCATOR(ReadableByteStreamTeeDefaultReadRequest);
GC_DEFINE_ALLOCATOR(ReadableByteStreamTeeBYOBReadRequest);
ReadableByteStreamTeeParams::ReadableByteStreamTeeParams(ReadableStreamReader reader)
: reader(move(reader))
{
}
void ReadableByteStreamTeeParams::visit_edges(Visitor& visitor)
{
Base::visit_edges(visitor);
visitor.visit(reason1);
visitor.visit(reason2);
visitor.visit(branch1);
visitor.visit(branch2);
visitor.visit(pull1_algorithm);
visitor.visit(pull2_algorithm);
reader.visit([&](auto underlying_reader) { visitor.visit(underlying_reader); });
}
// https://streams.spec.whatwg.org/#ref-for-read-request④
ReadableByteStreamTeeDefaultReadRequest::ReadableByteStreamTeeDefaultReadRequest(
JS::Realm& realm,
GC::Ref<ReadableStream> stream,
GC::Ref<ReadableByteStreamTeeParams> params,
GC::Ref<WebIDL::Promise> cancel_promise)
: m_realm(realm)
, m_stream(stream)
, m_params(params)
, m_cancel_promise(cancel_promise)
{
}
void ReadableByteStreamTeeDefaultReadRequest::visit_edges(Visitor& visitor)
{
Base::visit_edges(visitor);
visitor.visit(m_realm);
visitor.visit(m_stream);
visitor.visit(m_params);
visitor.visit(m_cancel_promise);
}
// https://streams.spec.whatwg.org/#ref-for-read-request-chunk-steps④
void ReadableByteStreamTeeDefaultReadRequest::on_chunk(JS::Value chunk)
{
// 1. Queue a microtask to perform the following steps:
HTML::queue_a_microtask(nullptr, GC::create_function(m_realm->heap(), [this, chunk]() mutable {
HTML::TemporaryExecutionContext execution_context { m_realm, HTML::TemporaryExecutionContext::CallbacksEnabled::Yes };
auto controller1 = m_params->branch1->controller()->get<GC::Ref<ReadableByteStreamController>>();
auto controller2 = m_params->branch2->controller()->get<GC::Ref<ReadableByteStreamController>>();
// 1. Set readAgainForBranch1 to false.
m_params->read_again_for_branch1 = false;
// 2. Set readAgainForBranch2 to false.
m_params->read_again_for_branch2 = false;
// 3. Let chunk1 and chunk2 be chunk.
auto chunk1 = chunk;
auto chunk2 = chunk;
// 4. If canceled1 is false and canceled2 is false,
if (!m_params->canceled1 && !m_params->canceled2) {
// 1. Let cloneResult be CloneAsUint8Array(chunk).
auto chunk_view = m_realm->create<WebIDL::ArrayBufferView>(chunk.as_object());
auto clone_result = clone_as_uint8_array(m_realm, chunk_view);
// 2. If cloneResult is an abrupt completion,
if (clone_result.is_exception()) {
auto completion = Bindings::exception_to_throw_completion(m_realm->vm(), clone_result.release_error());
// 1. Perform ! ReadableByteStreamControllerError(branch1.[[controller]], cloneResult.[[Value]]).
readable_byte_stream_controller_error(controller1, completion.value());
// 2. Perform ! ReadableByteStreamControllerError(branch2.[[controller]], cloneResult.[[Value]]).
readable_byte_stream_controller_error(controller2, completion.value());
// 3. Resolve cancelPromise with ! ReadableStreamCancel(stream, cloneResult.[[Value]]).
auto cancel_result = readable_stream_cancel(m_stream, completion.value());
WebIDL::resolve_promise(m_realm, m_cancel_promise, cancel_result->promise());
// 4. Return.
return;
}
// 3. Otherwise, set chunk2 to cloneResult.[[Value]].
chunk2 = clone_result.release_value();
}
// 5. If canceled1 is false, perform ! ReadableByteStreamControllerEnqueue(branch1.[[controller]], chunk1).
if (!m_params->canceled1) {
MUST(readable_byte_stream_controller_enqueue(controller1, chunk1));
}
// 6. If canceled2 is false, perform ! ReadableByteStreamControllerEnqueue(branch2.[[controller]], chunk2).
if (!m_params->canceled2) {
MUST(readable_byte_stream_controller_enqueue(controller2, chunk2));
}
// 7. Set reading to false.
m_params->reading = false;
// 8. If readAgainForBranch1 is true, perform pull1Algorithm.
if (m_params->read_again_for_branch1) {
m_params->pull1_algorithm->function()();
}
// 9. Otherwise, if readAgainForBranch2 is true, perform pull2Algorithm.
else if (m_params->read_again_for_branch2) {
m_params->pull2_algorithm->function()();
}
}));
// NOTE: The microtask delay here is necessary because it takes at least a microtask to detect errors, when we
// use reader.[[closedPromise]] below. We want errors in stream to error both branches immediately, so we
// cannot let successful synchronously-available reads happen ahead of asynchronously-available errors.
}
// https://streams.spec.whatwg.org/#ref-for-read-request-close-steps③
void ReadableByteStreamTeeDefaultReadRequest::on_close()
{
auto controller1 = m_params->branch1->controller()->get<GC::Ref<ReadableByteStreamController>>();
auto controller2 = m_params->branch2->controller()->get<GC::Ref<ReadableByteStreamController>>();
// 1. Set reading to false.
m_params->reading = false;
// 2. If canceled1 is false, perform ! ReadableByteStreamControllerClose(branch1.[[controller]]).
if (!m_params->canceled1) {
MUST(readable_byte_stream_controller_close(controller1));
}
// 3. If canceled2 is false, perform ! ReadableByteStreamControllerClose(branch2.[[controller]]).
if (!m_params->canceled2) {
MUST(readable_byte_stream_controller_close(controller2));
}
// 4. If branch1.[[controller]].[[pendingPullIntos]] is not empty, perform ! ReadableByteStreamControllerRespond(branch1.[[controller]], 0).
if (!controller1->pending_pull_intos().is_empty()) {
MUST(readable_byte_stream_controller_respond(controller1, 0));
}
// 5. If branch2.[[controller]].[[pendingPullIntos]] is not empty, perform ! ReadableByteStreamControllerRespond(branch2.[[controller]], 0).
if (!controller2->pending_pull_intos().is_empty()) {
MUST(readable_byte_stream_controller_respond(controller2, 0));
}
// 6. If canceled1 is false or canceled2 is false, resolve cancelPromise with undefined.
if (!m_params->canceled1 || !m_params->canceled2) {
WebIDL::resolve_promise(m_realm, m_cancel_promise, JS::js_undefined());
}
}
// https://streams.spec.whatwg.org/#ref-for-read-request-error-steps④
void ReadableByteStreamTeeDefaultReadRequest::on_error(JS::Value)
{
// 1. Set reading to false.
m_params->reading = false;
}
// https://streams.spec.whatwg.org/#ref-for-read-into-request②
ReadableByteStreamTeeBYOBReadRequest::ReadableByteStreamTeeBYOBReadRequest(
JS::Realm& realm,
GC::Ref<ReadableStream> stream,
GC::Ref<ReadableByteStreamTeeParams> params,
GC::Ref<WebIDL::Promise> cancel_promise,
GC::Ref<ReadableStream> byob_branch,
GC::Ref<ReadableStream> other_branch,
bool for_branch2)
: m_realm(realm)
, m_stream(stream)
, m_params(params)
, m_cancel_promise(cancel_promise)
, m_byob_branch(byob_branch)
, m_other_branch(other_branch)
, m_for_branch2(for_branch2)
{
}
void ReadableByteStreamTeeBYOBReadRequest::visit_edges(Visitor& visitor)
{
Base::visit_edges(visitor);
visitor.visit(m_realm);
visitor.visit(m_stream);
visitor.visit(m_params);
visitor.visit(m_cancel_promise);
visitor.visit(m_byob_branch);
visitor.visit(m_other_branch);
}
// https://streams.spec.whatwg.org/#ref-for-read-into-request-chunk-steps①
void ReadableByteStreamTeeBYOBReadRequest::on_chunk(JS::Value chunk)
{
auto chunk_view = m_realm->create<WebIDL::ArrayBufferView>(chunk.as_object());
// 1. Queue a microtask to perform the following steps:
HTML::queue_a_microtask(nullptr, GC::create_function(m_realm->heap(), [this, chunk = chunk_view]() {
HTML::TemporaryExecutionContext execution_context { m_realm, HTML::TemporaryExecutionContext::CallbacksEnabled::Yes };
auto byob_controller = m_byob_branch->controller()->get<GC::Ref<ReadableByteStreamController>>();
auto other_controller = m_other_branch->controller()->get<GC::Ref<ReadableByteStreamController>>();
// 1. Set readAgainForBranch1 to false.
m_params->read_again_for_branch1 = false;
// 2. Set readAgainForBranch2 to false.
m_params->read_again_for_branch2 = false;
// 3. Let byobCanceled be canceled2 if forBranch2 is true, and canceled1 otherwise.
auto byob_cancelled = m_for_branch2 ? m_params->canceled2 : m_params->canceled1;
// 4. Let otherCanceled be canceled2 if forBranch2 is false, and canceled1 otherwise.
auto other_cancelled = !m_for_branch2 ? m_params->canceled2 : m_params->canceled1;
// 5. If otherCanceled is false,
if (!other_cancelled) {
// 1. Let cloneResult be CloneAsUint8Array(chunk).
auto clone_result = clone_as_uint8_array(m_realm, chunk);
// 2. If cloneResult is an abrupt completion,
if (clone_result.is_exception()) {
auto completion = Bindings::exception_to_throw_completion(m_realm->vm(), clone_result.release_error());
// 1. Perform ! ReadableByteStreamControllerError(byobBranch.[[controller]], cloneResult.[[Value]]).
readable_byte_stream_controller_error(byob_controller, completion.value());
// 2. Perform ! ReadableByteStreamControllerError(otherBranch.[[controller]], cloneResult.[[Value]]).
readable_byte_stream_controller_error(other_controller, completion.value());
// 3. Resolve cancelPromise with ! ReadableStreamCancel(stream, cloneResult.[[Value]]).
auto cancel_result = readable_stream_cancel(m_stream, completion.value());
WebIDL::resolve_promise(m_realm, m_cancel_promise, cancel_result->promise());
// 4. Return.
return;
}
// 3. Otherwise, let clonedChunk be cloneResult.[[Value]].
auto cloned_chunk = clone_result.release_value();
// 4. If byobCanceled is false, perform ! ReadableByteStreamControllerRespondWithNewView(byobBranch.[[controller]], chunk).
if (!byob_cancelled) {
MUST(readable_byte_stream_controller_respond_with_new_view(m_realm, byob_controller, chunk));
}
// 5. Perform ! ReadableByteStreamControllerEnqueue(otherBranch.[[controller]], clonedChunk).
MUST(readable_byte_stream_controller_enqueue(other_controller, cloned_chunk));
}
// 6. Otherwise, if byobCanceled is false, perform ! ReadableByteStreamControllerRespondWithNewView(byobBranch.[[controller]], chunk).
else if (!byob_cancelled) {
MUST(readable_byte_stream_controller_respond_with_new_view(m_realm, byob_controller, chunk));
}
// 7. Set reading to false.
m_params->reading = false;
// 8. If readAgainForBranch1 is true, perform pull1Algorithm.
if (m_params->read_again_for_branch1) {
m_params->pull1_algorithm->function()();
}
// 9. Otherwise, if readAgainForBranch2 is true, perform pull2Algorithm.
else if (m_params->read_again_for_branch2) {
m_params->pull2_algorithm->function()();
}
}));
// NOTE: The microtask delay here is necessary because it takes at least a microtask to detect errors, when we
// use reader.[[closedPromise]] below. We want errors in stream to error both branches immediately, so we
// cannot let successful synchronously-available reads happen ahead of asynchronously-available errors.
}
// https://streams.spec.whatwg.org/#ref-for-read-into-request-close-steps②
void ReadableByteStreamTeeBYOBReadRequest::on_close(JS::Value chunk)
{
auto byob_controller = m_byob_branch->controller()->get<GC::Ref<ReadableByteStreamController>>();
auto other_controller = m_other_branch->controller()->get<GC::Ref<ReadableByteStreamController>>();
// 1. Set reading to false.
m_params->reading = false;
// 2. Let byobCanceled be canceled2 if forBranch2 is true, and canceled1 otherwise.
auto byob_cancelled = m_for_branch2 ? m_params->canceled2 : m_params->canceled1;
// 3. Let otherCanceled be canceled2 if forBranch2 is false, and canceled1 otherwise.
auto other_cancelled = !m_for_branch2 ? m_params->canceled2 : m_params->canceled1;
// 4. If byobCanceled is false, perform ! ReadableByteStreamControllerClose(byobBranch.[[controller]]).
if (!byob_cancelled) {
MUST(readable_byte_stream_controller_close(byob_controller));
}
// 5. If otherCanceled is false, perform ! ReadableByteStreamControllerClose(otherBranch.[[controller]]).
if (!other_cancelled) {
MUST(readable_byte_stream_controller_close(other_controller));
}
// 6. If chunk is not undefined,
if (!chunk.is_undefined()) {
// 1. Assert: chunk.[[ByteLength]] is 0.
// 2. If byobCanceled is false, perform ! ReadableByteStreamControllerRespondWithNewView(byobBranch.[[controller]], chunk).
if (!byob_cancelled) {
auto array_buffer_view = m_realm->create<WebIDL::ArrayBufferView>(chunk.as_object());
MUST(readable_byte_stream_controller_respond_with_new_view(m_realm, byob_controller, array_buffer_view));
}
// 3. If otherCanceled is false and otherBranch.[[controller]].[[pendingPullIntos]] is not empty,
// perform ! ReadableByteStreamControllerRespond(otherBranch.[[controller]], 0).
if (!other_cancelled && !other_controller->pending_pull_intos().is_empty()) {
MUST(readable_byte_stream_controller_respond(other_controller, 0));
}
}
// 7. If byobCanceled is false or otherCanceled is false, resolve cancelPromise with undefined.
if (!byob_cancelled || !other_cancelled) {
WebIDL::resolve_promise(m_realm, m_cancel_promise, JS::js_undefined());
}
}
// https://streams.spec.whatwg.org/#ref-for-read-into-request-error-steps①
void ReadableByteStreamTeeBYOBReadRequest::on_error(JS::Value)
{
// 1. Set reading to false.
m_params->reading = false;
}
}

139
Libraries/LibWeb/Streams/ReadableStreamTee.h Normal file
View file

@ -0,0 +1,139 @@
/*
* Copyright (c) 2025, Tim Flynn <trflynn89@ladybird.org>
*
* SPDX-License-Identifier: BSD-2-Clause
*/
#pragma once
#include <LibGC/Ptr.h>
#include <LibJS/Forward.h>
#include <LibJS/Heap/Cell.h>
#include <LibWeb/Forward.h>
#include <LibWeb/Streams/Algorithms.h>
#include <LibWeb/Streams/ReadableStream.h>
#include <LibWeb/Streams/ReadableStreamBYOBReader.h>
#include <LibWeb/Streams/ReadableStreamDefaultReader.h>
namespace Web::Streams::Detail {
struct ReadableStreamTeeParams final : public JS::Cell {
GC_CELL(ReadableStreamTeeParams, JS::Cell);
GC_DECLARE_ALLOCATOR(ReadableStreamTeeParams);
virtual void visit_edges(Visitor& visitor) override;
bool reading { false };
bool read_again { false };
bool canceled1 { false };
bool canceled2 { false };
JS::Value reason1 { JS::js_undefined() };
JS::Value reason2 { JS::js_undefined() };
GC::Ptr<ReadableStream> branch1;
GC::Ptr<ReadableStream> branch2;
GC::Ptr<PullAlgorithm> pull_algorithm;
};
// https://streams.spec.whatwg.org/#ref-for-read-request③
class ReadableStreamTeeReadRequest final : public ReadRequest {
GC_CELL(ReadableStreamTeeReadRequest, ReadRequest);
GC_DECLARE_ALLOCATOR(ReadableStreamTeeReadRequest);
private:
ReadableStreamTeeReadRequest(
JS::Realm& realm,
GC::Ref<ReadableStream> stream,
GC::Ref<ReadableStreamTeeParams> params,
GC::Ref<WebIDL::Promise> cancel_promise,
bool clone_for_branch2);
virtual void visit_edges(Visitor& visitor) override;
virtual void on_chunk(JS::Value chunk) override;
virtual void on_close() override;
virtual void on_error(JS::Value) override;
GC::Ref<JS::Realm> m_realm;
GC::Ref<ReadableStream> m_stream;
GC::Ref<ReadableStreamTeeParams> m_params;
GC::Ref<WebIDL::Promise> m_cancel_promise;
bool m_clone_for_branch2 { false };
};
struct ReadableByteStreamTeeParams final : public JS::Cell {
GC_CELL(ReadableByteStreamTeeParams, JS::Cell);
GC_DECLARE_ALLOCATOR(ReadableByteStreamTeeParams);
explicit ReadableByteStreamTeeParams(ReadableStreamReader reader);
virtual void visit_edges(Visitor& visitor) override;
bool reading { false };
bool read_again_for_branch1 { false };
bool read_again_for_branch2 { false };
bool canceled1 { false };
bool canceled2 { false };
JS::Value reason1 { JS::js_undefined() };
JS::Value reason2 { JS::js_undefined() };
GC::Ptr<ReadableStream> branch1;
GC::Ptr<ReadableStream> branch2;
GC::Ptr<PullAlgorithm> pull1_algorithm;
GC::Ptr<PullAlgorithm> pull2_algorithm;
ReadableStreamReader reader;
};
// https://streams.spec.whatwg.org/#ref-for-read-request④
class ReadableByteStreamTeeDefaultReadRequest final : public ReadRequest {
GC_CELL(ReadableByteStreamTeeDefaultReadRequest, ReadRequest);
GC_DECLARE_ALLOCATOR(ReadableByteStreamTeeDefaultReadRequest);
private:
ReadableByteStreamTeeDefaultReadRequest(
JS::Realm& realm,
GC::Ref<ReadableStream> stream,
GC::Ref<ReadableByteStreamTeeParams> params,
GC::Ref<WebIDL::Promise> cancel_promise);
virtual void visit_edges(Visitor& visitor) override;
virtual void on_chunk(JS::Value chunk) override;
virtual void on_close() override;
virtual void on_error(JS::Value) override;
GC::Ref<JS::Realm> m_realm;
GC::Ref<ReadableStream> m_stream;
GC::Ref<ReadableByteStreamTeeParams> m_params;
GC::Ref<WebIDL::Promise> m_cancel_promise;
};
// https://streams.spec.whatwg.org/#ref-for-read-into-request②
class ReadableByteStreamTeeBYOBReadRequest final : public ReadIntoRequest {
GC_CELL(ReadableByteStreamTeeBYOBReadRequest, ReadIntoRequest);
GC_DECLARE_ALLOCATOR(ReadableByteStreamTeeBYOBReadRequest);
private:
ReadableByteStreamTeeBYOBReadRequest(
JS::Realm& realm,
GC::Ref<ReadableStream> stream,
GC::Ref<ReadableByteStreamTeeParams> params,
GC::Ref<WebIDL::Promise> cancel_promise,
GC::Ref<ReadableStream> byob_branch,
GC::Ref<ReadableStream> other_branch,
bool for_branch2);
virtual void visit_edges(Visitor& visitor) override;
virtual void on_chunk(JS::Value chunk) override;
virtual void on_close(JS::Value chunk) override;
virtual void on_error(JS::Value) override;
GC::Ref<JS::Realm> m_realm;
GC::Ref<ReadableStream> m_stream;
GC::Ref<ReadableByteStreamTeeParams> m_params;
GC::Ref<WebIDL::Promise> m_cancel_promise;
GC::Ref<ReadableStream> m_byob_branch;
GC::Ref<ReadableStream> m_other_branch;
bool m_for_branch2 { false };
};
}

1
Libraries/LibWeb/Streams/TransformStream.cpp
View file

@ -7,6 +7,7 @@
#include <LibWeb/Bindings/Intrinsics.h> #include <LibWeb/Bindings/Intrinsics.h>
#include <LibWeb/Bindings/TransformStreamPrototype.h> #include <LibWeb/Bindings/TransformStreamPrototype.h>
#include <LibWeb/Streams/AbstractOperations.h> #include <LibWeb/Streams/AbstractOperations.h>
#include <LibWeb/Streams/ReadableStream.h>
#include <LibWeb/Streams/TransformStream.h> #include <LibWeb/Streams/TransformStream.h>
#include <LibWeb/Streams/TransformStreamDefaultController.h> #include <LibWeb/Streams/TransformStreamDefaultController.h>
#include <LibWeb/Streams/Transformer.h> #include <LibWeb/Streams/Transformer.h>

2
Libraries/LibWeb/Streams/TransformStreamDefaultController.cpp
View file

@ -6,6 +6,8 @@
#include <LibWeb/Bindings/Intrinsics.h> #include <LibWeb/Bindings/Intrinsics.h>
#include <LibWeb/Bindings/TransformStreamDefaultControllerPrototype.h> #include <LibWeb/Bindings/TransformStreamDefaultControllerPrototype.h>
#include <LibWeb/Streams/AbstractOperations.h>
#include <LibWeb/Streams/ReadableStreamOperations.h>
#include <LibWeb/Streams/TransformStream.h> #include <LibWeb/Streams/TransformStream.h>
#include <LibWeb/Streams/TransformStreamDefaultController.h> #include <LibWeb/Streams/TransformStreamDefaultController.h>

2
Libraries/LibWeb/Streams/TransformStreamDefaultController.h
View file

@ -7,7 +7,7 @@
#pragma once #pragma once
#include <LibWeb/Bindings/PlatformObject.h> #include <LibWeb/Bindings/PlatformObject.h>
#include <LibWeb/Streams/AbstractOperations.h> #include <LibWeb/Streams/Algorithms.h>
namespace Web::Streams { namespace Web::Streams {

1
Libraries/LibWeb/Streams/Transformer.cpp
View file

@ -5,7 +5,6 @@
*/ */
#include <LibJS/Runtime/VM.h> #include <LibJS/Runtime/VM.h>
#include <LibWeb/Streams/AbstractOperations.h>
#include <LibWeb/Streams/Transformer.h> #include <LibWeb/Streams/Transformer.h>
#include <LibWeb/WebIDL/CallbackType.h> #include <LibWeb/WebIDL/CallbackType.h>

1
Libraries/LibWeb/Streams/UnderlyingSink.cpp
View file

@ -5,7 +5,6 @@
*/ */
#include <LibJS/Runtime/VM.h> #include <LibJS/Runtime/VM.h>
#include <LibWeb/Streams/AbstractOperations.h>
#include <LibWeb/Streams/UnderlyingSink.h> #include <LibWeb/Streams/UnderlyingSink.h>
#include <LibWeb/WebIDL/CallbackType.h> #include <LibWeb/WebIDL/CallbackType.h>

1
Libraries/LibWeb/Streams/UnderlyingSource.cpp
View file

@ -6,7 +6,6 @@
*/ */
#include <LibJS/Runtime/VM.h> #include <LibJS/Runtime/VM.h>
#include <LibWeb/Streams/AbstractOperations.h>
#include <LibWeb/Streams/UnderlyingSource.h> #include <LibWeb/Streams/UnderlyingSource.h>
#include <LibWeb/WebIDL/AbstractOperations.h> #include <LibWeb/WebIDL/AbstractOperations.h>
#include <LibWeb/WebIDL/CallbackType.h> #include <LibWeb/WebIDL/CallbackType.h>