observable.cc | Explore in Territory

// Copyright 2023 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/dom/observable.h"

#include "base/types/pass_key.h"
#include "third_party/blink/renderer/bindings/core/v8/script_function.h"
#include "third_party/blink/renderer/bindings/core/v8/script_promise.h"
#include "third_party/blink/renderer/bindings/core/v8/script_promise_resolver.h"
#include "third_party/blink/renderer/bindings/core/v8/v8_catch_callback.h"
#include "third_party/blink/renderer/bindings/core/v8/v8_mapper.h"
#include "third_party/blink/renderer/bindings/core/v8/v8_observable_inspector.h"
#include "third_party/blink/renderer/bindings/core/v8/v8_observable_inspector_abort_handler.h"
#include "third_party/blink/renderer/bindings/core/v8/v8_observer.h"
#include "third_party/blink/renderer/bindings/core/v8/v8_observer_callback.h"
#include "third_party/blink/renderer/bindings/core/v8/v8_observer_complete_callback.h"
#include "third_party/blink/renderer/bindings/core/v8/v8_predicate.h"
#include "third_party/blink/renderer/bindings/core/v8/v8_reducer.h"
#include "third_party/blink/renderer/bindings/core/v8/v8_subscribe_callback.h"
#include "third_party/blink/renderer/bindings/core/v8/v8_subscribe_options.h"
#include "third_party/blink/renderer/bindings/core/v8/v8_union_observableinspector_observercallback.h"
#include "third_party/blink/renderer/bindings/core/v8/v8_union_observer_observercallback.h"
#include "third_party/blink/renderer/bindings/core/v8/v8_visitor.h"
#include "third_party/blink/renderer/bindings/core/v8/v8_void_function.h"
#include "third_party/blink/renderer/core/dom/abort_controller.h"
#include "third_party/blink/renderer/core/dom/dom_exception.h"
#include "third_party/blink/renderer/core/dom/observable_internal_observer.h"
#include "third_party/blink/renderer/core/dom/subscriber.h"
#include "third_party/blink/renderer/core/execution_context/execution_context.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/instrumentation/use_counter.h"
#include "third_party/blink/renderer/platform/runtime_enabled_features.h"

