Issue #1587 Part 11 (followup 1): Implement multithreaded signals for workers.

8 files changed, 813 insertions, 722 deletions

diff --git a/dom/fetch/Fetch.cpp b/dom/fetch/Fetch.cpp
index 191f4cfc3..f0350fbce 100644
--- a/dom/fetch/Fetch.cpp
+++ b/dom/fetch/Fetch.cpp
@@ -111,6 +111,12 @@ public:
     return mSignalMainThread;
   }
 
+  AbortSignal*
+  GetSignalForTargetThread()
+  {
+    return mFollowingSignal;
+  }
+
   void
   Shutdown()
   {
@@ -161,7 +167,7 @@ public:
   }
 
   AbortSignal*
-  GetAbortSignal()
+  GetAbortSignalForMainThread()
   {
     MOZ_ASSERT(NS_IsMainThread());
 
@@ -172,6 +178,18 @@ public:
     return mSignalProxy->GetOrCreateSignalForMainThread();
   }
 
+  AbortSignal*
+  GetAbortSignalForTargetThread()
+  {
+    mPromiseProxy->GetWorkerPrivate()->AssertIsOnWorkerThread();
+
+    if (!mSignalProxy) {
+      return nullptr;
+    }
+
+    return mSignalProxy->GetSignalForTargetThread();
+  }
+
   void
   OnResponseAvailableInternal(InternalResponse* aResponse) override;
 
@@ -205,14 +223,16 @@ class MainThreadFetchResolver final : public FetchDriverObserver
   RefPtr<Promise> mPromise;
   RefPtr<Response> mResponse;
   RefPtr<FetchObserver> mFetchObserver;
+  RefPtr<AbortSignal> mSignal;
 
   nsCOMPtr<nsIDocument> mDocument;
 
   NS_DECL_OWNINGTHREAD
 public:
-  MainThreadFetchResolver(Promise* aPromise, FetchObserver* aObserver)
+  MainThreadFetchResolver(Promise* aPromise, FetchObserver* aObserver, AbortSignal* aSignal)
     : mPromise(aPromise)
     , mFetchObserver(aObserver)
+    , mSignal(aSignal)
   {}
 
   void
@@ -287,7 +307,7 @@ public:
       fetch->SetWorkerScript(spec);
     }
 
-    RefPtr<AbortSignal> signal = mResolver->GetAbortSignal();
+    RefPtr<AbortSignal> signal = mResolver->GetAbortSignalForMainThread();
 
     // ...but release it before calling Fetch, because mResolver's callback can
     // be called synchronously and they want the mutex, too.
