dom/webtransport/api/WebTransportStreams.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/WebTransportStreams.h"
   8
   9 #include "mozilla/dom/WebTransportLog.h"
  10 #include "mozilla/dom/Promise-inl.h"
  11 #include "mozilla/dom/WebTransport.h"
  12 #include "mozilla/dom/WebTransportBidirectionalStream.h"
  13 #include "mozilla/dom/WebTransportReceiveStream.h"
  14 #include "mozilla/dom/WebTransportSendStream.h"
  15 #include "mozilla/Result.h"
  16
  17 using namespace mozilla::ipc;
  18
  19 namespace mozilla::dom {
  20 NS_IMPL_CYCLE_COLLECTION_INHERITED(WebTransportIncomingStreamsAlgorithms,
  21                                    UnderlyingSourceAlgorithmsWrapper,
  22                                    mTransport, mCallback)
  23 NS_IMPL_ADDREF_INHERITED(WebTransportIncomingStreamsAlgorithms,
  24                          UnderlyingSourceAlgorithmsWrapper)
  25 NS_IMPL_RELEASE_INHERITED(WebTransportIncomingStreamsAlgorithms,
  26                           UnderlyingSourceAlgorithmsWrapper)
  27 NS_INTERFACE_MAP_BEGIN_CYCLE_COLLECTION(WebTransportIncomingStreamsAlgorithms)
  28 NS_INTERFACE_MAP_END_INHERITING(UnderlyingSourceAlgorithmsWrapper)
  29
  30 WebTransportIncomingStreamsAlgorithms::WebTransportIncomingStreamsAlgorithms(
  31     StreamType aUnidirectional, WebTransport* aTransport)
  32     : mUnidirectional(aUnidirectional), mTransport(aTransport) {}
  33
  34 WebTransportIncomingStreamsAlgorithms::
  35     ~WebTransportIncomingStreamsAlgorithms() = default;
  36
  37 already_AddRefed<Promise>
  38 WebTransportIncomingStreamsAlgorithms::PullCallbackImpl(
  39     JSContext* aCx, ReadableStreamController& aController, ErrorResult& aRv) {
  40   // https://w3c.github.io/webtransport/#pullbidirectionalstream and
  41   // https://w3c.github.io/webtransport/#pullunidirectionalstream
  42
  43   // Step 1: If transport.[[State]] is "connecting", then return the result
  44   // of performing the following steps upon fulfillment of
  45   // transport.[[Ready]]:
  46   // We don't explicitly check mState here, since we'll reject
  47   // mIncomingStreamPromise if we go to FAILED or CLOSED
  48   //
  49   // Step 2: Let session be transport.[[Session]].
  50   // Step 3: Let p be a new promise.
  51   RefPtr<Promise> promise =
  52       Promise::CreateInfallible(mTransport->GetParentObject());
  53   RefPtr<WebTransportIncomingStreamsAlgorithms> self(this);
  54   // The real work of PullCallback()
  55   // Step 5: Wait until there is an available incoming unidirectional stream.
  56   auto length = (mUnidirectional == StreamType::Unidirectional)
  57                     ? mTransport->mUnidirectionalStreams.Length()
  58                     : mTransport->mBidirectionalStreams.Length();
  59   if (length == 0) {
  60     // We need to wait.
  61     // Per
  62     // https://streams.spec.whatwg.org/#readablestreamdefaultcontroller-pulling
  63     // we can't be called again until the promise is resolved
  64     MOZ_ASSERT(!mCallback);
  65     mCallback = promise;
  66
  67     LOG(("Incoming%sDirectionalStreams Pull waiting for a stream",
  68          mUnidirectional == StreamType::Unidirectional ? "Uni" : "Bi"));
  69     Result<RefPtr<Promise>, nsresult> returnResult =
  70         promise->ThenWithCycleCollectedArgs(
  71             [](JSContext* aCx, JS::Handle<JS::Value>, ErrorResult& aRv,
  72                RefPtr<WebTransportIncomingStreamsAlgorithms> self,
  73                RefPtr<Promise> aPromise) -> already_AddRefed<Promise> {
  74               self->BuildStream(aCx, aRv);
  75               return nullptr;
  76             },
  77             self, promise);
  78     if (returnResult.isErr()) {
  79       // XXX Reject?
  80       aRv.Throw(returnResult.unwrapErr());
  81       return nullptr;
  82     }
  83     // Step 4: Return p and run the remaining steps in parallel.
  84     return returnResult.unwrap().forget();
  85   }
  86   self->BuildStream(aCx, aRv);
  87   // Step 4: Return p and run the remaining steps in parallel.
  88   return promise.forget();
  89 }
  90
  91 // Note: fallible
  92 void WebTransportIncomingStreamsAlgorithms::BuildStream(JSContext* aCx,
  93                                                         ErrorResult& aRv) {
  94   // https://w3c.github.io/webtransport/#pullbidirectionalstream and
  95   // https://w3c.github.io/webtransport/#pullunidirectionalstream
  96   LOG(("Incoming%sDirectionalStreams Pull building a stream",
  97        mUnidirectional == StreamType::Unidirectional ? "Uni" : "Bi"));
  98   if (mUnidirectional == StreamType::Unidirectional) {
  99     // Step 6: Let internalStream be the result of receiving an incoming
 100     // unidirectional stream.
 101     MOZ_ASSERT(mTransport->mUnidirectionalStreams.Length() > 0);
 102     std::tuple<uint64_t, RefPtr<mozilla::ipc::DataPipeReceiver>> tuple =
 103         mTransport->mUnidirectionalStreams[0];
 104     mTransport->mUnidirectionalStreams.RemoveElementAt(0);
 105
 106     // Step 7.1: Let stream be the result of creating a
 107     // WebTransportReceiveStream with internalStream and transport
 108     RefPtr<WebTransportReceiveStream> readableStream =
 109         WebTransportReceiveStream::Create(mTransport, mTransport->mGlobal,
 110                                           std::get<0>(tuple),
 111                                           std::get<1>(tuple), aRv);
 112     if (MOZ_UNLIKELY(!readableStream)) {
 113       aRv.ThrowUnknownError("Internal error");
 114       return;
 115     }
 116     // Step 7.2 Enqueue stream to transport.[[IncomingUnidirectionalStreams]].
 117     JS::Rooted<JS::Value> jsStream(aCx);
 118     if (MOZ_UNLIKELY(!ToJSValue(aCx, readableStream, &jsStream))) {
 119       aRv.ThrowUnknownError("Internal error");
 120       return;
 121     }
 122     // EnqueueNative is CAN_RUN_SCRIPT
 123     RefPtr<ReadableStream> incomingStream =
 124         mTransport->mIncomingUnidirectionalStreams;
 125     incomingStream->EnqueueNative(aCx, jsStream, aRv);
 126     if (MOZ_UNLIKELY(aRv.Failed())) {
 127       aRv.ThrowUnknownError("Internal error");
 128       return;
 129     }
 130   } else {
 131     // Step 6: Let internalStream be the result of receiving a bidirectional
 132     // stream
 133     MOZ_ASSERT(mTransport->mBidirectionalStreams.Length() > 0);
 134     std::tuple<uint64_t, UniquePtr<BidirectionalPair>> tuple =
 135         std::move(mTransport->mBidirectionalStreams.ElementAt(0));
 136     mTransport->mBidirectionalStreams.RemoveElementAt(0);
 137     RefPtr<DataPipeReceiver> input = std::get<1>(tuple)->first.forget();
 138     RefPtr<DataPipeSender> output = std::get<1>(tuple)->second.forget();
 139
 140     RefPtr<WebTransportBidirectionalStream> stream =
 141         WebTransportBidirectionalStream::Create(mTransport, mTransport->mGlobal,
 142                                                 std::get<0>(tuple), input,
 143                                                 output, Nothing(), aRv);
 144
 145     // Step 7.2 Enqueue stream to transport.[[IncomingBidirectionalStreams]].
 146     JS::Rooted<JS::Value> jsStream(aCx);
 147     if (MOZ_UNLIKELY(!ToJSValue(aCx, stream, &jsStream))) {
 148       return;
 149     }
 150     LOG(("Enqueuing bidirectional stream\n"));
 151     // EnqueueNative is CAN_RUN_SCRIPT
 152     RefPtr<ReadableStream> incomingStream =
 153         mTransport->mIncomingBidirectionalStreams;
 154     incomingStream->EnqueueNative(aCx, jsStream, aRv);
 155     if (MOZ_UNLIKELY(aRv.Failed())) {
 156       return;
 157     }
 158   }
 159   // Step 7.3: Resolve p with undefined.
 160 }
 161
 162 void WebTransportIncomingStreamsAlgorithms::NotifyIncomingStream() {
 163   if (mUnidirectional == StreamType::Unidirectional) {
 164     LOG(("NotifyIncomingStream: %zu Unidirectional ",
 165          mTransport->mUnidirectionalStreams.Length()));
 166 #ifdef DEBUG
 167     auto number = mTransport->mUnidirectionalStreams.Length();
 168     MOZ_ASSERT(number > 0);
 169 #endif
 170     RefPtr<Promise> promise = mCallback.forget();
 171     if (promise) {
 172       promise->MaybeResolveWithUndefined();
 173     }
 174   } else {
 175     LOG(("NotifyIncomingStream: %zu Bidirectional ",
 176          mTransport->mBidirectionalStreams.Length()));
 177 #ifdef DEBUG
 178     auto number = mTransport->mBidirectionalStreams.Length();
 179     MOZ_ASSERT(number > 0);
 180 #endif
 181     RefPtr<Promise> promise = mCallback.forget();
 182     if (promise) {
 183       promise->MaybeResolveWithUndefined();
 184     }
 185   }
 186 }
 187
 188 void WebTransportIncomingStreamsAlgorithms::NotifyRejectAll() {
 189   // cancel all pulls
 190   LOG(("Cancel all WebTransport Pulls"));
 191   // Ensure we clear the callback before resolving/rejecting it
 192   if (RefPtr<Promise> promise = mCallback.forget()) {
 193     promise->MaybeReject(NS_ERROR_FAILURE);
 194   }
 195 }
 196
 197 }  // namespace mozilla::dom