DevTools: introduce CrossThreadProtocolCallback

... to assure DevTools protocol callbacks are always invoked (and destroyed) on the
correct thread.

Change-Id: Ie94c67d754138161ba22192bb76891f4021ad280
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/1775501
Commit-Queue: Andrey Kosyakov <caseq@chromium.org>
Reviewed-by: Alexei Filippov <alph@chromium.org>
Cr-Commit-Position: refs/heads/master@{#691793}

content/browser/BUILD.gn

@@ -673,6 +673,7 @@ jumbo_source_set("browser") {
     "data_url_loader_factory.h",
     "devtools/browser_devtools_agent_host.cc",
     "devtools/browser_devtools_agent_host.h",
+    "devtools/cross_thread_protocol_callback.h",
     "devtools/devtools_agent_host_impl.cc",
     "devtools/devtools_agent_host_impl.h",
     "devtools/devtools_background_services_context_impl.cc",

content/browser/devtools/cross_thread_protocol_callback.h

+// Copyright 2019 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef CONTENT_BROWSER_DEVTOOLS_CROSS_THREAD_PROTOCOL_CALLBACK_H_
+#define CONTENT_BROWSER_DEVTOOLS_CROSS_THREAD_PROTOCOL_CALLBACK_H_
+
+#include <memory>
+
+#include "base/task/post_task.h"
+#include "content/public/browser/browser_thread.h"
+
+namespace content {
+
+// A wrapper for a DevTools protocol method callback that assures the
+// underlying callback is called on the correct thread. Use this to pass
+// the protocol callback to methods handling DevTools commands on threads
+// other than UI.
+template <typename ProtocolCallback>
+class CrossThreadProtocolCallback {
+ public:
+  explicit CrossThreadProtocolCallback(
+      std::unique_ptr<ProtocolCallback> callback)
+      : callback_(std::move(callback)) {}
+  CrossThreadProtocolCallback(CrossThreadProtocolCallback&& r) = default;
+
+  template <typename... Args>
+  void sendSuccess(Args&&... args) {
+    base::PostTask(
+        FROM_HERE, {BrowserThread::UI},
+        base::BindOnce(&ProtocolCallback::sendSuccess, std::move(callback_),
+                       std::forward<Args>(args)...));
+  }
+
+  void sendFailure(protocol::DispatchResponse response) {
+    base::PostTask(FROM_HERE, {BrowserThread::UI},
+                   base::BindOnce(&ProtocolCallback::sendFailure,
+                                  std::move(callback_), std::move(response)));
+  }
+
+  ~CrossThreadProtocolCallback() {
+    BrowserThread::DeleteSoon({BrowserThread::UI}, FROM_HERE,
+                              std::move(callback_));
+  }
+
+ private:
+  std::unique_ptr<ProtocolCallback> callback_;
+};
+
+template <typename ProtocolCallback>
+CrossThreadProtocolCallback<ProtocolCallback> WrapForAnotherThread(
+    std::unique_ptr<ProtocolCallback> callback) {
+  return CrossThreadProtocolCallback<ProtocolCallback>(std::move(callback));
+}
+
+}  // namespace content
+
+#endif  // CONTENT_BROWSER_DEVTOOLS_CROSS_THREAD_PROTOCOL_CALLBACK_H_
\ No newline at end of file

content/browser/devtools/devtools_url_loader_interceptor.cc

@@ -11,6 +11,7 @@
 #include "base/task/post_task.h"
 #include "base/time/time.h"
 #include "base/unguessable_token.h"
+#include "content/browser/devtools/cross_thread_protocol_callback.h"
 #include "content/browser/devtools/protocol/network.h"
 #include "content/browser/devtools/protocol/network_handler.h"
 #include "content/browser/frame_host/frame_tree_node.h"
@@ -37,7 +38,7 @@ using RequestInterceptedCallback =
     DevToolsNetworkInterceptor::RequestInterceptedCallback;
 using ContinueInterceptedRequestCallback =
     DevToolsNetworkInterceptor::ContinueInterceptedRequestCallback;
-using GetResponseBodyForInterceptionCallback =
+using GetResponseBodyCallback =
     DevToolsNetworkInterceptor::GetResponseBodyForInterceptionCallback;
 using TakeResponseBodyPipeCallback =
     DevToolsNetworkInterceptor::TakeResponseBodyPipeCallback;
@@ -74,15 +75,14 @@ class BodyReader : public mojo::DataPipeDrainer::Client {
 
   void StartReading(mojo::ScopedDataPipeConsumerHandle body);
 
-  void AddCallback(
-      std::unique_ptr<GetResponseBodyForInterceptionCallback> callback) {
-    callbacks_.push_back(std::move(callback));
+  using GetBodyCallback = CrossThreadProtocolCallback<GetResponseBodyCallback>;
+  void AddCallback(GetBodyCallback callback) {
     if (data_complete_) {
-      DCHECK_EQ(1UL, callbacks_.size());
-      base::PostTask(FROM_HERE, {BrowserThread::UI},
-                     base::BindOnce(&BodyReader::DispatchBodyOnUI,
-                                    std::move(callbacks_), encoded_body_));
+      DCHECK(callbacks_.empty());
+      callback.sendSuccess(encoded_body_, true);
+      return;
     }
+    callbacks_.push_back(std::move(callback));
   }
 
   bool data_complete() const { return data_complete_; }
@@ -93,19 +93,12 @@ class BodyReader : public mojo::DataPipeDrainer::Client {
   }
 
   void CancelWithError(std::string error) {
-    base::PostTask(FROM_HERE, {BrowserThread::UI},
-                   base::BindOnce(&BodyReader::DispatchErrorOnUI,
-                                  std::move(callbacks_), std::move(error)));
+    for (auto& cb : callbacks_)
+      cb.sendFailure(Response::Error(error));
+    callbacks_.clear();
   }
 
  private:
-  using CallbackVector =
-      std::vector<std::unique_ptr<GetResponseBodyForInterceptionCallback>>;
-  static void DispatchBodyOnUI(const CallbackVector& callbacks,
-                               const std::string& body);
-  static void DispatchErrorOnUI(const CallbackVector& callbacks,
-                                const std::string& error);
-
   void OnDataAvailable(const void* data, size_t num_bytes) override {
     DCHECK(!data_complete_);
     body_->data().append(
@@ -115,7 +108,7 @@ class BodyReader : public mojo::DataPipeDrainer::Client {
   void OnDataComplete() override;
 
   std::unique_ptr<mojo::DataPipeDrainer> body_pipe_drainer_;
-  CallbackVector callbacks_;
+  std::vector<GetBodyCallback> callbacks_;
   base::OnceClosure download_complete_callback_;
   scoped_refptr<base::RefCountedString> body_;
   std::string encoded_body_;
@@ -136,26 +129,12 @@ void BodyReader::OnDataComplete() {
   body_pipe_drainer_.reset();
   // TODO(caseq): only encode if necessary.
   base::Base64Encode(body_->data(), &encoded_body_);
-  base::PostTask(FROM_HERE, {BrowserThread::UI},
-                 base::BindOnce(&BodyReader::DispatchBodyOnUI,
-                                std::move(callbacks_), encoded_body_));
+  for (auto& cb : callbacks_)
+    cb.sendSuccess(encoded_body_, true);
+  callbacks_.clear();
   std::move(download_complete_callback_).Run();
 }
 
-// static
-void BodyReader::DispatchBodyOnUI(const CallbackVector& callbacks,
-                                  const std::string& encoded_body) {
-  for (const auto& cb : callbacks)
-    cb->sendSuccess(encoded_body, true);
-}
-
-// static
-void BodyReader::DispatchErrorOnUI(const CallbackVector& callbacks,
-                                   const std::string& error) {
-  for (const auto& cb : callbacks)
-    cb->sendFailure(Response::Error(error));
-}
-
 struct ResponseMetadata {
   ResponseMetadata() = default;
   explicit ResponseMetadata(const network::ResourceResponseHead& head)
@@ -193,11 +172,11 @@ class InterceptionJob : public network::mojom::URLLoaderClient,
       mojo::PendingRemote<network::mojom::CookieManager> cookie_manager);
 
   void GetResponseBody(
-      std::unique_ptr<GetResponseBodyForInterceptionCallback> callback);
+      CrossThreadProtocolCallback<GetResponseBodyCallback> callback);
   void TakeResponseBodyPipe(TakeResponseBodyPipeCallback callback);
   void ContinueInterceptedRequest(
       std::unique_ptr<Modifications> modifications,
-      std::unique_ptr<ContinueInterceptedRequestCallback> callback);
+      CrossThreadProtocolCallback<ContinueInterceptedRequestCallback> callback);
   void Detach();
 
   void OnAuthRequest(
@@ -389,7 +368,7 @@ class DevToolsURLLoaderInterceptor::Impl
 
   void GetResponseBody(
       const std::string& interception_id,
-      std::unique_ptr<GetResponseBodyForInterceptionCallback> callback) {
+      CrossThreadProtocolCallback<GetResponseBodyCallback> callback) {
     if (InterceptionJob* job = FindJob(interception_id, &callback))
       job->GetResponseBody(std::move(callback));
   }
@@ -413,7 +392,8 @@ class DevToolsURLLoaderInterceptor::Impl
   void ContinueInterceptedRequest(
       const std::string& interception_id,
       std::unique_ptr<Modifications> modifications,
-      std::unique_ptr<ContinueInterceptedRequestCallback> callback) {
+      CrossThreadProtocolCallback<ContinueInterceptedRequestCallback>
+          callback) {
     if (InterceptionJob* job = FindJob(interception_id, &callback)) {
       job->ContinueInterceptedRequest(std::move(modifications),
                                       std::move(callback));
@@ -425,14 +405,12 @@ class DevToolsURLLoaderInterceptor::Impl
 
   template <typename Callback>
   InterceptionJob* FindJob(const std::string& id,
-                           std::unique_ptr<Callback>* callback) {
+                           CrossThreadProtocolCallback<Callback>* callback) {
     auto it = jobs_.find(id);
     if (it != jobs_.end())
       return it->second;
-    base::PostTask(FROM_HERE, {BrowserThread::UI},
-                   base::BindOnce(&Callback::sendFailure, std::move(*callback),
-                                  protocol::Response::InvalidParams(
-                                      "Invalid InterceptionId.")));
+    callback->sendFailure(
+        protocol::Response::InvalidParams("Invalid InterceptionId."));
     return nullptr;
   }
 
@@ -617,11 +595,11 @@ void DevToolsURLLoaderInterceptor::SetPatterns(
 
 void DevToolsURLLoaderInterceptor::GetResponseBody(
     const std::string& interception_id,
-    std::unique_ptr<GetResponseBodyForInterceptionCallback> callback) {
-  base::PostTask(
-      FROM_HERE, {BrowserThread::IO},
-      base::BindOnce(&Impl::GetResponseBody, base::Unretained(impl_.get()),
-                     interception_id, std::move(callback)));
+    std::unique_ptr<GetResponseBodyCallback> callback) {
+  base::PostTask(FROM_HERE, {BrowserThread::IO},
+                 base::BindOnce(&Impl::GetResponseBody,
+                                base::Unretained(impl_.get()), interception_id,
+                                WrapForAnotherThread(std::move(callback))));
 }
 
 void DevToolsURLLoaderInterceptor::TakeResponseBodyPipe(
@@ -640,7 +618,8 @@ void DevToolsURLLoaderInterceptor::ContinueInterceptedRequest(
   base::PostTask(FROM_HERE, {BrowserThread::IO},
                  base::BindOnce(&Impl::ContinueInterceptedRequest,
                                 base::Unretained(impl_.get()), interception_id,
-                                std::move(modifications), std::move(callback)));
+                                std::move(modifications),
+                                WrapForAnotherThread(std::move(callback))));
 }
 
 bool DevToolsURLLoaderInterceptor::CreateProxyForInterception(
@@ -754,14 +733,10 @@ bool InterceptionJob::CanGetResponseBody(std::string* error_reason) {
 }
 
 void InterceptionJob::GetResponseBody(
-    std::unique_ptr<GetResponseBodyForInterceptionCallback> callback) {
+    CrossThreadProtocolCallback<GetResponseBodyCallback> callback) {
   std::string error_reason;
   if (!CanGetResponseBody(&error_reason)) {
-    base::PostTask(
-        FROM_HERE, {BrowserThread::UI},
-        base::BindOnce(&GetResponseBodyForInterceptionCallback::sendFailure,
-                       std::move(callback),
-                       Response::Error(std::move(error_reason))));
+    callback.sendFailure(Response::Error(std::move(error_reason)));
     return;
   }
   if (!body_reader_) {
@@ -794,16 +769,13 @@ void InterceptionJob::TakeResponseBodyPipe(
 
 void InterceptionJob::ContinueInterceptedRequest(
     std::unique_ptr<Modifications> modifications,
-    std::unique_ptr<ContinueInterceptedRequestCallback> callback) {
+    CrossThreadProtocolCallback<ContinueInterceptedRequestCallback> callback) {
   Response response = InnerContinueRequest(std::move(modifications));
   // |this| may be destroyed at this point.
-  bool success = response.isSuccess();
-  base::OnceClosure task =
-      success ? base::BindOnce(&ContinueInterceptedRequestCallback::sendSuccess,
-                               std::move(callback))
-              : base::BindOnce(&ContinueInterceptedRequestCallback::sendFailure,
-                               std::move(callback), std::move(response));
-  base::PostTask(FROM_HERE, {BrowserThread::UI}, std::move(task));
+  if (response.isSuccess())
+    callback.sendSuccess();
+  else
+    callback.sendFailure(std::move(response));
 }
 
 void InterceptionJob::Detach() {