@@ -329,10 +349,7 @@ FetchRequest(nsIGlobalObject* aGlobal, const RequestOrUSVString& aInput,
 
   RefPtr<InternalRequest> r = request->GetInternalRequest();
 
-  RefPtr<AbortSignal> signal;
-  if (aInit.mSignal.WasPassed()) {
-    signal = &aInit.mSignal.Value();
-  }
+  RefPtr<AbortSignal> signal = request->GetSignal();
   
   if (signal && signal->Aborted()) {
     // An already aborted signal should reject immediately.
@@ -373,7 +390,7 @@ FetchRequest(nsIGlobalObject* aGlobal, const RequestOrUSVString& aInput,
     }
 
     RefPtr<MainThreadFetchResolver> resolver =
-      new MainThreadFetchResolver(p, observer);
+      new MainThreadFetchResolver(p, observer, signal);
     RefPtr<FetchDriver> fetch = new FetchDriver(r, principal, loadGroup);
     fetch->SetDocument(doc);
     resolver->SetDocument(doc);
@@ -416,7 +433,7 @@ MainThreadFetchResolver::OnResponseAvailableInternal(InternalResponse* aResponse
     }
 
     nsCOMPtr<nsIGlobalObject> go = mPromise->GetParentObject();
-    mResponse = new Response(go, aResponse);
+    mResponse = new Response(go, aResponse, mSignal);
     mPromise->MaybeResolve(mResponse);
   } else {
     if (mFetchObserver) {
@@ -479,7 +496,7 @@ public:
       }
 
       RefPtr<nsIGlobalObject> global = aWorkerPrivate->GlobalScope();
-      RefPtr<Response> response = new Response(global, mInternalResponse);
+      RefPtr<Response> response = new Response(global, mInternalResponse, mResolver->GetAbortSignalForTargetThread());
       promise->MaybeResolve(response);
     } else {
       if (mResolver->mFetchObserver) {
@@ -926,6 +943,12 @@ template <class Derived>
 already_AddRefed<Promise>
 FetchBody<Derived>::ConsumeBody(FetchConsumeType aType, ErrorResult& aRv)
 {
+  RefPtr<AbortSignal> signal = DerivedClass()->GetSignal();
+  if (signal && signal->Aborted()) {
+    aRv.Throw(NS_ERROR_DOM_ABORT_ERR);
+    return nullptr;
+  }
+
   if (BodyUsed()) {
     aRv.ThrowTypeError<MSG_FETCH_BODY_CONSUMED_ERROR>();
     return nullptr;
@@ -935,7 +958,7 @@ FetchBody<Derived>::ConsumeBody(FetchConsumeType aType, ErrorResult& aRv)
 
   RefPtr<Promise> promise =
     FetchBodyConsumer<Derived>::Create(DerivedClass()->GetParentObject(),
-                                       this, aType, aRv);
+                                       this, signal, aType, aRv);
   if (NS_WARN_IF(aRv.Failed())) {
     return nullptr;
   }
diff --git a/dom/fetch/Fetch.h b/dom/fetch/Fetch.h
index fc50d3fda..842066152 100644
--- a/dom/fetch/Fetch.h
+++ b/dom/fetch/Fetch.h
@@ -162,6 +162,9 @@ public:
     return mMimeType;
   }
 
+  virtual AbortSignal*
+  GetSignal() const = 0;
+
 protected:
   FetchBody();
 
diff --git a/dom/fetch/FetchConsumer.cpp b/dom/fetch/FetchConsumer.cpp
index e82e5ec51..581f014d9 100644
--- a/dom/fetch/FetchConsumer.cpp
+++ b/dom/fetch/FetchConsumer.cpp
@@ -1,699 +1,714 @@
-/* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
-/* vim: set ts=8 sts=2 et sw=2 tw=80: */
-/* 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 "Fetch.h"
-#include "FetchConsumer.h"
-
-#include "nsIInputStreamPump.h"
-#include "nsProxyRelease.h"
-#include "WorkerPrivate.h"
-#include "WorkerRunnable.h"
-#include "WorkerScope.h"
-#include "Workers.h"
-
-namespace mozilla {
-namespace dom {
-
-using namespace workers;
-
-namespace {
-
-template <class Derived>
-class FetchBodyWorkerHolder final : public workers::WorkerHolder
-{
-  RefPtr<FetchBodyConsumer<Derived>> mConsumer;
-  bool mWasNotified;
-
-public:
-  explicit FetchBodyWorkerHolder(FetchBodyConsumer<Derived>* aConsumer)
-    : mConsumer(aConsumer)
-    , mWasNotified(false)
-  {
-    MOZ_ASSERT(aConsumer);
-  }
-
-  ~FetchBodyWorkerHolder() = default;
-
-  bool Notify(workers::Status aStatus) override
-  {
-    MOZ_ASSERT(aStatus > workers::Running);
-    if (!mWasNotified) {
-      mWasNotified = true;
-      mConsumer->ShutDownMainThreadConsuming();
-    }
-
-    return true;
-  }
-};
-
-template <class Derived>
-class BeginConsumeBodyRunnable final : public Runnable
-{
-  RefPtr<FetchBodyConsumer<Derived>> mFetchBodyConsumer;
-
-public:
-  explicit BeginConsumeBodyRunnable(FetchBodyConsumer<Derived>* aConsumer)
-    : mFetchBodyConsumer(aConsumer)
-  { }
-
-  NS_IMETHOD
-  Run() override
-  {
-    mFetchBodyConsumer->BeginConsumeBodyMainThread();
-    return NS_OK;
-  }
-};
-
-/*
- * Called on successfully reading the complete stream.
- */
-template <class Derived>
-class ContinueConsumeBodyRunnable final : public MainThreadWorkerRunnable
-{
-  RefPtr<FetchBodyConsumer<Derived>> mFetchBodyConsumer;
-  nsresult mStatus;
-  uint32_t mLength;
-  uint8_t* mResult;
-
-public:
-  ContinueConsumeBodyRunnable(FetchBodyConsumer<Derived>* aFetchBodyConsumer,
-                              nsresult aStatus, uint32_t aLength,
-                              uint8_t* aResult)
-    : MainThreadWorkerRunnable(aFetchBodyConsumer->GetWorkerPrivate())
-    , mFetchBodyConsumer(aFetchBodyConsumer)
-    , mStatus(aStatus)
-    , mLength(aLength)
-    , mResult(aResult)
-  {
-    MOZ_ASSERT(NS_IsMainThread());
-  }
-
-  bool
-  WorkerRun(JSContext* aCx, WorkerPrivate* aWorkerPrivate) override
-  {
-    mFetchBodyConsumer->ContinueConsumeBody(mStatus, mLength, mResult);
-    return true;
-  }
-};
-
-template <class Derived>
-class FailConsumeBodyWorkerRunnable : public MainThreadWorkerControlRunnable
-{
-  RefPtr<FetchBodyConsumer<Derived>> mBodyConsumer;
-
-public:
-  explicit FailConsumeBodyWorkerRunnable(FetchBodyConsumer<Derived>* aBodyConsumer)
-    : MainThreadWorkerControlRunnable(aBodyConsumer->GetWorkerPrivate())
-    , mBodyConsumer(aBodyConsumer)
-  {
-    AssertIsOnMainThread();
-  }
-
-  bool
-  WorkerRun(JSContext* aCx, WorkerPrivate* aWorkerPrivate) override
-  {
-    mBodyConsumer->ContinueConsumeBody(NS_ERROR_FAILURE, 0, nullptr);
-    return true;
-  }
-};
-
-/*
- * In case of failure to create a stream pump or dispatch stream completion to
- * worker, ensure we cleanup properly. Thread agnostic.
- */
-template <class Derived>
-class MOZ_STACK_CLASS AutoFailConsumeBody final
-{
-  RefPtr<FetchBodyConsumer<Derived>> mBodyConsumer;
-
-public:
-  explicit AutoFailConsumeBody(FetchBodyConsumer<Derived>* aBodyConsumer)
-    : mBodyConsumer(aBodyConsumer)
-  {}
-
-  ~AutoFailConsumeBody()
-  {
-    AssertIsOnMainThread();
-
-    if (mBodyConsumer) {
-      if (mBodyConsumer->GetWorkerPrivate()) {
-        RefPtr<FailConsumeBodyWorkerRunnable<Derived>> r =
-          new FailConsumeBodyWorkerRunnable<Derived>(mBodyConsumer);
-        if (!r->Dispatch()) {
-          MOZ_CRASH("We are going to leak");
-        }
-      } else {
-        mBodyConsumer->ContinueConsumeBody(NS_ERROR_FAILURE, 0, nullptr);
-      }
-    }
-  }
-
-  void
-  DontFail()
-  {
-    mBodyConsumer = nullptr;
-  }
-};
-
-/*
- * Called on successfully reading the complete stream for Blob.
- */
-template <class Derived>
-class ContinueConsumeBlobBodyRunnable final : public MainThreadWorkerRunnable
-{
-  RefPtr<FetchBodyConsumer<Derived>> mFetchBodyConsumer;
-  RefPtr<BlobImpl> mBlobImpl;
-
-public:
-  ContinueConsumeBlobBodyRunnable(FetchBodyConsumer<Derived>* aFetchBodyConsumer,
-                                  BlobImpl* aBlobImpl)
-    : MainThreadWorkerRunnable(aFetchBodyConsumer->GetWorkerPrivate())
-    , mFetchBodyConsumer(aFetchBodyConsumer)
-    , mBlobImpl(aBlobImpl)
-  {
-    MOZ_ASSERT(NS_IsMainThread());
-    MOZ_ASSERT(mBlobImpl);
-  }
-
-  bool
-  WorkerRun(JSContext* aCx, WorkerPrivate* aWorkerPrivate) override
-  {
-    mFetchBodyConsumer->ContinueConsumeBlobBody(mBlobImpl);
-    return true;
-  }
-};
-
-template <class Derived>
-class ConsumeBodyDoneObserver : public nsIStreamLoaderObserver
-                              , public MutableBlobStorageCallback
-{
-  RefPtr<FetchBodyConsumer<Derived>> mFetchBodyConsumer;
-
-public:
-  NS_DECL_THREADSAFE_ISUPPORTS
-
-  explicit ConsumeBodyDoneObserver(FetchBodyConsumer<Derived>* aFetchBodyConsumer)
-    : mFetchBodyConsumer(aFetchBodyConsumer)
-  { }
-
-  NS_IMETHOD
-  OnStreamComplete(nsIStreamLoader* aLoader,
-                   nsISupports* aCtxt,
-                   nsresult aStatus,
-                   uint32_t aResultLength,
-                   const uint8_t* aResult) override
-  {
-    MOZ_ASSERT(NS_IsMainThread());
-
-    // The loading is completed. Let's nullify the pump before continuing the
-    // consuming of the body.
-    mFetchBodyConsumer->NullifyConsumeBodyPump();
-
-    uint8_t* nonconstResult = const_cast<uint8_t*>(aResult);
-    if (mFetchBodyConsumer->GetWorkerPrivate()) {
-      RefPtr<ContinueConsumeBodyRunnable<Derived>> r =
-        new ContinueConsumeBodyRunnable<Derived>(mFetchBodyConsumer,
-                                                 aStatus,
-                                                 aResultLength,
-                                                 nonconstResult);
-      if (!r->Dispatch()) {
-        NS_WARNING("Could not dispatch ConsumeBodyRunnable");
-        // Return failure so that aResult is freed.
-        return NS_ERROR_FAILURE;
-      }
-    } else {
-      mFetchBodyConsumer->ContinueConsumeBody(aStatus, aResultLength,
-                                              nonconstResult);
-    }
-
-    // FetchBody is responsible for data.
-    return NS_SUCCESS_ADOPTED_DATA;
-  }
-
-  virtual void BlobStoreCompleted(MutableBlobStorage* aBlobStorage,
-                                  Blob* aBlob,
-                                  nsresult aRv) override
-  {
-    // On error.
-    if (NS_FAILED(aRv)) {
-      OnStreamComplete(nullptr, nullptr, aRv, 0, nullptr);
-      return;
-    }
-
-    // The loading is completed. Let's nullify the pump before continuing the
-    // consuming of the body.
-    mFetchBodyConsumer->NullifyConsumeBodyPump();
-
-    MOZ_ASSERT(aBlob);
-
-    if (mFetchBodyConsumer->GetWorkerPrivate()) {
-      RefPtr<ContinueConsumeBlobBodyRunnable<Derived>> r =
-        new ContinueConsumeBlobBodyRunnable<Derived>(mFetchBodyConsumer,
-                                                     aBlob->Impl());
-
-      if (!r->Dispatch()) {
-        NS_WARNING("Could not dispatch ConsumeBlobBodyRunnable");
-        return;
-      }
-    } else {
-      mFetchBodyConsumer->ContinueConsumeBlobBody(aBlob->Impl());
-    }
-  }
-
-private:
-  virtual ~ConsumeBodyDoneObserver()
-  { }
-};
-
-template <class Derived>
-NS_IMPL_ADDREF(ConsumeBodyDoneObserver<Derived>)
-template <class Derived>
-NS_IMPL_RELEASE(ConsumeBodyDoneObserver<Derived>)
-template <class Derived>
-NS_INTERFACE_MAP_BEGIN(ConsumeBodyDoneObserver<Derived>)
-  NS_INTERFACE_MAP_ENTRY(nsIStreamLoaderObserver)
-  NS_INTERFACE_MAP_ENTRY_AMBIGUOUS(nsISupports, nsIStreamLoaderObserver)
-NS_INTERFACE_MAP_END
-
-} // anonymous
-
-template <class Derived>
-/* static */ already_AddRefed<Promise>
-FetchBodyConsumer<Derived>::Create(nsIGlobalObject* aGlobal,
-                                   FetchBody<Derived>* aBody,
-                                   FetchConsumeType aType,
-                                   ErrorResult& aRv)
-{
-  MOZ_ASSERT(aBody);
-
-  nsCOMPtr<nsIInputStream> bodyStream;
-  aBody->DerivedClass()->GetBody(getter_AddRefs(bodyStream));
-  if (!bodyStream) {
-    aRv = NS_NewCStringInputStream(getter_AddRefs(bodyStream), EmptyCString());
-    if (NS_WARN_IF(aRv.Failed())) {
-      return nullptr;
-    }
-  }
-
-  RefPtr<Promise> promise = Promise::Create(aGlobal, aRv);
-  if (aRv.Failed()) {
-    return nullptr;
-  }
-
-  WorkerPrivate* workerPrivate = nullptr;
-  if (!NS_IsMainThread()) {
-    workerPrivate = GetCurrentThreadWorkerPrivate();
-    MOZ_ASSERT(workerPrivate);
-  }
-
-  RefPtr<FetchBodyConsumer<Derived>> consumer =
-    new FetchBodyConsumer<Derived>(aGlobal, workerPrivate, aBody, bodyStream,
-                                   promise, aType);
-
-  if (!NS_IsMainThread()) {
-    MOZ_ASSERT(workerPrivate);
-    if (NS_WARN_IF(!consumer->RegisterWorkerHolder())) {
-      aRv.Throw(NS_ERROR_FAILURE);
-      return nullptr;
-    }
-  } else {
-    nsCOMPtr<nsIObserverService> os = mozilla::services::GetObserverService();
-    if (NS_WARN_IF(!os)) {
-      aRv.Throw(NS_ERROR_FAILURE);
-      return nullptr;
-    }
-
-    aRv = os->AddObserver(consumer, DOM_WINDOW_DESTROYED_TOPIC, true);
-    if (NS_WARN_IF(aRv.Failed())) {
-      return nullptr;
-    }
-
-    aRv = os->AddObserver(consumer, DOM_WINDOW_FROZEN_TOPIC, true);
-    if (NS_WARN_IF(aRv.Failed())) {
-      return nullptr;
-    }
-  }
-
-  nsCOMPtr<nsIRunnable> r = new BeginConsumeBodyRunnable<Derived>(consumer);
-
-  aRv = NS_DispatchToMainThread(r.forget());
-
-  if (NS_WARN_IF(aRv.Failed())) {
-    return nullptr;
-  }
-
-  return promise.forget();
-}
-
-template <class Derived>
-void
-FetchBodyConsumer<Derived>::ReleaseObject()
-{
-  AssertIsOnTargetThread();
-
-  if (NS_IsMainThread()) {
-    nsCOMPtr<nsIObserverService> os = mozilla::services::GetObserverService();
-    if (os) {
-      os->RemoveObserver(this, DOM_WINDOW_DESTROYED_TOPIC);
-      os->RemoveObserver(this, DOM_WINDOW_FROZEN_TOPIC);
-    }
-  }
-
-  mGlobal = nullptr;
-  mWorkerHolder = nullptr;
-
-#ifdef DEBUG
-  mBody = nullptr;
-#endif
-}
-
-template <class Derived>
-FetchBodyConsumer<Derived>::FetchBodyConsumer(nsIGlobalObject* aGlobalObject,
-                                              WorkerPrivate* aWorkerPrivate,
-                                              FetchBody<Derived>* aBody,
-                                              nsIInputStream* aBodyStream,
-                                              Promise* aPromise,
-                                              FetchConsumeType aType)
-  : mTargetThread(NS_GetCurrentThread())
-#ifdef DEBUG
-  , mBody(aBody)
-#endif
-  , mBodyStream(aBodyStream)
-  , mBlobStorageType(MutableBlobStorage::eOnlyInMemory)
-  , mGlobal(aGlobalObject)
-  , mWorkerPrivate(aWorkerPrivate)
-  , mConsumeType(aType)
-  , mConsumePromise(aPromise)
-  , mBodyConsumed(false)
-  , mShuttingDown(false)
-{
-  MOZ_ASSERT(aBody);
-  MOZ_ASSERT(aBodyStream);
-  MOZ_ASSERT(aPromise);
-
-  const mozilla::UniquePtr<mozilla::ipc::PrincipalInfo>& principalInfo =
-    aBody->DerivedClass()->GetPrincipalInfo();
-  // We support temporary file for blobs only if the principal is known and
-  // it's system or content not in private Browsing.
-  if (principalInfo &&
-      (principalInfo->type() == mozilla::ipc::PrincipalInfo::TSystemPrincipalInfo ||
-       (principalInfo->type() == mozilla::ipc::PrincipalInfo::TContentPrincipalInfo &&
-        principalInfo->get_ContentPrincipalInfo().attrs().mPrivateBrowsingId == 0))) {
-    mBlobStorageType = MutableBlobStorage::eCouldBeInTemporaryFile;
-  }
-
-  mBodyMimeType = aBody->MimeType();
-}
-
-template <class Derived>
-FetchBodyConsumer<Derived>::~FetchBodyConsumer()
-{
-}
-
-template <class Derived>
-void
-FetchBodyConsumer<Derived>::AssertIsOnTargetThread() const
-{
-  MOZ_ASSERT(NS_GetCurrentThread() == mTargetThread);
-}
-
-template <class Derived>
-bool
-FetchBodyConsumer<Derived>::RegisterWorkerHolder()
-{
-  MOZ_ASSERT(mWorkerPrivate);
-  mWorkerPrivate->AssertIsOnWorkerThread();
-
-  MOZ_ASSERT(!mWorkerHolder);
-  mWorkerHolder.reset(new FetchBodyWorkerHolder<Derived>(this));
-
-  if (!mWorkerHolder->HoldWorker(mWorkerPrivate, Closing)) {
-    NS_WARNING("Failed to add workerHolder");
-    mWorkerHolder = nullptr;
-    return false;
-  }
-
-  return true;
-}
-
-/*
- * BeginConsumeBodyMainThread() will automatically reject the consume promise
- * and clean up on any failures, so there is no need for callers to do so,
- * reflected in a lack of error return code.
- */
-template <class Derived>
-void
-FetchBodyConsumer<Derived>::BeginConsumeBodyMainThread()
-{
-  AssertIsOnMainThread();
-
-  AutoFailConsumeBody<Derived> autoReject(this);
-
-  if (mShuttingDown) {
-    // We haven't started yet, but we have been terminated. AutoFailConsumeBody
-    // will dispatch a runnable to release resources.
-    return;
-  }
-
-  nsCOMPtr<nsIInputStreamPump> pump;
-  nsresult rv = NS_NewInputStreamPump(getter_AddRefs(pump),
-                                      mBodyStream, -1, -1, 0, 0, false);
-  if (NS_WARN_IF(NS_FAILED(rv))) {
-    return;
-  }
-
-  RefPtr<ConsumeBodyDoneObserver<Derived>> p =
-   new ConsumeBodyDoneObserver<Derived>(this);
-
-  nsCOMPtr<nsIStreamListener> listener;
-  if (mConsumeType == CONSUME_BLOB) {
-    listener = new MutableBlobStreamListener(mBlobStorageType, nullptr,
-                                             mBodyMimeType, p);
-  } else {
-    nsCOMPtr<nsIStreamLoader> loader;
-    rv = NS_NewStreamLoader(getter_AddRefs(loader), p);
-    if (NS_WARN_IF(NS_FAILED(rv))) {
-      return;
-    }
-
-    listener = loader;
-  }
-
-  rv = pump->AsyncRead(listener, nullptr);
-  if (NS_WARN_IF(NS_FAILED(rv))) {
-    return;
-  }
-
-  // Now that everything succeeded, we can assign the pump to a pointer that
-  // stays alive for the lifetime of the FetchConsumer.
-  mConsumeBodyPump = pump;
-
-  // It is ok for retargeting to fail and reads to happen on the main thread.
-  autoReject.DontFail();
-
-  // Try to retarget, otherwise fall back to main thread.
-  nsCOMPtr<nsIThreadRetargetableRequest> rr = do_QueryInterface(pump);
-  if (rr) {
-    nsCOMPtr<nsIEventTarget> sts = do_GetService(NS_STREAMTRANSPORTSERVICE_CONTRACTID);
-    rv = rr->RetargetDeliveryTo(sts);
-    if (NS_WARN_IF(NS_FAILED(rv))) {
-      NS_WARNING("Retargeting failed");
-    }
-  }
-}
-
-template <class Derived>
-void
-FetchBodyConsumer<Derived>::ContinueConsumeBody(nsresult aStatus,
-                                                uint32_t aResultLength,
-                                                uint8_t* aResult)
-{
-  AssertIsOnTargetThread();
-
-  if (mBodyConsumed) {
-    return;
-  }
-  mBodyConsumed = true;
-
-  // Just a precaution to ensure ContinueConsumeBody is not called out of
-  // sync with a body read.
-  MOZ_ASSERT(mBody->BodyUsed());
-
-  auto autoFree = mozilla::MakeScopeExit([&] {
-    free(aResult);
-  });
-
-  MOZ_ASSERT(mConsumePromise);
-  RefPtr<Promise> localPromise = mConsumePromise.forget();
-
-  RefPtr<FetchBodyConsumer<Derived>> self = this;
-  auto autoReleaseObject = mozilla::MakeScopeExit([&] {
-    self->ReleaseObject();
-  });
-
-  if (NS_WARN_IF(NS_FAILED(aStatus))) {
-    localPromise->MaybeReject(NS_ERROR_DOM_ABORT_ERR);
-  }
-
-  // Don't warn here since we warned above.
-  if (NS_FAILED(aStatus)) {
-    return;
-  }
-
-  // Finish successfully consuming body according to type.
-  MOZ_ASSERT(aResult);
-
-  AutoJSAPI jsapi;
-  if (!jsapi.Init(mGlobal)) {
-    localPromise->MaybeReject(NS_ERROR_UNEXPECTED);
-    return;
-  }
-
-  JSContext* cx = jsapi.cx();
-  ErrorResult error;
-
-  switch (mConsumeType) {
-    case CONSUME_ARRAYBUFFER: {
-      JS::Rooted<JSObject*> arrayBuffer(cx);
-      BodyUtil::ConsumeArrayBuffer(cx, &arrayBuffer, aResultLength, aResult,
-                                   error);
-
-      if (!error.Failed()) {
-        JS::Rooted<JS::Value> val(cx);
-        val.setObjectOrNull(arrayBuffer);
-
-        localPromise->MaybeResolve(cx, val);
-        // ArrayBuffer takes over ownership.
-        aResult = nullptr;
-      }
-      break;
-    }
-    case CONSUME_BLOB: {
-      MOZ_CRASH("This should not happen.");
-      break;
-    }
-    case CONSUME_FORMDATA: {
-      nsCString data;
-      data.Adopt(reinterpret_cast<char*>(aResult), aResultLength);
-      aResult = nullptr;
-
-      RefPtr<dom::FormData> fd =
-        BodyUtil::ConsumeFormData(mGlobal, mBodyMimeType, data, error);
-      if (!error.Failed()) {
-        localPromise->MaybeResolve(fd);
-      }
-      break;
-    }
-    case CONSUME_TEXT:
-      // fall through handles early exit.
-    case CONSUME_JSON: {
-      nsString decoded;
-      if (NS_SUCCEEDED(BodyUtil::ConsumeText(aResultLength, aResult, decoded))) {
-        if (mConsumeType == CONSUME_TEXT) {
-          localPromise->MaybeResolve(decoded);
-        } else {
-          JS::Rooted<JS::Value> json(cx);
-          BodyUtil::ConsumeJson(cx, &json, decoded, error);
-          if (!error.Failed()) {
-            localPromise->MaybeResolve(cx, json);
-          }
-        }
-      };
-      break;
-    }
-    default:
-      NS_NOTREACHED("Unexpected consume body type");
-  }
-
-  error.WouldReportJSException();
-  if (error.Failed()) {
-    localPromise->MaybeReject(error);
-  }
-}
-
-template <class Derived>
-void
-FetchBodyConsumer<Derived>::ContinueConsumeBlobBody(BlobImpl* aBlobImpl)
-{
-  AssertIsOnTargetThread();
-  MOZ_ASSERT(mConsumeType == CONSUME_BLOB);
-
-  if (mBodyConsumed) {
-    return;
-  }
-  mBodyConsumed = true;
-
-  // Just a precaution to ensure ContinueConsumeBody is not called out of
-  // sync with a body read.
-  MOZ_ASSERT(mBody->BodyUsed());
-
-  MOZ_ASSERT(mConsumePromise);
-  RefPtr<Promise> localPromise = mConsumePromise.forget();
-
-  RefPtr<dom::Blob> blob = dom::Blob::Create(mGlobal, aBlobImpl);
-  MOZ_ASSERT(blob);
-
-  localPromise->MaybeResolve(blob);
-
-  ReleaseObject();
-}
-
-template <class Derived>
-void
-FetchBodyConsumer<Derived>::ShutDownMainThreadConsuming()
-{
-  if (!NS_IsMainThread()) {
-    RefPtr<FetchBodyConsumer<Derived>> self = this;
-
-    nsCOMPtr<nsIRunnable> r = NS_NewRunnableFunction(
-      [self] () { self->ShutDownMainThreadConsuming(); });
-
-    MOZ_ALWAYS_SUCCEEDS(NS_DispatchToMainThread(r.forget()));
-    return;
-  }
-
-  // We need this because maybe, mConsumeBodyPump has not been created yet. We
-  // must be sure that we don't try to do it.
-  mShuttingDown = true;
-
-  if (mConsumeBodyPump) {
-    mConsumeBodyPump->Cancel(NS_BINDING_ABORTED);
-    mConsumeBodyPump = nullptr;
-  }
-}
-
-template <class Derived>
-NS_IMETHODIMP
-FetchBodyConsumer<Derived>::Observe(nsISupports* aSubject,
-                                    const char* aTopic,
-                                    const char16_t* aData)
-{
-  AssertIsOnMainThread();
-
-  MOZ_ASSERT((strcmp(aTopic, DOM_WINDOW_FROZEN_TOPIC) == 0) ||
-             (strcmp(aTopic, DOM_WINDOW_DESTROYED_TOPIC) == 0));
-
-  nsCOMPtr<nsPIDOMWindowInner> window = do_QueryInterface(mGlobal);
-  if (SameCOMIdentity(aSubject, window)) {
-    ContinueConsumeBody(NS_BINDING_ABORTED, 0, nullptr);
-  }
-
-  return NS_OK;
-}
-
-template <class Derived>
-NS_IMPL_ADDREF(FetchBodyConsumer<Derived>)
-
-template <class Derived>
-NS_IMPL_RELEASE(FetchBodyConsumer<Derived>)
-
-template <class Derived>
-NS_IMPL_QUERY_INTERFACE(FetchBodyConsumer<Derived>,
-                        nsIObserver,
-                        nsISupportsWeakReference)
-
-} // namespace dom
-} // namespace mozilla
+/* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */

+/* vim: set ts=8 sts=2 et sw=2 tw=80: */

+/* 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 "Fetch.h"

+#include "FetchConsumer.h"

+

+#include "nsIInputStreamPump.h"

+#include "nsProxyRelease.h"

+#include "WorkerPrivate.h"

+#include "WorkerRunnable.h"

+#include "WorkerScope.h"

+#include "Workers.h"

+

+namespace mozilla {

+namespace dom {

+

+using namespace workers;

+

+namespace {

+

+template <class Derived>

+class FetchBodyWorkerHolder final : public workers::WorkerHolder

+{

+  RefPtr<FetchBodyConsumer<Derived>> mConsumer;

+  bool mWasNotified;

+

+public:

+  explicit FetchBodyWorkerHolder(FetchBodyConsumer<Derived>* aConsumer)

+    : mConsumer(aConsumer)

+    , mWasNotified(false)

+  {

+    MOZ_ASSERT(aConsumer);

+  }

+

+  ~FetchBodyWorkerHolder() = default;

+

+  bool Notify(workers::Status aStatus) override

+  {

+    MOZ_ASSERT(aStatus > workers::Running);

+    if (!mWasNotified) {

+      mWasNotified = true;

+      mConsumer->ShutDownMainThreadConsuming();

+    }

+

+    return true;

+  }

+};

+

+template <class Derived>

+class BeginConsumeBodyRunnable final : public Runnable

+{

+  RefPtr<FetchBodyConsumer<Derived>> mFetchBodyConsumer;

+

+public:

+  explicit BeginConsumeBodyRunnable(FetchBodyConsumer<Derived>* aConsumer)

+    : mFetchBodyConsumer(aConsumer)

+  { }

+

+  NS_IMETHOD

+  Run() override

+  {

+    mFetchBodyConsumer->BeginConsumeBodyMainThread();

+    return NS_OK;

+  }

+};

+

+/*

+ * Called on successfully reading the complete stream.

+ */

