chromium/third_party/blink/renderer/core/streams/readable_stream.cc

// Copyright 2019 The Chromium Authors
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#include "third_party/blink/renderer/core/streams/readable_stream.h"

#include "third_party/blink/renderer/bindings/core/v8/native_value_traits_impl.h"
#include "third_party/blink/renderer/bindings/core/v8/script_function.h"
#include "third_party/blink/renderer/bindings/core/v8/script_promise_resolver.h"
#include "third_party/blink/renderer/bindings/core/v8/to_v8_traits.h"
#include "third_party/blink/renderer/bindings/core/v8/v8_readable_stream.h"
#include "third_party/blink/renderer/bindings/core/v8/v8_readable_stream_get_reader_options.h"
#include "third_party/blink/renderer/bindings/core/v8/v8_readable_writable_pair.h"
#include "third_party/blink/renderer/bindings/core/v8/v8_stream_pipe_options.h"
#include "third_party/blink/renderer/bindings/core/v8/v8_underlying_source.h"
#include "third_party/blink/renderer/bindings/core/v8/v8_union_readablestreambyobreader_readablestreamdefaultreader.h"
#include "third_party/blink/renderer/core/execution_context/execution_context.h"
#include "third_party/blink/renderer/core/frame/web_feature.h"
#include "third_party/blink/renderer/core/streams/byte_stream_tee_engine.h"
#include "third_party/blink/renderer/core/streams/miscellaneous_operations.h"
#include "third_party/blink/renderer/core/streams/pipe_options.h"
#include "third_party/blink/renderer/core/streams/pipe_to_engine.h"
#include "third_party/blink/renderer/core/streams/promise_handler.h"
#include "third_party/blink/renderer/core/streams/read_into_request.h"
#include "third_party/blink/renderer/core/streams/read_request.h"
#include "third_party/blink/renderer/core/streams/readable_byte_stream_controller.h"
#include "third_party/blink/renderer/core/streams/readable_stream_byob_reader.h"
#include "third_party/blink/renderer/core/streams/readable_stream_controller.h"
#include "third_party/blink/renderer/core/streams/readable_stream_default_controller.h"
#include "third_party/blink/renderer/core/streams/readable_stream_generic_reader.h"
#include "third_party/blink/renderer/core/streams/readable_stream_transferring_optimizer.h"
#include "third_party/blink/renderer/core/streams/stream_algorithms.h"
#include "third_party/blink/renderer/core/streams/stream_promise_resolver.h"
#include "third_party/blink/renderer/core/streams/tee_engine.h"
#include "third_party/blink/renderer/core/streams/transferable_streams.h"
#include "third_party/blink/renderer/core/streams/underlying_byte_source_base.h"
#include "third_party/blink/renderer/core/streams/underlying_source_base.h"
#include "third_party/blink/renderer/core/streams/writable_stream.h"
#include "third_party/blink/renderer/core/streams/writable_stream_default_controller.h"
#include "third_party/blink/renderer/core/streams/writable_stream_default_writer.h"
#include "third_party/blink/renderer/core/streams/writable_stream_transferring_optimizer.h"
#include "third_party/blink/renderer/platform/bindings/exception_code.h"
#include "third_party/blink/renderer/platform/bindings/exception_state.h"
#include "third_party/blink/renderer/platform/bindings/script_state.h"
#include "third_party/blink/renderer/platform/bindings/v8_binding.h"
#include "third_party/blink/renderer/platform/bindings/v8_throw_exception.h"
#include "third_party/blink/renderer/platform/heap/collection_support/heap_vector.h"
#include "third_party/blink/renderer/platform/heap/garbage_collected.h"
#include "third_party/blink/renderer/platform/heap/persistent.h"
#include "third_party/blink/renderer/platform/instrumentation/use_counter.h"
#include "third_party/blink/renderer/platform/scheduler/public/event_loop.h"
#include "third_party/blink/renderer/platform/wtf/deque.h"
#include "third_party/blink/renderer/platform/wtf/functional.h"

