Quelle ReadableStreamPipeTo.cpp Sprache: C

/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
/* vim:set ts=2 sw=2 sts=2 et cindent: */
/* This Source Code Form is subject to the terms of the Mozilla Public
* License, v. 2.0. If a copy of the MPL was not distributed with this
* file, You can obtain one at http://mozilla.org/MPL/2.0/. */

#include "ReadableStreamPipeTo.h"

#include "mozilla/dom/AbortFollower.h"
#include "mozilla/dom/AbortSignal.h"
#include "mozilla/dom/ReadableStream.h"
#include "mozilla/dom/ReadableStreamDefaultReader.h"
#include "mozilla/dom/WritableStream.h"
#include "mozilla/dom/WritableStreamDefaultWriter.h"
#include "mozilla/dom/Promise.h"
#include "mozilla/dom/Promise-inl.h"
#include "mozilla/dom/PromiseNativeHandler.h"
#include "mozilla/AlreadyAddRefed.h"
#include "mozilla/ErrorResult.h"
#include "nsCycleCollectionParticipant.h"
#include "nsISupportsImpl.h"

#include "js/Exception.h"

namespace mozilla::dom {

using namespace streams_abstract;

struct PipeToReadRequest;
class WriteFinishedPromiseHandler;
class ShutdownActionFinishedPromiseHandler;

// https://streams.spec.whatwg.org/#readable-stream-pipe-to (Steps 14-15.)
//
// This class implements everything that is required to read all chunks from
// the reader (source) and write them to writer (destination), while
// following the constraints given in the spec using our implementation-defined
// behavior.
//
// The cycle-collected references look roughly like this:
// clang-format off
//
// Closed promise <-- ReadableStreamDefaultReader <--> ReadableStream
//         |                  ^              |
//         |(PromiseHandler)  |(mReader)     |(ReadRequest)
//         |                  |              |
//         |-------------> PipeToPump <-------
//                         ^  |   |
//         |---------------|  |   |
//         |                  |   |-------(mLastWrite) -------->
//         |(PromiseHandler)  |   |< ---- (PromiseHandler) ---- Promise
//         |                  |                                   ^
//         |                  |(mWriter)                          |(mWriteRequests)
//         |                  v                                   |
// Closed promise <-- WritableStreamDefaultWriter <--------> WritableStream
//
// clang-format on
class PipeToPump final : public AbortFollower {
  NS_DECL_CYCLE_COLLECTING_ISUPPORTS
  NS_DECL_CYCLE_COLLECTION_CLASS(PipeToPump)

  friend struct PipeToReadRequest;
  friend class WriteFinishedPromiseHandler;
  friend class ShutdownActionFinishedPromiseHandler;

  PipeToPump(Promise* aPromise, ReadableStreamDefaultReader* aReader,
             WritableStreamDefaultWriter* aWriter, bool aPreventClose,
             bool aPreventAbort, bool aPreventCancel)
      : mPromise(aPromise),
        mReader(aReader),
        mWriter(aWriter),
        mPreventClose(aPreventClose),
        mPreventAbort(aPreventAbort),
        mPreventCancel(aPreventCancel) {}

  MOZ_CAN_RUN_SCRIPT void Start(JSContext* aCx, AbortSignal* aSignal);

  MOZ_CAN_RUN_SCRIPT_BOUNDARY void RunAbortAlgorithm() override;

private:
  ~PipeToPump() override = default;

  MOZ_CAN_RUN_SCRIPT void PerformAbortAlgorithm(JSContext* aCx,
                                                AbortSignalImpl* aSignal);

  MOZ_CAN_RUN_SCRIPT bool SourceOrDestErroredOrClosed(JSContext* aCx);

  using ShutdownAction = already_AddRefed<Promise> (*)(
      JSContext*, PipeToPump*, JS::Handle<mozilla::Maybe<JS::Value>>,
      ErrorResult&);

  MOZ_CAN_RUN_SCRIPT void ShutdownWithAction(
      JSContext* aCx, ShutdownAction aAction,
      JS::Handle<mozilla::Maybe<JS::Value>> aError);
  MOZ_CAN_RUN_SCRIPT void ShutdownWithActionAfterFinishedWrite(
      JSContext* aCx, ShutdownAction aAction,
      JS::Handle<mozilla::Maybe<JS::Value>> aError);

  MOZ_CAN_RUN_SCRIPT void Shutdown(
      JSContext* aCx, JS::Handle<mozilla::Maybe<JS::Value>> aError);

  void Finalize(JSContext* aCx, JS::Handle<mozilla::Maybe<JS::Value>> aError);

  MOZ_CAN_RUN_SCRIPT void OnReadFulfilled(JSContext* aCx,
                                          JS::Handle<JS::Value> aChunk,
                                          ErrorResult& aRv);
  MOZ_CAN_RUN_SCRIPT void OnWriterReady(JSContext* aCx, JS::Handle<JS::Value>);
  MOZ_CAN_RUN_SCRIPT void Read(JSContext* aCx);

  MOZ_CAN_RUN_SCRIPT void OnSourceClosed(JSContext* aCx, JS::Handle<JS::Value>);
  MOZ_CAN_RUN_SCRIPT void OnSourceErrored(
      JSContext* aCx, JS::Handle<JS::Value> aSourceStoredError);

  MOZ_CAN_RUN_SCRIPT void OnDestClosed(JSContext* aCx, JS::Handle<JS::Value>);
  MOZ_CAN_RUN_SCRIPT void OnDestErrored(JSContext* aCx,
                                        JS::Handle<JS::Value> aDestStoredError);