+template <class Derived>

+class ContinueConsumeBodyRunnable final : public MainThreadWorkerRunnable

+{

+  RefPtr<FetchBodyConsumer<Derived>> mFetchBodyConsumer;

+  nsresult mStatus;

+  uint32_t mLength;

+  uint8_t* mResult;

+

+public:

+  ContinueConsumeBodyRunnable(FetchBodyConsumer<Derived>* aFetchBodyConsumer,

+                              nsresult aStatus, uint32_t aLength,

+                              uint8_t* aResult)

+    : MainThreadWorkerRunnable(aFetchBodyConsumer->GetWorkerPrivate())

+    , mFetchBodyConsumer(aFetchBodyConsumer)

+    , mStatus(aStatus)

+    , mLength(aLength)

+    , mResult(aResult)

+  {

+    MOZ_ASSERT(NS_IsMainThread());

+  }

+

+  bool

+  WorkerRun(JSContext* aCx, WorkerPrivate* aWorkerPrivate) override

+  {

+    mFetchBodyConsumer->ContinueConsumeBody(mStatus, mLength, mResult);

+    return true;

+  }

+};

+

+template <class Derived>

+class FailConsumeBodyWorkerRunnable : public MainThreadWorkerControlRunnable

+{

+  RefPtr<FetchBodyConsumer<Derived>> mBodyConsumer;

+

+public:

+  explicit FailConsumeBodyWorkerRunnable(FetchBodyConsumer<Derived>* aBodyConsumer)

+    : MainThreadWorkerControlRunnable(aBodyConsumer->GetWorkerPrivate())

+    , mBodyConsumer(aBodyConsumer)

+  {

+    AssertIsOnMainThread();

+  }

+

+  bool

+  WorkerRun(JSContext* aCx, WorkerPrivate* aWorkerPrivate) override

+  {

+    mBodyConsumer->ContinueConsumeBody(NS_ERROR_FAILURE, 0, nullptr);

+    return true;

+  }

+};