namespace blink {

namespace {

// A helper wrapper since we cannot hold `Member<ScriptValue>` directly.
class ScriptValueHolder final : public GarbageCollected<ScriptValueHolder> { … };

class RejectPromiseAbortAlgorithm final : public AbortSignal::Algorithm { … };

class ScriptCallbackInternalObserver final : public ObservableInternalObserver { … };

class ToArrayInternalObserver final : public ObservableInternalObserver { … };

// This is the internal observer associated with the `reduce()` operator. See
// https://wicg.github.io/observable/#dom-observable-reduce for its definition
// and spec prose.
class OperatorReduceInternalObserver final : public ObservableInternalObserver { … };

// This is the internal observer associated with the `find()` operator. See
// https://wicg.github.io/observable/#dom-observable-find for its definition
// and spec prose quoted below.
class OperatorFindInternalObserver final : public ObservableInternalObserver { … };

// This is the internal observer associated with the `every()` operator. See
// https://wicg.github.io/observable/#dom-observable-every for its definition
// and spec prose quoted below.
class OperatorEveryInternalObserver final : public ObservableInternalObserver { … };

// This is the internal observer associated with the `some()` operator. See
// https://wicg.github.io/observable/#dom-observable-some for its definition
// and spec prose quoted below.
class OperatorSomeInternalObserver final : public ObservableInternalObserver { … };

// This is the internal observer associated with the `last()` operator. See
// https://wicg.github.io/observable/#dom-observable-last for its definition
// and spec prose quoted below.
class OperatorLastInternalObserver final : public ObservableInternalObserver { … };

// This is the internal observer associated with the `first()` operator. See
// https://wicg.github.io/observable/#dom-observable-first for its definition
// and spec prose quoted below.
class OperatorFirstInternalObserver final : public ObservableInternalObserver { … };

class OperatorForEachInternalObserver final
    : public ObservableInternalObserver { … };

// This delegate is used by the `Observer#from()` operator, in the case where
// the given `any` value is a `Promise`. It simply utilizes the promise's
// then/catch handlers to pipe the corresponding fulfilled/rejection value to
// the Observable in a one-shot manner.
class OperatorFromPromiseSubscribeDelegate final
    : public Observable::SubscribeDelegate { … };

// This is the subscribe delegate for the `catch()` operator. It allows one to
// "catch" errors pushed from upstream Observables, and handle them by returning
// a new Observable derived from that error. The Observable returned from the
// catch handler is immediately subscribed to, and its values are plumbed
// downstream. See https://wicg.github.io/observable/#dom-observable-catch.
class OperatorCatchSubscribeDelegate final
    : public Observable::SubscribeDelegate { … };

// This is the subscribe delegate for the `inspect()` operator. It allows one to
// supply a pseudo "Observer" dictionary, specifically an `ObservableInspector`,
// which can tap into the direct outputs of a source Observable. It mirrors its
// `next()`, `error()`, and `complete()` handlers, as well as letting you pass
// in two supplemental callbacks:
//   1. A `subscribe()` callback, which runs immediately when the
//      `Observable`-returned-from-`inspect()` is subscribed to, and just before
//      *it* subscribes to its source Observable. Errors from this callback are
//      piped to the consumer Subscriber's `error()` handler, and the
//      subscription is promptly closed.
//   2. An `abort()` callback, which is run specifically for consumer-initiated
//      unsubscriptions/aborts, NOT producer (source-Observable-initiated)
//      unsubscriptions (via `complete()` or `error()`). See the documentation
//      in `OperatorInspectSubscribeDelegate::SourceInternalObserver::Error()`.
class OperatorInspectSubscribeDelegate final
    : public Observable::SubscribeDelegate { … };

class OperatorSwitchMapSubscribeDelegate final
    : public Observable::SubscribeDelegate { … };

// This class is the subscriber delegate for Observables returned by
// `flatMap()`. Flat map is a tricky operator, so here's how the flow works.
// Upon subscription, `this` subscribes to the "source" Observable, that had its
// `flatMap()` method called. All values that the source Observable emits, get
// piped to its subscription's internal observer, which is
// `OperatorFlatMapSubscribeDelegate::SourceInternalObserver`. It is that class
// that is responsible for mapping each of the individual source Observable, via
// `mapper`, to an Observable (that we call the "inner" Observable), which then
// gets subscribed to. Through the remainder the "inner" Observable's lifetime,
// its values are exclusively piped to the "outer" Subscriber — this allows the
// IDL `Observer` handlers associated with the Observable returned from
// `flatMap()` to observe the inner Observable's values.
//
// Once the inner Observable completes, the focus is transferred to the *next*
// value that the outer Observable has emitted, if one such exists. That value
// too gets mapped and converted to an Observable, and subscribed to, and so on.
// See also, the documentation above
// `OperatorFlatMapSubscribeDelegate::SourceInternalObserver::InnerFlatMapObserver`.
class OperatorFlatMapSubscribeDelegate final
    : public Observable::SubscribeDelegate { … };

// This delegate is used by the `Observer#from()` operator, in the case where
// the given `any` value is an iterable. In that case, we store the iterable in
// `this` delegate, and upon subscription, synchronously push to the subscriber
// all of the iterable's values.
class OperatorFromIterableSubscribeDelegate final
    : public Observable::SubscribeDelegate { … };

class OperatorDropSubscribeDelegate final
    : public Observable::SubscribeDelegate { … };

class OperatorTakeSubscribeDelegate final
    : public Observable::SubscribeDelegate { … };

class OperatorFilterSubscribeDelegate final
    : public Observable::SubscribeDelegate { … };

class OperatorMapSubscribeDelegate final
    : public Observable::SubscribeDelegate { … };

class OperatorTakeUntilSubscribeDelegate final
    : public Observable::SubscribeDelegate { … };

}  // namespace

PassKey;

// static
Observable* Observable::Create(ScriptState* script_state,
                               V8SubscribeCallback* subscribe_callback) { … }

Observable::Observable(ExecutionContext* execution_context,
                       V8SubscribeCallback* subscribe_callback)
    : … { … }

Observable::Observable(ExecutionContext* execution_context,
                       SubscribeDelegate* subscribe_delegate)
    : … { … }

void Observable::subscribe(ScriptState* script_state,
                           V8UnionObserverOrObserverCallback* observer_union,
                           SubscribeOptions* options) { … }

void Observable::SubscribeWithNativeObserver(
    ScriptState* script_state,
    ObservableInternalObserver* internal_observer,
    SubscribeOptions* options) { … }

void Observable::SubscribeInternal(
    ScriptState* script_state,
    V8UnionObserverOrObserverCallback* observer_union,
    ObservableInternalObserver* internal_observer,
    SubscribeOptions* options) { … }

// static
Observable* Observable::from(ScriptState* script_state,
                             ScriptValue value,
                             ExceptionState& exception_state) { … }

Observable* Observable::takeUntil(ScriptState*, Observable* notifier) { … }

Observable* Observable::map(ScriptState*, V8Mapper* mapper) { … }

Observable* Observable::filter(ScriptState*, V8Predicate* predicate) { … }

Observable* Observable::take(ScriptState*, uint64_t number_to_take) { … }

Observable* Observable::drop(ScriptState*, uint64_t number_to_drop) { … }

Observable* Observable::flatMap(ScriptState*,
                                V8Mapper* mapper,
                                ExceptionState& exception_state) { … }

Observable* Observable::switchMap(ScriptState*,
                                  V8Mapper* mapper,
                                  ExceptionState& exception_state) { … }

Observable* Observable::inspect(
    ScriptState* script_state,
    V8UnionObservableInspectorOrObserverCallback* inspector_union) { … }

Observable* Observable::catchImpl(ScriptState*,
                                  V8CatchCallback* callback,
                                  ExceptionState& exception_state) { … }

ScriptPromise<IDLSequence<IDLAny>> Observable::toArray(
    ScriptState* script_state,
    SubscribeOptions* options) { … }

ScriptPromise<IDLUndefined> Observable::forEach(ScriptState* script_state,
                                                V8Visitor* callback,
                                                SubscribeOptions* options) { … }

ScriptPromise<IDLAny> Observable::first(ScriptState* script_state,
                                        SubscribeOptions* options) { … }

ScriptPromise<IDLAny> Observable::last(ScriptState* script_state,
                                       SubscribeOptions* options) { … }

ScriptPromise<IDLBoolean> Observable::some(ScriptState* script_state,
                                           V8Predicate* predicate,
                                           SubscribeOptions* options) { … }

ScriptPromise<IDLBoolean> Observable::every(ScriptState* script_state,
                                            V8Predicate* predicate,
                                            SubscribeOptions* options) { … }

ScriptPromise<IDLAny> Observable::find(ScriptState* script_state,
                                       V8Predicate* predicate,
                                       SubscribeOptions* options) { … }

ScriptPromise<IDLAny> Observable::reduce(ScriptState* script_state,
                                         V8Reducer* reducer) { … }

ScriptPromise<IDLAny> Observable::reduce(ScriptState* script_state,
                                         V8Reducer* reducer,
                                         v8::Local<v8::Value> initialValue,
                                         SubscribeOptions* options) { … }

ScriptPromise<IDLAny> Observable::ReduceInternal(
    ScriptState* script_state,
    V8Reducer* reducer,
    std::optional<ScriptValue> initial_value,
    SubscribeOptions* options) { … }

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

}  // namespace blink
chromium/third_party/blink/renderer/core/dom/observable.cc