  RefPtr<Promise> mPromise;
  RefPtr<ReadableStreamDefaultReader> mReader;
  RefPtr<WritableStreamDefaultWriter> mWriter;
  RefPtr<Promise> mLastWritePromise;
  const bool mPreventClose;
  const bool mPreventAbort;
  const bool mPreventCancel;
  bool mShuttingDown = false;
#ifdef DEBUG
  bool mReadChunk = false;
#endif
};

// This is a helper class for PipeToPump that allows it to attach
// member functions as promise handlers.
class PipeToPumpHandler final : public PromiseNativeHandler {
  virtual ~PipeToPumpHandler() = default;

  using FunPtr = void (PipeToPump::*)(JSContext*, JS::Handle<JS::Value>);

  RefPtr<PipeToPump> mPipeToPump;
  FunPtr mResolved;
  FunPtr mRejected;

public:
  NS_DECL_CYCLE_COLLECTING_ISUPPORTS
  NS_DECL_CYCLE_COLLECTION_CLASS(PipeToPumpHandler)

  explicit PipeToPumpHandler(PipeToPump* aPipeToPump, FunPtr aResolved,
                             FunPtr aRejected)
      : mPipeToPump(aPipeToPump), mResolved(aResolved), mRejected(aRejected) {}

  void ResolvedCallback(JSContext* aCx, JS::Handle<JS::Value> aValue,
                        ErrorResult&) override {
    if (mResolved) {
      (mPipeToPump->*mResolved)(aCx, aValue);
    }
  }

  void RejectedCallback(JSContext* aCx, JS::Handle<JS::Value> aReason,
                        ErrorResult&) override {
    if (mRejected) {
      (mPipeToPump->*mRejected)(aCx, aReason);
    }
  }
};

NS_IMPL_CYCLE_COLLECTION(PipeToPumpHandler, mPipeToPump)
NS_IMPL_CYCLE_COLLECTING_ADDREF(PipeToPumpHandler)
NS_IMPL_CYCLE_COLLECTING_RELEASE(PipeToPumpHandler)
NS_INTERFACE_MAP_BEGIN_CYCLE_COLLECTION(PipeToPumpHandler)
  NS_INTERFACE_MAP_ENTRY(nsISupports)
NS_INTERFACE_MAP_END

void PipeToPump::RunAbortAlgorithm() {
  AutoJSAPI jsapi;
  if (!jsapi.Init(mReader->GetStream()->GetParentObject())) {
    NS_WARNING(
        "Failed to initialize AutoJSAPI in PipeToPump::RunAbortAlgorithm");
    return;
  }
  JSContext* cx = jsapi.cx();

  RefPtr<AbortSignalImpl> signal = Signal();
  PerformAbortAlgorithm(cx, signal);
}

void PipeToPump::PerformAbortAlgorithm(JSContext* aCx,
                                       AbortSignalImpl* aSignal) {
  MOZ_ASSERT(aSignal->Aborted());

  // https://streams.spec.whatwg.org/#readable-stream-pipe-to
  // Step 14.1. Let abortAlgorithm be the following steps:
  // Note: All the following steps are 14.1.xx

  // Step 1. Let error be signal’s abort reason.
  JS::Rooted<JS::Value> error(aCx);
  aSignal->GetReason(aCx, &error);

  auto action = [](JSContext* aCx, PipeToPump* aPipeToPump,
                   JS::Handle<mozilla::Maybe<JS::Value>> aError,
                   ErrorResult& aRv) MOZ_CAN_RUN_SCRIPT {
    JS::Rooted<JS::Value> error(aCx, *aError);

    // Step 2. Let actions be an empty ordered set.
    nsTArray<RefPtr<Promise>> actions;

    // Step 3. If preventAbort is false, append the following action to actions:
    if (!aPipeToPump->mPreventAbort) {
      RefPtr<WritableStream> dest = aPipeToPump->mWriter->GetStream();

      // Step 3.1. If dest.[[state]] is "writable", return !
      // WritableStreamAbort(dest, error).
      if (dest->State() == WritableStream::WriterState::Writable) {
        RefPtr<Promise> p = WritableStreamAbort(aCx, dest, error, aRv);
        if (aRv.Failed()) {
          return already_AddRefed<Promise>();
        }
        actions.AppendElement(p);
      }

      // Step 3.2. Otherwise, return a promise resolved with undefined.
      // Note: This is basically a no-op.
    }

    // Step 4. If preventCancel is false, append the following action action to
    // actions:
    if (!aPipeToPump->mPreventCancel) {
      RefPtr<ReadableStream> source = aPipeToPump->mReader->GetStream();

      // Step 4.1. If source.[[state]] is "readable", return !
      // ReadableStreamCancel(source, error).
      if (source->State() == ReadableStream::ReaderState::Readable) {
        RefPtr<Promise> p = ReadableStreamCancel(aCx, source, error, aRv);
        if (aRv.Failed()) {
          return already_AddRefed<Promise>();
        }
        actions.AppendElement(p);
      }

      // Step 4.2. Otherwise, return a promise resolved with undefined.
      // No-op again.
    }

    // Step 5. .. action consisting of getting a promise to wait for
    // all of the actions in actions ...
    return Promise::All(aCx, actions, aRv);
  };

  // Step 5. Shutdown with an action consisting of getting a promise to wait for
  // all of the actions in actions, and with error.
  JS::Rooted<Maybe<JS::Value>> someError(aCx, Some(error.get()));
  ShutdownWithAction(aCx, action, someError);
}

bool PipeToPump::SourceOrDestErroredOrClosed(JSContext* aCx) {
  // (Constraint) Error and close states must be propagated:
  // the following conditions must be applied in order.
  RefPtr<ReadableStream> source = mReader->GetStream();
  RefPtr<WritableStream> dest = mWriter->GetStream();

  // Step 1. Errors must be propagated forward: if source.[[state]] is or
  // becomes "errored", then
  if (source->State() == ReadableStream::ReaderState::Errored) {
    JS::Rooted<JS::Value> storedError(aCx, source->StoredError());
    OnSourceErrored(aCx, storedError);
    return true;
  }

  // Step 2. Errors must be propagated backward: if dest.[[state]] is or becomes
  // "errored", then
  if (dest->State() == WritableStream::WriterState::Errored) {
    JS::Rooted<JS::Value> storedError(aCx, dest->StoredError());
    OnDestErrored(aCx, storedError);
    return true;
  }

  // Step 3. Closing must be propagated forward: if source.[[state]] is or
  // becomes "closed", then
  if (source->State() == ReadableStream::ReaderState::Closed) {
    OnSourceClosed(aCx, JS::UndefinedHandleValue);
    return true;
  }

  // Step 4. Closing must be propagated backward:
  // if ! WritableStreamCloseQueuedOrInFlight(dest) is true
  // or dest.[[state]] is "closed", then
  if (dest->CloseQueuedOrInFlight() ||
      dest->State() == WritableStream::WriterState::Closed) {
    OnDestClosed(aCx, JS::UndefinedHandleValue);
    return true;
  }

  return false;
}

// https://streams.spec.whatwg.org/#readable-stream-pipe-to
// Steps 14-15.
void PipeToPump::Start(JSContext* aCx, AbortSignal* aSignal) {
  // Step 14. If signal is not undefined,
  if (aSignal) {
    // Step 14.1. Let abortAlgorithm be the following steps:
    // ... This is implemented by RunAbortAlgorithm.

    // Step 14.2. If signal is aborted, perform abortAlgorithm and
    // return promise.
    if (aSignal->Aborted()) {
      PerformAbortAlgorithm(aCx, aSignal);
      return;
    }

    // Step 14.3. Add abortAlgorithm to signal.
    Follow(aSignal);
  }

  // Step 15. In parallel but not really; see #905, using reader and writer,
  // read all chunks from source and write them to dest.
  // Due to the locking provided by the reader and writer,
  // the exact manner in which this happens is not observable to author code,
  // and so there is flexibility in how this is done.

  // (Constraint) Error and close states must be propagated

  // Before piping has started, we have to check for source/destination being
  // errored/closed manually.
  if (SourceOrDestErroredOrClosed(aCx)) {
    return;
  }

  // We use the following two promises to propagate error and close states
  // during piping.
  RefPtr<Promise> readerClosed = mReader->ClosedPromise();
  readerClosed->AppendNativeHandler(new PipeToPumpHandler(
      this, &PipeToPump::OnSourceClosed, &PipeToPump::OnSourceErrored));

  // Note: Because we control the destination/writer it should never be closed
  // after we did the initial check above with SourceOrDestErroredOrClosed.
  RefPtr<Promise> writerClosed = mWriter->ClosedPromise();
  writerClosed->AppendNativeHandler(new PipeToPumpHandler(
      this, &PipeToPump::OnDestClosed, &PipeToPump::OnDestErrored));

  Read(aCx);
}

class WriteFinishedPromiseHandler final : public PromiseNativeHandler {
  RefPtr<PipeToPump> mPipeToPump;
  PipeToPump::ShutdownAction mAction;
  bool mHasError;
  JS::Heap<JS::Value> mError;

  virtual ~WriteFinishedPromiseHandler() { mozilla::DropJSObjects(this); };

public:
  NS_DECL_CYCLE_COLLECTING_ISUPPORTS
  NS_DECL_CYCLE_COLLECTION_SCRIPT_HOLDER_CLASS(WriteFinishedPromiseHandler)

  explicit WriteFinishedPromiseHandler(
      JSContext* aCx, PipeToPump* aPipeToPump,
      PipeToPump::ShutdownAction aAction,
      JS::Handle<mozilla::Maybe<JS::Value>> aError)
      : mPipeToPump(aPipeToPump), mAction(aAction) {
    mHasError = aError.isSome();
    if (mHasError) {
      mError = *aError;
    }
    mozilla::HoldJSObjects(this);
  }

  MOZ_CAN_RUN_SCRIPT void WriteFinished(JSContext* aCx) {
    RefPtr<PipeToPump> pipeToPump = mPipeToPump;  // XXX known-live?
    JS::Rooted<Maybe<JS::Value>> error(aCx);
    if (mHasError) {
      error = Some(mError);
    }
    pipeToPump->ShutdownWithActionAfterFinishedWrite(aCx, mAction, error);
  }

  MOZ_CAN_RUN_SCRIPT void ResolvedCallback(JSContext* aCx,
                                           JS::Handle<JS::Value> aValue,
                                           ErrorResult&) override {
    WriteFinished(aCx);
  }

