Bug 1790678: Implement WebTransport CreateUni/BidirectionalStream r=kershaw,saschanaz,webidl

Differential Revision: https://phabricator.services.mozilla.com/D168410

dom/webidl/WebTransport.webidl

@@ -32,6 +32,11 @@ dictionary WebTransportCloseInfo {
   UTF8String reason = "";
 };
 
+/* https://w3c.github.io/webtransport/#uni-stream-options */
+dictionary WebTransportSendStreamOptions {
+  long long? sendOrder = null;
+};
+
 /* https://w3c.github.io/webtransport/#web-transport-stats */
 
 dictionary WebTransportStats {
@@ -76,14 +81,16 @@ interface WebTransport {
   [Throws] readonly attribute WebTransportDatagramDuplexStream datagrams;
 
   [NewObject]
-  Promise<WebTransportBidirectionalStream> createBidirectionalStream();
+  Promise<WebTransportBidirectionalStream> createBidirectionalStream(
+     optional WebTransportSendStreamOptions options = {});
   /* a ReadableStream of WebTransportBidirectionalStream objects */
   readonly attribute ReadableStream incomingBidirectionalStreams;
 
 
   /* XXX spec says this should be WebTransportSendStream */
   [NewObject]
-  Promise<WritableStream> createUnidirectionalStream();
+  Promise<WritableStream> createUnidirectionalStream(
+    optional WebTransportSendStreamOptions options = {});
   /* a ReadableStream of WebTransportReceiveStream objects */
   readonly attribute ReadableStream incomingUnidirectionalStreams;
 };

dom/webtransport/api/WebTransport.cpp

@@ -6,6 +6,7 @@
 
 #include "WebTransport.h"
 
+#include "WebTransportBidirectionalStream.h"
 #include "mozilla/RefPtr.h"
 #include "nsUTF8Utils.h"
 #include "nsIURL.h"
@@ -526,10 +527,61 @@ already_AddRefed<WebTransportDatagramDuplexStream> WebTransport::GetDatagrams(
 }
 
 already_AddRefed<Promise> WebTransport::CreateBidirectionalStream(
-    ErrorResult& aError) {
+    const WebTransportSendStreamOptions& aOptions, ErrorResult& aRv) {
   LOG(("CreateBidirectionalStream() called"));
-  aError.Throw(NS_ERROR_NOT_IMPLEMENTED);
-  return nullptr;
+  // https://w3c.github.io/webtransport/#dom-webtransport-createbidirectionalstream
+  RefPtr<Promise> promise = Promise::CreateInfallible(GetParentObject());
+
+  // Step 2: If transport.[[State]] is "closed" or "failed", return a new
+  // rejected promise with an InvalidStateError.
+  if (mState == WebTransportState::CLOSED ||
+      mState == WebTransportState::FAILED) {
+    aRv.ThrowInvalidStateError("WebTransport close or failed");
+    return nullptr;
+  }
+
+  // Step 3: Let sendOrder be options's sendOrder.
+  Maybe<int64_t> sendOrder;
+  if (!aOptions.mSendOrder.IsNull()) {
+    sendOrder = Some(aOptions.mSendOrder.Value());
+  }
+  // Step 4: Let p be a new promise.
+  // Step 5: Run the following steps in parallel, but abort them whenever
+  // transport’s [[State]] becomes "closed" or "failed", and instead queue
+  // a network task with transport to reject p with an InvalidStateError.
+
+  // Ask the parent to create the stream and send us the DataPipeSender/Receiver
+  // pair
+  mChild->SendCreateBidirectionalStream(
+      sendOrder,
+      [self = RefPtr{this}, promise](
+          BidirectionalStreamResponse&& aPipes) MOZ_CAN_RUN_SCRIPT_BOUNDARY {
+        LOG(("CreateBidirectionalStream response"));
+        // Step 5.2.1: If transport.[[State]] is "closed" or "failed",
+        // reject p with an InvalidStateError and abort these steps.
+        if (self->mState == WebTransportState::CLOSED ||
+            self->mState == WebTransportState::FAILED) {
+          promise->MaybeRejectWithInvalidStateError(
+              "Transport close/errored before CreateBidirectional finished");
+          return;
+        }
+        ErrorResult error;
+        RefPtr<WebTransportBidirectionalStream> newStream =
+            WebTransportBidirectionalStream::Create(
+                self, self->mGlobal,
+                aPipes.get_BidirectionalStream().inStream(),
+                aPipes.get_BidirectionalStream().outStream(), error);
+        LOG(("Returning a bidirectionalStream"));
+        promise->MaybeResolve(newStream);
+      },
+      [self = RefPtr{this}, promise](mozilla::ipc::ResponseRejectReason) {
+        LOG(("CreateBidirectionalStream reject"));
+        promise->MaybeRejectWithInvalidStateError(
+            "Transport close/errored before CreateBidirectional started");
+      });
+
+  // Step 6: return p
+  return promise.forget();
 }
 
 already_AddRefed<ReadableStream> WebTransport::IncomingBidirectionalStreams() {
@@ -537,10 +589,74 @@ already_AddRefed<ReadableStream> WebTransport::IncomingBidirectionalStreams() {
 }
 
 already_AddRefed<Promise> WebTransport::CreateUnidirectionalStream(
-    ErrorResult& aError) {
+    const WebTransportSendStreamOptions& aOptions, ErrorResult& aRv) {
   LOG(("CreateUnidirectionalStream() called"));
-  aError.Throw(NS_ERROR_NOT_IMPLEMENTED);
-  return nullptr;
+  // https://w3c.github.io/webtransport/#dom-webtransport-createunidirectionalstream
+  // Step 2: If transport.[[State]] is "closed" or "failed", return a new
+  // rejected promise with an InvalidStateError.
+  if (mState == WebTransportState::CLOSED ||
+      mState == WebTransportState::FAILED) {
+    aRv.ThrowInvalidStateError("WebTransport close or failed");
+    return nullptr;
+  }
+
+  // Step 3: Let sendOrder be options's sendOrder.
+  Maybe<int64_t> sendOrder;
+  if (!aOptions.mSendOrder.IsNull()) {
+    sendOrder = Some(aOptions.mSendOrder.Value());
+  }
+  // Step 4: Let p be a new promise.
+  RefPtr<Promise> promise = Promise::CreateInfallible(GetParentObject());
+
+  // Step 5: Run the following steps in parallel, but abort them whenever
+  // transport’s [[State]] becomes "closed" or "failed", and instead queue
+  // a network task with transport to reject p with an InvalidStateError.
+
+  // Ask the parent to create the stream and send us the DataPipeSender
+  mChild->SendCreateUnidirectionalStream(
+      sendOrder,
+      [self = RefPtr{this},
+       promise](RefPtr<::mozilla::ipc::DataPipeSender>&& aPipe)
+          MOZ_CAN_RUN_SCRIPT_BOUNDARY {
+            LOG(("CreateUnidirectionalStream response"));
+            // Step 5.1: Let internalStream be the result of creating an
+            // outgoing unidirectional stream with transport.[[Session]].
+            // Step 5.2: Queue a network task with transport to run the
+            // following steps:
+            // Step 5.2.1 If transport.[[State]] is "closed" or "failed",
+            // reject p with an InvalidStateError and abort these steps.
+            if (self->mState == WebTransportState::CLOSED ||
+                self->mState == WebTransportState::FAILED) {
+              promise->MaybeRejectWithInvalidStateError(
+                  "Transport close/errored during CreateUnidirectional");
+              return;
+            }
+
+            // Step 5.2.2.: Let stream be the result of creating a
+            // WebTransportSendStream with internalStream, transport, and
+            // sendOrder.
+            ErrorResult error;
+            RefPtr<WebTransportSendStream> writableStream =
+                WebTransportSendStream::Create(self, self->mGlobal, aPipe,
+                                               error);
+            if (!writableStream) {
+              promise->MaybeReject(std::move(error));
+              return;
+            }
+            LOG(("Returning a writableStream"));
+            // https://w3c.github.io/webtransport/#send-stream-procedures step 7
+            self->mSendStreams.AppendElement(writableStream);
+            // Step 5.2.3: Resolve p with stream.
+            promise->MaybeResolve(writableStream);
+          },
+      [self = RefPtr{this}, promise](mozilla::ipc::ResponseRejectReason) {
+        LOG(("CreateUnidirectionalStream reject"));
+        promise->MaybeRejectWithInvalidStateError(
+            "Transport close/errored during CreateUnidirectional");
+      });
+
+  // Step 6: return p
+  return promise.forget();
 }
 
 already_AddRefed<ReadableStream> WebTransport::IncomingUnidirectionalStreams() {
@@ -600,7 +716,7 @@ void WebTransport::Cleanup(WebTransportError* aError,
   if (aCloseInfo) {
     // 12.1: Resolve closed with closeInfo.
     LOG(("Resolving mClosed with closeinfo"));
-    mClosed->MaybeResolve(aCloseInfo);
+    mClosed->MaybeResolve(*aCloseInfo);
     // 12.2: Assert: ready is settled.
     MOZ_ASSERT(mReady->State() != Promise::PromiseState::Pending);
     // 12.3: Close incomingBidirectionalStreams

dom/webtransport/api/WebTransport.h

@@ -91,10 +91,12 @@ class WebTransport final : public nsISupports, public nsWrapperCache {
                                 ErrorResult& aRv);
   already_AddRefed<WebTransportDatagramDuplexStream> GetDatagrams(
       ErrorResult& aRv);
-  already_AddRefed<Promise> CreateBidirectionalStream(ErrorResult& aError);
+  already_AddRefed<Promise> CreateBidirectionalStream(
+      const WebTransportSendStreamOptions& aOptions, ErrorResult& aRv);
+  already_AddRefed<Promise> CreateUnidirectionalStream(
+      const WebTransportSendStreamOptions& aOptions, ErrorResult& aRv);
   MOZ_CAN_RUN_SCRIPT_BOUNDARY already_AddRefed<ReadableStream>
   IncomingBidirectionalStreams();
-  already_AddRefed<Promise> CreateUnidirectionalStream(ErrorResult& aError);
   MOZ_CAN_RUN_SCRIPT_BOUNDARY already_AddRefed<ReadableStream>
   IncomingUnidirectionalStreams();
 

dom/webtransport/api/WebTransportBidirectionalStream.cpp

@@ -9,6 +9,8 @@
 
 namespace mozilla::dom {
 
+using namespace mozilla::ipc;
+
 NS_IMPL_CYCLE_COLLECTION_WRAPPERCACHE(WebTransportBidirectionalStream, mGlobal,
                                       mReadable, mWritable)
 
@@ -30,6 +32,35 @@ JSObject* WebTransportBidirectionalStream::WrapObject(
   return WebTransportBidirectionalStream_Binding::Wrap(aCx, this, aGivenProto);
 }
 
+// static
+already_AddRefed<WebTransportBidirectionalStream>
+WebTransportBidirectionalStream::Create(WebTransport* aWebTransport,
+                                        nsIGlobalObject* aGlobal,
+                                        DataPipeReceiver* receiver,
+                                        DataPipeSender* sender,
+                                        ErrorResult& aRv) {
+  // https://w3c.github.io/webtransport/#pullbidirectionalstream (and
+  // createBidirectionalStream)
+
+  // Step 7.1: Let stream be the result of creating a
+  // WebTransportBidirectionalStream with internalStream and transport
+  RefPtr<WebTransportReceiveStream> readableStream =
+      WebTransportReceiveStream::Create(aWebTransport, aGlobal, receiver, aRv);
+  if (!readableStream) {
+    return nullptr;
+  }
+  RefPtr<WebTransportSendStream> writableStream =
+      WebTransportSendStream::Create(aWebTransport, aGlobal, sender, aRv);
+  if (!writableStream) {
+    return nullptr;
+    ;
+  }
+
+  auto stream = MakeRefPtr<WebTransportBidirectionalStream>(
+      aGlobal, readableStream, writableStream);
+  return stream.forget();
+}
+
 // WebIDL Interface
 
 }  // namespace mozilla::dom

dom/webtransport/api/WebTransportBidirectionalStream.h

@@ -32,6 +32,11 @@ class WebTransportBidirectionalStream final : public nsISupports,
   NS_DECL_CYCLE_COLLECTING_ISUPPORTS
   NS_DECL_CYCLE_COLLECTION_WRAPPERCACHE_CLASS(WebTransportBidirectionalStream)
 
+  static already_AddRefed<WebTransportBidirectionalStream> Create(
+      WebTransport* aWebTransport, nsIGlobalObject* aGlobal,
+      ::mozilla::ipc::DataPipeReceiver* receiver,
+      ::mozilla::ipc::DataPipeSender* sender, ErrorResult& aRv);
+
   // WebIDL Boilerplate
   nsIGlobalObject* GetParentObject() const;
 

dom/webtransport/api/WebTransportStreams.cpp

@@ -132,23 +132,9 @@ void WebTransportIncomingStreamsAlgorithms::BuildStream(JSContext* aCx,
     RefPtr<DataPipeReceiver> input = pipes->first.forget();
     RefPtr<DataPipeSender> output = pipes->second.forget();
 
-    // Step 7.1: Let stream be the result of creating a
-    // WebTransportBidirectionalStream with internalStream and transport
-    RefPtr<WebTransportReceiveStream> readableStream =
-        WebTransportReceiveStream::Create(mTransport, mTransport->mGlobal,
-                                          input, aRv);
-    if (!readableStream) {
-      return;
-    }
-    RefPtr<WebTransportSendStream> writableStream =
-        WebTransportSendStream::Create(mTransport, mTransport->mGlobal, output,
-                                       aRv);
-    if (!writableStream) {
-      return;
-    }
-
-    auto stream = MakeRefPtr<WebTransportBidirectionalStream>(
-        mTransport->mGlobal, readableStream, writableStream);
+    RefPtr<WebTransportBidirectionalStream> stream =
+        WebTransportBidirectionalStream::Create(mTransport, mTransport->mGlobal,
+                                                input, output, aRv);
 
     // Step 7.2 Enqueue stream to transport.[[IncomingBidirectionalStreams]].
     JS::Rooted<JS::Value> jsStream(aCx);
@@ -163,11 +149,6 @@ void WebTransportIncomingStreamsAlgorithms::BuildStream(JSContext* aCx,
     if (MOZ_UNLIKELY(aRv.Failed())) {
       return;
     }
-    // Add to ReceiveStreams & SendStreams
-    // https://w3c.github.io/webtransport/#send-stream-procedures Step 7:
-    mTransport->mSendStreams.AppendElement(writableStream);
-    // https://w3c.github.io/webtransport/#receive-stream-procedures Step 5:
-    mTransport->mReceiveStreams.AppendElement(readableStream);
   }
   // Step 7.3: Resolve p with undefined.
 }

dom/webtransport/parent/WebTransportParent.cpp

@@ -143,13 +143,155 @@ IPCResult WebTransportParent::RecvClose(const uint32_t& aCode,
   return IPC_OK();
 }
 
+class ReceiveStream final : public nsIWebTransportStreamCallback {
+ public:
+  NS_DECL_THREADSAFE_ISUPPORTS
+  NS_DECL_NSIWEBTRANSPORTSTREAMCALLBACK
+
+  ReceiveStream(
+      WebTransportParent::CreateUnidirectionalStreamResolver&& aResolver,
+      nsCOMPtr<nsISerialEventTarget>& aSocketThread)
+      : mUniResolver(aResolver), mSocketThread(aSocketThread) {}
+  ReceiveStream(
+      WebTransportParent::CreateBidirectionalStreamResolver&& aResolver,
+      nsCOMPtr<nsISerialEventTarget>& aSocketThread)
+      : mBiResolver(aResolver), mSocketThread(aSocketThread) {}
+
+ private:
+  ~ReceiveStream() = default;
+  std::function<void(::mozilla::ipc::DataPipeSender*)> mUniResolver;
+  WebTransportParent::CreateBidirectionalStreamResolver mBiResolver;
+  nsCOMPtr<nsISerialEventTarget> mSocketThread;
+};
+
+NS_IMPL_ISUPPORTS(ReceiveStream, nsIWebTransportStreamCallback)
+
+// nsIWebTransportStreamCallback:
+NS_IMETHODIMP ReceiveStream::OnBidirectionalStreamReady(
+    nsIWebTransportBidirectionalStream* aStream) {
+  LOG(("Bidirectional stream ready!"));
+  MOZ_ASSERT(mSocketThread->IsOnCurrentThread());
+
+  RefPtr<mozilla::ipc::DataPipeSender> inputsender;
+  RefPtr<mozilla::ipc::DataPipeReceiver> inputreceiver;
+  nsresult rv =
+      NewDataPipe(mozilla::ipc::kDefaultDataPipeCapacity,
+                  getter_AddRefs(inputsender), getter_AddRefs(inputreceiver));
+  if (NS_WARN_IF(NS_FAILED(rv))) {
+    mBiResolver(rv);
+    return rv;
+  }
+
+  nsCOMPtr<nsIAsyncInputStream> inputStream;
+  aStream->GetInputStream(getter_AddRefs(inputStream));
+  MOZ_ASSERT(inputStream);
+  rv = NS_AsyncCopy(inputStream, inputsender, mSocketThread,
+                    NS_ASYNCCOPY_VIA_WRITESEGMENTS,  // can we use READSEGMENTS?
+                    mozilla::ipc::kDefaultDataPipeCapacity, nullptr, nullptr);
+  if (NS_WARN_IF(NS_FAILED(rv))) {
+    mBiResolver(rv);
+    return rv;
+  }
+
+  RefPtr<mozilla::ipc::DataPipeSender> outputsender;
+  RefPtr<mozilla::ipc::DataPipeReceiver> outputreceiver;
+  rv =
+      NewDataPipe(mozilla::ipc::kDefaultDataPipeCapacity,
+                  getter_AddRefs(outputsender), getter_AddRefs(outputreceiver));
+  if (NS_WARN_IF(NS_FAILED(rv))) {
+    mBiResolver(rv);
+    return rv;
+  }
+
+  nsCOMPtr<nsIAsyncOutputStream> outputStream;
+  aStream->GetOutputStream(getter_AddRefs(outputStream));
+  MOZ_ASSERT(outputStream);
+  rv = NS_AsyncCopy(outputreceiver, outputStream, mSocketThread,
+                    NS_ASYNCCOPY_VIA_READSEGMENTS,
+                    mozilla::ipc::kDefaultDataPipeCapacity, nullptr, nullptr);
+  if (NS_WARN_IF(NS_FAILED(rv))) {
+    mBiResolver(rv);
+    return rv;
+  }
+
+  LOG(("Returning BidirectionalStream pipe to content"));
+  mBiResolver(BidirectionalStream(inputreceiver, outputsender));
+  return NS_OK;
+}
+
+NS_IMETHODIMP
+ReceiveStream::OnUnidirectionalStreamReady(nsIWebTransportSendStream* aStream) {
+  LOG(("Unidirectional stream ready!"));
+  // We should be on the Socket Thread
+  MOZ_ASSERT(mSocketThread->IsOnCurrentThread());
+
+  RefPtr<::mozilla::ipc::DataPipeSender> sender;
+  RefPtr<::mozilla::ipc::DataPipeReceiver> receiver;
+  nsresult rv = NewDataPipe(mozilla::ipc::kDefaultDataPipeCapacity,
+                            getter_AddRefs(sender), getter_AddRefs(receiver));
+  if (NS_WARN_IF(NS_FAILED(rv))) {
+    mUniResolver(nullptr);
+    return rv;
+  }
+
+  nsCOMPtr<nsIAsyncOutputStream> outputStream;
+  aStream->GetOutputStream(getter_AddRefs(outputStream));
+  MOZ_ASSERT(outputStream);
+  rv = NS_AsyncCopy(receiver, outputStream, mSocketThread,
+                    NS_ASYNCCOPY_VIA_READSEGMENTS,
+                    mozilla::ipc::kDefaultDataPipeCapacity, nullptr, nullptr);
+  if (NS_WARN_IF(NS_FAILED(rv))) {
+    mUniResolver(nullptr);
+    return rv;
+  }
+
+  LOG(("Returning UnidirectionalStream pipe to content"));
+  // pass the DataPipeSender to the content process
+  mUniResolver(sender);
+  return NS_OK;
+}
+
+JS_HAZ_CAN_RUN_SCRIPT NS_IMETHODIMP ReceiveStream::OnError(uint8_t aError) {
+  nsresult rv = aError == nsIWebTransport::INVALID_STATE_ERROR
+                    ? NS_ERROR_DOM_INVALID_STATE_ERR
+                    : NS_ERROR_FAILURE;
+  if (mUniResolver) {
+    mUniResolver(nullptr);
+  } else if (mBiResolver) {
+    mBiResolver(rv);
+  }
+  return NS_OK;
+}
+
 IPCResult WebTransportParent::RecvCreateUnidirectionalStream(
-    CreateUnidirectionalStreamResolver&& aResolver) {
+    Maybe<int64_t> aSendOrder, CreateUnidirectionalStreamResolver&& aResolver) {
+  LOG(("%s for %p received, useSendOrder=%d, sendOrder=%" PRIi64, __func__,
+       this, aSendOrder.isSome(),
+       aSendOrder.isSome() ? aSendOrder.value() : 0));
+
+  RefPtr<ReceiveStream> callback =
+      new ReceiveStream(std::move(aResolver), mSocketThread);
+  nsresult rv;
+  rv = mWebTransport->CreateOutgoingUnidirectionalStream(callback);
+  if (NS_FAILED(rv)) {
+    callback->OnError(0);  // XXX
+  }
   return IPC_OK();
 }
 
 IPCResult WebTransportParent::RecvCreateBidirectionalStream(
-    CreateBidirectionalStreamResolver&& aResolver) {
+    Maybe<int64_t> aSendOrder, CreateBidirectionalStreamResolver&& aResolver) {
+  LOG(("%s for %p received, useSendOrder=%d, sendOrder=%" PRIi64, __func__,
+       this, aSendOrder.isSome(),
+       aSendOrder.isSome() ? aSendOrder.value() : 0));
+
+  RefPtr<ReceiveStream> callback =
+      new ReceiveStream(std::move(aResolver), mSocketThread);
+  nsresult rv;
+  rv = mWebTransport->CreateOutgoingBidirectionalStream(callback);
+  if (NS_FAILED(rv)) {
+    callback->OnError(0);  // XXX
+  }
   return IPC_OK();
 }
 

dom/webtransport/parent/WebTransportParent.h

@@ -40,9 +40,10 @@ class WebTransportParent : public PWebTransportParent,
   IPCResult RecvClose(const uint32_t& aCode, const nsACString& aReason);
 
   IPCResult RecvCreateUnidirectionalStream(
+      Maybe<int64_t> aSendOrder,
       CreateUnidirectionalStreamResolver&& aResolver);
   IPCResult RecvCreateBidirectionalStream(
-      CreateBidirectionalStreamResolver&& aResolver);
+      Maybe<int64_t> aSendOrder, CreateBidirectionalStreamResolver&& aResolver);
 
   void ActorDestroy(ActorDestroyReason aWhy) override;
 

dom/webtransport/shared/PWebTransport.ipdl

@@ -10,10 +10,15 @@ include PBackgroundSharedTypes;
 namespace mozilla {
 namespace dom {
 struct BidirectionalStream {
-  DataPipeReceiver incoming;
+  DataPipeReceiver inStream;
   DataPipeSender outStream;
 };
 
+union BidirectionalStreamResponse {
+  nsresult;
+  BidirectionalStream;
+};
+
 async protocol PWebTransport
 {
  parent:
@@ -21,10 +26,10 @@ async protocol PWebTransport
    * TODO: documentation
    */
   async Close(uint32_t code, nsCString reason);
-  async CreateUnidirectionalStream()
+  async CreateUnidirectionalStream(int64_t? sendOrder)
       returns(DataPipeSender sender);
-  async CreateBidirectionalStream()
-      returns(BidirectionalStream pipes);
+  async CreateBidirectionalStream(int64_t? sendOrder)
+      returns(BidirectionalStreamResponse response);
 
  child: