Bug 1378342 - AbortSignal/AbortController - part 8 - Aborting ReadableStream when AbortSignal is aborted, r=bkelly

dom/abort/AbortSignal.cpp

@@ -119,5 +119,11 @@ AbortSignal::Follower::Unfollow()
   }
 }
 
+bool
+AbortSignal::Follower::IsFollowing() const
+{
+  return !!mFollowingSignal;
+}
+
 } // dom namespace
 } // mozilla namespace

dom/abort/AbortSignal.h

@@ -33,6 +33,9 @@ public:
     void
     Unfollow();
 
+    bool
+    IsFollowing() const;
+
     RefPtr<AbortSignal> mFollowingSignal;
   };
 

dom/fetch/Fetch.cpp

@@ -26,7 +26,7 @@
 #include "mozilla/dom/BindingDeclarations.h"
 #include "mozilla/dom/BodyUtil.h"
 #include "mozilla/dom/Exceptions.h"
-#include "mozilla/dom/DOMError.h"
+#include "mozilla/dom/DOMException.h"
 #include "mozilla/dom/FetchDriver.h"
 #include "mozilla/dom/File.h"
 #include "mozilla/dom/FormData.h"
@@ -56,6 +56,27 @@ namespace dom {
 
 using namespace workers;
 
+namespace {
+
+void
+AbortStream(JSContext* aCx, JS::Handle<JSObject*> aStream)
+{
+  if (!JS::ReadableStreamIsReadable(aStream)) {
+    return;
+  }
+
+  RefPtr<DOMException> e = DOMException::Create(NS_ERROR_DOM_ABORT_ERR);
+
+  JS::Rooted<JS::Value> value(aCx);
+  if (!GetOrCreateDOMReflector(aCx, e, &value)) {
+    return;
+  }
+
+  JS::ReadableStreamError(aCx, aStream, value);
+}
+
+} // anonymous
+
 // This class helps the proxying of AbortSignal changes cross threads.
 class AbortSignalProxy final : public AbortSignal::Follower
 {
@@ -943,6 +964,7 @@ FetchBody<Response>::FetchBody(nsIGlobalObject* aOwner);
 template <class Derived>
 FetchBody<Derived>::~FetchBody()
 {
+  Unfollow();
 }
 
 template
@@ -1106,20 +1128,33 @@ FetchBody<Response>::SetMimeType();
 
 template <class Derived>
 void
-FetchBody<Derived>::SetReadableStreamBody(JSObject* aBody)
+FetchBody<Derived>::SetReadableStreamBody(JSContext* aCx, JSObject* aBody)
 {
   MOZ_ASSERT(!mReadableStreamBody);
   MOZ_ASSERT(aBody);
   mReadableStreamBody = aBody;
+
+  RefPtr<AbortSignal> signal = DerivedClass()->GetSignal();
+  if (!signal) {
+    return;
+  }
+
+  bool aborted = signal->Aborted();
+  if (aborted) {
+    JS::Rooted<JSObject*> body(aCx, mReadableStreamBody);
+    AbortStream(aCx, body);
+  } else if (!IsFollowing()) {
+    Follow(signal);
+  }
 }
 
 template
 void
-FetchBody<Request>::SetReadableStreamBody(JSObject* aBody);
+FetchBody<Request>::SetReadableStreamBody(JSContext* aCx, JSObject* aBody);
 
 template
 void
-FetchBody<Response>::SetReadableStreamBody(JSObject* aBody);
+FetchBody<Response>::SetReadableStreamBody(JSContext* aCx, JSObject* aBody);
 
 template <class Derived>
 void
@@ -1157,6 +1192,15 @@ FetchBody<Derived>::GetBody(JSContext* aCx,
     }
   }
 
+  RefPtr<AbortSignal> signal = DerivedClass()->GetSignal();
+  if (signal) {
+    if (signal->Aborted()) {
+      AbortStream(aCx, body);
+    } else if (!IsFollowing()) {
+      Follow(signal);
+    }
+  }
+
   mReadableStreamBody = body;
   aBodyOut.set(mReadableStreamBody);
 }
@@ -1269,5 +1313,30 @@ FetchBody<Response>::MaybeTeeReadableStreamBody(JSContext* aCx,
                                                 nsIInputStream** aInputStream,
                                                 ErrorResult& aRv);
 
+template <class Derived>
+void
+FetchBody<Derived>::Aborted()
+{
+  MOZ_ASSERT(mReadableStreamBody);
+
+  AutoJSAPI jsapi;
+  if (!jsapi.Init(mOwner)) {
+    return;
+  }
+
+  JSContext* cx = jsapi.cx();
+
+  JS::Rooted<JSObject*> body(cx, mReadableStreamBody);
+  AbortStream(cx, body);
+}
+
+template
+void
+FetchBody<Request>::Aborted();
+
+template
+void
+FetchBody<Response>::Aborted();
+
 } // namespace dom
 } // namespace mozilla

dom/fetch/Fetch.h

@@ -17,6 +17,7 @@
 
 #include "mozilla/DebugOnly.h"
 #include "mozilla/ErrorResult.h"