  MOZ_CAN_RUN_SCRIPT void RejectedCallback(JSContext* aCx,
                                           JS::Handle<JS::Value> aReason,
                                           ErrorResult&) override {
    WriteFinished(aCx);
  }
};

NS_IMPL_CYCLE_COLLECTION_WITH_JS_MEMBERS(WriteFinishedPromiseHandler,
                                         (mPipeToPump), (mError))
NS_IMPL_CYCLE_COLLECTING_ADDREF(WriteFinishedPromiseHandler)
NS_IMPL_CYCLE_COLLECTING_RELEASE(WriteFinishedPromiseHandler)
NS_INTERFACE_MAP_BEGIN_CYCLE_COLLECTION(WriteFinishedPromiseHandler)
  NS_INTERFACE_MAP_ENTRY(nsISupports)
NS_INTERFACE_MAP_END

// https://streams.spec.whatwg.org/#rs-pipeTo-shutdown-with-action
// Shutdown with an action: if any of the above requirements ask to shutdown
// with an action action, optionally with an error originalError, then:
void PipeToPump::ShutdownWithAction(
    JSContext* aCx, ShutdownAction aAction,
    JS::Handle<mozilla::Maybe<JS::Value>> aError) {
  // Step 1. If shuttingDown is true, abort these substeps.
  if (mShuttingDown) {
    return;
  }

  // Step 2. Set shuttingDown to true.
  mShuttingDown = true;

  // Step 3. If dest.[[state]] is "writable" and !
  // WritableStreamCloseQueuedOrInFlight(dest) is false,
  RefPtr<WritableStream> dest = mWriter->GetStream();
  if (dest->State() == WritableStream::WriterState::Writable &&
      !dest->CloseQueuedOrInFlight()) {
    // Step 3.1. If any chunks have been read but not yet written, write them to
    // dest.
    // Step 3.2. Wait until every chunk that has been read has been
    // written (i.e. the corresponding promises have settled).
    //
    // Note: Write requests are processed in order, so when the promise
    // for the last written chunk is settled all previous chunks have been
    // written as well.
    if (mLastWritePromise) {
      mLastWritePromise->AppendNativeHandler(
          new WriteFinishedPromiseHandler(aCx, this, aAction, aError));
      return;
    }
  }

  // Don't have to wait for last write, immediately continue.
  ShutdownWithActionAfterFinishedWrite(aCx, aAction, aError);
}

class ShutdownActionFinishedPromiseHandler final : public PromiseNativeHandler {
  RefPtr<PipeToPump> mPipeToPump;
  bool mHasError;
  JS::Heap<JS::Value> mError;

  virtual ~ShutdownActionFinishedPromiseHandler() {
    mozilla::DropJSObjects(this);
  }

public:
  NS_DECL_CYCLE_COLLECTING_ISUPPORTS
  NS_DECL_CYCLE_COLLECTION_SCRIPT_HOLDER_CLASS(
      ShutdownActionFinishedPromiseHandler)

  explicit ShutdownActionFinishedPromiseHandler(
      JSContext* aCx, PipeToPump* aPipeToPump,
      JS::Handle<mozilla::Maybe<JS::Value>> aError)
      : mPipeToPump(aPipeToPump) {
    mHasError = aError.isSome();
    if (mHasError) {
      mError = *aError;
    }
    mozilla::HoldJSObjects(this);
  }

  void ResolvedCallback(JSContext* aCx, JS::Handle<JS::Value> aValue,
                        ErrorResult&) override {
    // https://streams.spec.whatwg.org/#rs-pipeTo-shutdown-with-action
    // Step 5. Upon fulfillment of p, finalize, passing along originalError if
    // it was given.
    JS::Rooted<Maybe<JS::Value>> error(aCx);
    if (mHasError) {
      error = Some(mError);
    }
    mPipeToPump->Finalize(aCx, error);
  }

