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

1 files changed, 135 insertions, 35 deletions

diff --git a/dom/fetch/Fetch.cpp b/dom/fetch/Fetch.cpp
index 11e93205ce..04aa7fd915 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.