Issue #1587 - Part 5: Hook FetchObserver up to the Fetch API

dom/fetch/Fetch.cpp

@@ -39,6 +39,7 @@
 #include "mozilla/dom/URLSearchParams.h"
 #include "mozilla/dom/workers/ServiceWorkerManager.h"
 
+#include "FetchObserver.h"
 #include "InternalRequest.h"
 #include "InternalResponse.h"
 
@@ -126,18 +127,20 @@ private:
 class WorkerFetchResolver final : public FetchDriverObserver
 {
   friend class MainThreadFetchRunnable;
+  friend class WorkerDataAvailableRunnable;
   friend class WorkerFetchResponseEndBase;
   friend class WorkerFetchResponseEndRunnable;
   friend class WorkerFetchResponseRunnable;
 
   RefPtr<PromiseWorkerProxy> mPromiseProxy;
   RefPtr<FetchSignalProxy> mSignalProxy;
+  RefPtr<FetchObserver> mFetchObserver;
 
 public:
   // Returns null if worker is shutting down.
   static already_AddRefed<WorkerFetchResolver>
   Create(workers::WorkerPrivate* aWorkerPrivate, Promise* aPromise,
-         FetchSignal* aSignal)
+         FetchSignal* aSignal, FetchObserver* aObserver)
   {
     MOZ_ASSERT(aWorkerPrivate);
     aWorkerPrivate->AssertIsOnWorkerThread();
@@ -152,7 +155,8 @@ public:
       signalProxy = new FetchSignalProxy(aSignal);
     }
 
-    RefPtr<WorkerFetchResolver> r = new WorkerFetchResolver(proxy, signalProxy);
+    RefPtr<WorkerFetchResolver> r =
+      new WorkerFetchResolver(proxy, signalProxy, aObserver);
     return r.forget();
   }
 
@@ -174,15 +178,19 @@ public:
   void
   OnResponseEnd(FetchDriverObserver::EndReason eReason) override;
 
+  void
+  OnDataAvailable() override;
+
 private:
    WorkerFetchResolver(PromiseWorkerProxy* aProxy,
-                       FetchSignalProxy* aSignalProxy)
+                       FetchSignalProxy* aSignalProxy,
+                       FetchObserver* aObserver)
     : mPromiseProxy(aProxy)
     , mSignalProxy(aSignalProxy)
+    , mFetchObserver(aObserver)
   {
     MOZ_ASSERT(!NS_IsMainThread());
     MOZ_ASSERT(mPromiseProxy);
-
   }
 
   ~WorkerFetchResolver()
@@ -196,12 +204,16 @@ class MainThreadFetchResolver final : public FetchDriverObserver
 {
   RefPtr<Promise> mPromise;
   RefPtr<Response> mResponse;
+  RefPtr<FetchObserver> mFetchObserver;
 
   nsCOMPtr<nsIDocument> mDocument;
 
   NS_DECL_OWNINGTHREAD
 public:
-  explicit MainThreadFetchResolver(Promise* aPromise);
+  MainThreadFetchResolver(Promise* aPromise, FetchObserver* aObserver)
+    : mPromise(aPromise)
+    , mFetchObserver(aObserver)
+  {}
 
   void
   OnResponseAvailableInternal(InternalResponse* aResponse) override;
@@ -217,9 +229,14 @@ public:
       mPromise->MaybeReject(NS_ERROR_DOM_ABORT_ERR);
     }
 
+    mFetchObserver = nullptr;
+
     FlushConsoleReport();
   }
 
+  void
+  OnDataAvailable() override;
+
 private:
   ~MainThreadFetchResolver();
 
@@ -318,6 +335,12 @@ FetchRequest(nsIGlobalObject* aGlobal, const RequestOrUSVString& aInput,
     // Let's FetchDriver to deal with an already aborted signal.
   }
 
+  RefPtr<FetchObserver> observer;
+  if (aInit.mObserve.WasPassed()) {
+    observer = new FetchObserver(aGlobal, signal);
+    aInit.mObserve.Value().HandleEvent(*observer);
+  }
+
   if (NS_IsMainThread()) {
     nsCOMPtr<nsPIDOMWindowInner> window = do_QueryInterface(aGlobal);
     nsCOMPtr<nsIDocument> doc;
@@ -344,7 +367,8 @@ FetchRequest(nsIGlobalObject* aGlobal, const RequestOrUSVString& aInput,
       }
     }
 