  void RejectedCallback(JSContext* aCx, JS::Handle<JS::Value> aReason,
                        ErrorResult&) override {
    // https://streams.spec.whatwg.org/#rs-pipeTo-shutdown-with-action
    // Step 6. Upon rejection of p with reason newError, finalize with
    // newError.
    JS::Rooted<Maybe<JS::Value>> error(aCx, Some(aReason));
    mPipeToPump->Finalize(aCx, error);
  }
};

NS_IMPL_CYCLE_COLLECTION_WITH_JS_MEMBERS(ShutdownActionFinishedPromiseHandler,
                                         (mPipeToPump), (mError))
NS_IMPL_CYCLE_COLLECTING_ADDREF(ShutdownActionFinishedPromiseHandler)
NS_IMPL_CYCLE_COLLECTING_RELEASE(ShutdownActionFinishedPromiseHandler)
NS_INTERFACE_MAP_BEGIN_CYCLE_COLLECTION(ShutdownActionFinishedPromiseHandler)
  NS_INTERFACE_MAP_ENTRY(nsISupports)
NS_INTERFACE_MAP_END

// https://streams.spec.whatwg.org/#rs-pipeTo-shutdown-with-action
// Continuation after Step 3. triggered a promise resolution.
void PipeToPump::ShutdownWithActionAfterFinishedWrite(
    JSContext* aCx, ShutdownAction aAction,
    JS::Handle<mozilla::Maybe<JS::Value>> aError) {
  if (!aAction) {
    // Used to implement shutdown without action. Finalize immediately.
    Finalize(aCx, aError);
    return;
  }

  // Step 4. Let p be the result of performing action.
  RefPtr<PipeToPump> thisRefPtr = this;
  ErrorResult rv;
  RefPtr<Promise> p = aAction(aCx, thisRefPtr, aError, rv);

  // Error while calling actions above, continue immediately with finalization.
  if (rv.MaybeSetPendingException(aCx)) {
    JS::Rooted<Maybe<JS::Value>> someError(aCx);

    JS::Rooted<JS::Value> error(aCx);
    if (JS_GetPendingException(aCx, &error)) {
      someError = Some(error.get());
    }

    JS_ClearPendingException(aCx);

    Finalize(aCx, someError);
    return;
  }

  // Steps 5-6.
  p->AppendNativeHandler(
      new ShutdownActionFinishedPromiseHandler(aCx, this, aError));
}

// https://streams.spec.whatwg.org/#rs-pipeTo-shutdown
// Shutdown: if any of the above requirements or steps ask to shutdown,
// optionally with an error error, then:
void PipeToPump::Shutdown(JSContext* aCx,
                          JS::Handle<mozilla::Maybe<JS::Value>> aError) {
  // Note: We implement "shutdown" in terms of "shutdown with action".
  // We can observe that when passing along an action that always succeeds
  // shutdown with action and shutdown have the same behavior, when
  // Ignoring the potential micro task for the promise that we skip anyway.
  ShutdownWithAction(aCx, nullptr, aError);
}

// https://streams.spec.whatwg.org/#rs-pipeTo-finalize
// Finalize: both forms of shutdown will eventually ask to finalize,
// optionally with an error error, which means to perform the following steps:
void PipeToPump::Finalize(JSContext* aCx,
                          JS::Handle<mozilla::Maybe<JS::Value>> aError) {
  IgnoredErrorResult rv;
  // Step 1. Perform ! WritableStreamDefaultWriterRelease(writer).
  WritableStreamDefaultWriterRelease(aCx, mWriter);

  // Step 2. If reader implements ReadableStreamBYOBReader,
  // perform ! ReadableStreamBYOBReaderRelease(reader).
  // Note: We always use a default reader.
  MOZ_ASSERT(!mReader->IsBYOB());

  // Step 3. Otherwise, perform ! ReadableStreamDefaultReaderRelease(reader).
  ReadableStreamDefaultReaderRelease(aCx, mReader, rv);
  NS_WARNING_ASSERTION(!rv.Failed(),
                       "ReadableStreamReaderGenericRelease should not fail.");

  // Step 3. If signal is not undefined, remove abortAlgorithm from signal.
  if (IsFollowing()) {
    Unfollow();
  }

  // Step 4. If error was given, reject promise with error.
  if (aError.isSome()) {
    JS::Rooted<JS::Value> error(aCx, *aError);
    mPromise->MaybeReject(error);
  } else {
    // Step 5. Otherwise, resolve promise with undefined.
    mPromise->MaybeResolveWithUndefined();
  }

  // Remove all references.
  mPromise = nullptr;
  mReader = nullptr;
  mWriter = nullptr;
  mLastWritePromise = nullptr;
  Unfollow();
}

void PipeToPump::OnReadFulfilled(JSContext* aCx, JS::Handle<JS::Value> aChunk,
                                 ErrorResult& aRv) {
  // (Constraint) 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 ...
  //
  // We may reach this point after |On{Source,Dest}{Clos,Error}ed| has responded
  // to an out-of-band change.  Per the comment in |OnSourceErrored|, we want to
  // allow the implicated shutdown to proceed, and we don't want to interfere
  // with or additionally alter its operation.  Particularly, we don't want to
  // queue up the successfully-read chunk (if there was one, and this isn't just
  // reporting "done") to be written: it wasn't "already-read" when that
  // error/closure happened.
  //
  // All specified reactions to a closure/error invoke either the shutdown, or
  // shutdown with an action, algorithms.  Those algorithms each abort if either
  // shutdown algorithm has already been invoked.  So we check for shutdown here
  // in case of asynchronous closure/error and abort if shutdown has already
  // started (and possibly finished).
  //
  // TODO: Implement the eventual resolution from
  // https://github.com/whatwg/streams/issues/1207
  if (mShuttingDown) {
    return;
  }

  // Write asynchronously. Roughly this is like:
  // `Promise.resolve().then(() => stream.write(chunk));`
  // XXX: The spec currently does not require asynchronicity, but this still
  // matches other engines' behavior. See
  // https://github.com/whatwg/streams/issues/1243.
  RefPtr<Promise> promise =
      Promise::CreateInfallible(xpc::CurrentNativeGlobal(aCx));
  promise->MaybeResolveWithUndefined();
  auto result = promise->ThenWithCycleCollectedArgsJS(
      [](JSContext* aCx, JS::Handle<JS::Value>, ErrorResult& aRv,
         const RefPtr<PipeToPump>& aSelf,
         const RefPtr<WritableStreamDefaultWriter>& aWriter,
         JS::Handle<JS::Value> aChunk)
          MOZ_CAN_RUN_SCRIPT_FOR_DEFINITION -> already_AddRefed<Promise> {
            RefPtr<Promise> promise =
                WritableStreamDefaultWriterWrite(aCx, aWriter, aChunk, aRv);

            // Last read has finished, so it's time to start reading again.
            aSelf->Read(aCx);

            return promise.forget();
          },
      std::make_tuple(RefPtr{this}, mWriter), std::make_tuple(aChunk));
  if (result.isErr()) {
    mLastWritePromise = nullptr;
    return;
  }
  mLastWritePromise = result.unwrap();

  mLastWritePromise->AppendNativeHandler(
      new PipeToPumpHandler(this, nullptr, &PipeToPump::OnDestErrored));
}

void PipeToPump::OnWriterReady(JSContext* aCx, JS::Handle<JS::Value>) {
  // Writer is ready again (i.e. backpressure was resolved), so read.
  Read(aCx);
}

struct PipeToReadRequest : public ReadRequest {
public:
  NS_DECL_ISUPPORTS_INHERITED
  NS_DECL_CYCLE_COLLECTION_CLASS_INHERITED(PipeToReadRequest, ReadRequest)

