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