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