  RefPtr<PipeToPump> mPipeToPump;

  explicit PipeToReadRequest(PipeToPump* aPipeToPump)
      : mPipeToPump(aPipeToPump) {}

  MOZ_CAN_RUN_SCRIPT void ChunkSteps(JSContext* aCx,
                                     JS::Handle<JS::Value> aChunk,
                                     ErrorResult& aRv) override {
    RefPtr<PipeToPump> pipeToPump = mPipeToPump;  // XXX known live?
    pipeToPump->OnReadFulfilled(aCx, aChunk, aRv);
  }

  // The reader's closed promise handlers will already call OnSourceClosed/
  // OnSourceErrored, so these steps can just be ignored.
  void CloseSteps(JSContext* aCx, ErrorResult& aRv) override {}
  void ErrorSteps(JSContext* aCx, JS::Handle<JS::Value> aError,
                  ErrorResult& aRv) override {}

protected:
  virtual ~PipeToReadRequest() = default;
};

NS_IMPL_CYCLE_COLLECTION_INHERITED(PipeToReadRequest, ReadRequest, mPipeToPump)

NS_IMPL_ADDREF_INHERITED(PipeToReadRequest, ReadRequest)
NS_IMPL_RELEASE_INHERITED(PipeToReadRequest, ReadRequest)

NS_INTERFACE_MAP_BEGIN_CYCLE_COLLECTION(PipeToReadRequest)
NS_INTERFACE_MAP_END_INHERITING(ReadRequest)

void PipeToPump::Read(JSContext* aCx) {
#ifdef DEBUG
  mReadChunk = true;
#endif

  // (Constraint) Shutdown must stop activity:
  // If shuttingDown becomes true, the user agent must not initiate
  // further reads from reader
  if (mShuttingDown) {
    return;
  }

  // (Constraint) Backpressure must be enforced:
  // While WritableStreamDefaultWriterGetDesiredSize(writer) is ≤ 0 or is null,
  // the user agent must not read from reader.
  Nullable<double> desiredSize =
      WritableStreamDefaultWriterGetDesiredSize(mWriter);
  if (desiredSize.IsNull()) {
    // This means the writer has errored. This is going to be handled
    // by the writer closed promise.
    return;
  }

  if (desiredSize.Value() <= 0) {
    // Wait for the writer to become ready before reading more data from
    // the reader. We don't care about rejections here, because those are
    // already handled by the writer closed promise.
    RefPtr<Promise> readyPromise = mWriter->Ready();
    readyPromise->AppendNativeHandler(
        new PipeToPumpHandler(this, &PipeToPump::OnWriterReady, nullptr));
    return;
  }

  RefPtr<ReadableStreamDefaultReader> reader = mReader;
  RefPtr<ReadRequest> request = new PipeToReadRequest(this);
  ErrorResult rv;
  ReadableStreamDefaultReaderRead(aCx, reader, request, rv);
  if (rv.MaybeSetPendingException(aCx)) {
    // XXX It's actually not quite obvious what we should do here.
    // We've got an error during reading, so on the surface it seems logical
    // to invoke `OnSourceErrored`. However in certain cases the required
    // condition > source.[[state]] is or becomes "errored" < won't actually
    // happen i.e. when `WritableStreamDefaultWriterWrite` called from
    // `OnReadFulfilled` (via PipeToReadRequest::ChunkSteps) fails in
    // a synchronous fashion.
    JS::Rooted<JS::Value> error(aCx);
    JS::Rooted<Maybe<JS::Value>> someError(aCx);

    // The error was moved to the JSContext by MaybeSetPendingException.
    if (JS_GetPendingException(aCx, &error)) {
      someError = Some(error.get());
    }

    JS_ClearPendingException(aCx);

    Shutdown(aCx, someError);
  }
}

// Step 3. Closing must be propagated forward: if source.[[state]] is or
// becomes "closed", then
void PipeToPump::OnSourceClosed(JSContext* aCx, JS::Handle<JS::Value>) {
  // Step 3.1. If preventClose is false, shutdown with an action of
  // ! WritableStreamDefaultWriterCloseWithErrorPropagation(writer).
  if (!mPreventClose) {
    ShutdownWithAction(
        aCx,
        [](JSContext* aCx, PipeToPump* aPipeToPump,
           JS::Handle<mozilla::Maybe<JS::Value>> aError, ErrorResult& aRv)
            MOZ_CAN_RUN_SCRIPT {
              RefPtr<WritableStreamDefaultWriter> writer = aPipeToPump->mWriter;
              return WritableStreamDefaultWriterCloseWithErrorPropagation(
                  aCx, writer, aRv);
            },
        JS::NothingHandleValue);
  } else {
    // Step 3.2 Otherwise, shutdown.
    Shutdown(aCx, JS::NothingHandleValue);
  }
}

// Step 1. Errors must be propagated forward: if source.[[state]] is or
// becomes "errored", then
void PipeToPump::OnSourceErrored(JSContext* aCx,
                                 JS::Handle<JS::Value> aSourceStoredError) {
  // If |source| becomes errored not during a pending read, it's clear we must
  // react immediately.
  //
  // But what if |source| becomes errored *during* a pending read?  Should this
  // first error, or the pending-read second error, predominate?  Two semantics
  // are possible when |source|/|dest| become closed or errored while there's a
  // pending read:
  //
  //   1. Wait until the read fulfills or rejects, then respond to the
  //      closure/error without regard to the read having fulfilled or rejected.
  //      (This will simply not react to the read being rejected, or it will
  //      queue up the read chunk to be written during shutdown.)
  //   2. React to the closure/error immediately per "Error and close states
  //      must be propagated".  Then when the read fulfills or rejects later, do
  //      nothing.
  //
  // The spec doesn't clearly require either semantics.  It requires that
  // *already-read* chunks be written (at least if |dest| didn't become errored
  // or closed such that no further writes can occur).  But it's silent as to
  // not-fully-read chunks.  (These semantic differences may only be observable
  // with very carefully constructed readable/writable streams.)
  //
  // It seems best, generally, to react to the temporally-earliest problem that
  // arises, so we implement option #2.  (Blink, in contrast, currently
  // implements option #1.)
  //
  // All specified reactions to a closure/error invoke either the shutdown, or
  // shutdown with an action, algorithms.  Those algorithms each abort if either
  // shutdown algorithm has already been invoked.  So we don't need to do
  // anything special here to deal with a pending read.
  //
  // TODO: Implement the eventual resolution from
  // https://github.com/whatwg/streams/issues/1207

  // Step 1.1 If preventAbort is false, shutdown with an action of
  // ! WritableStreamAbort(dest, source.[[storedError]])
  // and with source.[[storedError]].
  JS::Rooted<Maybe<JS::Value>> error(aCx, Some(aSourceStoredError));
  if (!mPreventAbort) {
    ShutdownWithAction(
        aCx,
        [](JSContext* aCx, PipeToPump* aPipeToPump,
           JS::Handle<mozilla::Maybe<JS::Value>> aError, ErrorResult& aRv)
            MOZ_CAN_RUN_SCRIPT {
              JS::Rooted<JS::Value> error(aCx, *aError);
              RefPtr<WritableStream> dest = aPipeToPump->mWriter->GetStream();
              return WritableStreamAbort(aCx, dest, error, aRv);
            },
        error);
  } else {
    // Step 1.1. Otherwise, shutdown with source.[[storedError]].
    Shutdown(aCx, error);
  }
}

// Step 4. Closing must be propagated backward:
// if ! WritableStreamCloseQueuedOrInFlight(dest) is true
// or dest.[[state]] is "closed", then
void PipeToPump::OnDestClosed(JSContext* aCx, JS::Handle<JS::Value>) {
  // Step 4.1. Assert: no chunks have been read or written.
  // Note: No reading automatically implies no writing.
  // In a perfect world OnDestClosed would only be called before we start
  // piping, because afterwards the writer has an exclusive lock on the stream.
  // In reality the closed promise can still be resolved after we release
  // the lock on the writer in Finalize.
  if (mShuttingDown) {
    return;
  }
  MOZ_ASSERT(!mReadChunk);

  // Step 4.2. Let destClosed be a new TypeError.
  JS::Rooted<Maybe<JS::Value>> destClosed(aCx, Nothing());
  {
    ErrorResult rv;
    rv.ThrowTypeError("Cannot pipe to closed stream");
    JS::Rooted<JS::Value> error(aCx);
    bool ok = ToJSValue(aCx, std::move(rv), &error);
    MOZ_RELEASE_ASSERT(ok, "must be ok");
    destClosed = Some(error.get());
  }

  // Step 4.3. If preventCancel is false, shutdown with an action of
  // ! ReadableStreamCancel(source, destClosed) and with destClosed.
  if (!mPreventCancel) {
    ShutdownWithAction(
        aCx,
        [](JSContext* aCx, PipeToPump* aPipeToPump,
           JS::Handle<mozilla::Maybe<JS::Value>> aError, ErrorResult& aRv)
            MOZ_CAN_RUN_SCRIPT {
              JS::Rooted<JS::Value> error(aCx, *aError);
              RefPtr<ReadableStream> dest = aPipeToPump->mReader->GetStream();
              return ReadableStreamCancel(aCx, dest, error, aRv);
            },
        destClosed);
  } else {
    // Step 4.4. Otherwise, shutdown with destClosed.
    Shutdown(aCx, destClosed);
  }
}

// Step 2. Errors must be propagated backward: if dest.[[state]] is or becomes
// "errored", then
void PipeToPump::OnDestErrored(JSContext* aCx,
                               JS::Handle<JS::Value> aDestStoredError) {
  // Step 2.1. If preventCancel is false, shutdown with an action of
  // ! ReadableStreamCancel(source, dest.[[storedError]])
  // and with dest.[[storedError]].
  JS::Rooted<Maybe<JS::Value>> error(aCx, Some(aDestStoredError));
  if (!mPreventCancel) {
    ShutdownWithAction(
        aCx,
        [](JSContext* aCx, PipeToPump* aPipeToPump,
           JS::Handle<mozilla::Maybe<JS::Value>> aError, ErrorResult& aRv)
            MOZ_CAN_RUN_SCRIPT {
              JS::Rooted<JS::Value> error(aCx, *aError);
              RefPtr<ReadableStream> dest = aPipeToPump->mReader->GetStream();
              return ReadableStreamCancel(aCx, dest, error, aRv);
            },
        error);
  } else {
    // Step 2.1. Otherwise, shutdown with dest.[[storedError]].
    Shutdown(aCx, error);
  }
}

NS_IMPL_CYCLE_COLLECTION_CLASS(PipeToPump)
NS_IMPL_CYCLE_COLLECTING_ADDREF(PipeToPump)
NS_IMPL_CYCLE_COLLECTING_RELEASE(PipeToPump)
NS_INTERFACE_MAP_BEGIN_CYCLE_COLLECTION(PipeToPump)
  NS_INTERFACE_MAP_ENTRY(nsISupports)
NS_INTERFACE_MAP_END

NS_IMPL_CYCLE_COLLECTION_TRAVERSE_BEGIN(PipeToPump)
  NS_IMPL_CYCLE_COLLECTION_TRAVERSE(mPromise)
  NS_IMPL_CYCLE_COLLECTION_TRAVERSE(mReader)
  NS_IMPL_CYCLE_COLLECTION_TRAVERSE(mWriter)
  NS_IMPL_CYCLE_COLLECTION_TRAVERSE(mLastWritePromise)
NS_IMPL_CYCLE_COLLECTION_TRAVERSE_END

NS_IMPL_CYCLE_COLLECTION_UNLINK_BEGIN(PipeToPump)
  NS_IMPL_CYCLE_COLLECTION_UNLINK(mPromise)
  NS_IMPL_CYCLE_COLLECTION_UNLINK(mReader)
  NS_IMPL_CYCLE_COLLECTION_UNLINK(mWriter)
  NS_IMPL_CYCLE_COLLECTION_UNLINK(mLastWritePromise)
NS_IMPL_CYCLE_COLLECTION_UNLINK_END

namespace streams_abstract {
// https://streams.spec.whatwg.org/#readable-stream-pipe-to
already_AddRefed<Promise> ReadableStreamPipeTo(
    ReadableStream* aSource, WritableStream* aDest, bool aPreventClose,
    bool aPreventAbort, bool aPreventCancel, AbortSignal* aSignal,
    mozilla::ErrorResult& aRv) {
  // Step 1. Assert: source implements ReadableStream. (Implicit)
  // Step 2. Assert: dest implements WritableStream. (Implicit)
  // Step 3. Assert: preventClose, preventAbort, and preventCancel are all
  //         booleans (Implicit)
  // Step 4. If signal was not given, let signal be
  //         undefined. (Implicit)
  // Step 5. Assert: either signal is undefined, or signal
  //         implements AbortSignal. (Implicit)
  // Step 6. Assert: !IsReadableStreamLocked(source) is false.
  MOZ_ASSERT(!IsReadableStreamLocked(aSource));

  // Step 7. Assert: !IsWritableStreamLocked(dest) is false.
  MOZ_ASSERT(!IsWritableStreamLocked(aDest));

  AutoJSAPI jsapi;
  if (!jsapi.Init(aSource->GetParentObject())) {
    aRv.ThrowUnknownError("Internal error");
    return nullptr;
  }
  JSContext* cx = jsapi.cx();

  // Step 8. If source.[[controller]] implements ReadableByteStreamController,
  //         let reader be either !AcquireReadableStreamBYOBReader(source) or
  //         !AcquireReadableStreamDefaultReader(source), at the user agent’s
  //         discretion.
  // Step 9. Otherwise, let reader be
  //         !AcquireReadableStreamDefaultReader(source).

  // Note: In the interests of simplicity, we choose here to always acquire
  // a default reader.
  RefPtr<ReadableStreamDefaultReader> reader =
      AcquireReadableStreamDefaultReader(aSource, aRv);
  if (aRv.Failed()) {
    return nullptr;
  }

  // Step 10. Let writer be ! AcquireWritableStreamDefaultWriter(dest).
  RefPtr<WritableStreamDefaultWriter> writer =
      AcquireWritableStreamDefaultWriter(aDest, aRv);
  if (aRv.Failed()) {
    return nullptr;
  }

  // Step 11. Set source.[[disturbed]] to true.
  aSource->SetDisturbed(true);

  // Step 12. Let shuttingDown be false.
  // Note: PipeToPump ensures this by construction.

  // Step 13. Let promise be a new promise.
  RefPtr<Promise> promise =
      Promise::CreateInfallible(aSource->GetParentObject());

  // Steps 14-15.
  RefPtr<PipeToPump> pump = new PipeToPump(
      promise, reader, writer, aPreventClose, aPreventAbort, aPreventCancel);
  pump->Start(cx, aSignal);

  // Step 16. Return promise.
  return promise.forget();
}
}  // namespace streams_abstract

}  // namespace mozilla::dom

quality100%

¤ Dauer der Verarbeitung: 0.21 Sekunden (vorverarbeitet) ¤

Wurzel

Suchen

Beweissystem der NASA

Beweissystem Isabelle

NIST Cobol Testsuite

Cephes Mathematical Library

Wiener Entwicklungsmethode

Haftungshinweis

Die Informationen auf dieser Webseite wurden nach bestem Wissen sorgfältig zusammengestellt. Es wird jedoch weder Vollständigkeit, noch Richtigkeit, noch Qualität der bereit gestellten Informationen zugesichert.

Bemerkung:

Die farbliche Syntaxdarstellung ist noch experimentell.