+

+/*

+ * In case of failure to create a stream pump or dispatch stream completion to

+ * worker, ensure we cleanup properly. Thread agnostic.

+ */

+template <class Derived>

+class MOZ_STACK_CLASS AutoFailConsumeBody final

+{

+  RefPtr<FetchBodyConsumer<Derived>> mBodyConsumer;

+

+public:

+  explicit AutoFailConsumeBody(FetchBodyConsumer<Derived>* aBodyConsumer)

+    : mBodyConsumer(aBodyConsumer)

+  {}

+

+  ~AutoFailConsumeBody()

+  {

+    AssertIsOnMainThread();

+

+    if (mBodyConsumer) {

+      if (mBodyConsumer->GetWorkerPrivate()) {

+        RefPtr<FailConsumeBodyWorkerRunnable<Derived>> r =

+          new FailConsumeBodyWorkerRunnable<Derived>(mBodyConsumer);

+        if (!r->Dispatch()) {

+          MOZ_CRASH("We are going to leak");

+        }

+      } else {

+        mBodyConsumer->ContinueConsumeBody(NS_ERROR_FAILURE, 0, nullptr);

+      }

+    }

+  }

+

+  void

+  DontFail()

+  {

+    mBodyConsumer = nullptr;

+  }

+};

+

+/*

+ * Called on successfully reading the complete stream for Blob.

+ */

+template <class Derived>

+class ContinueConsumeBlobBodyRunnable final : public MainThreadWorkerRunnable

+{

+  RefPtr<FetchBodyConsumer<Derived>> mFetchBodyConsumer;

+  RefPtr<BlobImpl> mBlobImpl;

+

+public:

+  ContinueConsumeBlobBodyRunnable(FetchBodyConsumer<Derived>* aFetchBodyConsumer,

+                                  BlobImpl* aBlobImpl)

+    : MainThreadWorkerRunnable(aFetchBodyConsumer->GetWorkerPrivate())

+    , mFetchBodyConsumer(aFetchBodyConsumer)

+    , mBlobImpl(aBlobImpl)

+  {

+    MOZ_ASSERT(NS_IsMainThread());

+    MOZ_ASSERT(mBlobImpl);

+  }

+

+  bool

+  WorkerRun(JSContext* aCx, WorkerPrivate* aWorkerPrivate) override

+  {

+    mFetchBodyConsumer->ContinueConsumeBlobBody(mBlobImpl);

+    return true;

+  }

+};

+

+template <class Derived>

+class ConsumeBodyDoneObserver : public nsIStreamLoaderObserver

+                              , public MutableBlobStorageCallback

+{

+  RefPtr<FetchBodyConsumer<Derived>> mFetchBodyConsumer;

+

+public:

+  NS_DECL_THREADSAFE_ISUPPORTS

+

+  explicit ConsumeBodyDoneObserver(FetchBodyConsumer<Derived>* aFetchBodyConsumer)

+    : mFetchBodyConsumer(aFetchBodyConsumer)

+  { }

+

+  NS_IMETHOD

+  OnStreamComplete(nsIStreamLoader* aLoader,

+                   nsISupports* aCtxt,

+                   nsresult aStatus,

+                   uint32_t aResultLength,

+                   const uint8_t* aResult) override

+  {

+    MOZ_ASSERT(NS_IsMainThread());

+

+    // The loading is completed. Let's nullify the pump before continuing the

+    // consuming of the body.

+    mFetchBodyConsumer->NullifyConsumeBodyPump();

+

+    uint8_t* nonconstResult = const_cast<uint8_t*>(aResult);

+    if (mFetchBodyConsumer->GetWorkerPrivate()) {

+      RefPtr<ContinueConsumeBodyRunnable<Derived>> r =

+        new ContinueConsumeBodyRunnable<Derived>(mFetchBodyConsumer,

+                                                 aStatus,

+                                                 aResultLength,

+                                                 nonconstResult);

+      if (!r->Dispatch()) {

+        NS_WARNING("Could not dispatch ConsumeBodyRunnable");

+        // Return failure so that aResult is freed.

+        return NS_ERROR_FAILURE;

+      }

+    } else {

+      mFetchBodyConsumer->ContinueConsumeBody(aStatus, aResultLength,

+                                              nonconstResult);

+    }

+

+    // FetchBody is responsible for data.

+    return NS_SUCCESS_ADOPTED_DATA;

+  }

+

+  virtual void BlobStoreCompleted(MutableBlobStorage* aBlobStorage,

+                                  Blob* aBlob,

+                                  nsresult aRv) override

+  {

+    // On error.

+    if (NS_FAILED(aRv)) {

+      OnStreamComplete(nullptr, nullptr, aRv, 0, nullptr);

+      return;

+    }

+

+    // The loading is completed. Let's nullify the pump before continuing the

+    // consuming of the body.

+    mFetchBodyConsumer->NullifyConsumeBodyPump();

+

+    MOZ_ASSERT(aBlob);

+

+    if (mFetchBodyConsumer->GetWorkerPrivate()) {

+      RefPtr<ContinueConsumeBlobBodyRunnable<Derived>> r =

+        new ContinueConsumeBlobBodyRunnable<Derived>(mFetchBodyConsumer,

+                                                     aBlob->Impl());

+

+      if (!r->Dispatch()) {

+        NS_WARNING("Could not dispatch ConsumeBlobBodyRunnable");

+        return;

+      }

+    } else {

+      mFetchBodyConsumer->ContinueConsumeBlobBody(aBlob->Impl());

+    }

+  }

+

+private:

+  virtual ~ConsumeBodyDoneObserver()

+  { }

+};

+

+template <class Derived>

+NS_IMPL_ADDREF(ConsumeBodyDoneObserver<Derived>)

+template <class Derived>

+NS_IMPL_RELEASE(ConsumeBodyDoneObserver<Derived>)

+template <class Derived>

+NS_INTERFACE_MAP_BEGIN(ConsumeBodyDoneObserver<Derived>)

+  NS_INTERFACE_MAP_ENTRY(nsIStreamLoaderObserver)

+  NS_INTERFACE_MAP_ENTRY_AMBIGUOUS(nsISupports, nsIStreamLoaderObserver)

+NS_INTERFACE_MAP_END

+

+} // anonymous

+

+template <class Derived>

+/* static */ already_AddRefed<Promise>

+FetchBodyConsumer<Derived>::Create(nsIGlobalObject* aGlobal,

+                                   FetchBody<Derived>* aBody,

+                                   AbortSignal* aSignal,

+                                   FetchConsumeType aType,

+                                   ErrorResult& aRv)

