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("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("AbortConsumeBodyControlRunnable"),
 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(mWorkerRef->Private())) {
 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("ContinueConsumeBlobBodyRunnable"),
 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("AbortConsumeBlobBodyControlRunnable"),
 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(mWorkerRef->Private())) {
 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(mWorkerRef->Private()))) {
 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     consumer->GlobalTeardownObserver::BindToOwner(aGlobal);
 317     consumer->GlobalFreezeObserver::BindToOwner(aGlobal);
 318   }
 319
 320   nsCOMPtr<nsIRunnable> r = new BeginConsumeBodyRunnable(consumer, workerRef);
 321   aRv = aMainThreadEventTarget->Dispatch(r.forget(), NS_DISPATCH_NORMAL);
 322   if (NS_WARN_IF(aRv.Failed())) {
 323     return nullptr;
 324   }
 325
 326   if (aSignalImpl) {
 327     consumer->Follow(aSignalImpl);
 328   }
 329
 330   return promise.forget();
 331 }
 332
 333 void BodyConsumer::ReleaseObject() {
 334   AssertIsOnTargetThread();
 335
 336   if (NS_IsMainThread()) {
 337     GlobalTeardownObserver::DisconnectFromOwner();
 338     DisconnectFreezeObserver();
 339   }
 340
 341   mGlobal = nullptr;
 342
 343   Unfollow();
 344 }
 345
 346 BodyConsumer::BodyConsumer(
 347     nsISerialEventTarget* aMainThreadEventTarget,
 348     nsIGlobalObject* aGlobalObject, nsIInputStream* aBodyStream,
 349     Promise* aPromise, ConsumeType aType, const nsACString& aBodyBlobURISpec,
 350     const nsAString& aBodyLocalPath, const nsACString& aBodyMimeType,
 351     const nsACString& aMixedCaseMimeType,
 352     MutableBlobStorage::MutableBlobStorageType aBlobStorageType)
 353     : mTargetThread(NS_GetCurrentThread()),
 354       mMainThreadEventTarget(aMainThreadEventTarget),
 355       mBodyStream(aBodyStream),
 356       mBlobStorageType(aBlobStorageType),
 357       mBodyMimeType(aBodyMimeType),
 358       mMixedCaseMimeType(aMixedCaseMimeType),
 359       mBodyBlobURISpec(aBodyBlobURISpec),
 360       mBodyLocalPath(aBodyLocalPath),
 361       mGlobal(aGlobalObject),
 362       mConsumeType(aType),
 363       mConsumePromise(aPromise),
 364       mBodyConsumed(false),
 365       mShuttingDown(false) {
 366   MOZ_ASSERT(aMainThreadEventTarget);
 367   MOZ_ASSERT(aBodyStream);
 368   MOZ_ASSERT(aPromise);
 369 }
 370
 371 BodyConsumer::~BodyConsumer() = default;
 372
 373 void BodyConsumer::AssertIsOnTargetThread() const {
 374   MOZ_ASSERT(NS_GetCurrentThread() == mTargetThread);
 375 }
 376
 377 namespace {
 378
 379 class FileCreationHandler final : public PromiseNativeHandler {
 380  public:
 381   NS_DECL_THREADSAFE_ISUPPORTS
 382
 383   static void Create(Promise* aPromise, BodyConsumer* aConsumer,
 384                      ThreadSafeWorkerRef* aWorkerRef) {
 385     AssertIsOnMainThread();
 386     MOZ_ASSERT(aPromise);
 387
 388     RefPtr<FileCreationHandler> handler =
 389         new FileCreationHandler(aConsumer, aWorkerRef);
 390     aPromise->AppendNativeHandler(handler);
 391   }
 392
 393   void ResolvedCallback(JSContext* aCx, JS::Handle<JS::Value> aValue,
 394                         ErrorResult& aRv) override {
 395     AssertIsOnMainThread();
 396
 397     if (NS_WARN_IF(!aValue.isObject())) {
 398       mConsumer->OnBlobResult(nullptr, mWorkerRef);
 399       return;
 400     }
 401
 402     RefPtr<Blob> blob;
 403     if (NS_WARN_IF(NS_FAILED(UNWRAP_OBJECT(Blob, &aValue.toObject(), blob)))) {
 404       mConsumer->OnBlobResult(nullptr, mWorkerRef);
 405       return;
 406     }
 407
 408     mConsumer->OnBlobResult(blob->Impl(), mWorkerRef);
 409   }
 410
 411   void RejectedCallback(JSContext* aCx, JS::Handle<JS::Value> aValue,
 412                         ErrorResult& aRv) override {
 413     AssertIsOnMainThread();
 414
 415     mConsumer->OnBlobResult(nullptr, mWorkerRef);
 416   }
 417
 418  private:
 419   FileCreationHandler(BodyConsumer* aConsumer, ThreadSafeWorkerRef* aWorkerRef)
 420       : mConsumer(aConsumer), mWorkerRef(aWorkerRef) {
 421     AssertIsOnMainThread();
 422     MOZ_ASSERT(aConsumer);
 423   }
 424
 425   ~FileCreationHandler() = default;
 426
 427   RefPtr<BodyConsumer> mConsumer;
 428   RefPtr<ThreadSafeWorkerRef> mWorkerRef;
 429 };
 430
 431 NS_IMPL_ISUPPORTS0(FileCreationHandler)
 432
 433 }  // namespace
 434
 435 nsresult BodyConsumer::GetBodyLocalFile(nsIFile** aFile) const {
 436   AssertIsOnMainThread();
 437
 438   if (!mBodyLocalPath.Length()) {
 439     return NS_OK;
 440   }
 441
 442   nsCOMPtr<nsIFile> file;
 443   MOZ_TRY(NS_NewLocalFile(mBodyLocalPath, getter_AddRefs(file)));
 444
 445   bool exists;
 446   MOZ_TRY(file->Exists(&exists));
 447   if (!exists) {
 448     return NS_ERROR_FILE_NOT_FOUND;
 449   }
 450
 451   bool isDir;
 452   MOZ_TRY(file->IsDirectory(&isDir));
 453   if (isDir) {
 454     return NS_ERROR_FILE_IS_DIRECTORY;
 455   }
 456
 457   file.forget(aFile);
 458   return NS_OK;
 459 }
 460
 461 /*
 462  * BeginConsumeBodyMainThread() will automatically reject the consume promise
 463  * and clean up on any failures, so there is no need for callers to do so,
 464  * reflected in a lack of error return code.
 465  */
 466 void BodyConsumer::BeginConsumeBodyMainThread(ThreadSafeWorkerRef* aWorkerRef) {
 467   AssertIsOnMainThread();
 468
 469   AutoFailConsumeBody autoReject(this, aWorkerRef);
 470
 471   if (mShuttingDown) {
 472     // We haven't started yet, but we have been terminated. AutoFailConsumeBody
 473     // will dispatch a runnable to release resources.
 474     return;
 475   }
 476
 477   if (mConsumeType == ConsumeType::Blob) {
 478     nsresult rv;
 479
 480     // If we're trying to consume a blob, and the request was for a blob URI,
 481     // then just consume that URI's blob instance.
 482     if (!mBodyBlobURISpec.IsEmpty()) {
 483       RefPtr<BlobImpl> blobImpl;
 484       rv = NS_GetBlobForBlobURISpec(mBodyBlobURISpec, getter_AddRefs(blobImpl));
 485       if (NS_WARN_IF(NS_FAILED(rv)) || !blobImpl) {
 486         return;
 487       }
 488       autoReject.DontFail();
 489       DispatchContinueConsumeBlobBody(blobImpl, aWorkerRef);
 490       return;
 491     }
 492
 493     // If we're trying to consume a blob, and the request was for a local
 494     // file, then generate and return a File blob.
 495     nsCOMPtr<nsIFile> file;
 496     rv = GetBodyLocalFile(getter_AddRefs(file));
 497     if (!NS_WARN_IF(NS_FAILED(rv)) && file && !aWorkerRef) {
 498       ChromeFilePropertyBag bag;
 499       CopyUTF8toUTF16(mBodyMimeType, bag.mType);
 500
 501       ErrorResult error;
 502       RefPtr<Promise> promise =
 503           FileCreatorHelper::CreateFile(mGlobal, file, bag, true, error);
 504       if (NS_WARN_IF(error.Failed())) {
 505         return;
 506       }
 507
 508       autoReject.DontFail();
 509       FileCreationHandler::Create(promise, this, aWorkerRef);
 510       return;
 511     }
 512   }
 513
 514   nsCOMPtr<nsIInputStreamPump> pump;
 515   nsresult rv =
 516       NS_NewInputStreamPump(getter_AddRefs(pump), mBodyStream.forget(), 0, 0,
 517                             false, mMainThreadEventTarget);
 518   if (NS_WARN_IF(NS_FAILED(rv))) {
 519     return;
 520   }
 521
 522   RefPtr<ConsumeBodyDoneObserver> p =
 523       new ConsumeBodyDoneObserver(this, aWorkerRef);
 524
 525   nsCOMPtr<nsIStreamListener> listener;
 526   if (mConsumeType == ConsumeType::Blob) {
 527     listener = new MutableBlobStreamListener(mBlobStorageType, mBodyMimeType, p,
 528                                              mMainThreadEventTarget);
 529   } else {
 530     nsCOMPtr<nsIStreamLoader> loader;
 531     rv = NS_NewStreamLoader(getter_AddRefs(loader), p);
 532     if (NS_WARN_IF(NS_FAILED(rv))) {
 533       return;
 534     }
 535
 536     listener = loader;
 537   }
 538
 539   rv = pump->AsyncRead(listener);
 540   if (NS_WARN_IF(NS_FAILED(rv))) {
 541     return;
 542   }
 543
 544   // Now that everything succeeded, we can assign the pump to a pointer that
 545   // stays alive for the lifetime of the BodyConsumer.
 546   mConsumeBodyPump = pump;
 547
 548   // It is ok for retargeting to fail and reads to happen on the main thread.
 549   autoReject.DontFail();
 550
 551   // Try to retarget, otherwise fall back to main thread.
 552   nsCOMPtr<nsIThreadRetargetableRequest> rr = do_QueryInterface(pump);
 553   if (rr) {
 554     nsCOMPtr<nsIEventTarget> sts =
 555         do_GetService(NS_STREAMTRANSPORTSERVICE_CONTRACTID);
 556     RefPtr<TaskQueue> queue =
 557         TaskQueue::Create(sts.forget(), "BodyConsumer STS Delivery Queue");
 558     rv = rr->RetargetDeliveryTo(queue);
 559     if (NS_FAILED(rv)) {
 560       NS_WARNING("Retargeting failed");
 561     }
 562   }
 563 }
 564
 565 /*
 566  * OnBlobResult() is called when a blob body is ready to be consumed (when its
 567  * network transfer completes in BeginConsumeBodyRunnable or its local File has
 568  * been wrapped by FileCreationHandler). The blob is sent to the target thread
 569  * and ContinueConsumeBody is called.
 570  */
 571 void BodyConsumer::OnBlobResult(BlobImpl* aBlobImpl,
 572                                 ThreadSafeWorkerRef* aWorkerRef) {
 573   AssertIsOnMainThread();
 574
 575   DispatchContinueConsumeBlobBody(aBlobImpl, aWorkerRef);
 576 }
 577
 578 void BodyConsumer::DispatchContinueConsumeBlobBody(
 579     BlobImpl* aBlobImpl, ThreadSafeWorkerRef* aWorkerRef) {
 580   AssertIsOnMainThread();
 581
 582   // Main-thread.
 583   if (!aWorkerRef) {
 584     if (aBlobImpl) {
 585       ContinueConsumeBlobBody(aBlobImpl);
 586     } else {
 587       ContinueConsumeBody(NS_ERROR_DOM_ABORT_ERR, 0, nullptr);
 588     }
 589     return;
 590   }
 591
 592   // Web Worker.
 593   if (aBlobImpl) {
 594     RefPtr<ContinueConsumeBlobBodyRunnable> r =
 595         new ContinueConsumeBlobBodyRunnable(this, aWorkerRef->Private(),
 596                                             aBlobImpl);
 597
 598     if (r->Dispatch(aWorkerRef->Private())) {
 599       return;
 600     }
 601   } else {
 602     RefPtr<ContinueConsumeBodyRunnable> r = new ContinueConsumeBodyRunnable(
 603         this, aWorkerRef->Private(), NS_ERROR_DOM_ABORT_ERR, 0, nullptr);
 604
 605     if (r->Dispatch(aWorkerRef->Private())) {
 606       return;
 607     }
 608   }
 609
 610   // The worker is shutting down. Let's use a control runnable to complete the
 611   // shutting down procedure.
 612
 613   RefPtr<AbortConsumeBlobBodyControlRunnable> r =
 614       new AbortConsumeBlobBodyControlRunnable(this, aWorkerRef->Private());
 615
 616   Unused << NS_WARN_IF(!r->Dispatch(aWorkerRef->Private()));
 617 }
 618
 619 /*
 620  * ContinueConsumeBody() is to be called on the target thread whenever the
 621  * final result of the fetch is known. The fetch promise is resolved or
 622  * rejected based on whether the fetch succeeded, and the body can be
 623  * converted into the expected type of JS object.
 624  */
 625 void BodyConsumer::ContinueConsumeBody(nsresult aStatus, uint32_t aResultLength,
 626                                        uint8_t* aResult, bool aShuttingDown) {
 627   AssertIsOnTargetThread();
 628
 629   // This makes sure that we free the data correctly.
 630   UniquePtr<uint8_t[], JS::FreePolicy> resultPtr{aResult};
 631
 632   if (mBodyConsumed) {
 633     return;
 634   }
 635   mBodyConsumed = true;
 636
 637   MOZ_ASSERT(mConsumePromise);
 638   RefPtr<Promise> localPromise = std::move(mConsumePromise);
 639
 640   RefPtr<BodyConsumer> self = this;
 641   auto autoReleaseObject =
 642       mozilla::MakeScopeExit([self] { self->ReleaseObject(); });
 643
 644   if (aShuttingDown) {
 645     // If shutting down, we don't want to resolve any promise.
 646     return;
 647   }
 648
 649   if (NS_WARN_IF(NS_FAILED(aStatus))) {
 650     // Per
 651     // https://streams.spec.whatwg.org/#readablestreamdefaultreader-read-all-bytes
 652     // Decoding errors should reject with a TypeError
 653     if (aStatus == NS_ERROR_INVALID_CONTENT_ENCODING) {
 654       localPromise->MaybeRejectWithTypeError<MSG_DOM_DECODING_FAILED>();
 655     } else if (aStatus == NS_ERROR_DOM_WRONG_TYPE_ERR) {
 656       localPromise->MaybeRejectWithTypeError<MSG_FETCH_BODY_WRONG_TYPE>();
 657     } else if (aStatus == NS_ERROR_NET_PARTIAL_TRANSFER) {
 658       localPromise->MaybeRejectWithTypeError<MSG_FETCH_PARTIAL>();
 659     } else {
 660       localPromise->MaybeReject(NS_ERROR_DOM_ABORT_ERR);
 661     }
 662
 663     return;
 664   }
 665
 666   // Finish successfully consuming body according to type.
 667   MOZ_ASSERT(resultPtr);
 668
 669   AutoJSAPI jsapi;
 670   if (!jsapi.Init(mGlobal)) {
 671     localPromise->MaybeReject(NS_ERROR_UNEXPECTED);
 672     return;
 673   }
 674
 675   JSContext* cx = jsapi.cx();
 676   ErrorResult error;
 677
 678   switch (mConsumeType) {
 679     case ConsumeType::ArrayBuffer: {
 680       JS::Rooted<JSObject*> arrayBuffer(cx);
 681       BodyUtil::ConsumeArrayBuffer(cx, &arrayBuffer, aResultLength,
 682                                    std::move(resultPtr), error);
 683       if (!error.Failed()) {
 684         JS::Rooted<JS::Value> val(cx, JS::ObjectValue(*arrayBuffer));
 685         localPromise->MaybeResolve(val);
 686       }
 687       break;
 688     }
 689     case ConsumeType::Blob: {
 690       MOZ_CRASH("This should not happen.");
 691       break;
 692     }
 693     case ConsumeType::Bytes: {
 694       JS::Rooted<JSObject*> bytes(cx);
 695       BodyUtil::ConsumeBytes(cx, &bytes, aResultLength, std::move(resultPtr),
 696                              error);
 697       if (!error.Failed()) {
 698         JS::Rooted<JS::Value> val(cx, JS::ObjectValue(*bytes));
 699         localPromise->MaybeResolve(val);
 700       }
 701       break;
 702     }
 703     case ConsumeType::FormData: {
 704       nsCString data;
 705       data.Adopt(reinterpret_cast<char*>(resultPtr.release()), aResultLength);
 706
 707       RefPtr<dom::FormData> fd = BodyUtil::ConsumeFormData(
 708           mGlobal, mBodyMimeType, mMixedCaseMimeType, data, error);
 709       if (!error.Failed()) {
 710         localPromise->MaybeResolve(fd);
 711       }
 712       break;
 713     }
 714     case ConsumeType::Text:
 715       // fall through handles early exit.
 716     case ConsumeType::JSON: {
 717       nsString decoded;
 718       if (NS_SUCCEEDED(
 719               BodyUtil::ConsumeText(aResultLength, resultPtr.get(), decoded))) {
 720         if (mConsumeType == ConsumeType::Text) {
 721           localPromise->MaybeResolve(decoded);
 722         } else {
 723           JS::Rooted<JS::Value> json(cx);
 724           BodyUtil::ConsumeJson(cx, &json, decoded, error);
 725           if (!error.Failed()) {
 726             localPromise->MaybeResolve(json);
 727           }
 728         }
 729       };
 730       break;
 731     }
 732     default:
 733       MOZ_ASSERT_UNREACHABLE("Unexpected consume body type");
 734   }
 735
 736   error.WouldReportJSException();
 737   if (error.Failed()) {
 738     localPromise->MaybeReject(std::move(error));
 739   }
 740 }
 741
 742 void BodyConsumer::ContinueConsumeBlobBody(BlobImpl* aBlobImpl,
 743                                            bool aShuttingDown) {
 744   AssertIsOnTargetThread();
 745   MOZ_ASSERT(mConsumeType == ConsumeType::Blob);
 746
 747   if (mBodyConsumed) {
 748     return;
 749   }
 750   mBodyConsumed = true;
 751
 752   MOZ_ASSERT(mConsumePromise);
 753   RefPtr<Promise> localPromise = std::move(mConsumePromise);
 754
 755   if (!aShuttingDown) {
 756     RefPtr<dom::Blob> blob = dom::Blob::Create(mGlobal, aBlobImpl);
 757     if (NS_WARN_IF(!blob)) {
 758       localPromise->MaybeReject(NS_ERROR_FAILURE);
 759       return;
 760     }
 761
 762     localPromise->MaybeResolve(blob);
 763   }
 764
 765   ReleaseObject();
 766 }
 767
 768 void BodyConsumer::ShutDownMainThreadConsuming() {
 769   if (!NS_IsMainThread()) {
 770     RefPtr<BodyConsumer> self = this;
 771
 772     nsCOMPtr<nsIRunnable> r = NS_NewRunnableFunction(
 773         "BodyConsumer::ShutDownMainThreadConsuming",
 774         [self]() { self->ShutDownMainThreadConsuming(); });
 775
 776     mMainThreadEventTarget->Dispatch(r.forget(), NS_DISPATCH_NORMAL);
 777     return;
 778   }
 779
 780   // We need this because maybe, mConsumeBodyPump has not been created yet. We
 781   // must be sure that we don't try to do it.
 782   mShuttingDown = true;
 783
 784   if (mConsumeBodyPump) {
 785     mConsumeBodyPump->CancelWithReason(
 786         NS_BINDING_ABORTED, "BodyConsumer::ShutDownMainThreadConsuming"_ns);
 787     mConsumeBodyPump = nullptr;
 788   }
 789 }
 790
 791 void BodyConsumer::MaybeAbortConsumption() {
 792   AssertIsOnMainThread();
 793
 794   ContinueConsumeBody(NS_BINDING_ABORTED, 0, nullptr);
 795 }
 796
 797 void BodyConsumer::RunAbortAlgorithm() {
 798   AssertIsOnTargetThread();
 799   ShutDownMainThreadConsuming();
 800   ContinueConsumeBody(NS_ERROR_DOM_ABORT_ERR, 0, nullptr);
 801 }
 802
 803 NS_IMPL_ADDREF(BodyConsumer)
 804 NS_IMPL_RELEASE(BodyConsumer)
 805 NS_INTERFACE_TABLE_HEAD(BodyConsumer)
 806   NS_INTERFACE_TABLE_BEGIN
 807     NS_INTERFACE_TABLE_ENTRY_AMBIGUOUS(BodyConsumer, nsISupports,
 808                                        GlobalTeardownObserver)
 809   NS_INTERFACE_TABLE_END
 810 NS_INTERFACE_TABLE_TAIL
 811
 812 }  // namespace mozilla::dom