Issue #1442 - WIP: Response.body handling.fetchstreams-work

This doesn't build due to unknown code state at Mozilla. The BZ bug landed hg
commit has typoes in it so can't have been what landed because it wouldn't be
compile-able. hg history lists a backup but no re-land on the JS part as well
that doesn't make sense.

On top, further work apparently involved 100 refactoring commits to come to
Mozilla's final state. I'm signing off here.

9 files changed, 385 insertions, 60 deletions

diff --git a/dom/bindings/Bindings.conf b/dom/bindings/Bindings.conf
index 4e94a7bcdf..0e2cd7b8ec 100644
--- a/dom/bindings/Bindings.conf
+++ b/dom/bindings/Bindings.conf
@@ -734,7 +734,8 @@ DOMInterfaces = {
 
 'Response': {
     'binaryNames': { 'headers': 'headers_' },
-    'implicitJSContext': [ 'arrayBuffer', 'blob', 'formData', 'json', 'text' ],
+    'implicitJSContext': [ 'arrayBuffer', 'blob', 'formData', 'json', 'text',
+                           'clone', 'cloneUnfiltered' ],
 },
 
 'RGBColor': {
diff --git a/dom/fetch/Fetch.cpp b/dom/fetch/Fetch.cpp
index 8f67c8f385..dbc940ca56 100644
--- a/dom/fetch/Fetch.cpp
+++ b/dom/fetch/Fetch.cpp
@@ -843,7 +843,7 @@ ExtractFromURLSearchParams(const URLSearchParams& aParams,
 } // namespace
 
 nsresult
-ExtractByteStreamFromBody(const OwningArrayBufferOrArrayBufferViewOrBlobOrFormDataOrUSVStringOrURLSearchParams& aBodyInit,
+ExtractByteStreamFromBody(const fetch::OwningBodyInit& aBodyInit,
                           nsIInputStream** aStream,
                           nsCString& aContentType,
                           uint64_t& aContentLength)
@@ -881,7 +881,7 @@ ExtractByteStreamFromBody(const OwningArrayBufferOrArrayBufferViewOrBlobOrFormDa
 }
 
 nsresult
-ExtractByteStreamFromBody(const ArrayBufferOrArrayBufferViewOrBlobOrFormDataOrUSVStringOrURLSearchParams& aBodyInit,
+ExtractByteStreamFromBody(const fetch::BodyInit& aBodyInit,
                           nsIInputStream** aStream,
                           nsCString& aContentType,
                           uint64_t& aContentLength)
@@ -919,6 +919,50 @@ ExtractByteStreamFromBody(const ArrayBufferOrArrayBufferViewOrBlobOrFormDataOrUS
   return NS_ERROR_FAILURE;
 }
 
+nsresult
+ExtractByteStreamFromBody(const fetch::ResponseBodyInit& aBodyInit,
+                          nsIInputStream** aStream,
+                          nsCString& aContentType,
+                          uint64_t& aContentLength)
+{
+  MOZ_ASSERT(aStream);
+  MOZ_ASSERT(!*aStream);
+
+  // ReadableStreams should be handled by
+  // BodyExtractorReadableStream::GetAsStream.
+  MOZ_ASSERT(!aBodyInit.IsReadableStream());
+
+  if (aBodyInit.IsArrayBuffer()) {
+    const ArrayBuffer& buf = aBodyInit.GetAsArrayBuffer();
+    return ExtractFromArrayBuffer(buf, aStream, aContentLength);
+  }
+  if (aBodyInit.IsArrayBufferView()) {
+    const ArrayBufferView& buf = aBodyInit.GetAsArrayBufferView();
+    return ExtractFromArrayBufferView(buf, aStream, aContentLength);
+  }
+  if (aBodyInit.IsBlob()) {
+    const Blob& blob = aBodyInit.GetAsBlob();
+    return ExtractFromBlob(blob, aStream, aContentType, aContentLength);
+  }
+  if (aBodyInit.IsFormData()) {
+    FormData& form = aBodyInit.GetAsFormData();
+    return ExtractFromFormData(form, aStream, aContentType, aContentLength);
+  }
+  if (aBodyInit.IsUSVString()) {
+    nsAutoString str;
+    str.Assign(aBodyInit.GetAsUSVString());
+    return ExtractFromUSVString(str, aStream, aContentType, aContentLength);
+  }
+  if (aBodyInit.IsURLSearchParams()) {
+    URLSearchParams& params = aBodyInit.GetAsURLSearchParams();
+    return ExtractFromURLSearchParams(params, aStream, aContentType, aContentLength);
+  }
+
+  NS_NOTREACHED("Should never reach here");
+  return NS_ERROR_FAILURE;
+}
+
+
 template <class Derived>
 FetchBody<Derived>::FetchBody()
   : mWorkerPrivate(nullptr)
@@ -962,7 +1006,8 @@ FetchBody<Derived>::BodyUsed() const
 
     JS::Rooted<JSObject*> body(cx, mReadableStreamBody);
     if (JS::ReadableStreamIsDisturbed(body) ||
-        JS::ReadableStreamIsLocked(body)) {
+        JS::ReadableStreamIsLocked(body) ||
+        !JS::ReadableStreamIsReadable(body)) {
       return true;
     }
   }
@@ -995,14 +1040,18 @@ FetchBody<Derived>::ConsumeBody(JSContext* aCx, FetchConsumeType aType, ErrorRes
 
   SetBodyUsed();
 
-  nsCOMPtr<nsIGlobalObject> global = DerivedClass()->GetParentObject();
-
-  // If we already created a ReadableStreamBody we have to close it now.
+  // If we already created a ReadableStreamBody we have to lock it now because
+  // it may have been shared with other objects..
   if (mReadableStreamBody) {
     JS::Rooted<JSObject*> body(aCx, mReadableStreamBody);
-    JS::ReadableStreamClose(aCx, body);
+    LockStream(aCx, body, aRv);
+    if (NS_WARN_IF(aRv.Failed())) {
+      return nullptr;
+    }
   }
 
+  nsCOMPtr<nsIGlobalObject> global = DerivedClass()->GetParentObject();
+
   RefPtr<Promise> promise =
     FetchBodyConsumer<Derived>::Create(global, this, signal, aType, aRv);
   if (NS_WARN_IF(aRv.Failed())) {
@@ -1050,6 +1099,23 @@ FetchBody<Response>::SetMimeType();
 
 template <class Derived>
 void
+FetchBody<Derived>::SetReadableStreamBody(JSObject* aBody)
+{
+  MOZ_ASSERT(!mReadableStreamBody);
+  MOZ_ASSERT(aBody);
+  mReadableStreamBody = aBody;
+}
+
+template
+void
+FetchBody<Request>::SetReadableStreamBody(JSObject* aBody);
+
+template
+void
+FetchBody<Response>::SetReadableStreamBody(JSObject* aBody);
+
+template <class Derived>
+void
 FetchBody<Derived>::GetBody(JSContext* aCx,
                             JS::MutableHandle<JSObject*> aBodyOut,
                             ErrorResult& aRv)
@@ -1075,9 +1141,11 @@ FetchBody<Derived>::GetBody(JSContext* aCx,
     MOZ_ASSERT(body);
 
     // If the body has been already consumed, we close the stream.
-    if (BodyUsed() && !JS::ReadableStreamClose(aCx, body)) {
-      aRv.StealExceptionFromJSContext(aCx);
-      return;
+    if (BodyUsed()) {
+      LockStream(aCx, body, aRv);
+      if (NS_WARN_IF(aRv.Failed())) {
+        return;
+      }
     }
 
     mReadableStreamBody = body;
@@ -1098,6 +1166,72 @@ FetchBody<Response>::GetBody(JSContext* aCx,
                              JS::MutableHandle<JSObject*> aMessage,
                              ErrorResult& aRv);
 
+template <class Derived>
+void
+FetchBody<Derived>::LockStream(JSContext* aCx,
+                               JS::HandleObject aStream,
+                               ErrorResult& aRv)
+{
+  // XXXMC: TODO
+}
+
+template
+void
+FetchBody<Request>::LockStream(JSContext* aCx,
+                               JS::HandleObject aStream,
+                               ErrorResult& aRv);
+
+template
+void
+FetchBody<Response>::LockStream(JSContext* aCx,
+                                JS::HandleObject aStream,
+                                ErrorResult& aRv);
+
+template <class Derived>
+void
+FetchBody<Derived>::MaybeTeeReadableStreamBody(JSContext* aCx,
+                                               JS::MutableHandle<JSObject*> aBodyOut,
+                                               ErrorResult& aRv)
+{
+  MOZ_DIAGNOSTIC_ASSERT(!BodyUsed());
+
+  if (!mReadableStreamBody) {
+    return;
+  }
+
+  JS::Rooted<JSObject*> stream(aCx, mReadableStreamBody);
+
+  // If this is a ReadableStream with an external source, this has been
+  // generated by a Fetch. In this case, Fetch will be able to recreate it
+  // again when GetBody() is called.
+  if (JS::ReadableStreamGetMode(stream) == JS::ReadableStreamMode::ExternalSource) {
+    aBodyOut.set(nullptr);
+    return;
+  }
+
+  JS::Rooted<JSObject*> branch1(aCx);
+  JS::Rooted<JSObject*> branch2(aCx);
+
+  if (!JS::ReadableStreamTee(aCx, stream, &branch1, &branch2)) {
+    aRv.StealExceptionFromJSContext(aCx);
+    return;
+  }
+
+  mReadableStreamBody = branch1;
+  aBodyOut.set(branch2);
+}
+
+template
+void
+FetchBody<Request>::MaybeTeeReadableStreamBody(JSContext* aCx,
+                                               JS::MutableHandle<JSObject*> aMessage,
+                                               ErrorResult& aRv);
+
+template
+void
+FetchBody<Response>::MaybeTeeReadableStreamBody(JSContext* aCx,
+                                                JS::MutableHandle<JSObject*> aMessage,
+                                                ErrorResult& aRv);
 
 } // namespace dom
 } // namespace mozilla
diff --git a/dom/fetch/Fetch.h b/dom/fetch/Fetch.h
index c605e45bc5..cac72ede81 100644
--- a/dom/fetch/Fetch.h
+++ b/dom/fetch/Fetch.h
@@ -25,9 +25,11 @@ namespace mozilla {
 namespace dom {

 

 class ArrayBufferOrArrayBufferViewOrBlobOrFormDataOrUSVStringOrURLSearchParams;

+class BlobOrArrayBufferViewOrArrayBufferOrFormDataOrURLSearchParamsOrReadableStreamOrUSVString;

 class BlobImpl;

 class InternalRequest;

 class OwningArrayBufferOrArrayBufferViewOrBlobOrFormDataOrUSVStringOrURLSearchParams;

+struct ReadableStream;

 class RequestOrUSVString;

 

 namespace workers {

@@ -41,13 +43,20 @@ FetchRequest(nsIGlobalObject* aGlobal, const RequestOrUSVString& aInput,
 nsresult

 UpdateRequestReferrer(nsIGlobalObject* aGlobal, InternalRequest* aRequest);

 

+/* Deal with unwieldy long webIDL-generated type names */

+namespace fetch {

+  typedef ArrayBufferOrArrayBufferViewOrBlobOrFormDataOrUSVStringOrURLSearchParams BodyInit;

+  typedef BlobOrArrayBufferViewOrArrayBufferOrFormDataOrURLSearchParamsOrReadableStreamOrUSVString ResponseBodyInit;

+  typedef OwningArrayBufferOrArrayBufferViewOrBlobOrFormDataOrUSVStringOrURLSearchParams OwningBodyInit;

+};

+

 /*

  * Creates an nsIInputStream based on the fetch specifications 'extract a byte

  * stream algorithm' - http://fetch.spec.whatwg.org/#concept-bodyinit-extract.

  * Stores content type in out param aContentType.

  */

 nsresult

-ExtractByteStreamFromBody(const OwningArrayBufferOrArrayBufferViewOrBlobOrFormDataOrUSVStringOrURLSearchParams& aBodyInit,

+ExtractByteStreamFromBody(const fetch::OwningBodyInit& aBodyInit,

                           nsIInputStream** aStream,

                           nsCString& aContentType,

                           uint64_t& aContentLength);

@@ -56,7 +65,17 @@ ExtractByteStreamFromBody(const OwningArrayBufferOrArrayBufferViewOrBlobOrFormDa
  * Non-owning version.

  */

 nsresult

-ExtractByteStreamFromBody(const ArrayBufferOrArrayBufferViewOrBlobOrFormDataOrUSVStringOrURLSearchParams& aBodyInit,

+ExtractByteStreamFromBody(const fetch::BodyInit& aBodyInit,

+                          nsIInputStream** aStream,

+                          nsCString& aContentType,

+                          uint64_t& aContentLength);

+

+/*

+ * Non-owning version. This method should go away when BodyInit will contain

+ * ReadableStream.

+ */

+nsresult

+ExtractByteStreamFromBody(const fetch::ResponseBodyInit& aBodyInit,

                           nsIInputStream** aStream,

                           nsCString& aContentType,

                           uint64_t& aContentLength);

@@ -72,6 +91,15 @@ enum FetchConsumeType
   CONSUME_TEXT,

 };

 

+class FetchStreamHolder

+{

+public:

+  NS_IMETHOD_(MozExternalRefCountType) AddRef(void) = 0;

+  NS_IMETHOD_(MozExternalRefCountType) Release(void) = 0;

+

+  virtual void NullifyStream() = 0;

+};

+

 /*

  * FetchBody's body consumption uses nsIInputStreamPump to read from the

  * underlying stream to a block of memory, which is then adopted by

@@ -106,14 +134,11 @@ enum FetchConsumeType
  * The pump is always released on the main thread.

  */

 template <class Derived>

-class FetchBody

+class FetchBody : public FetchStreamHolder

 {

 public:

   friend class FetchBodyConsumer<Derived>;

 

-  NS_IMETHOD_(MozExternalRefCountType) AddRef(void) = 0;

-  NS_IMETHOD_(MozExternalRefCountType) Release(void) = 0;

-

   bool

   BodyUsed() const;

 

@@ -152,6 +177,13 @@ public:
           JS::MutableHandle<JSObject*> aBodyOut,

           ErrorResult& aRv);

 

+  // If the body contains a ReadableStream body object, this method produces a

+  // tee() of it.

+  void

+  MaybeTeeReadableStreamBody(JSContext* aCx,

+                             JS::MutableHandle<JSObject*> aBodyOut,

+                             ErrorResult& aRv);

+

   // Utility public methods accessed by various runnables.

 

   void

@@ -166,6 +198,12 @@ public:
     return mMimeType;

   }

 

+  void

+  NullifyStream() override

+  {

+    mReadableStreamBody = nullptr;

+  }

+

   virtual AbortSignal*

   GetSignal() const = 0;

 

@@ -182,6 +220,10 @@ protected:
 

   void

   SetMimeType();

+

+  void

+  SetReadableStreamBody(JSObject* aBody);

+

 private:

   Derived*

   DerivedClass() const

@@ -192,6 +234,9 @@ private:
   already_AddRefed<Promise>

   ConsumeBody(JSContext* aCx, FetchConsumeType aType, ErrorResult& aRv);

 

+  void

+  LockStream(JSContext* aCx, JS::HandleObject aStream, ErrorResult& aRv);

+

   bool

   IsOnTargetThread()

   {

diff --git a/dom/fetch/FetchStream.cpp b/dom/fetch/FetchStream.cpp
index e57991fcc2..94837504f1 100644
--- a/dom/fetch/FetchStream.cpp
+++ b/dom/fetch/FetchStream.cpp
@@ -60,10 +60,12 @@ class FetchStreamWorkerHolderShutdown final : public WorkerControlRunnable
 public:
   FetchStreamWorkerHolderShutdown(WorkerPrivate* aWorkerPrivate,
                                   UniquePtr<WorkerHolder>&& aHolder,
-                                  nsCOMPtr<nsIGlobalObject>&& aGlobal)
-    : WorkerControlRunnable(aWorkerPrivate)
+                                  nsCOMPtr<nsIGlobalObject>&& aGlobal,
+                                  RefPtr<FetchStreamHolder>&& aStreamHolder)
+    : WorkerControlRunnable(aWorkerPrivate, WorkerThreadUnchangedBusyCount)
     , mHolder(Move(aHolder))
     , mGlobal(Move(aGlobal))
+    , mStreamHolder(Move(aStreamHolder))
   {}
 
   bool
@@ -71,12 +73,30 @@ public:
   {
     mHolder = nullptr;
     mGlobal = nullptr;
+
+    mStreamHolder->NullifyStream();
+    mStreamHolder = nullptr;
+
+    return true;
+  }
+
+  // This runnable starts from a JS Thread. We need to disable a couple of
+  // assertions by overriding the following methods.
+
+  bool
+  PreDispatch(WorkerPrivate* aWorkerPrivate) override
+  {
     return true;
   }
 
+  void
+  PostDispatch(WorkerPrivate* aWorkerPrivate, bool aDispatchResult) override
+  {}
+
 private:
   UniquePtr<WorkerHolder> mHolder;
   nsCOMPtr<nsIGlobalObject> mGlobal;
+  RefPtr<FetchStreamHolder> mStreamHolder;
 };
 
 } // anonymous
@@ -85,13 +105,15 @@ NS_IMPL_ISUPPORTS(FetchStream, nsIInputStreamCallback, nsIObserver,
                   nsISupportsWeakReference)
 
 /* static */ JSObject*
-FetchStream::Create(JSContext* aCx, nsIGlobalObject* aGlobal,
-                    nsIInputStream* aInputStream, ErrorResult& aRv)
+FetchStream::Create(JSContext* aCx, FetchStreamHolder* aStreamHolder,
+                    nsIGlobalObject* aGlobal, nsIInputStream* aInputStream,
+                    ErrorResult& aRv)
 {
   MOZ_DIAGNOSTIC_ASSERT(aCx);
   MOZ_DIAGNOSTIC_ASSERT(aInputStream);
+  MOZ_DIAGNOSTIC_ASSERT(aStreamHolder);
 
-  RefPtr<FetchStream> stream = new FetchStream(aGlobal, aInputStream);
+  RefPtr<FetchStream> stream = new FetchStream(aGlobal, aStreamHolder, aInputStream);
 
   if (NS_IsMainThread()) {
     nsCOMPtr<nsIObserverService> os = mozilla::services::GetObserverService();
@@ -226,7 +248,7 @@ FetchStream::RequestDataCallback(JSContext* aCx,
   MOZ_DIAGNOSTIC_ASSERT(!stream->mOriginalInputStream);
 
   nsresult rv =
-    stream->mInputStream->AsyncWait(stream, 0, 0, nullptr);
+    stream->mInputStream->AsyncWait(stream, 0, 0, stream->mOwningEventTarget);
   if (NS_WARN_IF(NS_FAILED(rv))) {
     stream->ErrorPropagation(aCx, aStream, rv);
     return;
@@ -264,12 +286,11 @@ FetchStream::WriteIntoReadRequestCallback(JSContext* aCx,
   *aByteWritten = written;
 
   if (written == 0) {
-    stream->mState = eClosed;
-    JS::ReadableStreamClose(aCx, aStream);
+    stream->CloseAndReleaseObjects(aCx, aStream);
     return;
   }
 
-  rv = stream->mInputStream->AsyncWait(stream, 0, 0, nullptr);
+  rv = stream->mInputStream->AsyncWait(stream, 0, 0, stream->mOwningEventTarget);
   if (NS_WARN_IF(NS_FAILED(rv))) {
     stream->ErrorPropagation(aCx, aStream, rv);
     return;
@@ -292,7 +313,7 @@ FetchStream::CancelCallback(JSContext* aCx, JS::HandleObject aStream,
     stream->mInputStream->CloseWithStatus(NS_BASE_STREAM_CLOSED);
   }
 
-  stream->mState = eClosed;
+  stream->ReleaseObjects();
 
   return JS::UndefinedValue();
 }
@@ -325,22 +346,21 @@ FetchStream::FinalizeCallback(void* aUnderlyingSource, uint8_t aFlags)
   RefPtr<FetchStream> stream =
     dont_AddRef(static_cast<FetchStream*>(aUnderlyingSource));
 
-  if (stream->mState == eClosed) {
-    return;
-  }
-
-  stream->CloseAndReleaseObjects();
+  stream->ReleaseObjects();
 }
 
 FetchStream::FetchStream(nsIGlobalObject* aGlobal,
+                         FetchStreamHolder* aStreamHolder,
                          nsIInputStream* aInputStream)
   : mState(eWaiting)
   , mGlobal(aGlobal)
+  , mStreamHolder(aStreamHolder)
   , mOriginalInputStream(aInputStream)
   , mOwningEventTarget(nullptr)
   , mReadableStream(nullptr)
 {
   MOZ_DIAGNOSTIC_ASSERT(aInputStream);
+  MOZ_DIAGNOSTIC_ASSERT(aStreamHolder);
 }
 
 FetchStream::~FetchStream()
@@ -356,12 +376,9 @@ FetchStream::ErrorPropagation(JSContext* aCx, JS::HandleObject aStream,
     return;
   }
 
-  // We cannot continue with any other operation.
-  mState = eClosed;
-
   // Let's close the stream.
   if (aError == NS_BASE_STREAM_CLOSED) {
-    JS::ReadableStreamClose(aCx, aStream);
+    CloseAndReleaseObjects(aCx, aStream);
     return;
   }
 
@@ -428,6 +445,27 @@ FetchStream::OnInputStreamReady(nsIAsyncInputStream* aStream)
   return NS_OK;
 }
 
+/* static */ nsresult
+FetchStream::RetrieveInputStream(void* aUnderlyingReadableStreamSource,
+                                 nsIInputStream** aInputStream)
+{
+  MOZ_ASSERT(aUnderlyingReadableStreamSource);
+  MOZ_ASSERT(aInputStream);
+
+  RefPtr<FetchStream> stream =
+    static_cast<FetchStream*>(aUnderlyingReadableStreamSource);
+
+  // if mOriginalInputStream is null, the reading already started. We don't want
+  // to expose the internal inputStream.
+  if (NS_WARN_IF(!stream->mOriginalInputStream)) {
+    return NS_ERROR_DOM_INVALID_STATE_ERR;
+  }
+
+  nsCOMPtr<nsIInputStream> inputStream = stream->mOriginalInputStream;
+  inputStream.forget(aInputStream);
+  return NS_OK;
+}
+
 void
 FetchStream::Close()
 {
@@ -437,27 +475,42 @@ FetchStream::Close()
 
   AutoJSAPI jsapi;
   if (NS_WARN_IF(!jsapi.Init(mGlobal))) {
+    ReleaseObjects();
     return;
   }
 
   JSContext* cx = jsapi.cx();
   JS::Rooted<JSObject*> stream(cx, mReadableStream);
-  JS::ReadableStreamClose(cx, stream);
 
-  CloseAndReleaseObjects();
+  CloseAndReleaseObjects(cx, stream);
 }
 
 void
-FetchStream::CloseAndReleaseObjects()
+FetchStream::CloseAndReleaseObjects(JSContext* aCx, JS::HandleObject aStream)
 {
   MOZ_DIAGNOSTIC_ASSERT(mState != eClosed);
+
+  ReleaseObjects();
+
+  if (JS::ReadableStreamIsReadable(aStream)) {
+    JS::ReadableStreamClose(aCx, aStream);
+  }
+}
+
+void
+FetchStream::ReleaseObjects()
+{
+  if (mState == eClosed) {
+    return;
+  }
+
   mState = eClosed;
 
   if (mWorkerHolder) {
     RefPtr<FetchStreamWorkerHolderShutdown> r =
       new FetchStreamWorkerHolderShutdown(
         static_cast<FetchStreamWorkerHolder*>(mWorkerHolder.get())->GetWorkerPrivate(),
-        Move(mWorkerHolder), Move(mGlobal));
+        Move(mWorkerHolder), Move(mGlobal), Move(mStreamHolder));
     r->Dispatch();
   } else {
     RefPtr<FetchStream> self = this;
@@ -468,6 +521,9 @@ FetchStream::CloseAndReleaseObjects()
           os->RemoveObserver(self, DOM_WINDOW_DESTROYED_TOPIC);
         }
         self->mGlobal = nullptr;
+
+        self->mStreamHolder->NullifyStream();
+        self->mStreamHolder = nullptr;
       });
 
     NS_DispatchToMainThread(r);
diff --git a/dom/fetch/FetchStream.h b/dom/fetch/FetchStream.h
index b18b7356ff..b27588eda0 100644
--- a/dom/fetch/FetchStream.h
+++ b/dom/fetch/FetchStream.h
@@ -24,6 +24,8 @@ namespace workers {
 class WorkerHolder;
 }
 
+class FetchStreamHolder;
+
 class FetchStream final : public nsIInputStreamCallback
                         , public nsIObserver
                         , public nsSupportsWeakReference
@@ -34,14 +36,20 @@ public:
   NS_DECL_NSIOBSERVER
 
   static JSObject*
-  Create(JSContext* aCx, nsIGlobalObject* aGlobal,
-         nsIInputStream* aInputStream, ErrorResult& aRv);
+  Create(JSContext* aCx, FetchStreamHolder* aStreamHolder,
+         nsIGlobalObject* aGlobal, nsIInputStream* aInputStream,
+         ErrorResult& aRv);
 
   void
   Close();
 
+  static nsresult
+  RetrieveInputStream(void* aUnderlyingReadableStreamSource,
+                      nsIInputStream** aInputStream);
 private:
-  FetchStream(nsIGlobalObject* aGlobal, nsIInputStream* aInputStream);
+  FetchStream(nsIGlobalObject* aGlobal,
+              FetchStreamHolder* aStreamHolder,
+              nsIInputStream* aInputStream);
   ~FetchStream();
 
   static void
@@ -76,7 +84,10 @@ private:
   ErrorPropagation(JSContext* aCx, JS::HandleObject aStream, nsresult aRv);
 
   void
-  CloseAndReleaseObjects();
+  CloseAndReleaseObjects(JSContext* aCx, JS::HandleObject aStream);
+
+  void
+  ReleaseObjects();
 
   // Common methods
 
@@ -104,6 +115,8 @@ private:
   State mState;
 
   nsCOMPtr<nsIGlobalObject> mGlobal;
+  RefPtr<FetchStreamHolder> mStreamHolder;
+  nsCOMPtr<nsIEventTarget> mOwningEventTarget;
 
   // This is the original inputStream received during the CTOR. It will be
   // converted into an nsIAsyncInputStream and stored into mInputStream at the
@@ -111,8 +124,6 @@ private:
   nsCOMPtr<nsIInputStream> mOriginalInputStream;
   nsCOMPtr<nsIAsyncInputStream> mInputStream;
 
-  nsCOMPtr<nsIEventTarget> mOwningEventTarget;
-
   UniquePtr<workers::WorkerHolder> mWorkerHolder;
 
   JS::Heap<JSObject*> mReadableStream;
diff --git a/dom/fetch/Response.cpp b/dom/fetch/Response.cpp
index de38fa3e53..bc3d543663 100644
--- a/dom/fetch/Response.cpp
+++ b/dom/fetch/Response.cpp
@@ -12,12 +12,14 @@
 
 #include "mozilla/ErrorResult.h"
 #include "mozilla/dom/FetchBinding.h"
+#include "mozilla/dom/ResponseBinding.h"
 #include "mozilla/dom/Headers.h"
 #include "mozilla/dom/Promise.h"
 #include "mozilla/dom/URL.h"
 
 #include "nsDOMString.h"
 
+#include "FetchStream.h"
 #include "InternalResponse.h"
 #include "WorkerPrivate.h"
 
@@ -129,7 +131,7 @@ Response::Redirect(const GlobalObject& aGlobal, const nsAString& aUrl,
     return nullptr;
   }
 
-  Optional<Nullable<ArrayBufferOrArrayBufferViewOrBlobOrFormDataOrUSVStringOrURLSearchParams>> body;
+  Optional<Nullable<fetch::ResponseBodyInit>> body;
   ResponseInit init;
   init.mStatus = aStatus;
   RefPtr<Response> r = Response::Constructor(aGlobal, body, init, aRv);
@@ -150,7 +152,7 @@ Response::Redirect(const GlobalObject& aGlobal, const nsAString& aUrl,
 
 /*static*/ already_AddRefed<Response>
 Response::Constructor(const GlobalObject& aGlobal,
-                      const Optional<Nullable<ArrayBufferOrArrayBufferViewOrBlobOrFormDataOrUSVStringOrURLSearchParams>>& aBody,
+                      const Optional<Nullable<fetch::ResponseBodyInit>>& aBody,
                       const ResponseInit& aInit, ErrorResult& aRv)
 {
   nsCOMPtr<nsIGlobalObject> global = do_QueryInterface(aGlobal.GetAsSupports());
@@ -225,13 +227,56 @@ Response::Constructor(const GlobalObject& aGlobal,
     nsCOMPtr<nsIInputStream> bodyStream;
     nsCString contentType;
     uint64_t bodySize = 0;
-    aRv = ExtractByteStreamFromBody(aBody.Value().Value(),
-                                    getter_AddRefs(bodyStream),
-                                    contentType,
-                                    bodySize);
-    if (NS_WARN_IF(aRv.Failed())) {
-      return nullptr;
+    
+    if (aBody.Value().IsReadableStream()) {
+      const ReadableStream& readableStream =
+        aBody.Value().GetAsReadableStream();
+
+      JS::Rooted<JSObject*> readableStreamObj(aGlobal.Context(),
+                                              readableStream.Obj());
+
+      if (JS::ReadableStreamIsDisturbed(readableStreamObj) ||
+          JS::ReadableStreamIsLocked(readableStreamObj) ||
+          !JS::ReadableStreamIsReadable(readableStreamObj)) {
+        aRv.ThrowTypeError<MSG_FETCH_BODY_CONSUMED_ERROR>();
+        return nullptr;
+      }
+
+      r->SetReadableStreamBody(readableStreamObj);
+
+      // XXXMC: TODO
+      MOZ_ASSERT(JS::ReadableStreamGetMode(readableStreamObj) !=
+                   JS::ReadableStreamMode::ExternalSource);
+
+      void* underlyingSource = nullptr;
+      if (!JS::ReadableStreamGetExternalUnderlyingSource(aGlobal.Context(),
+                                                         readableStreamObj,
+                                                         &underlyingSource)) {
+        aRv.StealExceptionFromJSContext(aGlobal.Context());
+        return nullptr;
+      }
+
+      bodySize = InternalResponse::UNKNOWN_BODY_SIZE;
+
+      MOZ_ASSERT(underlyingSource);
+      aRv = FetchStream::RetrieveInputStream(underlyingSource,
+                                             getter_AddRefs(bodyStream));
+
+      // Releasing of the external source is needed in order to avoid an extra stream lock.
+      JS::ReadableStreamReleaseExternalUnderlyingSource(readableStreamObj);
+      if (NS_WARN_IF(aRv.Failed())) {
+        return nullptr;
+      }    
+    } else {
+      aRv = ExtractByteStreamFromBody(aBody.Value().Value(),
+                                      getter_AddRefs(bodyStream),
+                                      contentType,
+                                      bodySize);
+      if (NS_WARN_IF(aRv.Failed())) {
+        return nullptr;
+      }
     }
+    
     internalResponse->SetBody(bodyStream, bodySize);
 
     if (!contentType.IsVoid() &&
@@ -252,7 +297,7 @@ Response::Constructor(const GlobalObject& aGlobal,
 }
 
 already_AddRefed<Response>
-Response::Clone(ErrorResult& aRv) const
+Response::Clone(JSContext* aCx, ErrorResult& aRv) const
 {
   if (BodyUsed()) {
     aRv.ThrowTypeError<MSG_FETCH_BODY_CONSUMED_ERROR>();
@@ -261,11 +306,26 @@ Response::Clone(ErrorResult& aRv) const
 
   RefPtr<InternalResponse> ir = mInternalResponse->Clone();
   RefPtr<Response> response = new Response(mOwner, ir, mSignal);
+
+  JS::Rooted<JSObject*> body(aCx);
+  MaybeTeeReadableStreamBody(aCx, &body, aRv);
+  if (NS_WARN_IF(aRv.Failed())) {
+    return nullptr;
+  }
+
+  if (body) {
+    // Maybe we have a body, but we receive null from MaybeTeeReadableStreamBody
+    // if this body is a native stream. In this case, the InternalResponse will
+    // have a clone of the native body and the ReadableStream will be created
+    // lazily if needed.
+    response->SetReadableStreamBody(body);
+  }
+
   return response.forget();
 }
 
 already_AddRefed<Response>
-Response::CloneUnfiltered(ErrorResult& aRv) const
+Response::CloneUnfiltered(JSContext* aCx, ErrorResult& aRv) const
 {
   if (BodyUsed()) {
     aRv.ThrowTypeError<MSG_FETCH_BODY_CONSUMED_ERROR>();
@@ -275,6 +335,21 @@ Response::CloneUnfiltered(ErrorResult& aRv) const
   RefPtr<InternalResponse> clone = mInternalResponse->Clone();
   RefPtr<InternalResponse> ir = clone->Unfiltered();
   RefPtr<Response> ref = new Response(mOwner, ir, mSignal);
+
+  JS::Rooted<JSObject*> body(aCx);
+  MaybeTeeReadableStreamBody(aCx, &body, aRv);
+  if (NS_WARN_IF(aRv.Failed())) {
+    return nullptr;
+  }
+
+  if (body) {
+    // Maybe we have a body, but we receive null from MaybeTeeReadableStreamBody
+    // if this body is a native stream. In this case, the InternalResponse will
+    // have a clone of the native body and the ReadableStream will be created
+    // lazily if needed.
+    ref->SetReadableStreamBody(body);
+  }
+
   return ref.forget();
 }
 
diff --git a/dom/fetch/Response.h b/dom/fetch/Response.h
index 61f1427f93..5d6891dddd 100644
--- a/dom/fetch/Response.h
+++ b/dom/fetch/Response.h
@@ -115,7 +115,7 @@ public:
 
   static already_AddRefed<Response>
   Constructor(const GlobalObject& aGlobal,
-              const Optional<Nullable<ArrayBufferOrArrayBufferViewOrBlobOrFormDataOrUSVStringOrURLSearchParams>>& aBody,
+              const Optional<Nullable<fetch::ResponseBodyInit>>& aBody,
               const ResponseInit& aInit, ErrorResult& rv);
 
   nsIGlobalObject* GetParentObject() const
@@ -124,10 +124,10 @@ public:
   }
 
   already_AddRefed<Response>
-  Clone(ErrorResult& aRv) const;
+  Clone(JSContext* aCx, ErrorResult& aRv) const;
 
   already_AddRefed<Response>
-  CloneUnfiltered(ErrorResult& aRv) const;
+  CloneUnfiltered(JSContext* aCx, ErrorResult& aRv) const;
 
   void
   SetBody(nsIInputStream* aBody, int64_t aBodySize);
diff --git a/dom/webidl/Fetch.webidl b/dom/webidl/Fetch.webidl
index 4b2a0af7d0..4cae9b8509 100644
--- a/dom/webidl/Fetch.webidl
+++ b/dom/webidl/Fetch.webidl
@@ -9,6 +9,7 @@
 
 typedef object JSON;
 typedef (ArrayBuffer or ArrayBufferView or Blob or FormData or USVString or URLSearchParams) BodyInit;
+typedef (Blob or ArrayBufferView or ArrayBuffer or FormData or URLSearchParams or ReadableStream or USVString) ResponseBodyInit;
 
 [NoInterfaceObject, Exposed=(Window,Worker)]
 interface Body {
diff --git a/dom/webidl/Response.webidl b/dom/webidl/Response.webidl
index d02b714f7d..0bc1e81490 100644
--- a/dom/webidl/Response.webidl
+++ b/dom/webidl/Response.webidl
@@ -7,7 +7,9 @@
  * https://fetch.spec.whatwg.org/#response-class
  */
 
-[Constructor(optional BodyInit? body, optional ResponseInit init),
+// This should be Constructor(optional BodyInit... but BodyInit doesn't include
+// ReadableStream yet because we don't want to expose the Streams API to Request.
+[Constructor(optional ResponseBodyInit? body, optional ResponseInit init),
  Exposed=(Window,Worker)]
 interface Response {
   [NewObject] static Response error();