dom/streams/UnderlyingSinkCallbackHelpers.cpp

   1 /* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
   2 /* vim:set ts=2 sw=2 sts=2 et cindent: */
   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 "mozilla/dom/UnderlyingSinkCallbackHelpers.h"
   8 #include "StreamUtils.h"
   9 #include "mozilla/dom/UnionTypes.h"
  10 #include "mozilla/dom/WebTransportError.h"
  11 #include "nsHttp.h"
  12
  13 using namespace mozilla::dom;
  14
  15 NS_IMPL_CYCLE_COLLECTION(UnderlyingSinkAlgorithmsBase)
  16 NS_IMPL_CYCLE_COLLECTING_ADDREF(UnderlyingSinkAlgorithmsBase)
  17 NS_IMPL_CYCLE_COLLECTING_RELEASE(UnderlyingSinkAlgorithmsBase)
  18 NS_INTERFACE_MAP_BEGIN_CYCLE_COLLECTION(UnderlyingSinkAlgorithmsBase)
  19   NS_INTERFACE_MAP_ENTRY(nsISupports)
  20 NS_INTERFACE_MAP_END
  21
  22 NS_IMPL_CYCLE_COLLECTION_INHERITED_WITH_JS_MEMBERS(
  23     UnderlyingSinkAlgorithms, UnderlyingSinkAlgorithmsBase,
  24     (mGlobal, mStartCallback, mWriteCallback, mCloseCallback, mAbortCallback),
  25     (mUnderlyingSink))
  26 NS_IMPL_ADDREF_INHERITED(UnderlyingSinkAlgorithms, UnderlyingSinkAlgorithmsBase)
  27 NS_IMPL_RELEASE_INHERITED(UnderlyingSinkAlgorithms,
  28                           UnderlyingSinkAlgorithmsBase)
  29 NS_INTERFACE_MAP_BEGIN_CYCLE_COLLECTION(UnderlyingSinkAlgorithms)
  30 NS_INTERFACE_MAP_END_INHERITING(UnderlyingSinkAlgorithmsBase)
  31
  32 // https://streams.spec.whatwg.org/#set-up-writable-stream-default-controller-from-underlying-sink
  33 void UnderlyingSinkAlgorithms::StartCallback(
  34     JSContext* aCx, WritableStreamDefaultController& aController,
  35     JS::MutableHandle<JS::Value> aRetVal, ErrorResult& aRv) {
  36   if (!mStartCallback) {
  37     // Step 2: Let startAlgorithm be an algorithm that returns undefined.
  38     aRetVal.setUndefined();
  39     return;
  40   }
  41
  42   // Step 6: If underlyingSinkDict["start"] exists, then set startAlgorithm to
  43   // an algorithm which returns the result of invoking
  44   // underlyingSinkDict["start"] with argument list « controller » and callback
  45   // this value underlyingSink.
  46   JS::Rooted<JSObject*> thisObj(aCx, mUnderlyingSink);
  47   return mStartCallback->Call(thisObj, aController, aRetVal, aRv,
  48                               "UnderlyingSink.start",
  49                               CallbackFunction::eRethrowExceptions);
  50 }
  51
  52 // https://streams.spec.whatwg.org/#set-up-writable-stream-default-controller-from-underlying-sink
  53 already_AddRefed<Promise> UnderlyingSinkAlgorithms::WriteCallback(
  54     JSContext* aCx, JS::Handle<JS::Value> aChunk,
  55     WritableStreamDefaultController& aController, ErrorResult& aRv) {
  56   if (!mWriteCallback) {
  57     // Step 3: Let writeAlgorithm be an algorithm that returns a promise
  58     // resolved with undefined.
  59     return Promise::CreateResolvedWithUndefined(mGlobal, aRv);
  60   }
  61
  62   // Step 7: If underlyingSinkDict["write"] exists, then set writeAlgorithm to
  63   // an algorithm which takes an argument chunk and returns the result of
  64   // invoking underlyingSinkDict["write"] with argument list « chunk, controller
  65   // » and callback this value underlyingSink.
  66   JS::Rooted<JSObject*> thisObj(aCx, mUnderlyingSink);
  67   RefPtr<Promise> promise = mWriteCallback->Call(
  68       thisObj, aChunk, aController, aRv, "UnderlyingSink.write",
  69       CallbackFunction::eRethrowExceptions);
  70   return promise.forget();
  71 }
  72
  73 // https://streams.spec.whatwg.org/#set-up-writable-stream-default-controller-from-underlying-sink
  74 already_AddRefed<Promise> UnderlyingSinkAlgorithms::CloseCallback(
  75     JSContext* aCx, ErrorResult& aRv) {
  76   if (!mCloseCallback) {
  77     // Step 4: Let closeAlgorithm be an algorithm that returns a promise
  78     // resolved with undefined.
  79     return Promise::CreateResolvedWithUndefined(mGlobal, aRv);
  80   }
  81
  82   // Step 8: If underlyingSinkDict["close"] exists, then set closeAlgorithm to
  83   // an algorithm which returns the result of invoking
  84   // underlyingSinkDict["close"] with argument list «» and callback this value
  85   // underlyingSink.
  86   JS::Rooted<JSObject*> thisObj(aCx, mUnderlyingSink);
  87   RefPtr<Promise> promise =
  88       mCloseCallback->Call(thisObj, aRv, "UnderlyingSink.close",
  89                            CallbackFunction::eRethrowExceptions);
  90   return promise.forget();
  91 }
  92
  93 // https://streams.spec.whatwg.org/#set-up-writable-stream-default-controller-from-underlying-sink
  94 already_AddRefed<Promise> UnderlyingSinkAlgorithms::AbortCallback(
  95     JSContext* aCx, const Optional<JS::Handle<JS::Value>>& aReason,
  96     ErrorResult& aRv) {
  97   if (!mAbortCallback) {
  98     // Step 5: Let abortAlgorithm be an algorithm that returns a promise
  99     // resolved with undefined.
 100     return Promise::CreateResolvedWithUndefined(mGlobal, aRv);
 101   }
 102
 103   // Step 9: Let abortAlgorithm be an algorithm that returns a promise resolved
 104   // with undefined.
 105   JS::Rooted<JSObject*> thisObj(aCx, mUnderlyingSink);
 106   RefPtr<Promise> promise =
 107       mAbortCallback->Call(thisObj, aReason, aRv, "UnderlyingSink.abort",
 108                            CallbackFunction::eRethrowExceptions);
 109
 110   return promise.forget();
 111 }
 112
 113 // https://streams.spec.whatwg.org/#writable-set-up
 114 // Step 2.1: Let closeAlgorithmWrapper be an algorithm that runs these steps:
 115 already_AddRefed<Promise> UnderlyingSinkAlgorithmsWrapper::CloseCallback(
 116     JSContext* aCx, ErrorResult& aRv) {
 117   nsCOMPtr<nsIGlobalObject> global = xpc::CurrentNativeGlobal(aCx);
 118   return PromisifyAlgorithm(
 119       global, [&](ErrorResult& aRv) { return CloseCallbackImpl(aCx, aRv); },
 120       aRv);
 121 }
 122
 123 // https://streams.spec.whatwg.org/#writable-set-up
 124 // Step 3.1: Let abortAlgorithmWrapper be an algorithm that runs these steps:
 125 already_AddRefed<Promise> UnderlyingSinkAlgorithmsWrapper::AbortCallback(
 126     JSContext* aCx, const Optional<JS::Handle<JS::Value>>& aReason,
 127     ErrorResult& aRv) {
 128   nsCOMPtr<nsIGlobalObject> global = xpc::CurrentNativeGlobal(aCx);
 129   return PromisifyAlgorithm(
 130       global,
 131       [&](ErrorResult& aRv) { return AbortCallbackImpl(aCx, aReason, aRv); },
 132       aRv);
 133 }
 134
 135 NS_IMPL_ISUPPORTS_CYCLE_COLLECTION_INHERITED(WritableStreamToOutput,
 136                                              UnderlyingSinkAlgorithmsBase,
 137                                              nsIOutputStreamCallback)
 138 NS_IMPL_CYCLE_COLLECTION_INHERITED(WritableStreamToOutput,
 139                                    UnderlyingSinkAlgorithmsBase, mParent,
 140                                    mOutput, mPromise)
 141
 142 NS_IMETHODIMP
 143 WritableStreamToOutput::OnOutputStreamReady(nsIAsyncOutputStream* aStream) {
 144   if (!mData) {
 145     return NS_OK;
 146   }
 147   MOZ_ASSERT(mPromise);
 148   uint32_t written = 0;
 149   nsresult rv = mOutput->Write(
 150       reinterpret_cast<const char*>(mData->Elements() + mWritten),
 151       mData->Length() - mWritten, &written);
 152   if (NS_FAILED(rv) && rv != NS_BASE_STREAM_WOULD_BLOCK) {
 153     mPromise->MaybeRejectWithAbortError("Error writing to stream"_ns);
 154     ClearData();
 155     // XXX should we add mErrored and fail future calls immediately?
 156     // I presume new calls to Write() will fail though, too
 157     return rv;
 158   }
 159   if (NS_SUCCEEDED(rv)) {
 160     mWritten += written;
 161     MOZ_ASSERT(mWritten <= mData->Length());
 162     if (mWritten >= mData->Length()) {
 163       mPromise->MaybeResolveWithUndefined();
 164       ClearData();
 165       return NS_OK;
 166     }
 167     // more to write
 168   }
 169   // wrote partial or nothing
 170   // Wait for space
 171   nsCOMPtr<nsIEventTarget> target = mozilla::GetCurrentSerialEventTarget();
 172   rv = mOutput->AsyncWait(this, 0, 0, target);
 173   if (NS_FAILED(rv)) {
 174     mPromise->MaybeRejectWithUnknownError("error waiting to write data");
 175     ClearData();
 176     // XXX should we add mErrored and fail future calls immediately?
 177     // New calls to Write() will fail, note
 178     // See step 5.2 of
 179     // https://streams.spec.whatwg.org/#writable-stream-default-controller-process-write.
 180     return rv;
 181   }
 182   return NS_OK;
 183 }
 184
 185 already_AddRefed<Promise> WritableStreamToOutput::WriteCallback(
 186     JSContext* aCx, JS::Handle<JS::Value> aChunk,
 187     WritableStreamDefaultController& aController, ErrorResult& aError) {
 188   ArrayBufferViewOrArrayBuffer data;
 189   if (!data.Init(aCx, aChunk)) {
 190     aError.StealExceptionFromJSContext(aCx);
 191     return nullptr;
 192   }
 193   // buffer/bufferView
 194   MOZ_ASSERT(data.IsArrayBuffer() || data.IsArrayBufferView());
 195
 196   RefPtr<Promise> promise = Promise::Create(mParent, aError);
 197   if (NS_WARN_IF(aError.Failed())) {
 198     return nullptr;
 199   }
 200
 201   // Try to write first, and only enqueue data if we were already blocked
 202   // or the write didn't write it all.  This avoids allocations and copies
 203   // in common cases.
 204   MOZ_ASSERT(!mPromise);
 205   MOZ_ASSERT(mWritten == 0);
 206   uint32_t written = 0;
 207   ProcessTypedArraysFixed(data, [&](const Span<uint8_t>& aData) {
 208     Span<uint8_t> dataSpan = aData;
 209     nsresult rv = mOutput->Write(mozilla::AsChars(dataSpan).Elements(),
 210                                  dataSpan.Length(), &written);
 211     if (NS_FAILED(rv) && rv != NS_BASE_STREAM_WOULD_BLOCK) {
 212       promise->MaybeRejectWithAbortError("error writing data");
 213       return;
 214     }
 215     if (NS_SUCCEEDED(rv)) {
 216       if (written == dataSpan.Length()) {
 217         promise->MaybeResolveWithUndefined();
 218         return;
 219       }
 220       dataSpan = dataSpan.From(written);
 221     }
 222
 223     auto buffer = Buffer<uint8_t>::CopyFrom(dataSpan);
 224     if (buffer.isNothing()) {
 225       promise->MaybeReject(NS_ERROR_OUT_OF_MEMORY);
 226       return;
 227     }
 228     mData = std::move(buffer);
 229   });
 230
 231   if (promise->State() != Promise::PromiseState::Pending) {
 232     return promise.forget();
 233   }
 234
 235   mPromise = promise;
 236
 237   nsCOMPtr<nsIEventTarget> target = mozilla::GetCurrentSerialEventTarget();
 238   nsresult rv = mOutput->AsyncWait(this, 0, 0, target);
 239   if (NS_FAILED(rv)) {
 240     ClearData();
 241     promise->MaybeRejectWithUnknownError("error waiting to write data");
 242   }
 243   return promise.forget();
 244 }
 245
 246 already_AddRefed<Promise> WritableStreamToOutput::AbortCallbackImpl(
 247     JSContext* aCx, const Optional<JS::Handle<JS::Value>>& aReason,
 248     ErrorResult& aRv) {
 249   // https://streams.spec.whatwg.org/#writablestream-set-up
 250   // Step 3. Let abortAlgorithmWrapper be an algorithm that runs these steps:
 251
 252   if (aReason.WasPassed() && aReason.Value().isObject()) {
 253     JS::Rooted<JSObject*> obj(aCx, &aReason.Value().toObject());
 254     RefPtr<WebTransportError> error;
 255     UnwrapObject<prototypes::id::WebTransportError, WebTransportError>(
 256         obj, error, nullptr);
 257     if (error) {
 258       mOutput->CloseWithStatus(net::GetNSResultFromWebTransportError(
 259           error->GetStreamErrorCode().Value()));
 260       return nullptr;
 261     }
 262   }
 263
 264   // XXX The close or rather a dedicated abort should be async. For now we have
 265   // to always fall back to the Step 3.3 below.
 266   // XXX how do we know this stream is used by webtransport?
 267   mOutput->CloseWithStatus(NS_ERROR_WEBTRANSPORT_CODE_BASE);
 268
 269   // Step 3.3. Return a promise resolved with undefined.
 270   // Wrapper handles this
 271   return nullptr;
 272 }
 273
 274 void WritableStreamToOutput::ReleaseObjects() { mOutput->Close(); }