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

5 files changed, 237 insertions, 39 deletions

diff --git a/dom/fetch/Fetch.cpp b/dom/fetch/Fetch.cpp
index 11e93205c..04aa7fd91 100644
--- a/dom/fetch/Fetch.cpp
+++ b/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;
-  RefPtr<FetchSignalProxy> mSignalProxy;
+protected:
+  RefPtr<WorkerFetchResolver> mResolver;
 
 public:
-  WorkerFetchResponseEndBase(PromiseWorkerProxy* aPromiseProxy,
-                             FetchSignalProxy* aSignalProxy)
-    : mPromiseProxy(aPromiseProxy)
-    , mSignalProxy(aSignalProxy)
+  explicit WorkerFetchResponseEndBase(WorkerFetchResolver* aResolver)
+    : mResolver(aResolver)
   {
-    MOZ_ASSERT(mPromiseProxy);
+    MOZ_ASSERT(aResolver);
   }
 
   void
@@ -458,14 +533,13 @@ public:
     MOZ_ASSERT(aWorkerPrivate);
     aWorkerPrivate->AssertIsOnWorkerThread();
 
-    RefPtr<Promise> promise = mPromiseProxy->WorkerPromise();
-    promise->MaybeReject(NS_ERROR_DOM_ABORT_ERR);
+    mResolver->mPromiseProxy->CleanUp();
 
-    mPromiseProxy->CleanUp();
+    mResolver->mFetchObserver = nullptr;
 
-    if (mSignalProxy) {
-      mSignalProxy->Shutdown();
-      mSignalProxy = nullptr;
+    if (mResolver->mSignalProxy) {
+      mResolver->mSignalProxy->Shutdown();
+      mResolver->mSignalProxy = nullptr;
     }
   }
 };
@@ -473,17 +547,26 @@ public:
 class WorkerFetchResponseEndRunnable final : public MainThreadWorkerRunnable
                                            , public WorkerFetchResponseEndBase
 {
+  FetchDriverObserver::EndReason mReason;
+
 public:
-  WorkerFetchResponseEndRunnable(PromiseWorkerProxy* aPromiseProxy,
-                                 FetchSignalProxy* aSignalProxy)
-    : MainThreadWorkerRunnable(aPromiseProxy->GetWorkerPrivate())
-    , WorkerFetchResponseEndBase(aPromiseProxy, aSignalProxy)
+  WorkerFetchResponseEndRunnable(WorkerPrivate* aWorkerPrivate,
+                                 WorkerFetchResolver* aResolver,
+                                 FetchDriverObserver::EndReason aReason)
+    : 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,
-                                        FetchSignalProxy* aSignalProxy)
-    : MainThreadWorkerControlRunnable(aPromiseProxy->GetWorkerPrivate())
-    , WorkerFetchResponseEndBase(aPromiseProxy, aSignalProxy)
+  WorkerFetchResponseEndControlRunnable(WorkerPrivate* aWorkerPrivate,
+                                        WorkerFetchResolver* aResolver)
+    : MainThreadWorkerControlRunnable(aWorkerPrivate)
+    , WorkerFetchResponseEndBase(aResolver)
   {
   }
 
@@ -539,6 +622,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)
 {
   AssertIsOnMainThread();
@@ -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.
diff --git a/dom/fetch/FetchDriver.cpp b/dom/fetch/FetchDriver.cpp
index 448ec64cd..e8d726ce1 100644
--- a/dom/fetch/FetchDriver.cpp
+++ b/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);
diff --git a/dom/fetch/FetchDriver.h b/dom/fetch/FetchDriver.h
index 0ca9a34ee..e942aa242 100644
--- a/dom/fetch/FetchDriver.h
+++ b/dom/fetch/FetchDriver.h
@@ -67,6 +67,8 @@ public:
 
   virtual void FlushConsoleReport() = 0;
 
+  virtual void OnDataAvailable() = 0;
+
 protected:
   virtual ~FetchDriverObserver()
   { };
diff --git a/dom/fetch/FetchObserver.cpp b/dom/fetch/FetchObserver.cpp
index bc8c6fc2b..982f0ad49 100644
--- a/dom/fetch/FetchObserver.cpp
+++ b/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
diff --git a/dom/fetch/FetchObserver.h b/dom/fetch/FetchObserver.h
index 81f8e7b09..45adf2ba1 100644
--- a/dom/fetch/FetchObserver.h
+++ b/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;