+{

+  MOZ_ASSERT(aBody);

+

+  nsCOMPtr<nsIInputStream> bodyStream;

+  aBody->DerivedClass()->GetBody(getter_AddRefs(bodyStream));

+  if (!bodyStream) {

+    aRv = NS_NewCStringInputStream(getter_AddRefs(bodyStream), EmptyCString());

+    if (NS_WARN_IF(aRv.Failed())) {

+      return nullptr;

+    }

+  }

+

+  RefPtr<Promise> promise = Promise::Create(aGlobal, aRv);

+  if (aRv.Failed()) {

+    return nullptr;

+  }

+

+  WorkerPrivate* workerPrivate = nullptr;

+  if (!NS_IsMainThread()) {

+    workerPrivate = GetCurrentThreadWorkerPrivate();

+    MOZ_ASSERT(workerPrivate);

+  }

+

+  RefPtr<FetchBodyConsumer<Derived>> consumer =

+    new FetchBodyConsumer<Derived>(aGlobal, workerPrivate, aBody, bodyStream,

+                                   promise, aType);

+

+  if (!NS_IsMainThread()) {

+    MOZ_ASSERT(workerPrivate);

+    if (NS_WARN_IF(!consumer->RegisterWorkerHolder())) {

+      aRv.Throw(NS_ERROR_FAILURE);

+      return nullptr;

+    }

+  } else {

+    nsCOMPtr<nsIObserverService> os = mozilla::services::GetObserverService();

+    if (NS_WARN_IF(!os)) {

+      aRv.Throw(NS_ERROR_FAILURE);

+      return nullptr;

+    }

+

+    aRv = os->AddObserver(consumer, DOM_WINDOW_DESTROYED_TOPIC, true);

+    if (NS_WARN_IF(aRv.Failed())) {

+      return nullptr;

+    }

+

+    aRv = os->AddObserver(consumer, DOM_WINDOW_FROZEN_TOPIC, true);

+    if (NS_WARN_IF(aRv.Failed())) {

+      return nullptr;

+    }

+  }

+

+  nsCOMPtr<nsIRunnable> r = new BeginConsumeBodyRunnable<Derived>(consumer);

+

+  aRv = NS_DispatchToMainThread(r.forget());

+

+  if (NS_WARN_IF(aRv.Failed())) {

+    return nullptr;

+  }

+

+  if (aSignal) {

+    consumer->Follow(aSignal);

+  }

+

+  return promise.forget();

+}

+

+template <class Derived>

+void

+FetchBodyConsumer<Derived>::ReleaseObject()

+{

+  AssertIsOnTargetThread();

+

+  if (NS_IsMainThread()) {

+    nsCOMPtr<nsIObserverService> os = mozilla::services::GetObserverService();

+    if (os) {

+      os->RemoveObserver(this, DOM_WINDOW_DESTROYED_TOPIC);

+      os->RemoveObserver(this, DOM_WINDOW_FROZEN_TOPIC);

+    }

+  }

+

+  mGlobal = nullptr;

+  mWorkerHolder = nullptr;

+

+#ifdef DEBUG

+  mBody = nullptr;

+#endif

+

+  Unfollow();

+}

+

+template <class Derived>

+FetchBodyConsumer<Derived>::FetchBodyConsumer(nsIGlobalObject* aGlobalObject,

+                                              WorkerPrivate* aWorkerPrivate,

+                                              FetchBody<Derived>* aBody,

+                                              nsIInputStream* aBodyStream,

+                                              Promise* aPromise,

+                                              FetchConsumeType aType)

+  : mTargetThread(NS_GetCurrentThread())

+#ifdef DEBUG

+  , mBody(aBody)

+#endif

+  , mBodyStream(aBodyStream)

+  , mBlobStorageType(MutableBlobStorage::eOnlyInMemory)

+  , mGlobal(aGlobalObject)

+  , mWorkerPrivate(aWorkerPrivate)

+  , mConsumeType(aType)

+  , mConsumePromise(aPromise)

+  , mBodyConsumed(false)

+  , mShuttingDown(false)

+{

+  MOZ_ASSERT(aBody);

+  MOZ_ASSERT(aBodyStream);

+  MOZ_ASSERT(aPromise);

+

+  const mozilla::UniquePtr<mozilla::ipc::PrincipalInfo>& principalInfo =

+    aBody->DerivedClass()->GetPrincipalInfo();

+  // We support temporary file for blobs only if the principal is known and

+  // it's system or content not in private Browsing.

+  if (principalInfo &&

+      (principalInfo->type() == mozilla::ipc::PrincipalInfo::TSystemPrincipalInfo ||

+       (principalInfo->type() == mozilla::ipc::PrincipalInfo::TContentPrincipalInfo &&

+        principalInfo->get_ContentPrincipalInfo().attrs().mPrivateBrowsingId == 0))) {

+    mBlobStorageType = MutableBlobStorage::eCouldBeInTemporaryFile;

+  }

+

+  mBodyMimeType = aBody->MimeType();

+}

+

+template <class Derived>

+FetchBodyConsumer<Derived>::~FetchBodyConsumer()

+{

+}

+

+template <class Derived>

+void

+FetchBodyConsumer<Derived>::AssertIsOnTargetThread() const

+{

+  MOZ_ASSERT(NS_GetCurrentThread() == mTargetThread);

+}

+

+template <class Derived>

+bool

+FetchBodyConsumer<Derived>::RegisterWorkerHolder()

+{

+  MOZ_ASSERT(mWorkerPrivate);

+  mWorkerPrivate->AssertIsOnWorkerThread();

+

+  MOZ_ASSERT(!mWorkerHolder);

+  mWorkerHolder.reset(new FetchBodyWorkerHolder<Derived>(this));

+

+  if (!mWorkerHolder->HoldWorker(mWorkerPrivate, Closing)) {

+    NS_WARNING("Failed to add workerHolder");

+    mWorkerHolder = nullptr;

+    return false;

+  }

+

+  return true;

+}

+

+/*

+ * BeginConsumeBodyMainThread() will automatically reject the consume promise

+ * and clean up on any failures, so there is no need for callers to do so,

+ * reflected in a lack of error return code.

+ */

+template <class Derived>

+void

+FetchBodyConsumer<Derived>::BeginConsumeBodyMainThread()

+{

+  AssertIsOnMainThread();

+

+  AutoFailConsumeBody<Derived> autoReject(this);

+

+  if (mShuttingDown) {

+    // We haven't started yet, but we have been terminated. AutoFailConsumeBody

+    // will dispatch a runnable to release resources.

+    return;

+  }

+

+  nsCOMPtr<nsIInputStreamPump> pump;

+  nsresult rv = NS_NewInputStreamPump(getter_AddRefs(pump),

+                                      mBodyStream, -1, -1, 0, 0, false);

+  if (NS_WARN_IF(NS_FAILED(rv))) {

+    return;

+  }

+

+  RefPtr<ConsumeBodyDoneObserver<Derived>> p =

+   new ConsumeBodyDoneObserver<Derived>(this);

+

+  nsCOMPtr<nsIStreamListener> listener;

+  if (mConsumeType == CONSUME_BLOB) {

+    listener = new MutableBlobStreamListener(mBlobStorageType, nullptr,

+                                             mBodyMimeType, p);

+  } else {

+    nsCOMPtr<nsIStreamLoader> loader;

+    rv = NS_NewStreamLoader(getter_AddRefs(loader), p);

+    if (NS_WARN_IF(NS_FAILED(rv))) {

+      return;

+    }

+

+    listener = loader;

+  }

+

+  rv = pump->AsyncRead(listener, nullptr);

+  if (NS_WARN_IF(NS_FAILED(rv))) {

+    return;

+  }

+

+  // Now that everything succeeded, we can assign the pump to a pointer that

+  // stays alive for the lifetime of the FetchConsumer.

+  mConsumeBodyPump = pump;

+

+  // It is ok for retargeting to fail and reads to happen on the main thread.

+  autoReject.DontFail();

+

+  // Try to retarget, otherwise fall back to main thread.

+  nsCOMPtr<nsIThreadRetargetableRequest> rr = do_QueryInterface(pump);

+  if (rr) {

+    nsCOMPtr<nsIEventTarget> sts = do_GetService(NS_STREAMTRANSPORTSERVICE_CONTRACTID);

+    rv = rr->RetargetDeliveryTo(sts);

+    if (NS_WARN_IF(NS_FAILED(rv))) {

+      NS_WARNING("Retargeting failed");

+    }

+  }

+}

+

+template <class Derived>

+void

+FetchBodyConsumer<Derived>::ContinueConsumeBody(nsresult aStatus,

+                                                uint32_t aResultLength,

+                                                uint8_t* aResult)