+#include "mozilla/dom/AbortSignal.h"
 #include "mozilla/dom/Promise.h"
 #include "mozilla/dom/FetchStreamReader.h"
 #include "mozilla/dom/RequestBinding.h"
@@ -139,6 +140,7 @@ public:
  */
 template <class Derived>
 class FetchBody : public FetchStreamHolder
+                , public AbortSignal::Follower
 {
 public:
   friend class FetchBodyConsumer<Derived>;
@@ -233,6 +235,10 @@ public:
   virtual AbortSignal*
   GetSignal() const = 0;
 
+  // AbortSignal::Follower
+  void
+  Aborted() override;
+
 protected:
   nsCOMPtr<nsIGlobalObject> mOwner;
 
@@ -255,7 +261,7 @@ protected:
   SetMimeType();
 
   void
-  SetReadableStreamBody(JSObject* aBody);
+  SetReadableStreamBody(JSContext* aCx, JSObject* aBody);
 
 private:
   Derived*

dom/fetch/FetchStream.cpp

@@ -5,6 +5,7 @@
  * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
 
 #include "FetchStream.h"
+#include "mozilla/dom/DOMError.h"
 #include "nsITransport.h"
 #include "nsIStreamTransportService.h"
 #include "nsProxyRelease.h"
@@ -339,6 +340,18 @@ FetchStream::ErroredCallback(JSContext* aCx, JS::HandleObject aStream,
 {
   MOZ_DIAGNOSTIC_ASSERT(aUnderlyingSource);
   MOZ_DIAGNOSTIC_ASSERT(aFlags == FETCH_STREAM_FLAG);
+
+  // This is safe because we created an extra reference in FetchStream::Create()
+  // that won't be released until FetchStream::FinalizeCallback() is called.
+  // We are guaranteed that won't happen until the js ReadableStream object
+  // is finalized.
+  FetchStream* stream = static_cast<FetchStream*>(aUnderlyingSource);
+
+  if (stream->mInputStream) {
+    stream->mInputStream->CloseWithStatus(NS_BASE_STREAM_CLOSED);
+  }
+
+  stream->ReleaseObjects();
 }
 
 void

dom/fetch/Response.cpp

@@ -251,7 +251,7 @@ Response::Constructor(const GlobalObject& aGlobal,
         return nullptr;
       }
 
-      r->SetReadableStreamBody(readableStreamObj);
+      r->SetReadableStreamBody(aGlobal.Context(), readableStreamObj);
 
       if (JS::ReadableStreamGetMode(readableStreamObj) ==
             JS::ReadableStreamMode::ExternalSource) {
@@ -354,7 +354,7 @@ Response::Clone(JSContext* aCx, ErrorResult& aRv)
     // 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);
+    response->SetReadableStreamBody(aCx, body);
     response->mFetchStreamReader = streamReader;
     ir->SetBody(inputStream, InternalResponse::UNKNOWN_BODY_SIZE);
   }
@@ -397,7 +397,7 @@ Response::CloneUnfiltered(JSContext* aCx, ErrorResult& aRv)
     // 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);
+    ref->SetReadableStreamBody(aCx, body);
     ref->mFetchStreamReader = streamReader;
     ir->SetBody(inputStream, InternalResponse::UNKNOWN_BODY_SIZE);
   }

testing/web-platform/meta/fetch/api/abort/__dir__.ini

@@ -0,0 +1,2 @@
+prefs: [javascript.options.streams:true,
+        dom.streams.enabled:true]

testing/web-platform/meta/fetch/api/abort/general.html.ini

@@ -7,12 +7,6 @@
   [Window: Signal removed by setting to null]
     expected: FAIL
 
-  [Window: Stream errors once aborted. Underlying connection closed.]
-    expected: FAIL
-
-  [Window: Stream errors once aborted, after reading. Underlying connection closed.]
-    expected: FAIL
-
   [Window: Stream will not error if body is empty. It's closed with an empty queue before it errors.]
     expected: FAIL
 
@@ -28,12 +22,6 @@
   [DedicatedWorkerGlobalScope: Already aborted signal rejects immediately]
     expected: FAIL
 
-  [DedicatedWorkerGlobalScope: Stream errors once aborted. Underlying connection closed.]
-    expected: FAIL
-
-  [DedicatedWorkerGlobalScope: Stream errors once aborted, after reading. Underlying connection closed.]
-    expected: FAIL
-
   [DedicatedWorkerGlobalScope: Stream will not error if body is empty. It's closed with an empty queue before it errors.]
     expected: FAIL
 
@@ -49,12 +37,6 @@
   [SharedWorkerGlobalScope: Already aborted signal rejects immediately]
     expected: FAIL
 
-  [SharedWorkerGlobalScope: Stream errors once aborted. Underlying connection closed.]
-    expected: FAIL
-
-  [SharedWorkerGlobalScope: Stream errors once aborted, after reading. Underlying connection closed.]
-    expected: FAIL
-
   [SharedWorkerGlobalScope: Stream will not error if body is empty. It's closed with an empty queue before it errors.]
     expected: FAIL