dom/fetch/FetchStreamReader.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 "FetchStreamReader.h"
   8 #include "InternalResponse.h"
   9 #include "js/Stream.h"
  10 #include "mozilla/ConsoleReportCollector.h"
  11 #include "mozilla/dom/AutoEntryScript.h"
  12 #include "mozilla/dom/DOMException.h"
  13 #include "mozilla/dom/Promise.h"
  14 #include "mozilla/dom/PromiseBinding.h"
  15 #include "mozilla/dom/WorkerPrivate.h"
  16 #include "mozilla/dom/WorkerRef.h"
  17 #include "mozilla/HoldDropJSObjects.h"
  18 #include "mozilla/TaskCategory.h"
  19 #include "nsContentUtils.h"
  20 #include "nsIAsyncInputStream.h"
  21 #include "nsIPipe.h"
  22 #include "nsIScriptError.h"
  23 #include "nsPIDOMWindow.h"
  24 #include "jsapi.h"
  25
  26 namespace mozilla::dom {
  27
  28 NS_IMPL_CYCLE_COLLECTING_ADDREF(FetchStreamReader)
  29 NS_IMPL_CYCLE_COLLECTING_RELEASE(FetchStreamReader)
  30
  31 NS_IMPL_CYCLE_COLLECTION_CLASS(FetchStreamReader)
  32
  33 NS_IMPL_CYCLE_COLLECTION_UNLINK_BEGIN(FetchStreamReader)
  34   NS_IMPL_CYCLE_COLLECTION_UNLINK(mGlobal)
  35 NS_IMPL_CYCLE_COLLECTION_UNLINK_END
  36
  37 NS_IMPL_CYCLE_COLLECTION_TRAVERSE_BEGIN(FetchStreamReader)
  38   NS_IMPL_CYCLE_COLLECTION_TRAVERSE(mGlobal)
  39 NS_IMPL_CYCLE_COLLECTION_TRAVERSE_END
  40
  41 NS_IMPL_CYCLE_COLLECTION_TRACE_BEGIN(FetchStreamReader)
  42   NS_IMPL_CYCLE_COLLECTION_TRACE_JS_MEMBER_CALLBACK(mReader)
  43 NS_IMPL_CYCLE_COLLECTION_TRACE_END
  44
  45 NS_INTERFACE_MAP_BEGIN_CYCLE_COLLECTION(FetchStreamReader)
  46   NS_INTERFACE_MAP_ENTRY(nsIOutputStreamCallback)
  47   NS_INTERFACE_MAP_ENTRY_AMBIGUOUS(nsISupports, nsIOutputStreamCallback)
  48 NS_INTERFACE_MAP_END
  49
  50 /* static */
  51 nsresult FetchStreamReader::Create(JSContext* aCx, nsIGlobalObject* aGlobal,
  52                                    FetchStreamReader** aStreamReader,
  53                                    nsIInputStream** aInputStream) {
  54   MOZ_ASSERT(aCx);
  55   MOZ_ASSERT(aGlobal);
  56   MOZ_ASSERT(aStreamReader);
  57   MOZ_ASSERT(aInputStream);
  58
  59   RefPtr<FetchStreamReader> streamReader = new FetchStreamReader(aGlobal);
  60
  61   nsCOMPtr<nsIAsyncInputStream> pipeIn;
  62
  63   nsresult rv =
  64       NS_NewPipe2(getter_AddRefs(pipeIn),
  65                   getter_AddRefs(streamReader->mPipeOut), true, true, 0, 0);
  66   if (NS_WARN_IF(NS_FAILED(rv))) {
  67     return rv;
  68   }
  69
  70   if (!NS_IsMainThread()) {
  71     WorkerPrivate* workerPrivate = GetWorkerPrivateFromContext(aCx);
  72     MOZ_ASSERT(workerPrivate);
  73
  74     RefPtr<WeakWorkerRef> workerRef =
  75         WeakWorkerRef::Create(workerPrivate, [streamReader]() {
  76           MOZ_ASSERT(streamReader);
  77           MOZ_ASSERT(streamReader->mWorkerRef);
  78
  79           WorkerPrivate* workerPrivate = streamReader->mWorkerRef->GetPrivate();
  80           MOZ_ASSERT(workerPrivate);
  81
  82           streamReader->CloseAndRelease(workerPrivate->GetJSContext(),
  83                                         NS_ERROR_DOM_INVALID_STATE_ERR);
  84         });
  85
  86     if (NS_WARN_IF(!workerRef)) {
  87       streamReader->mPipeOut->CloseWithStatus(NS_ERROR_DOM_INVALID_STATE_ERR);
  88       return NS_ERROR_DOM_INVALID_STATE_ERR;
  89     }
  90
  91     // These 2 objects create a ref-cycle here that is broken when the stream is
  92     // closed or the worker shutsdown.
  93     streamReader->mWorkerRef = std::move(workerRef);
  94   }
  95
  96   pipeIn.forget(aInputStream);
  97   streamReader.forget(aStreamReader);
  98   return NS_OK;
  99 }
 100
 101 FetchStreamReader::FetchStreamReader(nsIGlobalObject* aGlobal)
 102     : mGlobal(aGlobal),
 103       mOwningEventTarget(mGlobal->EventTargetFor(TaskCategory::Other)),
 104       mBufferRemaining(0),
 105       mBufferOffset(0),
 106       mStreamClosed(false) {
 107   MOZ_ASSERT(aGlobal);
 108
 109   mozilla::HoldJSObjects(this);
 110 }
 111
 112 FetchStreamReader::~FetchStreamReader() {
 113   CloseAndRelease(nullptr, NS_BASE_STREAM_CLOSED);
 114
 115   mozilla::DropJSObjects(this);
 116 }
 117
 118 // If a context is provided, an attempt will be made to cancel the reader.  The
 119 // only situation where we don't expect to have a context is when closure is
 120 // being triggered from the destructor or the WorkerRef is notifying.  If
 121 // we're at the destructor, it's far too late to cancel anything.  And if the
 122 // WorkerRef is being notified, the global is going away, so there's also
 123 // no need to do further JS work.
 124 void FetchStreamReader::CloseAndRelease(JSContext* aCx, nsresult aStatus) {
 125   NS_ASSERT_OWNINGTHREAD(FetchStreamReader);
 126
 127   if (mStreamClosed) {
 128     // Already closed.
 129     return;
 130   }
 131
 132   RefPtr<FetchStreamReader> kungFuDeathGrip = this;
 133
 134   if (aCx && mReader) {
 135     RefPtr<DOMException> error = DOMException::Create(aStatus);
 136
 137     JS::Rooted<JS::Value> errorValue(aCx);
 138     if (ToJSValue(aCx, error, &errorValue)) {
 139       JS::Rooted<JSObject*> reader(aCx, mReader);
 140       // It's currently safe to cancel an already closed reader because, per the
 141       // comments in ReadableStream::cancel() conveying the spec, step 2 of
 142       // 3.4.3 that specified ReadableStreamCancel is: If stream.[[state]] is
 143       // "closed", return a new promise resolved with undefined.
 144       JS::ReadableStreamReaderCancel(aCx, reader, errorValue);
 145     }
 146
 147     // We don't want to propagate exceptions during the cleanup.
 148     JS_ClearPendingException(aCx);
 149   }
 150
 151   mStreamClosed = true;
 152
 153   mGlobal = nullptr;
 154
 155   mPipeOut->CloseWithStatus(aStatus);
 156   mPipeOut = nullptr;
 157
 158   mWorkerRef = nullptr;
 159
 160   mReader = nullptr;
 161   mBuffer.Clear();
 162 }
 163
 164 void FetchStreamReader::StartConsuming(JSContext* aCx, JS::HandleObject aStream,
 165                                        JS::MutableHandle<JSObject*> aReader,
 166                                        ErrorResult& aRv) {
 167   MOZ_DIAGNOSTIC_ASSERT(!mReader);
 168   MOZ_DIAGNOSTIC_ASSERT(aStream);
 169
 170   aRv.MightThrowJSException();
 171
 172   // Here, by spec, we can pick any global we want. Just to avoid extra
 173   // cross-compartment steps, we want to create the reader in the same
 174   // compartment of the owning Fetch Body object.
 175   // The same global will be used to retrieve data from this reader.
 176   JSAutoRealm ar(aCx, mGlobal->GetGlobalJSObject());
 177
 178   JS::Rooted<JSObject*> reader(
 179       aCx, JS::ReadableStreamGetReader(aCx, aStream,
 180                                        JS::ReadableStreamReaderMode::Default));
 181   if (!reader) {
 182     aRv.StealExceptionFromJSContext(aCx);
 183     CloseAndRelease(aCx, NS_ERROR_DOM_INVALID_STATE_ERR);
 184     return;
 185   }
 186
 187   mReader = reader;
 188   aReader.set(reader);
 189
 190   aRv = mPipeOut->AsyncWait(this, 0, 0, mOwningEventTarget);
 191   if (NS_WARN_IF(aRv.Failed())) {
 192     return;
 193   }
 194 }
 195
 196 // nsIOutputStreamCallback interface
 197
 198 NS_IMETHODIMP
 199 FetchStreamReader::OnOutputStreamReady(nsIAsyncOutputStream* aStream) {
 200   NS_ASSERT_OWNINGTHREAD(FetchStreamReader);
 201   MOZ_ASSERT(aStream == mPipeOut);
 202   MOZ_ASSERT(mReader);
 203
 204   if (mStreamClosed) {
 205     return NS_OK;
 206   }
 207
 208   if (!mBuffer.IsEmpty()) {
 209     return WriteBuffer();
 210   }
 211
 212   // Here we can retrieve data from the reader using any global we want because
 213   // it is not observable. We want to use the reader's global, which is also the
 214   // Response's one.
 215   AutoEntryScript aes(mGlobal, "ReadableStreamReader.read", !mWorkerRef);
 216
 217   JS::Rooted<JSObject*> reader(aes.cx(), mReader);
 218   JS::Rooted<JSObject*> promise(
 219       aes.cx(), JS::ReadableStreamDefaultReaderRead(aes.cx(), reader));
 220   if (NS_WARN_IF(!promise)) {
 221     // Let's close the stream.
 222     CloseAndRelease(aes.cx(), NS_ERROR_DOM_INVALID_STATE_ERR);
 223     return NS_ERROR_FAILURE;
 224   }
 225
 226   RefPtr<Promise> domPromise = Promise::CreateFromExisting(mGlobal, promise);
 227   if (NS_WARN_IF(!domPromise)) {
 228     // Let's close the stream.
 229     CloseAndRelease(aes.cx(), NS_ERROR_DOM_INVALID_STATE_ERR);
 230     return NS_ERROR_FAILURE;
 231   }
 232
 233   // Let's wait.
 234   domPromise->AppendNativeHandler(this);
 235   return NS_OK;
 236 }
 237
 238 void FetchStreamReader::ResolvedCallback(JSContext* aCx,
 239                                          JS::Handle<JS::Value> aValue) {
 240   if (mStreamClosed) {
 241     return;
 242   }
 243
 244   // This promise should be resolved with { done: boolean, value: something },
 245   // "value" is interesting only if done is false.
 246
 247   // We don't want to play with JS api, let's WebIDL bindings doing it for us.
 248   // FetchReadableStreamReadDataDone is a dictionary with just a boolean, if the
 249   // parsing succeeded, we can proceed with the parsing of the "value", which it
 250   // must be a Uint8Array.
 251   FetchReadableStreamReadDataDone valueDone;
 252   if (!valueDone.Init(aCx, aValue)) {
 253     JS_ClearPendingException(aCx);
 254     CloseAndRelease(aCx, NS_ERROR_DOM_INVALID_STATE_ERR);
 255     return;
 256   }
 257
 258   if (valueDone.mDone) {
 259     // Stream is completed.
 260     CloseAndRelease(aCx, NS_BASE_STREAM_CLOSED);
 261     return;
 262   }
 263
 264   RootedDictionary<FetchReadableStreamReadDataArray> value(aCx);
 265   if (!value.Init(aCx, aValue) || !value.mValue.WasPassed()) {
 266     JS_ClearPendingException(aCx);
 267     CloseAndRelease(aCx, NS_ERROR_DOM_INVALID_STATE_ERR);
 268     return;
 269   }
 270
 271   Uint8Array& array = value.mValue.Value();
 272   array.ComputeState();
 273   uint32_t len = array.Length();
 274
 275   if (len == 0) {
 276     // If there is nothing to read, let's do another reading.
 277     OnOutputStreamReady(mPipeOut);
 278     return;
 279   }
 280
 281   MOZ_DIAGNOSTIC_ASSERT(mBuffer.IsEmpty());
 282
 283   // Let's take a copy of the data.
 284   if (!mBuffer.AppendElements(array.Data(), len, fallible)) {
 285     CloseAndRelease(aCx, NS_ERROR_OUT_OF_MEMORY);
 286     return;
 287   }
 288
 289   mBufferOffset = 0;
 290   mBufferRemaining = len;
 291
 292   nsresult rv = WriteBuffer();
 293   if (NS_FAILED(rv)) {
 294     // DOMException only understands errors from domerr.msg, so we normalize to
 295     // identifying an abort if the write fails.
 296     CloseAndRelease(aCx, NS_ERROR_DOM_ABORT_ERR);
 297   }
 298 }
 299
 300 nsresult FetchStreamReader::WriteBuffer() {
 301   MOZ_ASSERT(!mBuffer.IsEmpty());
 302
 303   char* data = reinterpret_cast<char*>(mBuffer.Elements());
 304
 305   while (1) {
 306     uint32_t written = 0;
 307     nsresult rv =
 308         mPipeOut->Write(data + mBufferOffset, mBufferRemaining, &written);
 309
 310     if (rv == NS_BASE_STREAM_WOULD_BLOCK) {
 311       break;
 312     }
 313
 314     if (NS_WARN_IF(NS_FAILED(rv))) {
 315       return rv;
 316     }
 317
 318     MOZ_ASSERT(written <= mBufferRemaining);
 319     mBufferRemaining -= written;
 320     mBufferOffset += written;
 321
 322     if (mBufferRemaining == 0) {
 323       mBuffer.Clear();
 324       break;
 325     }
 326   }
 327
 328   nsresult rv = mPipeOut->AsyncWait(this, 0, 0, mOwningEventTarget);
 329   if (NS_WARN_IF(NS_FAILED(rv))) {
 330     return rv;
 331   }
 332
 333   return NS_OK;
 334 }
 335
 336 void FetchStreamReader::RejectedCallback(JSContext* aCx,
 337                                          JS::Handle<JS::Value> aValue) {
 338   ReportErrorToConsole(aCx, aValue);
 339   CloseAndRelease(aCx, NS_ERROR_FAILURE);
 340 }
 341
 342 void FetchStreamReader::ReportErrorToConsole(JSContext* aCx,
 343                                              JS::Handle<JS::Value> aValue) {
 344   nsCString sourceSpec;
 345   uint32_t line = 0;
 346   uint32_t column = 0;
 347   nsString valueString;
 348
 349   nsContentUtils::ExtractErrorValues(aCx, aValue, sourceSpec, &line, &column,
 350                                      valueString);
 351
 352   nsTArray<nsString> params;
 353   params.AppendElement(valueString);
 354
 355   RefPtr<ConsoleReportCollector> reporter = new ConsoleReportCollector();
 356   reporter->AddConsoleReport(nsIScriptError::errorFlag,
 357                              "ReadableStreamReader.read"_ns,
 358                              nsContentUtils::eDOM_PROPERTIES, sourceSpec, line,
 359                              column, "ReadableStreamReadingFailed"_ns, params);
 360
 361   uint64_t innerWindowId = 0;
 362
 363   if (NS_IsMainThread()) {
 364     nsCOMPtr<nsPIDOMWindowInner> window = do_QueryInterface(mGlobal);
 365     if (window) {
 366       innerWindowId = window->WindowID();
 367     }
 368     reporter->FlushReportsToConsole(innerWindowId);
 369     return;
 370   }
 371
 372   WorkerPrivate* workerPrivate = GetWorkerPrivateFromContext(aCx);
 373   if (workerPrivate) {
 374     innerWindowId = workerPrivate->WindowID();
 375   }
 376
 377   RefPtr<Runnable> r = NS_NewRunnableFunction(
 378       "FetchStreamReader::ReportErrorToConsole", [reporter, innerWindowId]() {
 379         reporter->FlushReportsToConsole(innerWindowId);
 380       });
 381
 382   workerPrivate->DispatchToMainThread(r.forget());
 383 }
 384
 385 }  // namespace mozilla::dom