dom/base/BodyConsumer.cpp

   1 /* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
   2 /* vim: set ts=8 sts=2 et sw=2 tw=80: */
   3 /* This Source Code Form is subject to the terms of the Mozilla Public
   4  * License, v. 2.0. If a copy of the MPL was not distributed with this
   5  * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
   6
   7 #include "BodyConsumer.h"
   8
   9 #include "mozilla/dom/BlobBinding.h"
  10 #include "mozilla/dom/BlobImpl.h"
  11 #include "mozilla/dom/BlobURLProtocolHandler.h"
  12 #include "mozilla/dom/BodyUtil.h"
  13 #include "mozilla/dom/File.h"
  14 #include "mozilla/dom/FileBinding.h"
  15 #include "mozilla/dom/FileCreatorHelper.h"
  16 #include "mozilla/dom/MutableBlobStreamListener.h"
  17 #include "mozilla/dom/Promise.h"
  18 #include "mozilla/dom/PromiseNativeHandler.h"
  19 #include "mozilla/dom/WorkerCommon.h"
  20 #include "mozilla/dom/WorkerPrivate.h"
  21 #include "mozilla/dom/WorkerRef.h"
  22 #include "mozilla/dom/WorkerRunnable.h"
  23 #include "mozilla/dom/WorkerScope.h"
  24 #include "mozilla/ipc/PBackgroundSharedTypes.h"
  25 #include "mozilla/ScopeExit.h"
  26 #include "mozilla/TaskQueue.h"
  27 #include "nsComponentManagerUtils.h"
  28 #include "nsIFile.h"
  29 #include "nsIThreadRetargetableRequest.h"
  30 #include "nsIStreamLoader.h"
  31 #include "nsNetUtil.h"
  32 #include "nsProxyRelease.h"
  33 #include "nsIInputStream.h"
  34
  35 // Undefine the macro of CreateFile to avoid FileCreatorHelper#CreateFile being
  36 // replaced by FileCreatorHelper#CreateFileW.
  37 #ifdef CreateFile
  38 #  undef CreateFile
  39 #endif
  40
  41 namespace mozilla::dom {
  42
  43 namespace {
  44
  45 class BeginConsumeBodyRunnable final : public Runnable {
  46  public:
  47   BeginConsumeBodyRunnable(BodyConsumer* aConsumer,
  48                            ThreadSafeWorkerRef* aWorkerRef)
  49       : Runnable("BeginConsumeBodyRunnable"),
  50         mBodyConsumer(aConsumer),
  51         mWorkerRef(aWorkerRef) {}
  52
  53   NS_IMETHOD
  54   Run() override {
  55     mBodyConsumer->BeginConsumeBodyMainThread(mWorkerRef);
  56     return NS_OK;
  57   }
  58
  59  private:
  60   RefPtr<BodyConsumer> mBodyConsumer;
  61   RefPtr<ThreadSafeWorkerRef> mWorkerRef;
  62 };
  63
  64 /*
  65  * Called on successfully reading the complete stream.
  66  */
  67 class ContinueConsumeBodyRunnable final : public MainThreadWorkerRunnable {
  68   RefPtr<BodyConsumer> mBodyConsumer;
  69   nsresult mStatus;
  70   uint32_t mLength;
  71   uint8_t* mResult;
  72
  73  public:
  74   ContinueConsumeBodyRunnable(BodyConsumer* aBodyConsumer,
  75                               WorkerPrivate* aWorkerPrivate, nsresult aStatus,
  76                               uint32_t aLength, uint8_t* aResult)
  77       : MainThreadWorkerRunnable(aWorkerPrivate, "ContinueConsumeBodyRunnable"),
  78         mBodyConsumer(aBodyConsumer),
  79         mStatus(aStatus),
  80         mLength(aLength),
  81         mResult(aResult) {
  82     MOZ_ASSERT(NS_IsMainThread());
  83   }
  84
  85   bool WorkerRun(JSContext* aCx, WorkerPrivate* aWorkerPrivate) override {
  86     mBodyConsumer->ContinueConsumeBody(mStatus, mLength, mResult);
  87     return true;
  88   }
  89 };
  90
  91 // ControlRunnable used to complete the releasing of resources on the worker
  92 // thread when already shutting down.
  93 class AbortConsumeBodyControlRunnable final
  94     : public MainThreadWorkerControlRunnable {
  95   RefPtr<BodyConsumer> mBodyConsumer;
  96
  97  public:
  98   AbortConsumeBodyControlRunnable(BodyConsumer* aBodyConsumer,
  99                                   WorkerPrivate* aWorkerPrivate)
 100       : MainThreadWorkerControlRunnable(aWorkerPrivate),
 101         mBodyConsumer(aBodyConsumer) {
 102     MOZ_ASSERT(NS_IsMainThread());
 103   }
 104
 105   bool WorkerRun(JSContext* aCx, WorkerPrivate* aWorkerPrivate) override {
 106     mBodyConsumer->ContinueConsumeBody(NS_BINDING_ABORTED, 0, nullptr,
 107                                        true /* shutting down */);
 108     return true;
 109   }
 110 };
 111
 112 /*
 113  * In case of failure to create a stream pump or dispatch stream completion to
 114  * worker, ensure we cleanup properly. Thread agnostic.
 115  */
 116 class MOZ_STACK_CLASS AutoFailConsumeBody final {
 117  public:
 118   AutoFailConsumeBody(BodyConsumer* aBodyConsumer,
 119                       ThreadSafeWorkerRef* aWorkerRef)
 120       : mBodyConsumer(aBodyConsumer), mWorkerRef(aWorkerRef) {}
 121
 122   ~AutoFailConsumeBody() {
 123     AssertIsOnMainThread();
 124
 125     if (!mBodyConsumer) {
 126       return;
 127     }
 128
 129     // Web Worker
 130     if (mWorkerRef) {
 131       RefPtr<AbortConsumeBodyControlRunnable> r =
 132           new AbortConsumeBodyControlRunnable(mBodyConsumer,
 133                                               mWorkerRef->Private());
 134       if (!r->Dispatch()) {
 135         MOZ_CRASH("We are going to leak");
 136       }
 137       return;
 138     }
 139
 140     // Main-thread
 141     mBodyConsumer->ContinueConsumeBody(NS_ERROR_FAILURE, 0, nullptr);
 142   }
 143
 144   void DontFail() { mBodyConsumer = nullptr; }
 145
 146  private:
 147   RefPtr<BodyConsumer> mBodyConsumer;
 148   RefPtr<ThreadSafeWorkerRef> mWorkerRef;
 149 };
 150
 151 /*
 152  * Called on successfully reading the complete stream for Blob.
 153  */
 154 class ContinueConsumeBlobBodyRunnable final : public MainThreadWorkerRunnable {
 155   RefPtr<BodyConsumer> mBodyConsumer;
 156   RefPtr<BlobImpl> mBlobImpl;
 157
 158  public:
 159   ContinueConsumeBlobBodyRunnable(BodyConsumer* aBodyConsumer,
 160                                   WorkerPrivate* aWorkerPrivate,
 161                                   BlobImpl* aBlobImpl)
 162       : MainThreadWorkerRunnable(aWorkerPrivate,
 163                                  "ContinueConsumeBlobBodyRunnable"),
 164         mBodyConsumer(aBodyConsumer),
 165         mBlobImpl(aBlobImpl) {
 166     MOZ_ASSERT(NS_IsMainThread());
 167     MOZ_ASSERT(mBlobImpl);
 168   }
 169
 170   bool WorkerRun(JSContext* aCx, WorkerPrivate* aWorkerPrivate) override {
 171     mBodyConsumer->ContinueConsumeBlobBody(mBlobImpl);
 172     return true;
 173   }
 174 };
 175
 176 // ControlRunnable used to complete the releasing of resources on the worker
 177 // thread when already shutting down.
 178 class AbortConsumeBlobBodyControlRunnable final
 179     : public MainThreadWorkerControlRunnable {
 180   RefPtr<BodyConsumer> mBodyConsumer;
 181
 182  public:
 183   AbortConsumeBlobBodyControlRunnable(BodyConsumer* aBodyConsumer,
 184                                       WorkerPrivate* aWorkerPrivate)
 185       : MainThreadWorkerControlRunnable(aWorkerPrivate),
 186         mBodyConsumer(aBodyConsumer) {
 187     MOZ_ASSERT(NS_IsMainThread());
 188   }
 189
 190   bool WorkerRun(JSContext* aCx, WorkerPrivate* aWorkerPrivate) override {
 191     mBodyConsumer->ContinueConsumeBlobBody(nullptr, true /* shutting down */);
 192     return true;
 193   }
 194 };
 195
 196 class ConsumeBodyDoneObserver final : public nsIStreamLoaderObserver,
 197                                       public MutableBlobStorageCallback {
 198  public:
 199   NS_DECL_THREADSAFE_ISUPPORTS
 200
 201   ConsumeBodyDoneObserver(BodyConsumer* aBodyConsumer,
 202                           ThreadSafeWorkerRef* aWorkerRef)
 203       : mBodyConsumer(aBodyConsumer), mWorkerRef(aWorkerRef) {}
 204
 205   NS_IMETHOD
 206   OnStreamComplete(nsIStreamLoader* aLoader, nsISupports* aCtxt,
 207                    nsresult aStatus, uint32_t aResultLength,
 208                    const uint8_t* aResult) override {
 209     MOZ_ASSERT(NS_IsMainThread());
 210
 211     // The loading is completed. Let's nullify the pump before continuing the
 212     // consuming of the body.
 213     mBodyConsumer->NullifyConsumeBodyPump();
 214
 215     uint8_t* nonconstResult = const_cast<uint8_t*>(aResult);
 216
 217     // Main-thread.
 218     if (!mWorkerRef) {
 219       mBodyConsumer->ContinueConsumeBody(aStatus, aResultLength,
 220                                          nonconstResult);
 221       // The caller is responsible for data.
 222       return NS_SUCCESS_ADOPTED_DATA;
 223     }
 224
 225     // Web Worker.
 226     {
 227       RefPtr<ContinueConsumeBodyRunnable> r = new ContinueConsumeBodyRunnable(
 228           mBodyConsumer, mWorkerRef->Private(), aStatus, aResultLength,
 229           nonconstResult);
 230       if (r->Dispatch()) {
 231         // The caller is responsible for data.
 232         return NS_SUCCESS_ADOPTED_DATA;
 233       }
 234     }
 235
 236     // The worker is shutting down. Let's use a control runnable to complete the
 237     // shutting down procedure.
 238
 239     RefPtr<AbortConsumeBodyControlRunnable> r =
 240         new AbortConsumeBodyControlRunnable(mBodyConsumer,
 241                                             mWorkerRef->Private());
 242     if (NS_WARN_IF(!r->Dispatch())) {
 243       return NS_ERROR_FAILURE;
 244     }
 245
 246     // We haven't taken ownership of the data.
 247     return NS_OK;
 248   }
 249
 250   virtual void BlobStoreCompleted(MutableBlobStorage* aBlobStorage,
 251                                   BlobImpl* aBlobImpl, nsresult aRv) override {
 252     // On error.
 253     if (NS_FAILED(aRv)) {
 254       OnStreamComplete(nullptr, nullptr, aRv, 0, nullptr);
 255       return;
 256     }
 257
 258     // The loading is completed. Let's nullify the pump before continuing the
 259     // consuming of the body.
 260     mBodyConsumer->NullifyConsumeBodyPump();
 261
 262     mBodyConsumer->OnBlobResult(aBlobImpl, mWorkerRef);
 263   }
 264
 265  private:
 266   ~ConsumeBodyDoneObserver() = default;
 267
 268   RefPtr<BodyConsumer> mBodyConsumer;
 269   RefPtr<ThreadSafeWorkerRef> mWorkerRef;
 270 };
 271
 272 NS_IMPL_ISUPPORTS(ConsumeBodyDoneObserver, nsIStreamLoaderObserver)
 273
 274 }  // namespace
 275
 276 /* static */ already_AddRefed<Promise> BodyConsumer::Create(
 277     nsIGlobalObject* aGlobal, nsISerialEventTarget* aMainThreadEventTarget,
 278     nsIInputStream* aBodyStream, AbortSignalImpl* aSignalImpl,
 279     ConsumeType aType, const nsACString& aBodyBlobURISpec,
 280     const nsAString& aBodyLocalPath, const nsACString& aBodyMimeType,
 281     const nsACString& aMixedCaseMimeType,
 282     MutableBlobStorage::MutableBlobStorageType aBlobStorageType,
 283     ErrorResult& aRv) {
 284   MOZ_ASSERT(aBodyStream);
 285   MOZ_ASSERT(aMainThreadEventTarget);
 286
 287   RefPtr<Promise> promise = Promise::Create(aGlobal, aRv);
 288   if (aRv.Failed()) {
 289     return nullptr;
 290   }
 291
 292   RefPtr<BodyConsumer> consumer =
 293       new BodyConsumer(aMainThreadEventTarget, aGlobal, aBodyStream, promise,
 294                        aType, aBodyBlobURISpec, aBodyLocalPath, aBodyMimeType,
 295                        aMixedCaseMimeType, aBlobStorageType);
 296
 297   RefPtr<ThreadSafeWorkerRef> workerRef;
 298
 299   if (!NS_IsMainThread()) {
 300     WorkerPrivate* workerPrivate = GetCurrentThreadWorkerPrivate();
 301     MOZ_ASSERT(workerPrivate);
 302
 303     RefPtr<StrongWorkerRef> strongWorkerRef =
 304         StrongWorkerRef::Create(workerPrivate, "BodyConsumer", [consumer]() {
 305           consumer->mConsumePromise = nullptr;
 306           consumer->mBodyConsumed = true;
 307           consumer->ReleaseObject();
 308           consumer->ShutDownMainThreadConsuming();
 309         });
 310     if (NS_WARN_IF(!strongWorkerRef)) {
 311       aRv.Throw(NS_ERROR_FAILURE);
 312       return nullptr;
 313     }
 314
 315     workerRef = new ThreadSafeWorkerRef(strongWorkerRef);
 316   } else {
 317     nsCOMPtr<nsIObserverService> os = mozilla::services::GetObserverService();
 318     if (NS_WARN_IF(!os)) {
 319       aRv.Throw(NS_ERROR_FAILURE);
 320       return nullptr;
 321     }
 322
 323     aRv = os->AddObserver(consumer, DOM_WINDOW_DESTROYED_TOPIC, true);
 324     if (NS_WARN_IF(aRv.Failed())) {
 325       return nullptr;
 326     }
 327
 328     aRv = os->AddObserver(consumer, DOM_WINDOW_FROZEN_TOPIC, true);
 329     if (NS_WARN_IF(aRv.Failed())) {
 330       return nullptr;
 331     }
 332   }
 333
 334   nsCOMPtr<nsIRunnable> r = new BeginConsumeBodyRunnable(consumer, workerRef);
 335   aRv = aMainThreadEventTarget->Dispatch(r.forget(), NS_DISPATCH_NORMAL);
 336   if (NS_WARN_IF(aRv.Failed())) {
 337     return nullptr;
 338   }
 339
 340   if (aSignalImpl) {
 341     consumer->Follow(aSignalImpl);
 342   }
 343
 344   return promise.forget();
 345 }
 346
 347 void BodyConsumer::ReleaseObject() {
 348   AssertIsOnTargetThread();
 349
 350   if (NS_IsMainThread()) {
 351     nsCOMPtr<nsIObserverService> os = mozilla::services::GetObserverService();
 352     if (os) {
 353       os->RemoveObserver(this, DOM_WINDOW_DESTROYED_TOPIC);
 354       os->RemoveObserver(this, DOM_WINDOW_FROZEN_TOPIC);
 355     }
 356   }
 357
 358   mGlobal = nullptr;
 359
 360   Unfollow();
 361 }
 362
 363 BodyConsumer::BodyConsumer(
 364     nsISerialEventTarget* aMainThreadEventTarget,
 365     nsIGlobalObject* aGlobalObject, nsIInputStream* aBodyStream,
 366     Promise* aPromise, ConsumeType aType, const nsACString& aBodyBlobURISpec,
 367     const nsAString& aBodyLocalPath, const nsACString& aBodyMimeType,
 368     const nsACString& aMixedCaseMimeType,
 369     MutableBlobStorage::MutableBlobStorageType aBlobStorageType)
 370     : mTargetThread(NS_GetCurrentThread()),
 371       mMainThreadEventTarget(aMainThreadEventTarget),
 372       mBodyStream(aBodyStream),
 373       mBlobStorageType(aBlobStorageType),
 374       mBodyMimeType(aBodyMimeType),
 375       mMixedCaseMimeType(aMixedCaseMimeType),
 376       mBodyBlobURISpec(aBodyBlobURISpec),
 377       mBodyLocalPath(aBodyLocalPath),
 378       mGlobal(aGlobalObject),
 379       mConsumeType(aType),
 380       mConsumePromise(aPromise),
 381       mBodyConsumed(false),
 382       mShuttingDown(false) {
 383   MOZ_ASSERT(aMainThreadEventTarget);
 384   MOZ_ASSERT(aBodyStream);
 385   MOZ_ASSERT(aPromise);
 386 }
 387
 388 BodyConsumer::~BodyConsumer() = default;
 389
 390 void BodyConsumer::AssertIsOnTargetThread() const {
 391   MOZ_ASSERT(NS_GetCurrentThread() == mTargetThread);
 392 }
 393
 394 namespace {
 395
 396 class FileCreationHandler final : public PromiseNativeHandler {
 397  public:
 398   NS_DECL_THREADSAFE_ISUPPORTS
 399
 400   static void Create(Promise* aPromise, BodyConsumer* aConsumer,
 401                      ThreadSafeWorkerRef* aWorkerRef) {
 402     AssertIsOnMainThread();
 403     MOZ_ASSERT(aPromise);
 404
 405     RefPtr<FileCreationHandler> handler =
 406         new FileCreationHandler(aConsumer, aWorkerRef);
 407     aPromise->AppendNativeHandler(handler);
 408   }
 409
 410   void ResolvedCallback(JSContext* aCx, JS::Handle<JS::Value> aValue,
 411                         ErrorResult& aRv) override {
 412     AssertIsOnMainThread();
 413
 414     if (NS_WARN_IF(!aValue.isObject())) {
 415       mConsumer->OnBlobResult(nullptr, mWorkerRef);
 416       return;
 417     }
 418
 419     RefPtr<Blob> blob;
 420     if (NS_WARN_IF(NS_FAILED(UNWRAP_OBJECT(Blob, &aValue.toObject(), blob)))) {
 421       mConsumer->OnBlobResult(nullptr, mWorkerRef);
 422       return;
 423     }
 424
 425     mConsumer->OnBlobResult(blob->Impl(), mWorkerRef);
 426   }
 427
 428   void RejectedCallback(JSContext* aCx, JS::Handle<JS::Value> aValue,
 429                         ErrorResult& aRv) override {
 430     AssertIsOnMainThread();
 431
 432     mConsumer->OnBlobResult(nullptr, mWorkerRef);
 433   }
 434
 435  private:
 436   FileCreationHandler(BodyConsumer* aConsumer, ThreadSafeWorkerRef* aWorkerRef)
 437       : mConsumer(aConsumer), mWorkerRef(aWorkerRef) {
 438     AssertIsOnMainThread();
 439     MOZ_ASSERT(aConsumer);
 440   }
 441
 442   ~FileCreationHandler() = default;
 443
 444   RefPtr<BodyConsumer> mConsumer;
 445   RefPtr<ThreadSafeWorkerRef> mWorkerRef;
 446 };
 447
 448 NS_IMPL_ISUPPORTS0(FileCreationHandler)
 449
 450 }  // namespace
 451
 452 nsresult BodyConsumer::GetBodyLocalFile(nsIFile** aFile) const {
 453   AssertIsOnMainThread();
 454
 455   if (!mBodyLocalPath.Length()) {
 456     return NS_OK;
 457   }
 458
 459   nsresult rv;
 460   nsCOMPtr<nsIFile> file = do_CreateInstance("@mozilla.org/file/local;1", &rv);
 461   if (NS_FAILED(rv)) {
 462     return rv;
 463   }
 464
 465   rv = file->InitWithPath(mBodyLocalPath);
 466   NS_ENSURE_SUCCESS(rv, rv);
 467
 468   bool exists;
 469   rv = file->Exists(&exists);
 470   NS_ENSURE_SUCCESS(rv, rv);
 471   if (!exists) {
 472     return NS_ERROR_FILE_NOT_FOUND;
 473   }
 474
 475   bool isDir;
 476   rv = file->IsDirectory(&isDir);
 477   NS_ENSURE_SUCCESS(rv, rv);
 478   if (isDir) {
 479     return NS_ERROR_FILE_IS_DIRECTORY;
 480   }
 481
 482   file.forget(aFile);
 483   return NS_OK;
 484 }
 485
 486 /*
 487  * BeginConsumeBodyMainThread() will automatically reject the consume promise
 488  * and clean up on any failures, so there is no need for callers to do so,
 489  * reflected in a lack of error return code.
 490  */
 491 void BodyConsumer::BeginConsumeBodyMainThread(ThreadSafeWorkerRef* aWorkerRef) {
 492   AssertIsOnMainThread();
 493
 494   AutoFailConsumeBody autoReject(this, aWorkerRef);
 495
 496   if (mShuttingDown) {
 497     // We haven't started yet, but we have been terminated. AutoFailConsumeBody
 498     // will dispatch a runnable to release resources.
 499     return;
 500   }
 501
 502   if (mConsumeType == CONSUME_BLOB) {
 503     nsresult rv;
 504
 505     // If we're trying to consume a blob, and the request was for a blob URI,
 506     // then just consume that URI's blob instance.
 507     if (!mBodyBlobURISpec.IsEmpty()) {
 508       RefPtr<BlobImpl> blobImpl;
 509       rv = NS_GetBlobForBlobURISpec(mBodyBlobURISpec, getter_AddRefs(blobImpl));
 510       if (NS_WARN_IF(NS_FAILED(rv)) || !blobImpl) {
 511         return;
 512       }
 513       autoReject.DontFail();
 514       DispatchContinueConsumeBlobBody(blobImpl, aWorkerRef);
 515       return;
 516     }
 517
 518     // If we're trying to consume a blob, and the request was for a local
 519     // file, then generate and return a File blob.
 520     nsCOMPtr<nsIFile> file;
 521     rv = GetBodyLocalFile(getter_AddRefs(file));
 522     if (!NS_WARN_IF(NS_FAILED(rv)) && file && !aWorkerRef) {
 523       ChromeFilePropertyBag bag;
 524       CopyUTF8toUTF16(mBodyMimeType, bag.mType);
 525
 526       ErrorResult error;
 527       RefPtr<Promise> promise =
 528           FileCreatorHelper::CreateFile(mGlobal, file, bag, true, error);
 529       if (NS_WARN_IF(error.Failed())) {
 530         return;
 531       }
 532
 533       autoReject.DontFail();
 534       FileCreationHandler::Create(promise, this, aWorkerRef);
 535       return;
 536     }
 537   }
 538
 539   nsCOMPtr<nsIInputStreamPump> pump;
 540   nsresult rv =
 541       NS_NewInputStreamPump(getter_AddRefs(pump), mBodyStream.forget(), 0, 0,
 542                             false, mMainThreadEventTarget);
 543   if (NS_WARN_IF(NS_FAILED(rv))) {
 544     return;
 545   }
 546
 547   RefPtr<ConsumeBodyDoneObserver> p =
 548       new ConsumeBodyDoneObserver(this, aWorkerRef);
 549
 550   nsCOMPtr<nsIStreamListener> listener;
 551   if (mConsumeType == CONSUME_BLOB) {
 552     listener = new MutableBlobStreamListener(mBlobStorageType, mBodyMimeType, p,
 553                                              mMainThreadEventTarget);
 554   } else {
 555     nsCOMPtr<nsIStreamLoader> loader;
 556     rv = NS_NewStreamLoader(getter_AddRefs(loader), p);
 557     if (NS_WARN_IF(NS_FAILED(rv))) {
 558       return;
 559     }
 560
 561     listener = loader;
 562   }
 563
 564   rv = pump->AsyncRead(listener);
 565   if (NS_WARN_IF(NS_FAILED(rv))) {
 566     return;
 567   }
 568
 569   // Now that everything succeeded, we can assign the pump to a pointer that
 570   // stays alive for the lifetime of the BodyConsumer.
 571   mConsumeBodyPump = pump;
 572
 573   // It is ok for retargeting to fail and reads to happen on the main thread.
 574   autoReject.DontFail();
 575
 576   // Try to retarget, otherwise fall back to main thread.
 577   nsCOMPtr<nsIThreadRetargetableRequest> rr = do_QueryInterface(pump);
 578   if (rr) {
 579     nsCOMPtr<nsIEventTarget> sts =
 580         do_GetService(NS_STREAMTRANSPORTSERVICE_CONTRACTID);
 581     RefPtr<TaskQueue> queue =
 582         TaskQueue::Create(sts.forget(), "BodyConsumer STS Delivery Queue");
 583     rv = rr->RetargetDeliveryTo(queue);
 584     if (NS_FAILED(rv)) {
 585       NS_WARNING("Retargeting failed");
 586     }
 587   }
 588 }
 589
 590 /*
 591  * OnBlobResult() is called when a blob body is ready to be consumed (when its
 592  * network transfer completes in BeginConsumeBodyRunnable or its local File has
 593  * been wrapped by FileCreationHandler). The blob is sent to the target thread
 594  * and ContinueConsumeBody is called.
 595  */
 596 void BodyConsumer::OnBlobResult(BlobImpl* aBlobImpl,
 597                                 ThreadSafeWorkerRef* aWorkerRef) {
 598   AssertIsOnMainThread();
 599
 600   DispatchContinueConsumeBlobBody(aBlobImpl, aWorkerRef);
 601 }
 602
 603 void BodyConsumer::DispatchContinueConsumeBlobBody(
 604     BlobImpl* aBlobImpl, ThreadSafeWorkerRef* aWorkerRef) {
 605   AssertIsOnMainThread();
 606
 607   // Main-thread.
 608   if (!aWorkerRef) {
 609     if (aBlobImpl) {
 610       ContinueConsumeBlobBody(aBlobImpl);
 611     } else {
 612       ContinueConsumeBody(NS_ERROR_DOM_ABORT_ERR, 0, nullptr);
 613     }
 614     return;
 615   }
 616
 617   // Web Worker.
 618   if (aBlobImpl) {
 619     RefPtr<ContinueConsumeBlobBodyRunnable> r =
 620         new ContinueConsumeBlobBodyRunnable(this, aWorkerRef->Private(),
 621                                             aBlobImpl);
 622
 623     if (r->Dispatch()) {
 624       return;
 625     }
 626   } else {
 627     RefPtr<ContinueConsumeBodyRunnable> r = new ContinueConsumeBodyRunnable(
 628         this, aWorkerRef->Private(), NS_ERROR_DOM_ABORT_ERR, 0, nullptr);
 629
 630     if (r->Dispatch()) {
 631       return;
 632     }
 633   }
 634
 635   // The worker is shutting down. Let's use a control runnable to complete the
 636   // shutting down procedure.
 637
 638   RefPtr<AbortConsumeBlobBodyControlRunnable> r =
 639       new AbortConsumeBlobBodyControlRunnable(this, aWorkerRef->Private());
 640
 641   Unused << NS_WARN_IF(!r->Dispatch());
 642 }
 643
 644 /*
 645  * ContinueConsumeBody() is to be called on the target thread whenever the
 646  * final result of the fetch is known. The fetch promise is resolved or
 647  * rejected based on whether the fetch succeeded, and the body can be
 648  * converted into the expected type of JS object.
 649  */
 650 void BodyConsumer::ContinueConsumeBody(nsresult aStatus, uint32_t aResultLength,
 651                                        uint8_t* aResult, bool aShuttingDown) {
 652   AssertIsOnTargetThread();
 653
 654   // This makes sure that we free the data correctly.
 655   UniquePtr<uint8_t[], JS::FreePolicy> resultPtr{aResult};
 656
 657   if (mBodyConsumed) {
 658     return;
 659   }
 660   mBodyConsumed = true;
 661
 662   MOZ_ASSERT(mConsumePromise);
 663   RefPtr<Promise> localPromise = std::move(mConsumePromise);
 664
 665   RefPtr<BodyConsumer> self = this;
 666   auto autoReleaseObject =
 667       mozilla::MakeScopeExit([self] { self->ReleaseObject(); });
 668
 669   if (aShuttingDown) {
 670     // If shutting down, we don't want to resolve any promise.
 671     return;
 672   }
 673
 674   if (NS_WARN_IF(NS_FAILED(aStatus))) {
 675     // Per
 676     // https://fetch.spec.whatwg.org/#concept-read-all-bytes-from-readablestream
 677     // Decoding errors should reject with a TypeError
 678     if (aStatus == NS_ERROR_INVALID_CONTENT_ENCODING) {
 679       localPromise->MaybeRejectWithTypeError<MSG_DOM_DECODING_FAILED>();
 680     } else if (aStatus == NS_ERROR_DOM_WRONG_TYPE_ERR) {
 681       localPromise->MaybeRejectWithTypeError<MSG_FETCH_BODY_WRONG_TYPE>();
 682     } else {
 683       localPromise->MaybeReject(NS_ERROR_DOM_ABORT_ERR);
 684     }
 685   }
 686
 687   // Don't warn here since we warned above.
 688   if (NS_FAILED(aStatus)) {
 689     return;
 690   }
 691
 692   // Finish successfully consuming body according to type.
 693   MOZ_ASSERT(resultPtr);
 694
 695   AutoJSAPI jsapi;
 696   if (!jsapi.Init(mGlobal)) {
 697     localPromise->MaybeReject(NS_ERROR_UNEXPECTED);
 698     return;
 699   }
 700
 701   JSContext* cx = jsapi.cx();
 702   ErrorResult error;
 703
 704   switch (mConsumeType) {
 705     case CONSUME_ARRAYBUFFER: {
 706       JS::Rooted<JSObject*> arrayBuffer(cx);
 707       BodyUtil::ConsumeArrayBuffer(cx, &arrayBuffer, aResultLength,
 708                                    std::move(resultPtr), error);
 709
 710       if (!error.Failed()) {
 711         JS::Rooted<JS::Value> val(cx);
 712         val.setObjectOrNull(arrayBuffer);
 713
 714         localPromise->MaybeResolve(val);
 715       }
 716       break;
 717     }
 718     case CONSUME_BLOB: {
 719       MOZ_CRASH("This should not happen.");
 720       break;
 721     }
 722     case CONSUME_FORMDATA: {
 723       nsCString data;
 724       data.Adopt(reinterpret_cast<char*>(resultPtr.release()), aResultLength);
 725
 726       RefPtr<dom::FormData> fd = BodyUtil::ConsumeFormData(
 727           mGlobal, mBodyMimeType, mMixedCaseMimeType, data, error);
 728       if (!error.Failed()) {
 729         localPromise->MaybeResolve(fd);
 730       }
 731       break;
 732     }
 733     case CONSUME_TEXT:
 734       // fall through handles early exit.
 735     case CONSUME_JSON: {
 736       nsString decoded;
 737       if (NS_SUCCEEDED(
 738               BodyUtil::ConsumeText(aResultLength, resultPtr.get(), decoded))) {
 739         if (mConsumeType == CONSUME_TEXT) {
 740           localPromise->MaybeResolve(decoded);
 741         } else {
 742           JS::Rooted<JS::Value> json(cx);
 743           BodyUtil::ConsumeJson(cx, &json, decoded, error);
 744           if (!error.Failed()) {
 745             localPromise->MaybeResolve(json);
 746           }
 747         }
 748       };
 749       break;
 750     }
 751     default:
 752       MOZ_ASSERT_UNREACHABLE("Unexpected consume body type");
 753   }
 754
 755   error.WouldReportJSException();
 756   if (error.Failed()) {
 757     localPromise->MaybeReject(std::move(error));
 758   }
 759 }
 760
 761 void BodyConsumer::ContinueConsumeBlobBody(BlobImpl* aBlobImpl,
 762                                            bool aShuttingDown) {
 763   AssertIsOnTargetThread();
 764   MOZ_ASSERT(mConsumeType == CONSUME_BLOB);
 765
 766   if (mBodyConsumed) {
 767     return;
 768   }
 769   mBodyConsumed = true;
 770
 771   MOZ_ASSERT(mConsumePromise);
 772   RefPtr<Promise> localPromise = std::move(mConsumePromise);
 773
 774   if (!aShuttingDown) {
 775     RefPtr<dom::Blob> blob = dom::Blob::Create(mGlobal, aBlobImpl);
 776     if (NS_WARN_IF(!blob)) {
 777       localPromise->MaybeReject(NS_ERROR_FAILURE);
 778       return;
 779     }
 780
 781     localPromise->MaybeResolve(blob);
 782   }
 783
 784   ReleaseObject();
 785 }
 786
 787 void BodyConsumer::ShutDownMainThreadConsuming() {
 788   if (!NS_IsMainThread()) {
 789     RefPtr<BodyConsumer> self = this;
 790
 791     nsCOMPtr<nsIRunnable> r = NS_NewRunnableFunction(
 792         "BodyConsumer::ShutDownMainThreadConsuming",
 793         [self]() { self->ShutDownMainThreadConsuming(); });
 794
 795     mMainThreadEventTarget->Dispatch(r.forget(), NS_DISPATCH_NORMAL);
 796     return;
 797   }
 798
 799   // We need this because maybe, mConsumeBodyPump has not been created yet. We
 800   // must be sure that we don't try to do it.
 801   mShuttingDown = true;
 802
 803   if (mConsumeBodyPump) {
 804     mConsumeBodyPump->CancelWithReason(
 805         NS_BINDING_ABORTED, "BodyConsumer::ShutDownMainThreadConsuming"_ns);
 806     mConsumeBodyPump = nullptr;
 807   }
 808 }
 809
 810 NS_IMETHODIMP BodyConsumer::Observe(nsISupports* aSubject, const char* aTopic,
 811                                     const char16_t* aData) {
 812   AssertIsOnMainThread();
 813
 814   MOZ_ASSERT((strcmp(aTopic, DOM_WINDOW_FROZEN_TOPIC) == 0) ||
 815              (strcmp(aTopic, DOM_WINDOW_DESTROYED_TOPIC) == 0));
 816
 817   nsCOMPtr<nsPIDOMWindowInner> window = do_QueryInterface(mGlobal);
 818   if (SameCOMIdentity(aSubject, window)) {
 819     ContinueConsumeBody(NS_BINDING_ABORTED, 0, nullptr);
 820   }
 821
 822   return NS_OK;
 823 }
 824
 825 void BodyConsumer::RunAbortAlgorithm() {
 826   AssertIsOnTargetThread();
 827   ShutDownMainThreadConsuming();
 828   ContinueConsumeBody(NS_ERROR_DOM_ABORT_ERR, 0, nullptr);
 829 }
 830
 831 NS_IMPL_ISUPPORTS(BodyConsumer, nsIObserver, nsISupportsWeakReference)
 832
 833 }  // namespace mozilla::dom