namespace blink {

// Implements a pull algorithm that delegates to an UnderlyingByteSourceBase.
// This is used when creating a ReadableByteStream from C++.
class ReadableStream::PullAlgorithm final : public StreamAlgorithm { … };

// Implements a cancel algorithm that delegates to an UnderlyingByteSourceBase.
class ReadableStream::CancelAlgorithm final : public StreamAlgorithm { … };

class ReadableStream::IterationSource final
    : public ReadableStream::IterationSourceBase { … };

class ReadableStream::IterationReadRequest final : public ReadRequest { … };

void ReadableStream::IterationSource::GetNextIterationResult() { … }

void ReadableStream::IterationSource::AsyncIteratorReturn(ScriptValue arg) { … }

ReadableStream* ReadableStream::Create(ScriptState* script_state,
                                       ExceptionState& exception_state) { … }

ReadableStream* ReadableStream::Create(ScriptState* script_state,
                                       ScriptValue underlying_source,
                                       ExceptionState& exception_state) { … }

ReadableStream* ReadableStream::Create(ScriptState* script_state,
                                       ScriptValue underlying_source,
                                       ScriptValue strategy,
                                       ExceptionState& exception_state) { … }

ReadableStream* ReadableStream::CreateWithCountQueueingStrategy(
    ScriptState* script_state,
    UnderlyingSourceBase* underlying_source,
    size_t high_water_mark) { … }

ReadableStream* ReadableStream::CreateWithCountQueueingStrategy(
    ScriptState* script_state,
    UnderlyingSourceBase* underlying_source,
    size_t high_water_mark,
    AllowPerChunkTransferring allow_per_chunk_transferring,
    std::unique_ptr<ReadableStreamTransferringOptimizer> optimizer) { … }

void ReadableStream::InitWithCountQueueingStrategy(
    ScriptState* script_state,
    UnderlyingSourceBase* underlying_source,
    size_t high_water_mark,
    AllowPerChunkTransferring allow_per_chunk_transferring,
    std::unique_ptr<ReadableStreamTransferringOptimizer> optimizer,
    ExceptionState& exception_state) { … }

ReadableStream* ReadableStream::Create(ScriptState* script_state,
                                       StreamStartAlgorithm* start_algorithm,
                                       StreamAlgorithm* pull_algorithm,
                                       StreamAlgorithm* cancel_algorithm,
                                       double high_water_mark,
                                       StrategySizeAlgorithm* size_algorithm,
                                       ExceptionState& exception_state) { … }

ReadableStream* ReadableStream::CreateByteStream(
    ScriptState* script_state,
    StreamStartAlgorithm* start_algorithm,
    StreamAlgorithm* pull_algorithm,
    StreamAlgorithm* cancel_algorithm,
    ExceptionState& exception_state) { … }

// static
ReadableStream* ReadableStream::CreateByteStream(
    ScriptState* script_state,
    UnderlyingByteSourceBase* underlying_byte_source) { … }

void ReadableStream::InitByteStream(
    ScriptState* script_state,
    ReadableStream* stream,
    UnderlyingByteSourceBase* underlying_byte_source,
    ExceptionState& exception_state) { … }

void ReadableStream::InitByteStream(ScriptState* script_state,
                                    ReadableStream* stream,
                                    ReadableByteStreamController* controller,
                                    StreamStartAlgorithm* start_algorithm,
                                    StreamAlgorithm* pull_algorithm,
                                    StreamAlgorithm* cancel_algorithm,
                                    ExceptionState& exception_state) { … }

ReadableStream::ReadableStream() = default;

ReadableStream::~ReadableStream() = default;

bool ReadableStream::locked() const { … }

ScriptPromise<IDLUndefined> ReadableStream::cancel(
    ScriptState* script_state,
    ExceptionState& exception_state) { … }

ScriptPromise<IDLUndefined> ReadableStream::cancel(
    ScriptState* script_state,
    ScriptValue reason,
    ExceptionState& exception_state) { … }

V8ReadableStreamReader* ReadableStream::getReader(
    ScriptState* script_state,
    ExceptionState& exception_state) { … }

V8ReadableStreamReader* ReadableStream::getReader(
    ScriptState* script_state,
    const ReadableStreamGetReaderOptions* options,
    ExceptionState& exception_state) { … }

ReadableStreamDefaultReader* ReadableStream::GetDefaultReaderForTesting(
    ScriptState* script_state,
    ExceptionState& exception_state) { … }

ReadableStreamBYOBReader* ReadableStream::GetBYOBReaderForTesting(
    ScriptState* script_state,
    ExceptionState& exception_state) { … }

ReadableStream* ReadableStream::pipeThrough(ScriptState* script_state,
                                            ReadableWritablePair* transform,
                                            ExceptionState& exception_state) { … }

// https://streams.spec.whatwg.org/#rs-pipe-through
ReadableStream* ReadableStream::pipeThrough(ScriptState* script_state,
                                            ReadableWritablePair* transform,
                                            const StreamPipeOptions* options,
                                            ExceptionState& exception_state) { … }

ScriptPromise<IDLUndefined> ReadableStream::pipeTo(
    ScriptState* script_state,
    WritableStream* destination,
    ExceptionState& exception_state) { … }

ScriptPromise<IDLUndefined> ReadableStream::pipeTo(
    ScriptState* script_state,
    WritableStream* destination,
    const StreamPipeOptions* options,
    ExceptionState& exception_state) { … }

HeapVector<Member<ReadableStream>> ReadableStream::tee(
    ScriptState* script_state,
    ExceptionState& exception_state) { … }

// Unlike in the standard, this is defined as a separate method from the
// constructor. This prevents problems when garbage collection happens
// re-entrantly during construction.
void ReadableStream::InitInternal(ScriptState* script_state,
                                  ScriptValue raw_underlying_source,
                                  ScriptValue raw_strategy,
                                  bool created_by_ua,
                                  ExceptionState& exception_state) { … }

//
// Readable stream abstract operations
//
ReadableStreamDefaultReader* ReadableStream::AcquireDefaultReader(
    ScriptState* script_state,
    ReadableStream* stream,
    ExceptionState& exception_state) { … }

ReadableStreamBYOBReader* ReadableStream::AcquireBYOBReader(
    ScriptState* script_state,
    ReadableStream* stream,
    ExceptionState& exception_state) { … }

void ReadableStream::Initialize(ReadableStream* stream) { … }

void ReadableStream::Tee(ScriptState* script_state,
                         ReadableStream** branch1,
                         ReadableStream** branch2,
                         bool clone_for_branch2,
                         ExceptionState& exception_state) { … }

void ReadableStream::ByteStreamTee(ScriptState* script_state,
                                   ReadableStream** branch1,
                                   ReadableStream** branch2,
                                   ExceptionState& exception_state) { … }

void ReadableStream::LockAndDisturb(ScriptState* script_state) { … }

void ReadableStream::CloseStream(ScriptState* script_state,
                                 ExceptionState& exception_state) { … }

void ReadableStream::Serialize(ScriptState* script_state,
                               MessagePort* port,
                               ExceptionState& exception_state) { … }

ReadableStream* ReadableStream::Deserialize(
    ScriptState* script_state,
    MessagePort* port,
    std::unique_ptr<ReadableStreamTransferringOptimizer> optimizer,
    ExceptionState& exception_state) { … }

ScriptPromise<IDLUndefined> ReadableStream::PipeTo(
    ScriptState* script_state,
    ReadableStream* readable,
    WritableStream* destination,
    PipeOptions* pipe_options,
    ExceptionState& exception_state) { … }

v8::Local<v8::Value> ReadableStream::GetStoredError(
    v8::Isolate* isolate) const { … }

std::unique_ptr<ReadableStreamTransferringOptimizer>
ReadableStream::TakeTransferringOptimizer() { … }

void ReadableStream::Trace(Visitor* visitor) const { … }

//
// Abstract Operations Used By Controllers
//

void ReadableStream::AddReadIntoRequest(ScriptState* script_state,
                                        ReadableStream* stream,
                                        ReadIntoRequest* readRequest) { … }

void ReadableStream::AddReadRequest(ScriptState* script_state,
                                    ReadableStream* stream,
                                    ReadRequest* read_request) { … }

ScriptPromise<IDLUndefined> ReadableStream::Cancel(
    ScriptState* script_state,
    ReadableStream* stream,
    v8::Local<v8::Value> reason) { … }

void ReadableStream::Close(ScriptState* script_state, ReadableStream* stream) { … }

void ReadableStream::Error(ScriptState* script_state,
                           ReadableStream* stream,
                           v8::Local<v8::Value> e) { … }

void ReadableStream::FulfillReadIntoRequest(ScriptState* script_state,
                                            ReadableStream* stream,
                                            DOMArrayBufferView* chunk,
                                            bool done,
                                            ExceptionState& exception_state) { … }

void ReadableStream::FulfillReadRequest(ScriptState* script_state,
                                        ReadableStream* stream,
                                        v8::Local<v8::Value> chunk,
                                        bool done,
                                        ExceptionState& exception_state) { … }

int ReadableStream::GetNumReadIntoRequests(const ReadableStream* stream) { … }

int ReadableStream::GetNumReadRequests(const ReadableStream* stream) { … }

bool ReadableStream::HasBYOBReader(const ReadableStream* stream) { … }

bool ReadableStream::HasDefaultReader(const ReadableStream* stream) { … }

HeapVector<Member<ReadableStream>> ReadableStream::CallTeeAndReturnBranchArray(
    ScriptState* script_state,
    ReadableStream* readable,
    bool clone_for_branch2,
    ExceptionState& exception_state) { … }

ReadableStream::IterationSourceBase* ReadableStream::CreateIterationSource(
    ScriptState* script_state,
    ReadableStream::IterationSourceBase::Kind kind,
    ReadableStreamIteratorOptions* options,
    ExceptionState& exception_state) { … }

}  // namespace blink