+{

+  AssertIsOnTargetThread();

+

+  if (mBodyConsumed) {

+    return;

+  }

+  mBodyConsumed = true;

+

+  // Just a precaution to ensure ContinueConsumeBody is not called out of

+  // sync with a body read.

+  MOZ_ASSERT(mBody->BodyUsed());

+

+  auto autoFree = mozilla::MakeScopeExit([&] {

+    free(aResult);

+  });

+

+  MOZ_ASSERT(mConsumePromise);

+  RefPtr<Promise> localPromise = mConsumePromise.forget();

+

+  RefPtr<FetchBodyConsumer<Derived>> self = this;

+  auto autoReleaseObject = mozilla::MakeScopeExit([&] {

+    self->ReleaseObject();

+  });

+

+  if (NS_WARN_IF(NS_FAILED(aStatus))) {

+    localPromise->MaybeReject(NS_ERROR_DOM_ABORT_ERR);

+  }

+

+  // Don't warn here since we warned above.

+  if (NS_FAILED(aStatus)) {

+    return;

+  }

+

+  // Finish successfully consuming body according to type.

+  MOZ_ASSERT(aResult);

+

+  AutoJSAPI jsapi;

+  if (!jsapi.Init(mGlobal)) {

+    localPromise->MaybeReject(NS_ERROR_UNEXPECTED);

+    return;

+  }

+

+  JSContext* cx = jsapi.cx();

+  ErrorResult error;

+

+  switch (mConsumeType) {

+    case CONSUME_ARRAYBUFFER: {

+      JS::Rooted<JSObject*> arrayBuffer(cx);

+      BodyUtil::ConsumeArrayBuffer(cx, &arrayBuffer, aResultLength, aResult,

+                                   error);

+

+      if (!error.Failed()) {

+        JS::Rooted<JS::Value> val(cx);

+        val.setObjectOrNull(arrayBuffer);

+

+        localPromise->MaybeResolve(cx, val);

+        // ArrayBuffer takes over ownership.

+        aResult = nullptr;

+      }

+      break;

+    }

+    case CONSUME_BLOB: {

+      MOZ_CRASH("This should not happen.");

+      break;

+    }

+    case CONSUME_FORMDATA: {

+      nsCString data;

+      data.Adopt(reinterpret_cast<char*>(aResult), aResultLength);

+      aResult = nullptr;

+

+      RefPtr<dom::FormData> fd =

+        BodyUtil::ConsumeFormData(mGlobal, mBodyMimeType, data, error);

+      if (!error.Failed()) {

+        localPromise->MaybeResolve(fd);

+      }

+      break;

+    }

+    case CONSUME_TEXT:

+      // fall through handles early exit.

+    case CONSUME_JSON: {

+      nsString decoded;

+      if (NS_SUCCEEDED(BodyUtil::ConsumeText(aResultLength, aResult, decoded))) {

+        if (mConsumeType == CONSUME_TEXT) {

+          localPromise->MaybeResolve(decoded);

+        } else {

+          JS::Rooted<JS::Value> json(cx);

+          BodyUtil::ConsumeJson(cx, &json, decoded, error);

+          if (!error.Failed()) {

+            localPromise->MaybeResolve(cx, json);

+          }

+        }

+      };

+      break;

+    }

+    default:

+      NS_NOTREACHED("Unexpected consume body type");

+  }

+

+  error.WouldReportJSException();

+  if (error.Failed()) {

+    localPromise->MaybeReject(error);

+  }

+}

+

+template <class Derived>

+void

+FetchBodyConsumer<Derived>::ContinueConsumeBlobBody(BlobImpl* aBlobImpl)

+{

+  AssertIsOnTargetThread();

+  MOZ_ASSERT(mConsumeType == CONSUME_BLOB);

+

+  if (mBodyConsumed) {

+    return;

+  }

+  mBodyConsumed = true;

+

+  // Just a precaution to ensure ContinueConsumeBody is not called out of

+  // sync with a body read.

+  MOZ_ASSERT(mBody->BodyUsed());

+

+  MOZ_ASSERT(mConsumePromise);

+  RefPtr<Promise> localPromise = mConsumePromise.forget();

+

+  RefPtr<dom::Blob> blob = dom::Blob::Create(mGlobal, aBlobImpl);

+  MOZ_ASSERT(blob);

+

+  localPromise->MaybeResolve(blob);

+

+  ReleaseObject();

+}

+

+template <class Derived>

+void

+FetchBodyConsumer<Derived>::ShutDownMainThreadConsuming()

+{

+  if (!NS_IsMainThread()) {

+    RefPtr<FetchBodyConsumer<Derived>> self = this;

+

+    nsCOMPtr<nsIRunnable> r = NS_NewRunnableFunction(

+      [self] () { self->ShutDownMainThreadConsuming(); });

+

+    MOZ_ALWAYS_SUCCEEDS(NS_DispatchToMainThread(r.forget()));

+    return;

+  }

+

+  // We need this because maybe, mConsumeBodyPump has not been created yet. We

+  // must be sure that we don't try to do it.

+  mShuttingDown = true;

+

+  if (mConsumeBodyPump) {

+    mConsumeBodyPump->Cancel(NS_BINDING_ABORTED);

+    mConsumeBodyPump = nullptr;

+  }

+}

+

+template <class Derived>

+NS_IMETHODIMP

+FetchBodyConsumer<Derived>::Observe(nsISupports* aSubject,

+                                    const char* aTopic,

+                                    const char16_t* aData)

+{

+  AssertIsOnMainThread();

+

+  MOZ_ASSERT((strcmp(aTopic, DOM_WINDOW_FROZEN_TOPIC) == 0) ||

+             (strcmp(aTopic, DOM_WINDOW_DESTROYED_TOPIC) == 0));

+

+  nsCOMPtr<nsPIDOMWindowInner> window = do_QueryInterface(mGlobal);

+  if (SameCOMIdentity(aSubject, window)) {

+    ContinueConsumeBody(NS_BINDING_ABORTED, 0, nullptr);

+  }

+

+  return NS_OK;

+}

+

+template <class Derived> 

+void

+FetchBodyConsumer<Derived>::Aborted()

+{

+  AssertIsOnTargetThread();

+  ContinueConsumeBody(NS_ERROR_DOM_ABORT_ERR, 0, nullptr);

+}

+

+template <class Derived>

+NS_IMPL_ADDREF(FetchBodyConsumer<Derived>)

+

+template <class Derived>

+NS_IMPL_RELEASE(FetchBodyConsumer<Derived>)

+

+template <class Derived>

+NS_IMPL_QUERY_INTERFACE(FetchBodyConsumer<Derived>,

+                        nsIObserver,

+                        nsISupportsWeakReference)

+

+} // namespace dom

+} // namespace mozilla

diff --git a/dom/fetch/FetchConsumer.h b/dom/fetch/FetchConsumer.h
index 2b5725342..77af09d9b 100644
--- a/dom/fetch/FetchConsumer.h
+++ b/dom/fetch/FetchConsumer.h
@@ -11,6 +11,7 @@
 #include "nsIInputStream.h"
 #include "nsIObserver.h"
 #include "nsWeakReference.h"
+#include "mozilla/dom/AbortSignal.h"
 #include "mozilla/dom/MutableBlobStorage.h"
 
 class nsIThread;
@@ -34,6 +35,7 @@ template <class Derived> class FetchBody;
 template <class Derived>
 class FetchBodyConsumer final : public nsIObserver
                               , public nsSupportsWeakReference
+                              , public AbortSignal::Follower
 {
 public:
   NS_DECL_THREADSAFE_ISUPPORTS
@@ -42,6 +44,7 @@ public:
   static already_AddRefed<Promise>
   Create(nsIGlobalObject* aGlobal,
          FetchBody<Derived>* aBody,
+         AbortSignal* aSignal,
          FetchConsumeType aType,
          ErrorResult& aRv);
 
@@ -73,6 +76,9 @@ public:
     mConsumeBodyPump = nullptr;
   }
 