-    RefPtr<MainThreadFetchResolver> resolver = new MainThreadFetchResolver(p);
+    RefPtr<MainThreadFetchResolver> resolver =
+      new MainThreadFetchResolver(p, observer);
     RefPtr<FetchDriver> fetch = new FetchDriver(r, principal, loadGroup);
     fetch->SetDocument(doc);
     resolver->SetDocument(doc);
@@ -360,7 +384,8 @@ FetchRequest(nsIGlobalObject* aGlobal, const RequestOrUSVString& aInput,
       r->SetSkipServiceWorker();
     }
 
-    RefPtr<WorkerFetchResolver> resolver = WorkerFetchResolver::Create(worker, p, signal);
+    RefPtr<WorkerFetchResolver> resolver =
+      WorkerFetchResolver::Create(worker, p, signal, observer);
     if (!resolver) {
       NS_WARNING("Could not add WorkerFetchResolver workerHolder to worker");
       aRv.Throw(NS_ERROR_DOM_ABORT_ERR);
@@ -374,11 +399,6 @@ FetchRequest(nsIGlobalObject* aGlobal, const RequestOrUSVString& aInput,
   return p.forget();
 }
 
-MainThreadFetchResolver::MainThreadFetchResolver(Promise* aPromise)
-  : mPromise(aPromise)
-{
-}
-
 void
 MainThreadFetchResolver::OnResponseAvailableInternal(InternalResponse* aResponse)
 {
@@ -386,16 +406,39 @@ MainThreadFetchResolver::OnResponseAvailableInternal(InternalResponse* aResponse
   AssertIsOnMainThread();
 
   if (aResponse->Type() != ResponseType::Error) {
+    if (mFetchObserver) {
+      mFetchObserver->SetState(FetchState::Complete);
+    }
+
     nsCOMPtr<nsIGlobalObject> go = mPromise->GetParentObject();
     mResponse = new Response(go, aResponse);
     mPromise->MaybeResolve(mResponse);
   } else {
+    if (mFetchObserver) {
+      mFetchObserver->SetState(FetchState::Errored);
+    }
+
     ErrorResult result;
     result.ThrowTypeError<MSG_FETCH_FAILED>();
     mPromise->MaybeReject(result);
   }
 }
 
+void
+MainThreadFetchResolver::OnDataAvailable()
+{
+  NS_ASSERT_OWNINGTHREAD(MainThreadFetchResolver);
+  AssertIsOnMainThread();
+
+  if (!mFetchObserver) {
+    return;
+  }
+
+  if (mFetchObserver->State() == FetchState::Requesting) {
+    mFetchObserver->SetState(FetchState::Responding);
+  }
+}
+
 MainThreadFetchResolver::~MainThreadFetchResolver()
 {
   NS_ASSERT_OWNINGTHREAD(MainThreadFetchResolver);
@@ -426,10 +469,18 @@ public:
     RefPtr<Promise> promise = mResolver->mPromiseProxy->WorkerPromise();
 
     if (mInternalResponse->Type() != ResponseType::Error) {
+      if (mResolver->mFetchObserver) {
+        mResolver->mFetchObserver->SetState(FetchState::Complete);
+      }
+
       RefPtr<nsIGlobalObject> global = aWorkerPrivate->GlobalScope();
       RefPtr<Response> response = new Response(global, mInternalResponse);
       promise->MaybeResolve(response);
     } else {
+      if (mResolver->mFetchObserver) {
+        mResolver->mFetchObserver->SetState(FetchState::Errored);
+      }
+
       ErrorResult result;
       result.ThrowTypeError<MSG_FETCH_FAILED>();
       promise->MaybeReject(result);
@@ -438,18 +489,42 @@ public:
   }
 };
 
+class WorkerDataAvailableRunnable final : public MainThreadWorkerRunnable
+{
+  RefPtr<WorkerFetchResolver> mResolver;
+public:
+  WorkerDataAvailableRunnable(WorkerPrivate* aWorkerPrivate,
+                              WorkerFetchResolver* aResolver)
+    : MainThreadWorkerRunnable(aWorkerPrivate)
+    , mResolver(aResolver)
+  {
+  }
+
+  bool
+  WorkerRun(JSContext* aCx, WorkerPrivate* aWorkerPrivate) override
+  {
+    MOZ_ASSERT(aWorkerPrivate);
+    aWorkerPrivate->AssertIsOnWorkerThread();
+
+    if (mResolver->mFetchObserver &&
+        mResolver->mFetchObserver->State() == FetchState::Requesting) {
+      mResolver->mFetchObserver->SetState(FetchState::Responding);
+    }
+
+    return true;
+  }
+};
+
 class WorkerFetchResponseEndBase
 {
-  RefPtr<PromiseWorkerProxy> mPromiseProxy;
+protected:
-  RefPtr<FetchSignalProxy> mSignalProxy;
+  RefPtr<WorkerFetchResolver> mResolver;
 
 public:
-  WorkerFetchResponseEndBase(PromiseWorkerProxy* aPromiseProxy,
+  explicit WorkerFetchResponseEndBase(WorkerFetchResolver* aResolver)
-                             FetchSignalProxy* aSignalProxy)
+    : mResolver(aResolver)
-    : mPromiseProxy(aPromiseProxy)
-    , mSignalProxy(aSignalProxy)
   {
-    MOZ_ASSERT(mPromiseProxy);
+    MOZ_ASSERT(aResolver);
   }
 
   void
@@ -458,14 +533,13 @@ public:
     MOZ_ASSERT(aWorkerPrivate);
     aWorkerPrivate->AssertIsOnWorkerThread();
 
-    RefPtr<Promise> promise = mPromiseProxy->WorkerPromise();
+    mResolver->mPromiseProxy->CleanUp();
-    promise->MaybeReject(NS_ERROR_DOM_ABORT_ERR);
 
-    mPromiseProxy->CleanUp();
+    mResolver->mFetchObserver = nullptr;
 
-    if (mSignalProxy) {
+    if (mResolver->mSignalProxy) {
-      mSignalProxy->Shutdown();
+      mResolver->mSignalProxy->Shutdown();
-      mSignalProxy = nullptr;
+      mResolver->mSignalProxy = nullptr;
     }
   }
 };
@@ -473,17 +547,26 @@ public:
 class WorkerFetchResponseEndRunnable final : public MainThreadWorkerRunnable
                                            , public WorkerFetchResponseEndBase
 {
+  FetchDriverObserver::EndReason mReason;
+
 public:
-  WorkerFetchResponseEndRunnable(PromiseWorkerProxy* aPromiseProxy,
+  WorkerFetchResponseEndRunnable(WorkerPrivate* aWorkerPrivate,
-                                 FetchSignalProxy* aSignalProxy)
+                                 WorkerFetchResolver* aResolver,
-    : MainThreadWorkerRunnable(aPromiseProxy->GetWorkerPrivate())
+                                 FetchDriverObserver::EndReason aReason)
-    , WorkerFetchResponseEndBase(aPromiseProxy, aSignalProxy)
+    : MainThreadWorkerRunnable(aWorkerPrivate)
+    , WorkerFetchResponseEndBase(aResolver)
+    , mReason(aReason)
   {
   }
 
   bool
   WorkerRun(JSContext* aCx, WorkerPrivate* aWorkerPrivate) override
   {
+    if (mReason == FetchDriverObserver::eAborted) {
+      RefPtr<Promise> promise = mResolver->mPromiseProxy->WorkerPromise();
+      promise->MaybeReject(NS_ERROR_DOM_ABORT_ERR);
+    }
+
     WorkerRunInternal(aWorkerPrivate);
     return true;
   }
@@ -502,10 +585,10 @@ class WorkerFetchResponseEndControlRunnable final : public MainThreadWorkerContr
                                                   , public WorkerFetchResponseEndBase
 {
 public:
-  WorkerFetchResponseEndControlRunnable(PromiseWorkerProxy* aPromiseProxy,
+  WorkerFetchResponseEndControlRunnable(WorkerPrivate* aWorkerPrivate,
-                                        FetchSignalProxy* aSignalProxy)
+                                        WorkerFetchResolver* aResolver)
-    : MainThreadWorkerControlRunnable(aPromiseProxy->GetWorkerPrivate())
+    : MainThreadWorkerControlRunnable(aWorkerPrivate)
-    , WorkerFetchResponseEndBase(aPromiseProxy, aSignalProxy)
+    , WorkerFetchResponseEndBase(aResolver)
   {
   }
 
@@ -538,6 +621,21 @@ WorkerFetchResolver::OnResponseAvailableInternal(InternalResponse* aResponse)
   }
 }
 
+void
+WorkerFetchResolver::OnDataAvailable()
+{
+  AssertIsOnMainThread();
+
+  MutexAutoLock lock(mPromiseProxy->Lock());
+  if (mPromiseProxy->CleanedUp()) {
+    return;
+  }
+
+  RefPtr<WorkerDataAvailableRunnable> r =
+    new WorkerDataAvailableRunnable(mPromiseProxy->GetWorkerPrivate(), this);
+  Unused << r->Dispatch();
+}
+
 void
 WorkerFetchResolver::OnResponseEnd(FetchDriverObserver::EndReason aReason)
 {
@@ -550,11 +648,13 @@ WorkerFetchResolver::OnResponseEnd(FetchDriverObserver::EndReason aReason)
   FlushConsoleReport();
 
   RefPtr<WorkerFetchResponseEndRunnable> r =
-    new WorkerFetchResponseEndRunnable(mPromiseProxy, mSignalProxy);
+    new WorkerFetchResponseEndRunnable(mPromiseProxy->GetWorkerPrivate(),
+                                       this, aReason);
 
   if (!r->Dispatch()) {
     RefPtr<WorkerFetchResponseEndControlRunnable> cr =
-      new WorkerFetchResponseEndControlRunnable(mPromiseProxy, mSignalProxy);
+      new WorkerFetchResponseEndControlRunnable(mPromiseProxy->GetWorkerPrivate(),
+                                                this);
     // This can fail if the worker thread is canceled or killed causing
     // the PromiseWorkerProxy to give up its WorkerHolder immediately,
     // allowing the worker thread to become Dead.

dom/fetch/FetchDriver.cpp

@@ -667,6 +667,31 @@ FetchDriver::OnStartRequest(nsIRequest* aRequest,
   return NS_OK;
 }
 
+namespace {
+
+// Runnable to call the observer OnDataAvailable on the main-thread.
+class DataAvailableRunnable final : public Runnable
+{
+  RefPtr<FetchDriverObserver> mObserver;
+
+public:
+  explicit DataAvailableRunnable(FetchDriverObserver* aObserver)
+    : mObserver(aObserver)
+  {
+     MOZ_ASSERT(aObserver);
+  }
+
+  NS_IMETHOD
+  Run() override
+  {
+    mObserver->OnDataAvailable();
+    mObserver = nullptr;
+    return NS_OK;
+  }
+};
+
+} // anonymous namespace
+
 NS_IMETHODIMP
 FetchDriver::OnDataAvailable(nsIRequest* aRequest,
                              nsISupports* aContext,
@@ -678,6 +703,18 @@ FetchDriver::OnDataAvailable(nsIRequest* aRequest,
   // called between OnStartRequest and OnStopRequest, so we don't need to worry
   // about races.
 
+  if (mObserver) {
+    if (NS_IsMainThread()) {
+      mObserver->OnDataAvailable();
+    } else {
+      RefPtr<Runnable> runnable = new DataAvailableRunnable(mObserver);
+      nsresult rv = NS_DispatchToMainThread(runnable);
+      if (NS_WARN_IF(NS_FAILED(rv))) {
+        return rv;
+      }
+    }
+  }
+
   uint32_t aRead;
   MOZ_ASSERT(mResponse);
   MOZ_ASSERT(mPipeOutputStream);

dom/fetch/FetchDriver.h

@@ -67,6 +67,8 @@ public:
 
   virtual void FlushConsoleReport() = 0;
 
+  virtual void OnDataAvailable() = 0;
+
 protected:
   virtual ~FetchDriverObserver()
   { };

dom/fetch/FetchObserver.cpp

@@ -6,6 +6,7 @@
 
 #include "FetchObserver.h"
 #include "WorkerPrivate.h"
+#include "mozilla/dom/Event.h"
 
 namespace mozilla {
 namespace dom {
@@ -45,10 +46,14 @@ FetchObserver::IsEnabled(JSContext* aCx, JSObject* aGlobal)
 }
 
 FetchObserver::FetchObserver(nsIGlobalObject* aGlobal,
-                             FetchState aState)
+                             FetchSignal* aSignal)
   : DOMEventTargetHelper(aGlobal)
-  , mState(aState)
+  , mState(FetchState::Requesting)
-{}
+{
+  if (aSignal) {
+    Follow(aSignal);
+  }
+}
 
 JSObject*
 FetchObserver::WrapObject(JSContext* aCx, JS::Handle<JSObject*> aGivenProto)
@@ -62,5 +67,51 @@ FetchObserver::State() const
   return mState;
 }
 
+void
+FetchObserver::Aborted()
+{
+  SetState(FetchState::Aborted);
+}
+
+void
+FetchObserver::SetState(FetchState aState)
+{
+  MOZ_ASSERT(mState < aState);
+
+  if (mState == FetchState::Aborted ||
+      mState == FetchState::Errored ||
+      mState == FetchState::Complete) {
+    // We are already in a final state.
+    return;
+  }
+
+  // We cannot pass from Requesting to Complete directly.
+  if (mState == FetchState::Requesting &&
+      aState == FetchState::Complete) {
+    SetState(FetchState::Responding);
+  }
+
+  mState = aState;
+
+  if (mState == FetchState::Aborted ||
+      mState == FetchState::Errored ||
+      mState == FetchState::Complete) {
+    Unfollow();
+  }
+
+  EventInit init;
+  init.mBubbles = false;
+  init.mCancelable = false;
+
+  // TODO which kind of event should we dispatch here?
+
+  RefPtr<Event> event =
+    Event::Constructor(this, NS_LITERAL_STRING("statechange"), init);
+  event->SetTrusted(true);
+
+  bool dummy;
+  DispatchEvent(event, &dummy);
+}
+
 } // dom namespace
 } // mozilla namespace

dom/fetch/FetchObserver.h

@@ -9,11 +9,13 @@
 
 #include "mozilla/DOMEventTargetHelper.h"
 #include "mozilla/dom/FetchObserverBinding.h"
+#include "mozilla/dom/FetchSignal.h"
 
 namespace mozilla {
 namespace dom {
 
 class FetchObserver final : public DOMEventTargetHelper
+                          , public FetchSignal::Follower
 {
 public:
   NS_DECL_ISUPPORTS_INHERITED
@@ -22,7 +24,7 @@ public:
   static bool
   IsEnabled(JSContext* aCx, JSObject* aGlobal);
 
-  FetchObserver(nsIGlobalObject* aGlobal, FetchState aState);
+  FetchObserver(nsIGlobalObject* aGlobal, FetchSignal* aSignal);
 
   JSObject*
   WrapObject(JSContext* aCx, JS::Handle<JSObject*> aGivenProto) override;
@@ -34,6 +36,12 @@ public:
   IMPL_EVENT_HANDLER(requestprogress);
   IMPL_EVENT_HANDLER(responseprogress);
 
+  void
+  Aborted() override;
+
+  void
+  SetState(FetchState aState);
+
 private:
   ~FetchObserver() = default;
 

dom/tests/mochitest/fetch/file_fetch_controller.html

@@ -88,7 +88,7 @@ function testAbortedFetch() {
   var fc = new FetchController();
   fc.abort();
 
-  fetch('data:,foo', { signal: fc.signal }).then(() => {
+  fetch('slow.sjs', { signal: fc.signal }).then(() => {
     ok(false, "Fetch should not return a resolved promise");
   }, e => {
     is(e.name, "AbortError", "We have an abort error");
@@ -98,7 +98,7 @@ function testAbortedFetch() {
 function testFetchAndAbort() {
   var fc = new FetchController();
 
-  var p = fetch('data:,foo', { signal: fc.signal });
+  var p = fetch('slow.sjs', { signal: fc.signal });
   fc.abort();
 
   p.then(() => {
@@ -109,49 +109,21 @@ function testFetchAndAbort() {
 }
 
 function testWorkerAbortedFetch() {
-  function worker() {
+  var w = new Worker('worker_fetch_controller.js');
-    var fc = new FetchController();
-    fc.abort();
-
-    fetch('data:,foo', { signal: fc.signal }).then(() => {
-      postMessage(false);
-    }, e => {
-      postMessage(e.name == "AbortError");
-    });
-  }
-
-  var str = worker.toString();
-  var content = str.substring(0, str.length - 1).split('\n').splice(1).join(' ');
-  var url = URL.createObjectURL(new Blob([content], { type: "application/javascript" }));
-  var w = new Worker(url);
   w.onmessage = function(e) {
     ok(e.data, "Abort + Fetch works in workers");
     next();
   }
+  w.postMessage('testWorkerAbortedFetch');
 }
 
 function testWorkerFetchAndAbort() {
-  function worker() {
+  var w = new Worker('worker_fetch_controller.js');
-    var fc = new FetchController();
-
-    var p = fetch('data:,foo', { signal: fc.signal });
-    fc.abort();
-
-    p.then(() => {
-      postMessage(false);
-    }, e => {
-      postMessage(e.name == "AbortError");
-    });
-  }
-
-  var str = worker.toString();
-  var content = str.substring(0, str.length - 1).split('\n').splice(1).join(' ');
-  var url = URL.createObjectURL(new Blob([content], { type: "application/javascript" }));
-  var w = new Worker(url);
   w.onmessage = function(e) {
     ok(e.data, "Abort + Fetch works in workers");
     next();
   }
+  w.postMessage('testWorkerFetchAndAbort');
 }
 
 var steps = [

dom/tests/mochitest/fetch/file_fetch_observer.html

@@ -10,12 +10,125 @@ function is(a, b, msg) {
 function testObserver() {
   ok("FetchObserver" in self, "We have a FetchObserver prototype");
 
-  fetch('data:,foo', { observe: o => {
+  fetch('http://mochi.test:8888/tests/dom/tests/mochitest/fetch/slow.sjs', { observe: o => {
+    ok(!!o, "We have an observer");
+    ok(o instanceof FetchObserver, "The correct object has been passed");
+    is(o.state, "requesting", "By default the state is requesting");
+    next();
   }});
 }
 
+function testObserveAbort() {
+  var fc = new FetchController();
+
+  fetch('http://mochi.test:8888/tests/dom/tests/mochitest/fetch/slow.sjs', {
+    signal: fc.signal,
+    observe: o => {
+      o.onstatechange = () => {
+        ok(true, "StateChange event dispatched");
+        if (o.state == "aborted") {
+          ok(true, "Aborted!");
+          next();
+        }
+      }
+      fc.abort();
+    }
+  });
+}
+
+function testObserveComplete() {
+  var fc = new FetchController();
+
+  fetch('http://mochi.test:8888/tests/dom/tests/mochitest/fetch/slow.sjs', {
+    signal: fc.signal,
+    observe: o => {
+      o.onstatechange = () => {
+        ok(true, "StateChange event dispatched");
+        if (o.state == "complete") {
+          ok(true, "Operation completed");
+          next();
+        }
+      }
+    }
+  });
+}
+
+function testObserveErrored() {
+  var fc = new FetchController();
+
+  fetch('foo: bar', {
+    signal: fc.signal,
+    observe: o => {
+      o.onstatechange = () => {
+        ok(true, "StateChange event dispatched");
+        if (o.state == "errored") {
+          ok(true, "Operation completed");
+          next();
+        }
+      }
+    }
+  });
+}
+
+function testObserveResponding() {
+  var fc = new FetchController();
+
+  fetch('http://mochi.test:8888/tests/dom/tests/mochitest/fetch/slow.sjs', {
+    signal: fc.signal,
+    observe: o => {
+      o.onstatechange = () => {
+        if (o.state == "responding") {
+          ok(true, "We have responding events");
+          next();
+        }
+      }
+    }
+  });
+}
+
+function workify(worker) {
+  function methods() {
+    function ok(a, msg) {
+      postMessage( { type: 'check', state: !!a, message: msg });
+    };
+    function is(a, b, msg) {
+      postMessage( { type: 'check', state: a === b, message: msg });
+    };
+    function next() {
+      postMessage( { type: 'finish' });
+    };
+  }
+
+  var str = methods.toString();
+  var methodsContent = str.substring(0, str.length - 1).split('\n').splice(1).join('\n');
+
+  str = worker.toString();
+  var workerContent = str.substring(0, str.length - 1).split('\n').splice(1).join('\n');
+
+  var content = methodsContent + workerContent;
+  var url = URL.createObjectURL(new Blob([content], { type: "application/javascript" }));
+  var w = new Worker(url);
+  w.onmessage = e => {
+    if (e.data.type == 'check') {
+      ok(e.data.state, "WORKER: " + e.data.message);
+    } else if (e.data.type == 'finish') {
+      next();
+    } else {
+      ok(false, "Something went wrong");
+    }
+  }
+}
+
 var steps = [
   testObserver,
+  testObserveAbort,
+  function() { workify(testObserveAbort); },
+  testObserveComplete,
+  function() { workify(testObserveComplete); },
+  testObserveErrored,
+  function() { workify(testObserveErrored); },
+  testObserveResponding,
+  function() { workify(testObserveResponding); },
 ];
 
 function next() {

dom/tests/mochitest/fetch/mochitest.ini

@@ -17,11 +17,13 @@ support-files =
   reroute.html
   reroute.js
   reroute.js^headers^
+  slow.sjs
   sw_reroute.js
   empty.js
   empty.js^headers^
   worker_temporaryFileBlob.js
   common_temporaryFileBlob.js
+  worker_fetch_controller.js
   !/dom/xhr/tests/file_XHR_binary1.bin
   !/dom/xhr/tests/file_XHR_binary1.bin^headers^
   !/dom/xhr/tests/file_XHR_binary2.bin

dom/tests/mochitest/fetch/slow.sjs

@@ -0,0 +1,11 @@
+function handleRequest(request, response)
+{
+  response.processAsync();
+
+  timer = Components.classes["@mozilla.org/timer;1"].
+          createInstance(Components.interfaces.nsITimer);
+  timer.init(function() {
+    response.write("Here the content. But slowly.");
+    response.finish();
+  }, 1000, Components.interfaces.nsITimer.TYPE_ONE_SHOT);
+}

dom/tests/mochitest/fetch/test_fetch_observer.html

@@ -12,7 +12,8 @@
 <body>
 <script class="testbody" type="text/javascript">
 
-SpecialPowers.pushPrefEnv({"set": [["dom.fetchObserver.enabled", true ]]}, () => {
+SpecialPowers.pushPrefEnv({"set": [["dom.fetchObserver.enabled", true ],
+                                   ["dom.fetchController.enabled", true ]]}, () => {
   let ifr = document.createElement('iframe');
   ifr.src = "file_fetch_observer.html";
   document.body.appendChild(ifr);

dom/tests/mochitest/fetch/worker_fetch_controller.js

@@ -0,0 +1,27 @@
+function testWorkerAbortedFetch() {
+  var fc = new FetchController();
+  fc.abort();
+
+  fetch('slow.sjs', { signal: fc.signal }).then(() => {
+    postMessage(false);
+  }, e => {
+    postMessage(e.name == "AbortError");
+  });
+}
+
+function testWorkerFetchAndAbort() {
+  var fc = new FetchController();
+
+  var p = fetch('slow.sjs', { signal: fc.signal });
+  fc.abort();
+
+  p.then(() => {
+    postMessage(false);
+  }, e => {
+    postMessage(e.name == "AbortError");
+  });
+}
+
+onmessage = function(e) {
+  self[e.data]();
+}