+  // Override AbortSignal::Follower::Aborted
+  void Aborted() override;
+
 private:
   FetchBodyConsumer(nsIGlobalObject* aGlobalObject,
                     workers::WorkerPrivate* aWorkerPrivate,
diff --git a/dom/fetch/Request.cpp b/dom/fetch/Request.cpp
index 6a7885b1a..ba268d331 100644
--- a/dom/fetch/Request.cpp
+++ b/dom/fetch/Request.cpp
@@ -37,15 +37,18 @@ NS_INTERFACE_MAP_BEGIN_CYCLE_COLLECTION(Request)
   NS_INTERFACE_MAP_ENTRY(nsISupports)
 NS_INTERFACE_MAP_END
 
-Request::Request(nsIGlobalObject* aOwner, InternalRequest* aRequest)
+Request::Request(nsIGlobalObject* aOwner, InternalRequest* aRequest, AbortSignal* aSignal)
   : FetchBody<Request>()
   , mOwner(aOwner)
   , mRequest(aRequest)
+  , mSignal(aSignal)
 {
   MOZ_ASSERT(aRequest->Headers()->Guard() == HeadersGuardEnum::Immutable ||
              aRequest->Headers()->Guard() == HeadersGuardEnum::Request ||
              aRequest->Headers()->Guard() == HeadersGuardEnum::Request_no_cors);
   SetMimeType();
+  
+  // aSignal can be null.
 }
 
 Request::~Request()
@@ -286,6 +289,8 @@ Request::Constructor(const GlobalObject& aGlobal,
   RefPtr<InternalRequest> request;
 
   nsCOMPtr<nsIGlobalObject> global = do_QueryInterface(aGlobal.GetAsSupports());
+  
+  RefPtr<AbortSignal> signal;
 
   if (aInput.IsRequest()) {
     RefPtr<Request> inputReq = &aInput.GetAsRequest();
@@ -300,6 +305,7 @@ Request::Constructor(const GlobalObject& aGlobal,
     }
 
     request = inputReq->GetInternalRequest();
+    signal = inputReq->GetOrCreateSignal();
   } else {
     // aInput is USVString.
     // We need to get url before we create a InternalRequest.
@@ -418,6 +424,10 @@ Request::Constructor(const GlobalObject& aGlobal,
     request->SetReferrerPolicy(aInit.mReferrerPolicy.Value());
   }
 
+  if (aInit.mSignal.WasPassed()) {
+    signal = &aInit.mSignal.Value();
+  }
+
   if (NS_IsMainThread()) {
     nsCOMPtr<nsPIDOMWindowInner> window = do_QueryInterface(global);
     if (window) {
@@ -579,7 +589,7 @@ Request::Constructor(const GlobalObject& aGlobal,
     }
   }
 
-  RefPtr<Request> domRequest = new Request(global, request);
+  RefPtr<Request> domRequest = new Request(global, request, signal);
   domRequest->SetMimeType();
 
   if (aInput.IsRequest()) {
@@ -595,7 +605,7 @@ Request::Constructor(const GlobalObject& aGlobal,
 }
 
 already_AddRefed<Request>
-Request::Clone(ErrorResult& aRv) const
+Request::Clone(ErrorResult& aRv)
 {
   if (BodyUsed()) {
     aRv.ThrowTypeError<MSG_FETCH_BODY_CONSUMED_ERROR>();
@@ -608,7 +618,7 @@ Request::Clone(ErrorResult& aRv) const
     return nullptr;
   }
 
-  RefPtr<Request> request = new Request(mOwner, ir);
+  RefPtr<Request> request = new Request(mOwner, ir, GetOrCreateSignal());
   return request.forget();
 }
 
@@ -622,5 +632,21 @@ Request::Headers_()
   return mHeaders;
 }
 
+AbortSignal*
+Request::GetOrCreateSignal()
+{
+  if (!mSignal) {
+    mSignal = new AbortSignal(false);
+  }
+
+  return mSignal;
+}
+
+AbortSignal*
+Request::GetSignal() const
+{
+  return mSignal;
+}
+
 } // namespace dom
 } // namespace mozilla
diff --git a/dom/fetch/Request.h b/dom/fetch/Request.h
index f6fe9be7b..34cbc52cf 100644
--- a/dom/fetch/Request.h
+++ b/dom/fetch/Request.h
@@ -33,7 +33,7 @@ class Request final : public nsISupports
   NS_DECL_CYCLE_COLLECTION_SCRIPT_HOLDER_CLASS(Request)
 
 public:
-  Request(nsIGlobalObject* aOwner, InternalRequest* aRequest);
+  Request(nsIGlobalObject* aOwner, InternalRequest* aRequest, AbortSignal* aSignal);
 
   static bool
   RequestContextEnabled(JSContext* aCx, JSObject* aObj);
@@ -142,7 +142,7 @@ public:
   }
 
   already_AddRefed<Request>
-  Clone(ErrorResult& aRv) const;
+  Clone(ErrorResult& aRv);
 
   already_AddRefed<InternalRequest>
   GetInternalRequest();
@@ -153,13 +153,22 @@ public:
     return mRequest->GetPrincipalInfo();
   }
 
+  AbortSignal*

+  GetOrCreateSignal();

+

+  // This can return a null AbortSignal.

+  AbortSignal*

+  GetSignal() const override;

+
 private:
   ~Request();
 
   nsCOMPtr<nsIGlobalObject> mOwner;
   RefPtr<InternalRequest> mRequest;
+  
   // Lazily created.
   RefPtr<Headers> mHeaders;
+  RefPtr<AbortSignal> mSignal;
 };
 
 } // namespace dom
diff --git a/dom/fetch/Response.cpp b/dom/fetch/Response.cpp
index e35de0e12..241614286 100644
--- a/dom/fetch/Response.cpp
+++ b/dom/fetch/Response.cpp
@@ -34,10 +34,11 @@ NS_INTERFACE_MAP_BEGIN_CYCLE_COLLECTION(Response)
   NS_INTERFACE_MAP_ENTRY(nsISupports)
 NS_INTERFACE_MAP_END
 
-Response::Response(nsIGlobalObject* aGlobal, InternalResponse* aInternalResponse)
+Response::Response(nsIGlobalObject* aGlobal, InternalResponse* aInternalResponse, AbortSignal* aSignal)
   : FetchBody<Response>()
   , mOwner(aGlobal)
   , mInternalResponse(aInternalResponse)
+  , mSignal(aSignal)
 {
   MOZ_ASSERT(aInternalResponse->Headers()->Guard() == HeadersGuardEnum::Immutable ||
              aInternalResponse->Headers()->Guard() == HeadersGuardEnum::Response);
@@ -53,7 +54,7 @@ Response::Error(const GlobalObject& aGlobal)
 {
   nsCOMPtr<nsIGlobalObject> global = do_QueryInterface(aGlobal.GetAsSupports());
   RefPtr<InternalResponse> error = InternalResponse::NetworkError();
-  RefPtr<Response> r = new Response(global, error);
+  RefPtr<Response> r = new Response(global, error, nullptr);
   return r.forget();
 }
 
@@ -173,7 +174,7 @@ Response::Constructor(const GlobalObject& aGlobal,
     internalResponse->InitChannelInfo(worker->GetChannelInfo());
   }
 
-  RefPtr<Response> r = new Response(global, internalResponse);
+  RefPtr<Response> r = new Response(global, internalResponse, nullptr);
 
   if (aInit.mHeaders.WasPassed()) {
     internalResponse->Headers()->Clear();
@@ -236,7 +237,7 @@ Response::Clone(ErrorResult& aRv) const
   }
 
   RefPtr<InternalResponse> ir = mInternalResponse->Clone();
-  RefPtr<Response> response = new Response(mOwner, ir);
+  RefPtr<Response> response = new Response(mOwner, ir, mSignal);
   return response.forget();
 }
 
@@ -250,7 +251,7 @@ Response::CloneUnfiltered(ErrorResult& aRv) const
 
   RefPtr<InternalResponse> clone = mInternalResponse->Clone();
   RefPtr<InternalResponse> ir = clone->Unfiltered();
-  RefPtr<Response> ref = new Response(mOwner, ir);
+  RefPtr<Response> ref = new Response(mOwner, ir, mSignal);
   return ref.forget();
 }
 
diff --git a/dom/fetch/Response.h b/dom/fetch/Response.h
index de367bef6..ca86c3458 100644
--- a/dom/fetch/Response.h
+++ b/dom/fetch/Response.h
@@ -33,7 +33,7 @@ class Response final : public nsISupports
   NS_DECL_CYCLE_COLLECTION_SCRIPT_HOLDER_CLASS(Response)
 
 public:
-  Response(nsIGlobalObject* aGlobal, InternalResponse* aInternalResponse);
+  Response(nsIGlobalObject* aGlobal, InternalResponse* aInternalResponse, AbortSignal* aSignal);
 
   Response(const Response& aOther) = delete;
 
@@ -134,13 +134,21 @@ public:
   already_AddRefed<InternalResponse>
   GetInternalResponse() const;
 
+  AbortSignal*
+  GetSignal() const override
+  {
+    return mSignal;
+  }
+
 private:
   ~Response();
 
   nsCOMPtr<nsIGlobalObject> mOwner;
   RefPtr<InternalResponse> mInternalResponse;
+
   // Lazily created
   RefPtr<Headers> mHeaders;
+  RefPtr<AbortSignal> mSignal;
 };
 
 } // namespace dom