Migrate gcm to network service's socket APIs

With network service, net:: C++ socket access will need to be changed to using
network service's socket APIs. This CL changes gcm to using
proxy_resolving_socket.mojom instead of the C++ ProxyResolvingClientSocket.

GCM files that are changed:
- google_apis/gcm/engine/connection_factory_impl.cc
  This will request and own a network::mojom::ProxyResolvingSocketPtr
- google_apis/gcm/engine/connection_handler_impl.cc
  This will pass the mojo read/write pipes to socket streams.
- google_apis/gcm/base/socket_stream.cc
  SocketInputStream/SocketOutputStream will read from/write to mojo data pipes
  instead of the C++ socket.

Network service changes
- services/network/public/mojom/proxy_resolving_socket.mojom
  Peer address is surfaced directly when connection is established. This is
  needed to support GCM's use case. They need the peer address synchronously
  when connection is done.

Browser changes:
- NetworkContext raw mojo pointer has to be queried on the UI thread on demand.
  Therefore changes are added to pass through a callback and ui task runner.

Bug: 862608
Cq-Include-Trybots: luci.chromium.try:ios-simulator-full-configs;luci.chromium.try:linux_mojo;master.tryserver.chromium.mac:ios-simulator-cronet
Change-Id: Ic0c72c8ee7d9bc7deb6e8109234227f80ea3bbe3
Reviewed-on: https://chromium-review.googlesource.com/1142344
Reviewed-by: Daniel Cheng <dcheng@chromium.org>
Reviewed-by: John Abd-El-Malek <jam@chromium.org>
Reviewed-by: Maks Orlovich <morlovich@chromium.org>
Reviewed-by: Sylvain Defresne <sdefresne@chromium.org>
Reviewed-by: Peter Beverloo <peter@chromium.org>
Commit-Queue: Helen Li <xunjieli@chromium.org>
Cr-Commit-Position: refs/heads/master@{#578386}
diff --git a/chrome/browser/browser_process_impl.cc b/chrome/browser/browser_process_impl.cc
index bc7a146..bf7f459 100644
--- a/chrome/browser/browser_process_impl.cc
+++ b/chrome/browser/browser_process_impl.cc
@@ -532,6 +532,23 @@
   return waitable_event_.TimedWaitUntil(end_time);
 }
 
+#if !defined(OS_ANDROID)
+void RequestProxyResolvingSocketFactoryOnUIThread(
+    network::mojom::ProxyResolvingSocketFactoryRequest request) {
+  network::mojom::NetworkContext* network_context =
+      g_browser_process->system_network_context_manager()->GetContext();
+  network_context->CreateProxyResolvingSocketFactory(std::move(request));
+}
+
+void RequestProxyResolvingSocketFactory(
+    network::mojom::ProxyResolvingSocketFactoryRequest request) {
+  BrowserThread::PostTask(
+      BrowserThread::UI, FROM_HERE,
+      base::BindOnce(&RequestProxyResolvingSocketFactoryOnUIThread,
+                     std::move(request)));
+}
+#endif
+
 }  // namespace
 
 void BrowserProcessImpl::FlushLocalStateAndReply(base::OnceClosure reply) {
@@ -1361,7 +1378,7 @@
 
   gcm_driver_ = gcm::CreateGCMDriverDesktop(
       base::WrapUnique(new gcm::GCMClientFactory), local_state(), store_path,
-      system_request_context(),
+      base::BindRepeating(&RequestProxyResolvingSocketFactory),
       system_network_context_manager()->GetSharedURLLoaderFactory(),
       chrome::GetChannel(), gcm::GetProductCategoryForSubtypes(local_state()),
       content::BrowserThread::GetTaskRunnerForThread(
diff --git a/chrome/browser/extensions/extension_gcm_app_handler_unittest.cc b/chrome/browser/extensions/extension_gcm_app_handler_unittest.cc
index 15fd662..1049110 100644
--- a/chrome/browser/extensions/extension_gcm_app_handler_unittest.cc
+++ b/chrome/browser/extensions/extension_gcm_app_handler_unittest.cc
@@ -76,6 +76,28 @@
 
 const char kTestExtensionName[] = "FooBar";
 
+void RequestProxyResolvingSocketFactoryOnUIThread(
+    Profile* profile,
+    base::WeakPtr<gcm::GCMProfileService> service,
+    network::mojom::ProxyResolvingSocketFactoryRequest request) {
+  if (!service)
+    return;
+  network::mojom::NetworkContext* network_context =
+      content::BrowserContext::GetDefaultStoragePartition(profile)
+          ->GetNetworkContext();
+  network_context->CreateProxyResolvingSocketFactory(std::move(request));
+}
+
+void RequestProxyResolvingSocketFactory(
+    Profile* profile,
+    base::WeakPtr<gcm::GCMProfileService> service,
+    network::mojom::ProxyResolvingSocketFactoryRequest request) {
+  content::BrowserThread::PostTask(
+      content::BrowserThread::UI, FROM_HERE,
+      base::BindOnce(&RequestProxyResolvingSocketFactoryOnUIThread, profile,
+                     service, std::move(request)));
+}
+
 }  // namespace
 
 // Helper class for asynchronous waiting.
@@ -211,6 +233,7 @@
             {base::MayBlock(), base::TaskShutdownBehavior::SKIP_ON_SHUTDOWN}));
     return std::make_unique<gcm::GCMProfileService>(
         profile->GetPrefs(), profile->GetPath(), profile->GetRequestContext(),
+        base::BindRepeating(&RequestProxyResolvingSocketFactory, profile),
         content::BrowserContext::GetDefaultStoragePartition(profile)
             ->GetURLLoaderFactoryForBrowserProcess(),
         chrome::GetChannel(),
diff --git a/chrome/browser/gcm/gcm_profile_service_factory.cc b/chrome/browser/gcm/gcm_profile_service_factory.cc
index 099d9fc..e798b79 100644
--- a/chrome/browser/gcm/gcm_profile_service_factory.cc
+++ b/chrome/browser/gcm/gcm_profile_service_factory.cc
@@ -38,6 +38,38 @@
 
 namespace gcm {
 
+namespace {
+
+#if !defined(OS_ANDROID)
+// Requests a ProxyResolvingSocketFactoryPtr on the UI thread. Note that a
+// WeakPtr of GCMProfileService is needed to detect when the KeyedService shuts
+// down, and avoid calling into |profile| which might have also been destroyed.
+void RequestProxyResolvingSocketFactoryOnUIThread(
+    Profile* profile,
+    base::WeakPtr<GCMProfileService> service,
+    network::mojom::ProxyResolvingSocketFactoryRequest request) {
+  if (!service)
+    return;
+  network::mojom::NetworkContext* network_context =
+      content::BrowserContext::GetDefaultStoragePartition(profile)
+          ->GetNetworkContext();
+  network_context->CreateProxyResolvingSocketFactory(std::move(request));
+}
+
+// A thread-safe wrapper to request a ProxyResolvingSocketFactoryPtr.
+void RequestProxyResolvingSocketFactory(
+    Profile* profile,
+    base::WeakPtr<GCMProfileService> service,
+    network::mojom::ProxyResolvingSocketFactoryRequest request) {
+  content::BrowserThread::PostTask(
+      content::BrowserThread::UI, FROM_HERE,
+      base::BindOnce(&RequestProxyResolvingSocketFactoryOnUIThread, profile,
+                     std::move(service), std::move(request)));
+}
+#endif
+
+}  // namespace
+
 // static
 GCMProfileService* GCMProfileServiceFactory::GetForProfile(
     content::BrowserContext* profile) {
@@ -83,8 +115,9 @@
   service = base::WrapUnique(
       new GCMProfileService(profile->GetPath(), blocking_task_runner));
 #else
-  service = base::WrapUnique(new GCMProfileService(
+  service = std::make_unique<GCMProfileService>(
       profile->GetPrefs(), profile->GetPath(), profile->GetRequestContext(),
+      base::BindRepeating(&RequestProxyResolvingSocketFactory, profile),
       content::BrowserContext::GetDefaultStoragePartition(profile)
           ->GetURLLoaderFactoryForBrowserProcess(),
       chrome::GetChannel(),
@@ -97,7 +130,7 @@
           content::BrowserThread::UI),
       content::BrowserThread::GetTaskRunnerForThread(
           content::BrowserThread::IO),
-      blocking_task_runner));
+      blocking_task_runner);
 #endif
 #if BUILDFLAG(ENABLE_OFFLINE_PAGES)
   offline_pages::PrefetchService* prefetch_service =
diff --git a/chrome/browser/gcm/gcm_profile_service_unittest.cc b/chrome/browser/gcm/gcm_profile_service_unittest.cc
index 28d1dd1..87feeaa 100644
--- a/chrome/browser/gcm/gcm_profile_service_unittest.cc
+++ b/chrome/browser/gcm/gcm_profile_service_unittest.cc
@@ -46,6 +46,27 @@
 const char kTestAppID[] = "TestApp";
 const char kUserID[] = "user";
 
+void RequestProxyResolvingSocketFactoryOnUIThread(
+    Profile* profile,
+    base::WeakPtr<gcm::GCMProfileService> service,
+    network::mojom::ProxyResolvingSocketFactoryRequest request) {
+  if (!service)
+    return;
+  return content::BrowserContext::GetDefaultStoragePartition(profile)
+      ->GetNetworkContext()
+      ->CreateProxyResolvingSocketFactory(std::move(request));
+}
+
+void RequestProxyResolvingSocketFactory(
+    Profile* profile,
+    base::WeakPtr<gcm::GCMProfileService> service,
+    network::mojom::ProxyResolvingSocketFactoryRequest request) {
+  content::BrowserThread::PostTask(
+      content::BrowserThread::UI, FROM_HERE,
+      base::BindOnce(&RequestProxyResolvingSocketFactoryOnUIThread, profile,
+                     service, std::move(request)));
+}
+
 std::unique_ptr<KeyedService> BuildGCMProfileService(
     content::BrowserContext* context) {
   Profile* profile = Profile::FromBrowserContext(context);
@@ -54,6 +75,7 @@
           {base::MayBlock(), base::TaskShutdownBehavior::SKIP_ON_SHUTDOWN}));
   return std::make_unique<gcm::GCMProfileService>(
       profile->GetPrefs(), profile->GetPath(), profile->GetRequestContext(),
+      base::BindRepeating(&RequestProxyResolvingSocketFactory, profile),
       content::BrowserContext::GetDefaultStoragePartition(profile)
           ->GetURLLoaderFactoryForBrowserProcess(),
       chrome::GetChannel(),
diff --git a/components/gcm_driver/BUILD.gn b/components/gcm_driver/BUILD.gn
index ac5404b..4455430 100644
--- a/components/gcm_driver/BUILD.gn
+++ b/components/gcm_driver/BUILD.gn
@@ -47,7 +47,9 @@
     "//components/gcm_driver/common",
     "//components/gcm_driver/crypto",
     "//components/gcm_driver/instance_id",
+    "//crypto",
   ]
+
   deps = [
     ":gcm_buildflags",
     "//base:i18n",
@@ -65,6 +67,7 @@
     "//net",
     "//services/identity/public/cpp",
     "//services/network/public/cpp",
+    "//services/network/public/mojom",
     "//url:url",
   ]
 
@@ -141,6 +144,7 @@
     "//net",
     "//services/network:test_support",
     "//services/network/public/cpp",
+    "//services/network/public/mojom",
     "//testing/gtest",
   ]
 
diff --git a/components/gcm_driver/DEPS b/components/gcm_driver/DEPS
index 952417e..d20d270 100644
--- a/components/gcm_driver/DEPS
+++ b/components/gcm_driver/DEPS
@@ -18,6 +18,7 @@
   "+net",
   "+services/identity/public",
   "+services/network/public/cpp",
+  "+services/network/public/mojom",
   "+services/network/test",
   "+third_party/leveldatabase",  # Only used for tests.
 ]
diff --git a/components/gcm_driver/fake_gcm_client.cc b/components/gcm_driver/fake_gcm_client.cc
index adbb8f9..6115c5d2 100644
--- a/components/gcm_driver/fake_gcm_client.cc
+++ b/components/gcm_driver/fake_gcm_client.cc
@@ -75,8 +75,9 @@
     const ChromeBuildInfo& chrome_build_info,
     const base::FilePath& store_path,
     const scoped_refptr<base::SequencedTaskRunner>& blocking_task_runner,
-    const scoped_refptr<net::URLRequestContextGetter>&
-        url_request_context_getter,
+    base::RepeatingCallback<
+        void(network::mojom::ProxyResolvingSocketFactoryRequest)>
+        get_socket_factory_callback,
     const scoped_refptr<network::SharedURLLoaderFactory>& url_loader_factory,
     std::unique_ptr<Encryptor> encryptor,
     Delegate* delegate) {
diff --git a/components/gcm_driver/fake_gcm_client.h b/components/gcm_driver/fake_gcm_client.h
index e1500d0..15be54d 100644
--- a/components/gcm_driver/fake_gcm_client.h
+++ b/components/gcm_driver/fake_gcm_client.h
@@ -46,8 +46,9 @@
       const ChromeBuildInfo& chrome_build_info,
       const base::FilePath& store_path,
       const scoped_refptr<base::SequencedTaskRunner>& blocking_task_runner,
-      const scoped_refptr<net::URLRequestContextGetter>&
-          url_request_context_getter,
+      base::RepeatingCallback<
+          void(network::mojom::ProxyResolvingSocketFactoryRequest)>
+          get_socket_factory_callback,
       const scoped_refptr<network::SharedURLLoaderFactory>& url_loader_factory,
       std::unique_ptr<Encryptor> encryptor,
       Delegate* delegate) override;
diff --git a/components/gcm_driver/gcm_client.h b/components/gcm_driver/gcm_client.h
index 79a7789..e8bd9d6 100644
--- a/components/gcm_driver/gcm_client.h
+++ b/components/gcm_driver/gcm_client.h
@@ -16,6 +16,7 @@
 #include "components/gcm_driver/common/gcm_messages.h"
 #include "components/gcm_driver/gcm_activity.h"
 #include "components/gcm_driver/registration_info.h"
+#include "services/network/public/mojom/proxy_resolving_socket.mojom.h"
 
 template <class T> class scoped_refptr;
 
@@ -27,7 +28,6 @@
 
 namespace net {
 class IPEndPoint;
-class URLRequestContextGetter;
 }
 
 namespace network {
@@ -235,16 +235,18 @@
   // |chrome_build_info|: chrome info, i.e., version, channel and etc.
   // |store_path|: path to the GCM store.
   // |blocking_task_runner|: for running blocking file tasks.
-  // |url_request_context_getter|: for url requests. The GCMClient must be
-  //     deleted before the Getter's underlying URLRequestContext.
+  // |get_socket_factory_callback|: a callback that can accept a request for a
+  //     network::mojom::ProxyResolvingSocketFactoryPtr. It needs to be safe to
+  //     run on any thread.
   // |delegate|: the delegate whose methods will be called asynchronously in
   //     response to events and messages.
   virtual void Initialize(
       const ChromeBuildInfo& chrome_build_info,
       const base::FilePath& store_path,
       const scoped_refptr<base::SequencedTaskRunner>& blocking_task_runner,
-      const scoped_refptr<net::URLRequestContextGetter>&
-          url_request_context_getter,
+      base::RepeatingCallback<
+          void(network::mojom::ProxyResolvingSocketFactoryRequest)>
+          get_socket_factory_callback,
       const scoped_refptr<network::SharedURLLoaderFactory>& url_loader_factory,
       std::unique_ptr<Encryptor> encryptor,
       Delegate* delegate) = 0;
diff --git a/components/gcm_driver/gcm_client_impl.cc b/components/gcm_driver/gcm_client_impl.cc
index 45c5cb4..44a050f 100644
--- a/components/gcm_driver/gcm_client_impl.cc
+++ b/components/gcm_driver/gcm_client_impl.cc
@@ -41,7 +41,6 @@
 #include "google_apis/gcm/monitoring/gcm_stats_recorder.h"
 #include "google_apis/gcm/protocol/checkin.pb.h"
 #include "google_apis/gcm/protocol/mcs.pb.h"
-#include "net/url_request/url_request_context.h"
 #include "services/network/public/cpp/shared_url_loader_factory.h"
 #include "url/gurl.h"
 
@@ -279,10 +278,13 @@
 std::unique_ptr<ConnectionFactory> GCMInternalsBuilder::BuildConnectionFactory(
     const std::vector<GURL>& endpoints,
     const net::BackoffEntry::Policy& backoff_policy,
-    net::URLRequestContext* url_request_context,
+    base::RepeatingCallback<
+        void(network::mojom::ProxyResolvingSocketFactoryRequest)>
+        get_socket_factory_callback,
     GCMStatsRecorder* recorder) {
-  return base::WrapUnique<ConnectionFactory>(new ConnectionFactoryImpl(
-      endpoints, backoff_policy, url_request_context, recorder));
+  return std::make_unique<ConnectionFactoryImpl>(
+      endpoints, backoff_policy, std::move(get_socket_factory_callback),
+      recorder);
 }
 
 GCMClientImpl::CheckinInfo::CheckinInfo()
@@ -318,7 +320,6 @@
       start_mode_(DELAYED_START),
       clock_(internals_builder_->GetClock()),
       gcm_store_reset_(false),
-      url_request_context_getter_(nullptr),
       periodic_checkin_ptr_factory_(this),
       destroying_gcm_store_ptr_factory_(this),
       weak_ptr_factory_(this) {}
@@ -330,16 +331,16 @@
     const ChromeBuildInfo& chrome_build_info,
     const base::FilePath& path,
     const scoped_refptr<base::SequencedTaskRunner>& blocking_task_runner,
-    const scoped_refptr<net::URLRequestContextGetter>&
-        url_request_context_getter,
+    base::RepeatingCallback<
+        void(network::mojom::ProxyResolvingSocketFactoryRequest)>
+        get_socket_factory_callback,
     const scoped_refptr<network::SharedURLLoaderFactory>& url_loader_factory,
     std::unique_ptr<Encryptor> encryptor,
     GCMClient::Delegate* delegate) {
   DCHECK_EQ(UNINITIALIZED, state_);
-  DCHECK(url_request_context_getter);
   DCHECK(delegate);
 
-  url_request_context_getter_ = url_request_context_getter;
+  get_socket_factory_callback_ = std::move(get_socket_factory_callback);
   url_loader_factory_ = url_loader_factory;
   chrome_build_info_ = chrome_build_info;
 
@@ -497,8 +498,8 @@
   if (fallback_endpoint.is_valid())
     endpoints.push_back(fallback_endpoint);
   connection_factory_ = internals_builder_->BuildConnectionFactory(
-      endpoints, GetGCMBackoffPolicy(),
-      url_request_context_getter_->GetURLRequestContext(), &recorder_);
+      endpoints, GetGCMBackoffPolicy(), get_socket_factory_callback_,
+      &recorder_);
   connection_factory_->SetConnectionListener(this);
   mcs_client_ = internals_builder_->BuildMCSClient(
       chrome_build_info_.version, clock_, connection_factory_.get(),
diff --git a/components/gcm_driver/gcm_client_impl.h b/components/gcm_driver/gcm_client_impl.h
index d2c730a..5f9edd6 100644
--- a/components/gcm_driver/gcm_client_impl.h
+++ b/components/gcm_driver/gcm_client_impl.h
@@ -29,7 +29,7 @@
 #include "google_apis/gcm/protocol/android_checkin.pb.h"
 #include "google_apis/gcm/protocol/checkin.pb.h"
 #include "net/http/http_status_code.h"
-#include "net/url_request/url_request_context_getter.h"
+#include "services/network/public/mojom/proxy_resolving_socket.mojom.h"
 
 class GURL;
 
@@ -42,10 +42,6 @@
 class DataMessageStanza;
 }  // namespace mcs_proto
 
-namespace net {
-class URLRequestContext;
-}  // namespace net
-
 namespace network {
 class SharedURLLoaderFactory;
 }  // namespace network
@@ -73,7 +69,9 @@
   virtual std::unique_ptr<ConnectionFactory> BuildConnectionFactory(
       const std::vector<GURL>& endpoints,
       const net::BackoffEntry::Policy& backoff_policy,
-      net::URLRequestContext* url_request_context,
+      base::RepeatingCallback<
+          void(network::mojom::ProxyResolvingSocketFactoryRequest)>
+          get_socket_factory_callback,
       GCMStatsRecorder* recorder);
 };
 
@@ -112,8 +110,9 @@
       const ChromeBuildInfo& chrome_build_info,
       const base::FilePath& store_path,
       const scoped_refptr<base::SequencedTaskRunner>& blocking_task_runner,
-      const scoped_refptr<net::URLRequestContextGetter>&
-          url_request_context_getter,
+      base::RepeatingCallback<
+          void(network::mojom::ProxyResolvingSocketFactoryRequest)>
+          get_socket_factory_callback,
       const scoped_refptr<network::SharedURLLoaderFactory>& url_loader_factory,
       std::unique_ptr<Encryptor> encryptor,
       GCMClient::Delegate* delegate) override;
@@ -369,7 +368,10 @@
   bool gcm_store_reset_;
 
   std::unique_ptr<ConnectionFactory> connection_factory_;
-  scoped_refptr<net::URLRequestContextGetter> url_request_context_getter_;
+  base::RepeatingCallback<void(
+      network::mojom::ProxyResolvingSocketFactoryRequest)>
+      get_socket_factory_callback_;
+
   scoped_refptr<network::SharedURLLoaderFactory> url_loader_factory_;
 
   // Controls receiving and sending of packets and reliable message queueing.
diff --git a/components/gcm_driver/gcm_client_impl_unittest.cc b/components/gcm_driver/gcm_client_impl_unittest.cc
index a2895c0..f440402 100644
--- a/components/gcm_driver/gcm_client_impl_unittest.cc
+++ b/components/gcm_driver/gcm_client_impl_unittest.cc
@@ -9,6 +9,7 @@
 #include <initializer_list>
 #include <memory>
 
+#include "base/bind_helpers.h"
 #include "base/command_line.h"
 #include "base/files/file_path.h"
 #include "base/files/file_util.h"
@@ -230,7 +231,9 @@
   std::unique_ptr<ConnectionFactory> BuildConnectionFactory(
       const std::vector<GURL>& endpoints,
       const net::BackoffEntry::Policy& backoff_policy,
-      net::URLRequestContext* url_request_context,
+      base::RepeatingCallback<
+          void(network::mojom::ProxyResolvingSocketFactoryRequest)>
+          get_socket_factory_callback,
       GCMStatsRecorder* recorder) override;
 
  private:
@@ -260,7 +263,9 @@
 FakeGCMInternalsBuilder::BuildConnectionFactory(
     const std::vector<GURL>& endpoints,
     const net::BackoffEntry::Policy& backoff_policy,
-    net::URLRequestContext* url_request_context,
+    base::RepeatingCallback<
+        void(network::mojom::ProxyResolvingSocketFactoryRequest)>
+        get_socket_factory_callback,
     GCMStatsRecorder* recorder) {
   return base::WrapUnique<ConnectionFactory>(new FakeConnectionFactory());
 }
@@ -622,8 +627,7 @@
   chrome_build_info.version = kChromeVersion;
   chrome_build_info.product_category_for_subtypes = kProductCategoryForSubtypes;
   gcm_client_->Initialize(
-      chrome_build_info, gcm_store_path(), task_runner_,
-      url_request_context_getter_,
+      chrome_build_info, gcm_store_path(), task_runner_, base::DoNothing(),
       base::MakeRefCounted<network::WeakWrapperSharedURLLoaderFactory>(
           &test_url_loader_factory_),
       base::WrapUnique<Encryptor>(new FakeEncryptor), this);
diff --git a/components/gcm_driver/gcm_desktop_utils.cc b/components/gcm_driver/gcm_desktop_utils.cc
index e1c9cfd..57c82121 100644
--- a/components/gcm_driver/gcm_desktop_utils.cc
+++ b/components/gcm_driver/gcm_desktop_utils.cc
@@ -93,7 +93,9 @@
     std::unique_ptr<GCMClientFactory> gcm_client_factory,
     PrefService* prefs,
     const base::FilePath& store_path,
-    const scoped_refptr<net::URLRequestContextGetter>& request_context,
+    base::RepeatingCallback<
+        void(network::mojom::ProxyResolvingSocketFactoryRequest)>
+        get_socket_factory_callback,
     scoped_refptr<network::SharedURLLoaderFactory> url_loader_factory,
     version_info::Channel channel,
     const std::string& product_category_for_subtypes,
@@ -104,8 +106,8 @@
       std::move(gcm_client_factory),
       GetChromeBuildInfo(channel, product_category_for_subtypes),
       GetChannelStatusRequestUrl(channel), GetUserAgent(channel), prefs,
-      store_path, request_context, url_loader_factory, ui_task_runner,
-      io_task_runner, blocking_task_runner));
+      store_path, get_socket_factory_callback, url_loader_factory,
+      ui_task_runner, io_task_runner, blocking_task_runner));
 }
 
 }  // namespace gcm
diff --git a/components/gcm_driver/gcm_desktop_utils.h b/components/gcm_driver/gcm_desktop_utils.h
index 8663fb4..105cd20 100644
--- a/components/gcm_driver/gcm_desktop_utils.h
+++ b/components/gcm_driver/gcm_desktop_utils.h
@@ -10,16 +10,13 @@
 #include "base/memory/ref_counted.h"
 #include "base/sequenced_task_runner.h"
 #include "components/version_info/version_info.h"
+#include "services/network/public/mojom/proxy_resolving_socket.mojom.h"
 
 class PrefService;
 namespace base {
 class FilePath;
 }
 
-namespace net {
-class URLRequestContextGetter;
-}
-
 namespace network {
 class SharedURLLoaderFactory;
 }
@@ -33,7 +30,9 @@
     std::unique_ptr<GCMClientFactory> gcm_client_factory,
     PrefService* prefs,
     const base::FilePath& store_path,
-    const scoped_refptr<net::URLRequestContextGetter>& request_context,
+    base::RepeatingCallback<
+        void(network::mojom::ProxyResolvingSocketFactoryRequest)>
+        get_socket_factory_callback,
     scoped_refptr<network::SharedURLLoaderFactory> url_loader_factory,
     version_info::Channel channel,
     const std::string& product_category_for_subtypes,
diff --git a/components/gcm_driver/gcm_driver_desktop.cc b/components/gcm_driver/gcm_driver_desktop.cc
index c8f6908..b8c20c0 100644
--- a/components/gcm_driver/gcm_driver_desktop.cc
+++ b/components/gcm_driver/gcm_driver_desktop.cc
@@ -26,7 +26,6 @@
 #include "components/gcm_driver/system_encryptor.h"
 #include "google_apis/gcm/engine/account_mapping.h"
 #include "net/base/ip_endpoint.h"
-#include "net/url_request/url_request_context_getter.h"
 #include "services/network/public/cpp/shared_url_loader_factory.h"
 
 #if defined(OS_CHROMEOS)
@@ -73,7 +72,9 @@
       std::unique_ptr<GCMClientFactory> gcm_client_factory,
       const GCMClient::ChromeBuildInfo& chrome_build_info,
       const base::FilePath& store_path,
-      const scoped_refptr<net::URLRequestContextGetter>& request_context,
+      base::RepeatingCallback<
+          void(network::mojom::ProxyResolvingSocketFactoryRequest)>
+          get_socket_factory_callback,
       std::unique_ptr<network::SharedURLLoaderFactoryInfo> loader_factory_info,
       const scoped_refptr<base::SequencedTaskRunner> blocking_task_runner);
   void Start(GCMClient::StartMode start_mode,
@@ -145,7 +146,9 @@
     std::unique_ptr<GCMClientFactory> gcm_client_factory,
     const GCMClient::ChromeBuildInfo& chrome_build_info,
     const base::FilePath& store_path,
-    const scoped_refptr<net::URLRequestContextGetter>& request_context,
+    base::RepeatingCallback<
+        void(network::mojom::ProxyResolvingSocketFactoryRequest)>
+        get_socket_factory_callback,
     std::unique_ptr<network::SharedURLLoaderFactoryInfo> loader_factory_info,
     const scoped_refptr<base::SequencedTaskRunner> blocking_task_runner) {
   DCHECK(io_thread_->RunsTasksInCurrentSequence());
@@ -156,7 +159,8 @@
       network::SharedURLLoaderFactory::Create(std::move(loader_factory_info));
 
   gcm_client_->Initialize(chrome_build_info, store_path, blocking_task_runner,
-                          request_context, url_loader_factory_for_io,
+                          std::move(get_socket_factory_callback),
+                          url_loader_factory_for_io,
                           std::make_unique<SystemEncryptor>(), this);
 }
 
@@ -527,7 +531,9 @@
     const std::string& user_agent,
     PrefService* prefs,
     const base::FilePath& store_path,
-    const scoped_refptr<net::URLRequestContextGetter>& request_context,
+    base::RepeatingCallback<
+        void(network::mojom::ProxyResolvingSocketFactoryRequest)>
+        get_socket_factory_callback,
     scoped_refptr<network::SharedURLLoaderFactory> url_loader_factory_for_ui,
     const scoped_refptr<base::SequencedTaskRunner>& ui_thread,
     const scoped_refptr<base::SequencedTaskRunner>& io_thread,
@@ -562,7 +568,7 @@
       base::BindOnce(&GCMDriverDesktop::IOWorker::Initialize,
                      base::Unretained(io_worker_.get()),
                      std::move(gcm_client_factory), chrome_build_info,
-                     store_path, request_context,
+                     store_path, std::move(get_socket_factory_callback),
                      // ->Clone() permits creation of an equivalent
                      // SharedURLLoaderFactory on IO thread.
                      url_loader_factory_for_ui->Clone(), blocking_task_runner));
diff --git a/components/gcm_driver/gcm_driver_desktop.h b/components/gcm_driver/gcm_driver_desktop.h
index c2713e12..697dcc8 100644
--- a/components/gcm_driver/gcm_driver_desktop.h
+++ b/components/gcm_driver/gcm_driver_desktop.h
@@ -22,6 +22,7 @@
 #include "components/gcm_driver/gcm_client.h"
 #include "components/gcm_driver/gcm_connection_observer.h"
 #include "components/gcm_driver/gcm_driver.h"
+#include "services/network/public/mojom/proxy_resolving_socket.mojom.h"
 
 class PrefService;
 
@@ -30,10 +31,6 @@
 class SequencedTaskRunner;
 }
 
-namespace net {
-class URLRequestContextGetter;
-}
-
 namespace network {
 class SharedURLLoaderFactory;
 }
@@ -57,7 +54,9 @@
       const std::string& user_agent,
       PrefService* prefs,
       const base::FilePath& store_path,
-      const scoped_refptr<net::URLRequestContextGetter>& request_context,
+      base::RepeatingCallback<
+          void(network::mojom::ProxyResolvingSocketFactoryRequest)>
+          get_socket_factory_callback,
       scoped_refptr<network::SharedURLLoaderFactory> url_loader_factory_for_ui,
       const scoped_refptr<base::SequencedTaskRunner>& ui_thread,
       const scoped_refptr<base::SequencedTaskRunner>& io_thread,
diff --git a/components/gcm_driver/gcm_driver_desktop_unittest.cc b/components/gcm_driver/gcm_driver_desktop_unittest.cc
index 3b90067..636fd29 100644
--- a/components/gcm_driver/gcm_driver_desktop_unittest.cc
+++ b/components/gcm_driver/gcm_driver_desktop_unittest.cc
@@ -243,15 +243,15 @@
       new net::TestURLRequestContextGetter(io_thread_.task_runner());
   GCMClient::ChromeBuildInfo chrome_build_info;
   chrome_build_info.product_category_for_subtypes = "com.chrome.macosx";
-  driver_.reset(new GCMDriverDesktop(
+  driver_ = std::make_unique<GCMDriverDesktop>(
       std::unique_ptr<GCMClientFactory>(new FakeGCMClientFactory(
           base::ThreadTaskRunnerHandle::Get(), io_thread_.task_runner())),
       chrome_build_info, kTestChannelStatusRequestURL, "user-agent-string",
-      &prefs_, temp_dir_.GetPath(), request_context,
+      &prefs_, temp_dir_.GetPath(), base::DoNothing(),
       base::MakeRefCounted<network::WeakWrapperSharedURLLoaderFactory>(
           &test_url_loader_factory_),
       base::ThreadTaskRunnerHandle::Get(), io_thread_.task_runner(),
-      message_loop_.task_runner()));
+      message_loop_.task_runner());
 
   gcm_app_handler_.reset(new FakeGCMAppHandler);
   gcm_connection_observer_.reset(new FakeGCMConnectionObserver);
diff --git a/components/gcm_driver/gcm_profile_service.cc b/components/gcm_driver/gcm_profile_service.cc
index 46e3cc4..abb9c5d 100644
--- a/components/gcm_driver/gcm_profile_service.cc
+++ b/components/gcm_driver/gcm_profile_service.cc
@@ -150,6 +150,10 @@
     PrefService* prefs,
     base::FilePath path,
     net::URLRequestContextGetter* request_context,
+    base::RepeatingCallback<
+        void(base::WeakPtr<GCMProfileService>,
+             network::mojom::ProxyResolvingSocketFactoryRequest)>
+        get_socket_factory_callback,
     scoped_refptr<network::SharedURLLoaderFactory> url_loader_factory,
     version_info::Channel channel,
     const std::string& product_category_for_subtypes,
@@ -166,7 +170,9 @@
       request_context_(request_context) {
   driver_ = CreateGCMDriverDesktop(
       std::move(gcm_client_factory), prefs,
-      path.Append(gcm_driver::kGCMStoreDirname), request_context_,
+      path.Append(gcm_driver::kGCMStoreDirname),
+      base::BindRepeating(get_socket_factory_callback,
+                          weak_ptr_factory_.GetWeakPtr()),
       url_loader_factory, channel, product_category_for_subtypes,
       ui_task_runner, io_task_runner, blocking_task_runner);
 
diff --git a/components/gcm_driver/gcm_profile_service.h b/components/gcm_driver/gcm_profile_service.h
index 800f1c4..d0afc44 100644
--- a/components/gcm_driver/gcm_profile_service.h
+++ b/components/gcm_driver/gcm_profile_service.h
@@ -13,12 +13,14 @@
 #include "base/files/file_path.h"
 #include "base/macros.h"
 #include "base/memory/ref_counted.h"
+#include "base/memory/weak_ptr.h"
 #include "build/build_config.h"
 #include "components/gcm_driver/gcm_buildflags.h"
 #include "components/keyed_service/core/keyed_service.h"
 #include "components/signin/core/browser/profile_oauth2_token_service.h"
 #include "components/signin/core/browser/signin_manager.h"
 #include "components/version_info/version_info.h"
+#include "services/network/public/mojom/proxy_resolving_socket.mojom.h"
 
 class PrefService;
 
@@ -46,6 +48,9 @@
 // Providing GCM service, via GCMDriver.
 class GCMProfileService : public KeyedService {
  public:
+  using GetProxyResolvingFactoryCallback = base::RepeatingCallback<void(
+      network::mojom::ProxyResolvingSocketFactoryRequest)>;
+
 #if BUILDFLAG(USE_GCM_FROM_PLATFORM)
   GCMProfileService(
       base::FilePath path,
@@ -55,6 +60,10 @@
       PrefService* prefs,
       base::FilePath path,
       net::URLRequestContextGetter* request_context,
+      base::RepeatingCallback<
+          void(base::WeakPtr<GCMProfileService>,
+               network::mojom::ProxyResolvingSocketFactoryRequest)>
+          get_socket_factory_callback,
       scoped_refptr<network::SharedURLLoaderFactory> url_loader_factory,
       version_info::Channel channel,
       const std::string& product_category_for_subtypes,
@@ -98,6 +107,11 @@
   std::unique_ptr<IdentityObserver> identity_observer_;
 #endif
 
+  GetProxyResolvingFactoryCallback get_socket_factory_callback_;
+
+  // WeakPtr generated by the factory must be dereferenced on the UI thread.
+  base::WeakPtrFactory<GCMProfileService> weak_ptr_factory_{this};
+
   DISALLOW_COPY_AND_ASSIGN(GCMProfileService);
 };
 
diff --git a/components/gcm_driver/instance_id/BUILD.gn b/components/gcm_driver/instance_id/BUILD.gn
index b0936a1..5a316155 100644
--- a/components/gcm_driver/instance_id/BUILD.gn
+++ b/components/gcm_driver/instance_id/BUILD.gn
@@ -22,6 +22,7 @@
     "//components/keyed_service/core:core",
     "//components/prefs:prefs",
     "//crypto",
+    "//services/network/public/mojom",
   ]
 
   if (use_gcm_from_platform) {
diff --git a/google_apis/gcm/BUILD.gn b/google_apis/gcm/BUILD.gn
index 97642da..bc2bef9 100644
--- a/google_apis/gcm/BUILD.gn
+++ b/google_apis/gcm/BUILD.gn
@@ -66,9 +66,10 @@
   deps = [
     "//base",
     "//base/third_party/dynamic_annotations",
+    "//mojo/public/cpp/system:system",
     "//net",
-    "//services/network:network_service",
     "//services/network/public/cpp",
+    "//services/network/public/mojom",
     "//third_party/leveldatabase",
     "//url",
   ]
@@ -107,6 +108,7 @@
   ]
   deps = [
     "//base",
+    "//mojo/public/cpp/system:system",
     "//net",
     "//net:test_support",
     "//testing/gtest",
@@ -164,8 +166,10 @@
     "//base",
     "//base/test:test_support",
     "//mojo/core/embedder",
+    "//mojo/public/cpp/system:system",
     "//net",
     "//net:test_support",
+    "//services/network:network_service",
     "//services/network:test_support",
     "//testing/gtest",
     "//third_party/protobuf:protobuf_lite",
diff --git a/google_apis/gcm/DEPS b/google_apis/gcm/DEPS
index a044cbe8..ad4f5f8 100644
--- a/google_apis/gcm/DEPS
+++ b/google_apis/gcm/DEPS
@@ -11,4 +11,5 @@
   "+net",
   "+third_party/leveldatabase",
   "+services/network",
+  "+mojo/public/cpp",
 ]
diff --git a/google_apis/gcm/base/socket_stream.cc b/google_apis/gcm/base/socket_stream.cc
index 6674d9c..964005c 100644
--- a/google_apis/gcm/base/socket_stream.cc
+++ b/google_apis/gcm/base/socket_stream.cc
@@ -21,15 +21,22 @@
 
 }  // namespace
 
-SocketInputStream::SocketInputStream(net::StreamSocket* socket)
-    : socket_(socket),
+SocketInputStream::SocketInputStream(mojo::ScopedDataPipeConsumerHandle stream)
+    : stream_(std::move(stream)),
+      stream_watcher_(FROM_HERE, mojo::SimpleWatcher::ArmingPolicy::MANUAL),
+      read_size_(0),
       io_buffer_(new net::IOBuffer(kDefaultBufferSize)),
-      read_buffer_(new net::DrainableIOBuffer(io_buffer_.get(),
-                                              kDefaultBufferSize)),
+      read_buffer_(
+          new net::DrainableIOBuffer(io_buffer_.get(), kDefaultBufferSize)),
       next_pos_(0),
       last_error_(net::OK),
       weak_ptr_factory_(this) {
-  DCHECK(socket->IsConnected());
+  stream_watcher_.Watch(
+      stream_.get(),
+      MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+      MOJO_TRIGGER_CONDITION_SIGNALS_SATISFIED,
+      base::BindRepeating(&SocketInputStream::ReadMore,
+                          base::Unretained(this)));
 }
 
 SocketInputStream::~SocketInputStream() {
@@ -85,39 +92,74 @@
   return read_buffer_->BytesConsumed() - next_pos_;
 }
 
-net::Error SocketInputStream::Refresh(const base::Closure& callback,
+net::Error SocketInputStream::Refresh(base::OnceClosure callback,
                                       int byte_limit) {
+  DCHECK(!read_callback_);
   DCHECK_NE(GetState(), CLOSED);
   DCHECK_NE(GetState(), READING);
   DCHECK_GT(byte_limit, 0);
 
   if (byte_limit > read_buffer_->BytesRemaining()) {
     LOG(ERROR) << "Out of buffer space, closing input stream.";
-    CloseStream(net::ERR_FILE_TOO_BIG, base::Closure());
+    CloseStream(net::ERR_FILE_TOO_BIG);
     return net::OK;
   }
 
-  if (!socket_->IsConnected()) {
-    LOG(ERROR) << "Socket was disconnected, closing input stream";
-    CloseStream(net::ERR_CONNECTION_CLOSED, base::Closure());
-    return net::OK;
+  read_size_ = byte_limit;
+  read_callback_ = std::move(callback);
+  stream_watcher_.ArmOrNotify();
+  last_error_ = net::ERR_IO_PENDING;
+  return net::ERR_IO_PENDING;
+}
+
+void SocketInputStream::ReadMore(
+    MojoResult result,
+    const mojo::HandleSignalsState& /* ignored */) {
+  DCHECK(read_callback_);
+  DCHECK_NE(0u, read_size_);
+
+  uint32_t num_bytes = read_size_;
+  if (result == MOJO_RESULT_OK) {
+    DVLOG(1) << "Refreshing input stream, limit of " << num_bytes << " bytes.";
+    result = stream_->ReadData(read_buffer_->data(), &num_bytes,
+                               MOJO_READ_DATA_FLAG_NONE);
+    DVLOG(1) << "Read returned mojo result" << result;
   }
 
-  DVLOG(1) << "Refreshing input stream, limit of " << byte_limit << " bytes.";
-  int result =
-      socket_->Read(read_buffer_.get(),
-                    byte_limit,
-                    base::Bind(&SocketInputStream::RefreshCompletionCallback,
-                               weak_ptr_factory_.GetWeakPtr(),
-                               callback));
-  DVLOG(1) << "Read returned " << result;
-  if (result == net::ERR_IO_PENDING) {
-    last_error_ = net::ERR_IO_PENDING;
-    return net::ERR_IO_PENDING;
+  if (result == MOJO_RESULT_SHOULD_WAIT) {
+    stream_watcher_.ArmOrNotify();
+    return;
   }
 
-  RefreshCompletionCallback(base::Closure(), result);
-  return net::OK;
+  read_size_ = 0;
+  if (result != MOJO_RESULT_OK) {
+    CloseStream(net::ERR_FAILED);
+    std::move(read_callback_).Run();
+    return;
+  }
+
+  // If an EOF has been received, close the stream.
+  if (result == MOJO_RESULT_OK && num_bytes == 0) {
+    CloseStream(net::ERR_CONNECTION_CLOSED);
+    std::move(read_callback_).Run();
+    return;
+  }
+
+  // If an error occurred before the completion callback could complete, ignore
+  // the result.
+  if (GetState() == CLOSED)
+    return;
+
+  last_error_ = net::OK;
+  read_buffer_->DidConsume(num_bytes);
+  // TODO(zea): investigating crbug.com/409985
+  CHECK_GT(UnreadByteCount(), 0);
+
+  DVLOG(1) << "Refresh complete with " << num_bytes << " new bytes. "
+           << "Current position " << next_pos_ << " of "
+           << read_buffer_->BytesConsumed() << ".";
+
+  std::move(read_callback_).Run();
 }
 
 void SocketInputStream::RebuildBuffer() {
@@ -162,39 +204,6 @@
   return READY;
 }
 
-void SocketInputStream::RefreshCompletionCallback(
-    const base::Closure& callback, int result) {
-  // If an error occurred before the completion callback could complete, ignore
-  // the result.
-  if (GetState() == CLOSED)
-    return;
-
-  // Result == 0 implies EOF, which is treated as an error.
-  if (result == 0)
-    result = net::ERR_CONNECTION_CLOSED;
-
-  DCHECK_NE(result, net::ERR_IO_PENDING);
-
-  if (result < net::OK) {
-    DVLOG(1) << "Failed to refresh socket: " << result;
-    CloseStream(static_cast<net::Error>(result), callback);
-    return;
-  }
-
-  DCHECK_GT(result, 0);
-  last_error_ = net::OK;
-  read_buffer_->DidConsume(result);
-  // TODO(zea): investigating crbug.com/409985
-  CHECK_GT(UnreadByteCount(), 0);
-
-  DVLOG(1) << "Refresh complete with " << result << " new bytes. "
-           << "Current position " << next_pos_
-           << " of " << read_buffer_->BytesConsumed() << ".";
-
-  if (!callback.is_null())
-    callback.Run();
-}
-
 void SocketInputStream::ResetInternal() {
   read_buffer_->SetOffset(0);
   next_pos_ = 0;
@@ -202,26 +211,27 @@
   weak_ptr_factory_.InvalidateWeakPtrs();  // Invalidate any callbacks.
 }
 
-void SocketInputStream::CloseStream(net::Error error,
-                                    const base::Closure& callback) {
+void SocketInputStream::CloseStream(net::Error error) {
   DCHECK_LT(error, net::ERR_IO_PENDING);
   ResetInternal();
   last_error_ = error;
   LOG(ERROR) << "Closing stream with result " << error;
-  if (!callback.is_null())
-    callback.Run();
 }
 
 SocketOutputStream::SocketOutputStream(
-    net::StreamSocket* socket,
-    const net::NetworkTrafficAnnotationTag& traffic_annotation)
-    : socket_(socket),
+    mojo::ScopedDataPipeProducerHandle stream)
+    : stream_(std::move(stream)),
+      stream_watcher_(FROM_HERE, mojo::SimpleWatcher::ArmingPolicy::MANUAL),
       io_buffer_(new net::IOBufferWithSize(kDefaultBufferSize)),
       next_pos_(0),
       last_error_(net::OK),
-      traffic_annotation_(traffic_annotation),
       weak_ptr_factory_(this) {
-  DCHECK(socket->IsConnected());
+  stream_watcher_.Watch(
+      stream_.get(),
+      MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+      MOJO_TRIGGER_CONDITION_SIGNALS_SATISFIED,
+      base::BindRepeating(&SocketOutputStream::WriteMore,
+                          base::Unretained(this)));
 }
 
 SocketOutputStream::~SocketOutputStream() {
@@ -254,7 +264,8 @@
   return next_pos_;
 }
 
-net::Error SocketOutputStream::Flush(const base::Closure& callback) {
+net::Error SocketOutputStream::Flush(base::OnceClosure callback) {
+  DCHECK(!write_callback_);
   DCHECK_EQ(GetState(), READY);
 
   if (!write_buffer_) {
@@ -262,27 +273,52 @@
         io_buffer_.get(), next_pos_);
   }
 
-  if (!socket_->IsConnected()) {
-    LOG(ERROR) << "Socket was disconnected, closing output stream";
-    last_error_ = net::ERR_CONNECTION_CLOSED;
-    return net::OK;
-  }
+  last_error_ = net::ERR_IO_PENDING;
+  stream_watcher_.ArmOrNotify();
+  write_callback_ = std::move(callback);
+  return net::ERR_IO_PENDING;
+}
 
-  DVLOG(1) << "Flushing " << write_buffer_->BytesRemaining()
-           << " bytes into socket.";
-  int result =
-      socket_->Write(write_buffer_.get(), write_buffer_->BytesRemaining(),
-                     base::Bind(&SocketOutputStream::FlushCompletionCallback,
-                                weak_ptr_factory_.GetWeakPtr(), callback),
-                     traffic_annotation_);
-  DVLOG(1) << "Write returned " << result;
-  if (result == net::ERR_IO_PENDING) {
-    last_error_ = net::ERR_IO_PENDING;
-    return net::ERR_IO_PENDING;
-  }
+void SocketOutputStream::WriteMore(MojoResult result,
+                                   const mojo::HandleSignalsState& state) {
+  DCHECK(write_callback_);
+  DCHECK(write_buffer_);
 
-  FlushCompletionCallback(base::Closure(), result);
-  return net::OK;
+  uint32_t num_bytes = write_buffer_->BytesRemaining();
+  DVLOG(1) << "Flushing " << num_bytes << " bytes into socket.";
+  if (result == MOJO_RESULT_OK) {
+    result = stream_->WriteData(write_buffer_->data(), &num_bytes,
+                                MOJO_WRITE_DATA_FLAG_NONE);
+  }
+  if (result == MOJO_RESULT_SHOULD_WAIT) {
+    stream_watcher_.ArmOrNotify();
+    return;
+  }
+  if (result != MOJO_RESULT_OK) {
+    LOG(ERROR) << "Failed to flush socket.";
+    last_error_ = net::ERR_FAILED;
+    std::move(write_callback_).Run();
+    return;
+  }
+  DVLOG(1) << "Wrote  " << num_bytes;
+  // If an error occurred before the completion callback could complete, ignore
+  // the result.
+  if (GetState() == CLOSED)
+    return;
+
+  DCHECK_GE(num_bytes, 0u);
+  last_error_ = net::OK;
+  write_buffer_->DidConsume(num_bytes);
+  if (write_buffer_->BytesRemaining() > 0) {
+    DVLOG(1) << "Partial flush complete. Retrying.";
+    // Only a partial write was completed. Flush again to finish the write.
+    Flush(std::move(write_callback_));
+    return;
+  }
+  DVLOG(1) << "Socket flush complete.";
+  write_buffer_ = nullptr;
+  next_pos_ = 0;
+  std::move(write_callback_).Run();
 }
 
 SocketOutputStream::State SocketOutputStream::GetState() const{
@@ -303,43 +339,4 @@
   return last_error_;
 }
 
-void SocketOutputStream::FlushCompletionCallback(
-    const base::Closure& callback, int result) {
-  // If an error occurred before the completion callback could complete, ignore
-  // the result.
-  if (GetState() == CLOSED)
-    return;
-
-  // Result == 0 implies EOF, which is treated as an error.
-  if (result == 0)
-    result = net::ERR_CONNECTION_CLOSED;
-
-  DCHECK_NE(result, net::ERR_IO_PENDING);
-
-  if (result < net::OK) {
-    LOG(ERROR) << "Failed to flush socket.";
-    last_error_ = static_cast<net::Error>(result);
-    if (!callback.is_null())
-      callback.Run();
-    return;
-  }
-
-  DCHECK_GT(result, net::OK);
-  last_error_ = net::OK;
-  write_buffer_->DidConsume(result);
-
-  if (write_buffer_->BytesRemaining() > 0) {
-    DVLOG(1) << "Partial flush complete. Retrying.";
-    // Only a partial write was completed. Flush again to finish the write.
-    Flush(callback);
-    return;
-  }
-
-  DVLOG(1) << "Socket flush complete.";
-  write_buffer_ = nullptr;
-  next_pos_ = 0;
-  if (!callback.is_null())
-    callback.Run();
-}
-
 }  // namespace gcm
diff --git a/google_apis/gcm/base/socket_stream.h b/google_apis/gcm/base/socket_stream.h
index c68acd00..63556b2 100644
--- a/google_apis/gcm/base/socket_stream.h
+++ b/google_apis/gcm/base/socket_stream.h
@@ -2,14 +2,15 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 //
-// Protobuf ZeroCopy[Input/Output]Stream implementations capable of using a
-// net::StreamSocket. Built to work with Protobuf CodedStreams.
+// Protobuf ZeroCopy[Input/Output]Stream implementations capable of using
+// mojo data pipes. Built to work with Protobuf CodedStreams.
 
 #ifndef GOOGLE_APIS_GCM_BASE_SOCKET_STREAM_H_
 #define GOOGLE_APIS_GCM_BASE_SOCKET_STREAM_H_
 
 #include <stdint.h>
 
+#include "base/callback.h"
 #include "base/callback_forward.h"
 #include "base/compiler_specific.h"
 #include "base/macros.h"
@@ -17,24 +18,21 @@
 #include "base/memory/weak_ptr.h"
 #include "google/protobuf/io/zero_copy_stream.h"
 #include "google_apis/gcm/base/gcm_export.h"
+#include "mojo/public/cpp/system/data_pipe.h"
+#include "mojo/public/cpp/system/simple_watcher.h"
 #include "net/base/net_errors.h"
-#include "net/traffic_annotation/network_traffic_annotation.h"
 
 namespace net {
 class DrainableIOBuffer;
 class IOBuffer;
 class IOBufferWithSize;
-class StreamSocket;
 }  // namespace net
 
 namespace gcm {
 
-// A helper class for interacting with a net::StreamSocket that is receiving
-// protobuf encoded messages. A SocketInputStream does not take ownership of
-// the socket itself, and it is expected that the life of the input stream
-// should match the life of the socket itself (while the socket remains
-// connected). If an error is encounters, the input stream will store the error
-// in |last_error_|, and GetState() will be set to CLOSED.
+// A helper class for interacting with a mojo consumer pipe that is receiving
+// protobuf encoded messages. If an error is encounters, the input stream will
+// store the error in |last_error_|, and GetState() will be set to CLOSED.
 // Typical usage:
 // 1. Check the GetState() of the input stream before using it. If CLOSED, the
 //    input stream must be rebuilt (and the socket likely needs to be
@@ -66,7 +64,7 @@
   };
 
   // |socket| should already be connected.
-  explicit SocketInputStream(net::StreamSocket* socket);
+  explicit SocketInputStream(mojo::ScopedDataPipeConsumerHandle stream);
   ~SocketInputStream() override;
 
   // ZeroCopyInputStream implementation.
@@ -85,7 +83,7 @@
   // net::OK without invoking callback.
   // Note: GetState() (and possibly last_error()) should be checked upon
   // completion to determine whether the Refresh encountered an error.
-  net::Error Refresh(const base::Closure& callback, int byte_limit);
+  net::Error Refresh(base::OnceClosure callback, int byte_limit);
 
   // Rebuilds the buffer state by copying over any unread data to the beginning
   // of the buffer and resetting the buffer read/write positions.
@@ -94,7 +92,9 @@
   void RebuildBuffer();
 
   // Returns the last fatal error encountered. Only valid if GetState() ==
-  // CLOSED.
+  // CLOSED. Note that all network read errors will be reported as
+  // net::ERR_FAILED, because mojo data pipe doesn't allow surfacing a more
+  // specific error code.
   net::Error last_error() const;
 
   // Returns the current state.
@@ -104,15 +104,17 @@
   // Clears the local state.
   void ResetInternal();
 
-  // Callback for Socket::Read calls.
-  void RefreshCompletionCallback(const base::Closure& callback, int result);
+  void ReadMore(MojoResult result, const mojo::HandleSignalsState& state);
 
   // Permanently closes the stream.
-  void CloseStream(net::Error error, const base::Closure& callback);
+  void CloseStream(net::Error error);
 
   // Internal net components.
-  net::StreamSocket* const socket_;
+  mojo::ScopedDataPipeConsumerHandle stream_;
+  mojo::SimpleWatcher stream_watcher_;
+  uint32_t read_size_;
   const scoped_refptr<net::IOBuffer> io_buffer_;
+  base::OnceClosure read_callback_;
   // IOBuffer implementation that wraps the data within |io_buffer_| that hasn't
   // been written to yet by Socket::Read calls.
   const scoped_refptr<net::DrainableIOBuffer> read_buffer_;
@@ -131,19 +133,16 @@
   DISALLOW_COPY_AND_ASSIGN(SocketInputStream);
 };
 
-// A helper class for writing to a SocketStream with protobuf encoded data.
-// A SocketOutputStream does not take ownership of the socket itself, and it is
-// expected that the life of the output stream should match the life of the
-// socket itself (while the socket remains connected).
-// Typical usage:
+// A helper class for writing to a mojo producer handle with protobuf encoded
+// data. Typical usage:
 // 1. Check the GetState() of the output stream before using it. If CLOSED, the
 //    output stream must be rebuilt (and the socket likely needs to be
 //    reconnected, as an error was encountered).
 // 2. If EMPTY, the output stream can be written via a CodedOutputStream using
 //    the ZeroCopyOutputStream interface.
 // 3. Once done writing, GetState() should be READY, so call Flush(..) to write
-//    the buffer into the StreamSocket. Wait for the callback to be invoked
-//    (it's invalid to write to an output stream while it's flushing).
+//    the buffer into the mojo producer handle. Wait for the callback to be
+//    invoked (it's invalid to write to an output stream while it's flushing).
 // 4. Check the GetState() again to ensure the Flush was successful. GetState()
 //    should be EMPTY again.
 // 5. Repeat.
@@ -161,10 +160,7 @@
     CLOSED,
   };
 
-  // |socket| should already be connected.
-  SocketOutputStream(
-      net::StreamSocket* socket,
-      const net::NetworkTrafficAnnotationTag& traffic_annotation);
+  explicit SocketOutputStream(mojo::ScopedDataPipeProducerHandle stream);
   ~SocketOutputStream() override;
 
   // ZeroCopyOutputStream implementation.
@@ -173,20 +169,24 @@
   int64_t ByteCount() const override;
 
   // Writes the buffer into the Socket.
-  net::Error Flush(const base::Closure& callback);
+  net::Error Flush(base::OnceClosure callback);
 
   // Returns the last fatal error encountered. Only valid if GetState() ==
-  // CLOSED.
+  // CLOSED. Note that All network read errors will be reported as
+  // net::ERR_FAILED, because mojo data pipe doesn't allow surfacing a more
+  // specific error code.
   net::Error last_error() const;
 
   // Returns the current state.
   State GetState() const;
 
  private:
-  void FlushCompletionCallback(const base::Closure& callback, int result);
+  void WriteMore(MojoResult result, const mojo::HandleSignalsState& state);
 
   // Internal net components.
-  net::StreamSocket* const socket_;
+  mojo::ScopedDataPipeProducerHandle stream_;
+  mojo::SimpleWatcher stream_watcher_;
+  base::OnceClosure write_callback_;
   const scoped_refptr<net::IOBufferWithSize> io_buffer_;
   // IOBuffer implementation that wraps the data within |io_buffer_| that hasn't
   // been written to the socket yet.
@@ -201,11 +201,6 @@
   // Note: last_error_ == net::ERR_IO_PENDING implies GetState() == FLUSHING.
   net::Error last_error_;
 
-  // Network traffic annotation for downstream socket write. SocketOutputStream
-  // is not reused, hence annotation can be added in constructor and used in all
-  // subsequent writes.
-  const net::NetworkTrafficAnnotationTag traffic_annotation_;
-
   base::WeakPtrFactory<SocketOutputStream> weak_ptr_factory_;
 
   DISALLOW_COPY_AND_ASSIGN(SocketOutputStream);
diff --git a/google_apis/gcm/base/socket_stream_unittest.cc b/google_apis/gcm/base/socket_stream_unittest.cc
index 7da26d1..b386443 100644
--- a/google_apis/gcm/base/socket_stream_unittest.cc
+++ b/google_apis/gcm/base/socket_stream_unittest.cc
@@ -16,10 +16,16 @@
 #include "base/message_loop/message_loop.h"
 #include "base/run_loop.h"
 #include "base/strings/string_piece.h"
+#include "base/test/bind_test_util.h"
+#include "base/test/scoped_task_environment.h"
 #include "net/base/ip_address.h"
 #include "net/log/net_log_source.h"
 #include "net/socket/socket_test_util.h"
 #include "net/traffic_annotation/network_traffic_annotation_test_helper.h"
+#include "net/url_request/url_request_test_util.h"
+#include "services/network/network_context.h"
+#include "services/network/network_service.h"
+#include "services/network/public/mojom/proxy_resolving_socket.mojom.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
 namespace gcm {
@@ -35,77 +41,6 @@
 const char kWriteData[] = "write_data";
 const int kWriteDataSize = arraysize(kWriteData) - 1;
 
-// A net::StreamSocket that returns a partial write only for the first time.
-class FirstWritePartialSocket : public net::StreamSocket {
- public:
-  FirstWritePartialSocket() {}
-  ~FirstWritePartialSocket() override {}
-
-  // Returns the data that is actually written to the socket.
-  const std::string& actual_data_written() { return actual_data_written_; }
-
-  // net::Socket implementation.
-  int Write(
-      net::IOBuffer* buf,
-      int buf_len,
-      net::CompletionOnceCallback callback,
-      const net::NetworkTrafficAnnotationTag& traffic_annotation) override {
-    // Make the first write as a partial write.
-    if (!write_invoked_) {
-      write_invoked_ = true;
-      actual_data_written_.append(buf->data(), buf_len / 2);
-      return buf_len / 2;
-    }
-    // For subsequent writes, write everything that caller has passed to us.
-    actual_data_written_.append(buf->data(), buf_len);
-    return buf_len;
-  }
-  int Read(net::IOBuffer* buf,
-           int buf_len,
-           net::CompletionOnceCallback callback) override {
-    return net::ERR_IO_PENDING;
-  }
-  int ReadIfReady(net::IOBuffer* buf,
-                  int buf_len,
-                  net::CompletionOnceCallback callback) override {
-    return net::ERR_IO_PENDING;
-  }
-  int CancelReadIfReady() override { return net::OK; }
-  int SetReceiveBufferSize(int32_t size) override { return net::OK; }
-  int SetSendBufferSize(int32_t size) override { return net::OK; }
-
-  // net::StreamSocket implementation.
-  int Connect(net::CompletionOnceCallback callback) override { return net::OK; }
-  void Disconnect() override {}
-  bool IsConnected() const override { return true; }
-  bool IsConnectedAndIdle() const override { return true; };
-  int GetPeerAddress(net::IPEndPoint* address) const override {
-    return net::OK;
-  }
-  int GetLocalAddress(net::IPEndPoint* address) const override {
-    return net::OK;
-  }
-  const net::NetLogWithSource& NetLog() const override { return net_log_; }
-  bool WasEverUsed() const override { return true; }
-  bool WasAlpnNegotiated() const override { return false; }
-  net::NextProto GetNegotiatedProtocol() const override {
-    return net::kProtoUnknown;
-  }
-  bool GetSSLInfo(net::SSLInfo* ssl_info) override { return false; }
-  void GetConnectionAttempts(net::ConnectionAttempts* out) const override {}
-  void ClearConnectionAttempts() override {}
-  void AddConnectionAttempts(const net::ConnectionAttempts& attempts) override {
-  }
-  int64_t GetTotalReceivedBytes() const override { return 0; }
-  void ApplySocketTag(const net::SocketTag& tag) override {}
-
- private:
-  net::NetLogWithSource net_log_;
-  std::string actual_data_written_;
-  // Whether Write() has been invoked before.
-  bool write_invoked_ = false;
-};
-
 class GCMSocketStreamTest : public testing::Test {
  public:
   GCMSocketStreamTest();
@@ -119,19 +54,21 @@
 
   // Simulates a google::protobuf::io::CodedInputStream read.
   base::StringPiece DoInputStreamRead(int bytes);
+
   // Simulates a google::protobuf::io::CodedOutputStream write.
   int DoOutputStreamWrite(const base::StringPiece& write_src);
 
+  // Simulates a google::protobuf::io::CodedOutputStream write, but do not call
+  // flush.
+  int DoOutputStreamWriteWithoutFlush(const base::StringPiece& write_src);
+
   // Synchronous Refresh wrapper.
   void WaitForData(int msg_size);
 
-  base::MessageLoop* message_loop() { return &message_loop_; };
-  net::StaticSocketDataProvider* data_provider() {
-    return data_provider_.get();
-  }
   SocketInputStream* input_stream() { return socket_input_stream_.get(); }
   SocketOutputStream* output_stream() { return socket_output_stream_.get(); }
-  net::StreamSocket* socket() { return socket_.get(); }
+
+  network::mojom::ProxyResolvingSocketPtr mojo_socket_ptr_;
 
   void set_socket_output_stream(std::unique_ptr<SocketOutputStream> stream) {
     socket_output_stream_ = std::move(stream);
@@ -142,26 +79,42 @@
   void ResetInputStream();
   void ResetOutputStream();
 
-  void ConnectCallback(int result);
+  base::test::ScopedTaskEnvironment scoped_task_environment_;
 
   // SocketStreams and their data providers.
   ReadList mock_reads_;
   WriteList mock_writes_;
   std::unique_ptr<net::StaticSocketDataProvider> data_provider_;
+  std::unique_ptr<net::SSLSocketDataProvider> ssl_data_provider_;
   std::unique_ptr<SocketInputStream> socket_input_stream_;
   std::unique_ptr<SocketOutputStream> socket_output_stream_;
 
   // net:: components.
-  std::unique_ptr<net::StreamSocket> socket_;
-  net::MockClientSocketFactory socket_factory_;
   net::AddressList address_list_;
-
-  base::MessageLoopForIO message_loop_;
+  std::unique_ptr<network::NetworkService> network_service_;
+  network::mojom::NetworkContextPtr network_context_ptr_;
+  net::MockClientSocketFactory socket_factory_;
+  net::TestURLRequestContext url_request_context_;
+  std::unique_ptr<network::NetworkContext> network_context_;
+  network::mojom::ProxyResolvingSocketFactoryPtr mojo_socket_factory_ptr_;
+  mojo::ScopedDataPipeConsumerHandle receive_pipe_handle_;
+  mojo::ScopedDataPipeProducerHandle send_pipe_handle_;
 };
 
-GCMSocketStreamTest::GCMSocketStreamTest() {
+GCMSocketStreamTest::GCMSocketStreamTest()
+    : scoped_task_environment_(
+          base::test::ScopedTaskEnvironment::MainThreadType::IO),
+      network_service_(network::NetworkService::CreateForTesting()),
+      url_request_context_(true /* delay_initialization */) {
   address_list_ = net::AddressList::CreateFromIPAddress(
       net::IPAddress::IPv4Localhost(), 5228);
+  socket_factory_.set_enable_read_if_ready(true);
+  url_request_context_.set_client_socket_factory(&socket_factory_);
+  url_request_context_.Init();
+
+  network_context_ = std::make_unique<network::NetworkContext>(
+      network_service_.get(), mojo::MakeRequest(&network_context_ptr_),
+      &url_request_context_);
 }
 
 GCMSocketStreamTest::~GCMSocketStreamTest() {}
@@ -170,9 +123,12 @@
                                       const WriteList& write_list) {
   mock_reads_ = read_list;
   mock_writes_ = write_list;
-  data_provider_.reset(
-      new net::StaticSocketDataProvider(mock_reads_, mock_writes_));
+  data_provider_ = std::make_unique<net::StaticSocketDataProvider>(
+      mock_reads_, mock_writes_);
+  ssl_data_provider_ =
+      std::make_unique<net::SSLSocketDataProvider>(net::SYNCHRONOUS, net::OK);
   socket_factory_.AddSocketDataProvider(data_provider_.get());
+  socket_factory_.AddSSLSocketDataProvider(ssl_data_provider_.get());
   OpenConnection();
   ResetInputStream();
   ResetOutputStream();
@@ -214,11 +170,23 @@
 
 int GCMSocketStreamTest::DoOutputStreamWrite(
     const base::StringPiece& write_src) {
+  int total_bytes_written = DoOutputStreamWriteWithoutFlush(write_src);
+  base::RunLoop run_loop;
+  if (socket_output_stream_->Flush(run_loop.QuitClosure()) ==
+      net::ERR_IO_PENDING) {
+    run_loop.Run();
+  }
+
+  return total_bytes_written;
+}
+
+int GCMSocketStreamTest::DoOutputStreamWriteWithoutFlush(
+    const base::StringPiece& write_src) {
   DCHECK_EQ(socket_output_stream_->GetState(), SocketOutputStream::EMPTY);
   int total_bytes_written = 0;
-  void* buffer = NULL;
+  void* buffer = nullptr;
   int size = 0;
-  int bytes = write_src.size();
+  const int bytes = write_src.size();
 
   do {
     if (!socket_output_stream_->Next(&buffer, &size))
@@ -232,12 +200,6 @@
     total_bytes_written += bytes_to_write;
   } while (total_bytes_written < bytes);
 
-  base::RunLoop run_loop;
-  if (socket_output_stream_->Flush(run_loop.QuitClosure()) ==
-          net::ERR_IO_PENDING) {
-    run_loop.Run();
-  }
-
   return total_bytes_written;
 }
 
@@ -255,33 +217,49 @@
 }
 
 void GCMSocketStreamTest::OpenConnection() {
-  socket_ = socket_factory_.CreateTransportClientSocket(
-      address_list_, NULL, NULL, net::NetLogSource());
-  socket_->Connect(
-      base::Bind(&GCMSocketStreamTest::ConnectCallback,
-                 base::Unretained(this)));
+  network_context_->CreateProxyResolvingSocketFactory(
+      mojo::MakeRequest(&mojo_socket_factory_ptr_));
+  base::RunLoop run_loop;
+  int net_error = net::ERR_FAILED;
+  const GURL kDestination("https://example.com");
+  mojo_socket_factory_ptr_->CreateProxyResolvingSocket(
+      kDestination, true /* use_tls */,
+      net::MutableNetworkTrafficAnnotationTag(TRAFFIC_ANNOTATION_FOR_TESTS),
+      mojo::MakeRequest(&mojo_socket_ptr_),
+      base::BindLambdaForTesting(
+          [&](int result, const base::Optional<net::IPEndPoint>& local_addr,
+              const base::Optional<net::IPEndPoint>& peer_addr,
+              mojo::ScopedDataPipeConsumerHandle receive_pipe_handle,
+              mojo::ScopedDataPipeProducerHandle send_pipe_handle) {
+            net_error = result;
+            receive_pipe_handle_ = std::move(receive_pipe_handle);
+            send_pipe_handle_ = std::move(send_pipe_handle);
+            run_loop.Quit();
+          }));
+  run_loop.Run();
+
   PumpLoop();
 }
 
-void GCMSocketStreamTest::ConnectCallback(int result) {}
-
 void GCMSocketStreamTest::ResetInputStream() {
-  DCHECK(socket_.get());
-  socket_input_stream_.reset(new SocketInputStream(socket_.get()));
+  DCHECK(mojo_socket_ptr_);
+  socket_input_stream_ =
+      std::make_unique<SocketInputStream>(std::move(receive_pipe_handle_));
 }
 
 void GCMSocketStreamTest::ResetOutputStream() {
-  DCHECK(socket_.get());
-  socket_output_stream_.reset(
-      new SocketOutputStream(socket_.get(), TRAFFIC_ANNOTATION_FOR_TESTS));
+  DCHECK(mojo_socket_ptr_);
+  socket_output_stream_ =
+      std::make_unique<SocketOutputStream>(std::move(send_pipe_handle_));
 }
 
 // A read where all data is already available.
 TEST_F(GCMSocketStreamTest, ReadDataSync) {
-  BuildSocket(ReadList(1, net::MockRead(net::SYNCHRONOUS,
-                                        kReadData,
-                                        kReadDataSize)),
-              WriteList());
+  ReadList read_list;
+  read_list.push_back(
+      net::MockRead(net::SYNCHRONOUS, kReadData, kReadDataSize));
+  read_list.push_back(net::MockRead(net::ASYNC, net::OK) /* EOF */);
+  BuildSocket(read_list, WriteList());
 
   WaitForData(kReadDataSize);
   ASSERT_EQ(std::string(kReadData, kReadDataSize),
@@ -301,6 +279,9 @@
       net::MockRead(net::SYNCHRONOUS,
                     &kReadData[first_read_len],
                     second_read_len));
+  // Add an EOF.
+  read_list.push_back(net::MockRead(net::SYNCHRONOUS, net::OK));
+
   BuildSocket(read_list, WriteList());
 
   WaitForData(kReadDataSize);
@@ -317,6 +298,7 @@
       net::MockRead(net::ASYNC, kReadData, first_read_len));
   read_list.push_back(
       net::MockRead(net::ASYNC, &kReadData[first_read_len], second_read_len));
+  read_list.push_back(net::MockRead(net::ASYNC, net::OK) /* EOF */);
   BuildSocket(read_list, WriteList());
   WaitForData(kReadDataSize);
   ASSERT_EQ(std::string(kReadData, kReadDataSize),
@@ -327,10 +309,13 @@
 TEST_F(GCMSocketStreamTest, TwoReadsAtOnce) {
   std::string long_data = std::string(kReadData, kReadDataSize) +
                           std::string(kReadData2, kReadData2Size);
-  BuildSocket(ReadList(1, net::MockRead(net::SYNCHRONOUS,
-                                        long_data.c_str(),
-                                        long_data.size())),
-              WriteList());
+  ReadList read_list;
+  read_list.push_back(
+      net::MockRead(net::SYNCHRONOUS, long_data.c_str(), long_data.size()));
+  // Add an EOF.
+  read_list.push_back(net::MockRead(net::SYNCHRONOUS, net::OK));
+
+  BuildSocket(read_list, WriteList());
 
   WaitForData(kReadDataSize);
   ASSERT_EQ(std::string(kReadData, kReadDataSize),
@@ -346,10 +331,14 @@
 TEST_F(GCMSocketStreamTest, TwoReadsAtOnceWithRebuild) {
   std::string long_data = std::string(kReadData, kReadDataSize) +
                           std::string(kReadData2, kReadData2Size);
-  BuildSocket(ReadList(1, net::MockRead(net::SYNCHRONOUS,
-                                        long_data.c_str(),
-                                        long_data.size())),
-              WriteList());
+  ReadList read_list;
+
+  read_list.push_back(
+      net::MockRead(net::SYNCHRONOUS, long_data.c_str(), long_data.size()));
+  // Add an EOF.
+  read_list.push_back(net::MockRead(net::SYNCHRONOUS, net::OK));
+
+  BuildSocket(read_list, WriteList());
 
   WaitForData(kReadDataSize);
   ASSERT_EQ(std::string(kReadData, kReadDataSize),
@@ -369,23 +358,23 @@
 
   WaitForData(kReadDataSize);
   ASSERT_EQ(SocketInputStream::CLOSED, input_stream()->GetState());
-  ASSERT_EQ(result, input_stream()->last_error());
+  ASSERT_EQ(net::ERR_FAILED, input_stream()->last_error());
 }
 
 // Simulate a read after the connection is closed.
 TEST_F(GCMSocketStreamTest, ReadDisconnected) {
-  BuildSocket(ReadList(), WriteList());
-  socket()->Disconnect();
+  BuildSocket(ReadList(1, net::MockRead(net::SYNCHRONOUS, net::ERR_IO_PENDING)),
+              WriteList());
+  mojo_socket_ptr_.reset();
   WaitForData(kReadDataSize);
   ASSERT_EQ(SocketInputStream::CLOSED, input_stream()->GetState());
-  ASSERT_EQ(net::ERR_CONNECTION_CLOSED, input_stream()->last_error());
+  ASSERT_EQ(net::ERR_FAILED, input_stream()->last_error());
 }
 
 // Write a full message in one go.
 TEST_F(GCMSocketStreamTest, WriteFull) {
-  BuildSocket(ReadList(),
-              WriteList(1, net::MockWrite(net::SYNCHRONOUS,
-                                          kWriteData,
+  BuildSocket(ReadList(1, net::MockRead(net::SYNCHRONOUS, net::ERR_IO_PENDING)),
+              WriteList(1, net::MockWrite(net::SYNCHRONOUS, kWriteData,
                                           kWriteDataSize)));
   ASSERT_EQ(kWriteDataSize,
             DoOutputStreamWrite(base::StringPiece(kWriteData,
@@ -401,7 +390,8 @@
   write_list.push_back(net::MockWrite(net::SYNCHRONOUS,
                                       kWriteData + kWriteDataSize / 2,
                                       kWriteDataSize / 2));
-  BuildSocket(ReadList(), write_list);
+  BuildSocket(ReadList(1, net::MockRead(net::SYNCHRONOUS, net::ERR_IO_PENDING)),
+              write_list);
   ASSERT_EQ(kWriteDataSize,
             DoOutputStreamWrite(base::StringPiece(kWriteData,
                                                   kWriteDataSize)));
@@ -409,13 +399,75 @@
 
 // Regression test for crbug.com/866635.
 TEST_F(GCMSocketStreamTest, WritePartialWithLengthChecking) {
-  auto socket = std::make_unique<FirstWritePartialSocket>();
-  auto socket_output_stream = std::make_unique<SocketOutputStream>(
-      socket.get(), TRAFFIC_ANNOTATION_FOR_TESTS);
+  // Add a prefix data in front of kWriteData.
+  std::string prefix_data("xxxxx");
+  const size_t kPrefixDataSize = 5;
+  // |pipe| has a capacity that is one byte smaller than |prefix_data.size()| +
+  // |kWriteDataSize|. This is so that the first write is a partial write
+  // of |prefix_data|, and the second write is a complete write of kWriteData.
+  // The 1 byte shortage is to simulate the partial write.
+  mojo::DataPipe pipe(kWriteDataSize + prefix_data.size() - 1 /* size */);
+  mojo::ScopedDataPipeConsumerHandle consumer_handle =
+      std::move(pipe.consumer_handle);
+  mojo::ScopedDataPipeProducerHandle producer_handle =
+      std::move(pipe.producer_handle);
+
+  // Prepopulate |producer_handle| of |prefix_data|, now the pipe's capacity is
+  // less than |kWriteDataSize|.
+  uint32_t num_bytes = prefix_data.size();
+  MojoResult r = producer_handle->WriteData(prefix_data.data(), &num_bytes,
+                                            MOJO_WRITE_DATA_FLAG_NONE);
+  ASSERT_EQ(MOJO_RESULT_OK, r);
+  ASSERT_EQ(prefix_data.size(), num_bytes);
+
+  // Create a SocketOutputStream from the producer pipe.
+  auto socket_output_stream =
+      std::make_unique<SocketOutputStream>(std::move(producer_handle));
   set_socket_output_stream(std::move(socket_output_stream));
-  ASSERT_EQ(kWriteDataSize,
-            DoOutputStreamWrite(base::StringPiece(kWriteData, kWriteDataSize)));
-  EXPECT_EQ(kWriteData, socket->actual_data_written());
+
+  // Write but do not flush.
+  EXPECT_EQ(kWriteDataSize, DoOutputStreamWriteWithoutFlush(kWriteData));
+
+  base::RunLoop run_loop;
+  output_stream()->Flush(run_loop.QuitClosure());
+  // Flush should be able to flush exactly 5 bytes, because of the data pipe
+  // capacity.
+  base::RunLoop().RunUntilIdle();
+
+  std::string contents;
+  // Read prefix.
+  char buffer[kPrefixDataSize];
+  uint32_t read_size = sizeof(buffer);
+  ASSERT_EQ(MOJO_RESULT_OK, consumer_handle->ReadData(
+                                buffer, &read_size, MOJO_READ_DATA_FLAG_NONE));
+  ASSERT_EQ(kPrefixDataSize, read_size);
+  contents += std::string(buffer, read_size);
+
+  base::RunLoop().RunUntilIdle();
+  // Flush now should complete.
+  run_loop.Run();
+
+  // Closes |producer_handle|.
+  set_socket_output_stream(nullptr);
+
+  // Read everything in |consumer_handle| now that |producer_handle| is closed
+  // to make sure data is as what we expected, and there is no trailing garbage
+  // data.
+  while (true) {
+    char buffer[5];
+    uint32_t read_size = sizeof(buffer);
+    MojoResult r =
+        consumer_handle->ReadData(buffer, &read_size, MOJO_READ_DATA_FLAG_NONE);
+    if (r == MOJO_RESULT_SHOULD_WAIT)
+      continue;
+    if (r != MOJO_RESULT_OK)
+      break;
+    ASSERT_EQ(MOJO_RESULT_OK, r);
+    contents += std::string(buffer, read_size);
+  }
+  std::string expected(prefix_data);
+  expected.append(kWriteData);
+  EXPECT_EQ(expected, contents);
 }
 
 // Write a message completely asynchronously (returns IO_PENDING before
@@ -428,7 +480,8 @@
   write_list.push_back(net::MockWrite(net::SYNCHRONOUS,
                                       kWriteData + kWriteDataSize / 2,
                                       kWriteDataSize / 2));
-  BuildSocket(ReadList(), write_list);
+  BuildSocket(ReadList(1, net::MockRead(net::SYNCHRONOUS, net::ERR_IO_PENDING)),
+              write_list);
   ASSERT_EQ(kWriteDataSize,
             DoOutputStreamWrite(base::StringPiece(kWriteData,
                                                   kWriteDataSize)));
@@ -436,11 +489,14 @@
 
 // Write a message then read a message.
 TEST_F(GCMSocketStreamTest, WriteThenRead) {
-  BuildSocket(ReadList(1, net::MockRead(net::SYNCHRONOUS,
-                                        kReadData,
-                                        kReadDataSize)),
-              WriteList(1, net::MockWrite(net::SYNCHRONOUS,
-                                          kWriteData,
+  ReadList read_list;
+  read_list.push_back(
+      net::MockRead(net::SYNCHRONOUS, kReadData, kReadDataSize));
+  // Add an EOF.
+  read_list.push_back(net::MockRead(net::SYNCHRONOUS, net::OK));
+
+  BuildSocket(read_list,
+              WriteList(1, net::MockWrite(net::SYNCHRONOUS, kWriteData,
                                           kWriteDataSize)));
 
   ASSERT_EQ(kWriteDataSize,
@@ -454,11 +510,14 @@
 
 // Read a message then write a message.
 TEST_F(GCMSocketStreamTest, ReadThenWrite) {
-  BuildSocket(ReadList(1, net::MockRead(net::SYNCHRONOUS,
-                                        kReadData,
-                                        kReadDataSize)),
-              WriteList(1, net::MockWrite(net::SYNCHRONOUS,
-                                          kWriteData,
+  ReadList read_list;
+  read_list.push_back(
+      net::MockRead(net::SYNCHRONOUS, kReadData, kReadDataSize));
+  // Add an EOF.
+  read_list.push_back(net::MockRead(net::SYNCHRONOUS, net::OK));
+
+  BuildSocket(read_list,
+              WriteList(1, net::MockWrite(net::SYNCHRONOUS, kWriteData,
                                           kWriteDataSize)));
 
   WaitForData(kReadDataSize);
@@ -473,20 +532,25 @@
 // Simulate a write that gets aborted.
 TEST_F(GCMSocketStreamTest, WriteError) {
   int result = net::ERR_ABORTED;
-  BuildSocket(ReadList(),
+  BuildSocket(ReadList(1, net::MockRead(net::SYNCHRONOUS, net::ERR_IO_PENDING)),
               WriteList(1, net::MockWrite(net::SYNCHRONOUS, result)));
-  DoOutputStreamWrite(base::StringPiece(kWriteData, kWriteDataSize));
+  // Mojo data pipe buffers data, so there is a delay before write error is
+  // observed.Continue writing if error is not observed.
+  while (output_stream()->GetState() != SocketOutputStream::CLOSED) {
+    DoOutputStreamWrite(base::StringPiece(kWriteData, kWriteDataSize));
+  }
   ASSERT_EQ(SocketOutputStream::CLOSED, output_stream()->GetState());
-  ASSERT_EQ(result, output_stream()->last_error());
+  ASSERT_EQ(net::ERR_FAILED, output_stream()->last_error());
 }
 
 // Simulate a write after the connection is closed.
 TEST_F(GCMSocketStreamTest, WriteDisconnected) {
-  BuildSocket(ReadList(), WriteList());
-  socket()->Disconnect();
+  BuildSocket(ReadList(1, net::MockRead(net::SYNCHRONOUS, net::ERR_IO_PENDING)),
+              WriteList());
+  mojo_socket_ptr_.reset();
   DoOutputStreamWrite(base::StringPiece(kWriteData, kWriteDataSize));
   ASSERT_EQ(SocketOutputStream::CLOSED, output_stream()->GetState());
-  ASSERT_EQ(net::ERR_CONNECTION_CLOSED, output_stream()->last_error());
+  ASSERT_EQ(net::ERR_FAILED, output_stream()->last_error());
 }
 
 }  // namespace
diff --git a/google_apis/gcm/engine/connection_factory.h b/google_apis/gcm/engine/connection_factory.h
index c837814..fad6843 100644
--- a/google_apis/gcm/engine/connection_factory.h
+++ b/google_apis/gcm/engine/connection_factory.h
@@ -7,9 +7,11 @@
 
 #include <string>
 
+#include "base/callback_forward.h"
 #include "base/time/time.h"
 #include "google_apis/gcm/base/gcm_export.h"
 #include "google_apis/gcm/engine/connection_handler.h"
+#include "services/network/public/mojom/proxy_resolving_socket.mojom.h"
 
 class GURL;
 
@@ -23,6 +25,9 @@
 
 namespace gcm {
 
+using GetProxyResolvingFactoryCallback = base::RepeatingCallback<void(
+    network::mojom::ProxyResolvingSocketFactoryRequest)>;
+
 // Factory for creating a ConnectionHandler and maintaining its connection.
 // The factory retains ownership of the ConnectionHandler and will enforce
 // backoff policies when attempting connections.
diff --git a/google_apis/gcm/engine/connection_factory_impl.cc b/google_apis/gcm/engine/connection_factory_impl.cc
index cde535e5..98c30af14 100644
--- a/google_apis/gcm/engine/connection_factory_impl.cc
+++ b/google_apis/gcm/engine/connection_factory_impl.cc
@@ -23,8 +23,7 @@
 #include "net/socket/client_socket_handle.h"
 #include "net/socket/client_socket_pool_manager.h"
 #include "net/ssl/ssl_config_service.h"
-#include "services/network/proxy_resolving_client_socket.h"
-#include "services/network/proxy_resolving_client_socket_factory.h"
+#include "net/traffic_annotation/network_traffic_annotation.h"
 
 namespace gcm {
 
@@ -52,15 +51,13 @@
 ConnectionFactoryImpl::ConnectionFactoryImpl(
     const std::vector<GURL>& mcs_endpoints,
     const net::BackoffEntry::Policy& backoff_policy,
-    net::URLRequestContext* url_request_context,
+    GetProxyResolvingFactoryCallback get_socket_factory_callback,
     GCMStatsRecorder* recorder)
     : mcs_endpoints_(mcs_endpoints),
       next_endpoint_(0),
       last_successful_endpoint_(0),
       backoff_policy_(backoff_policy),
-      socket_factory_(
-          std::make_unique<network::ProxyResolvingClientSocketFactory>(
-              url_request_context)),
+      get_socket_factory_callback_(get_socket_factory_callback),
       connecting_(false),
       waiting_for_backoff_(false),
       waiting_for_network_online_(false),
@@ -288,18 +285,6 @@
   return mcs_endpoints_[next_endpoint_];
 }
 
-net::IPEndPoint ConnectionFactoryImpl::GetPeerIP() {
-  if (!socket_)
-    return net::IPEndPoint();
-
-  net::IPEndPoint ip_endpoint;
-  int result = socket_->GetPeerAddress(&ip_endpoint);
-  if (result != net::OK)
-    return net::IPEndPoint();
-
-  return ip_endpoint;
-}
-
 void ConnectionFactoryImpl::ConnectImpl() {
   event_tracker_.StartConnectionAttempt();
   StartConnection();
@@ -316,21 +301,8 @@
   connecting_ = true;
   GURL current_endpoint = GetCurrentEndpoint();
   recorder_->RecordConnectionInitiated(current_endpoint.host());
-  socket_ = socket_factory_->CreateSocket(current_endpoint, true /*use_tls*/);
-  int status = socket_->Connect(base::BindRepeating(
-      &ConnectionFactoryImpl::OnConnectDone, weak_ptr_factory_.GetWeakPtr()));
-  if (status != net::ERR_IO_PENDING)
-    OnConnectDone(status);
-}
 
-void ConnectionFactoryImpl::InitHandler() {
-  // May be null in tests.
-  mcs_proto::LoginRequest login_request;
-  if (!request_builder_.is_null()) {
-    request_builder_.Run(&login_request);
-    DCHECK(login_request.IsInitialized());
-    event_tracker_.WriteToLoginRequest(&login_request);
-  }
+  get_socket_factory_callback_.Run(mojo::MakeRequest(&socket_factory_));
 
   net::NetworkTrafficAnnotationTag traffic_annotation =
       net::DefineNetworkTrafficAnnotation("gcm_connection_factory", R"(
@@ -368,7 +340,27 @@
           "but does not have any effect on other Google Cloud messages."
         )");
 
-  connection_handler_->Init(login_request, traffic_annotation, socket_.get());
+  socket_factory_->CreateProxyResolvingSocket(
+      current_endpoint, true /* use_tls */,
+      net::MutableNetworkTrafficAnnotationTag(traffic_annotation),
+      mojo::MakeRequest(&socket_),
+      base::BindOnce(&ConnectionFactoryImpl::OnConnectDone,
+                     base::Unretained(this)));
+}
+
+void ConnectionFactoryImpl::InitHandler(
+    mojo::ScopedDataPipeConsumerHandle receive_stream,
+    mojo::ScopedDataPipeProducerHandle send_stream) {
+  mcs_proto::LoginRequest login_request;
+  // May be null in tests.
+  if (!request_builder_.is_null()) {
+    request_builder_.Run(&login_request);
+    DCHECK(login_request.IsInitialized());
+    event_tracker_.WriteToLoginRequest(&login_request);
+  }
+
+  connection_handler_->Init(login_request, std::move(receive_stream),
+                            std::move(send_stream));
 }
 
 std::unique_ptr<net::BackoffEntry> ConnectionFactoryImpl::CreateBackoffEntry(
@@ -390,7 +382,12 @@
   return base::TimeTicks::Now();
 }
 
-void ConnectionFactoryImpl::OnConnectDone(int result) {
+void ConnectionFactoryImpl::OnConnectDone(
+    int result,
+    const base::Optional<net::IPEndPoint>& local_addr,
+    const base::Optional<net::IPEndPoint>& peer_addr,
+    mojo::ScopedDataPipeConsumerHandle receive_stream,
+    mojo::ScopedDataPipeProducerHandle send_stream) {
   DCHECK_NE(net::ERR_IO_PENDING, result);
   if (result != net::OK) {
     LOG(ERROR) << "Failed to connect to MCS endpoint with error " << result;
@@ -426,7 +423,11 @@
   connecting_ = false;
   handshake_in_progress_ = true;
   DVLOG(1) << "MCS endpoint socket connection success, starting login.";
-  InitHandler();
+  // |peer_addr| is only non-null if result == net::OK and the connection is not
+  // through a proxy.
+  if (peer_addr)
+    peer_addr_ = peer_addr.value();
+  InitHandler(std::move(receive_stream), std::move(send_stream));
 }
 
 void ConnectionFactoryImpl::ConnectionHandlerCallback(int result) {
@@ -451,7 +452,7 @@
   event_tracker_.ConnectionAttemptSucceeded();
 
   if (listener_)
-    listener_->OnConnected(GetCurrentEndpoint(), GetPeerIP());
+    listener_->OnConnected(GetCurrentEndpoint(), peer_addr_);
 }
 
 void ConnectionFactoryImpl::CloseSocket() {
@@ -460,9 +461,8 @@
   if (connection_handler_)
     connection_handler_->Reset();
 
-  if (socket_)
-    socket_->Disconnect();
-  socket_ = nullptr;
+  socket_.reset();
+  peer_addr_ = net::IPEndPoint();
 }
 
 }  // namespace gcm
diff --git a/google_apis/gcm/engine/connection_factory_impl.h b/google_apis/gcm/engine/connection_factory_impl.h
index 67c5412..4bd944c 100644
--- a/google_apis/gcm/engine/connection_factory_impl.h
+++ b/google_apis/gcm/engine/connection_factory_impl.h
@@ -9,26 +9,20 @@
 
 #include <stddef.h>
 
+#include "base/callback.h"
 #include "base/macros.h"
 #include "base/memory/weak_ptr.h"
 #include "base/time/time.h"
 #include "google_apis/gcm/engine/connection_event_tracker.h"
 #include "google_apis/gcm/engine/connection_handler.h"
 #include "google_apis/gcm/protocol/mcs.pb.h"
+#include "mojo/public/cpp/system/data_pipe.h"
 #include "net/base/backoff_entry.h"
 #include "net/base/network_change_notifier.h"
 #include "net/log/net_log_with_source.h"
+#include "services/network/public/mojom/proxy_resolving_socket.mojom.h"
 #include "url/gurl.h"
 
-namespace network {
-class ProxyResolvingClientSocket;
-class ProxyResolvingClientSocketFactory;
-}
-
-namespace net {
-class URLRequestContext;
-}
-
 namespace gcm {
 
 class GCMStatsRecorder;
@@ -37,12 +31,12 @@
     public ConnectionFactory,
     public net::NetworkChangeNotifier::NetworkChangeObserver {
  public:
-  // The caller is responsible for making sure the ConnectionFactoryImpl is
-  // destroyed before the |url_request_context|.
-  ConnectionFactoryImpl(const std::vector<GURL>& mcs_endpoints,
-                        const net::BackoffEntry::Policy& backoff_policy,
-                        net::URLRequestContext* url_request_context,
-                        GCMStatsRecorder* recorder);
+  ConnectionFactoryImpl(
+      const std::vector<GURL>& mcs_endpoints,
+      const net::BackoffEntry::Policy& backoff_policy,
+      GetProxyResolvingFactoryCallback get_socket_factory_callback,
+      // need task runner here.
+      GCMStatsRecorder* recorder);
   ~ConnectionFactoryImpl() override;
 
   // ConnectionFactory implementation.
@@ -67,10 +61,6 @@
   // attempt will be made.
   GURL GetCurrentEndpoint() const;
 
-  // Returns the IPEndpoint to which the factory is currently connected. If no
-  // connection is active, returns an empty IPEndpoint.
-  net::IPEndPoint GetPeerIP();
-
  protected:
   // Initiate the connection to the GCM server.
   // Virtual for testing.
@@ -78,7 +68,8 @@
 
   // Helper method for initalizing the connection hander.
   // Virtual for testing.
-  virtual void InitHandler();
+  virtual void InitHandler(mojo::ScopedDataPipeConsumerHandle receive_stream,
+                           mojo::ScopedDataPipeProducerHandle send_stream);
 
   // Helper method for creating a backoff entry.
   // Virtual for testing.
@@ -97,8 +88,12 @@
   // Virtual for testing.
   virtual base::TimeTicks NowTicks();
 
-  // Callback for Socket connection completion.
-  void OnConnectDone(int result);
+  // Callback for Socket connection completion. This is public for testing.
+  void OnConnectDone(int result,
+                     const base::Optional<net::IPEndPoint>& local_addr,
+                     const base::Optional<net::IPEndPoint>& peer_addr,
+                     mojo::ScopedDataPipeConsumerHandle receive_stream,
+                     mojo::ScopedDataPipeProducerHandle send_stream);
 
   // ConnectionHandler callback for connection issues.
   void ConnectionHandlerCallback(int result);
@@ -138,11 +133,15 @@
   // The backoff policy to use.
   const net::BackoffEntry::Policy backoff_policy_;
 
-  // ---- net:: components for establishing connections. ----
+  // ---- network:: components for establishing connections. ----
   // Socket factory for creating new GCM connections.
-  std::unique_ptr<network::ProxyResolvingClientSocketFactory> socket_factory_;
+  GetProxyResolvingFactoryCallback get_socket_factory_callback_;
+  network::mojom::ProxyResolvingSocketFactoryPtr socket_factory_;
   // The handle to the socket for the current connection, if one exists.
-  std::unique_ptr<network::ProxyResolvingClientSocket> socket_;
+  network::mojom::ProxyResolvingSocketPtr socket_;
+  // Peer address of |socket_|.
+  net::IPEndPoint peer_addr_;
+
   // Current backoff entry.
   std::unique_ptr<net::BackoffEntry> backoff_entry_;
   // Backoff entry from previous connection attempt. Updated on each login
diff --git a/google_apis/gcm/engine/connection_factory_impl_unittest.cc b/google_apis/gcm/engine/connection_factory_impl_unittest.cc
index a707f90..8440d03 100644
--- a/google_apis/gcm/engine/connection_factory_impl_unittest.cc
+++ b/google_apis/gcm/engine/connection_factory_impl_unittest.cc
@@ -9,7 +9,9 @@
 #include <utility>
 
 #include "base/message_loop/message_loop.h"
+#include "base/optional.h"
 #include "base/run_loop.h"
+#include "base/test/scoped_task_environment.h"
 #include "base/test/simple_test_tick_clock.h"
 #include "base/threading/thread_task_runner_handle.h"
 #include "google_apis/gcm/base/mcs_util.h"
@@ -18,6 +20,9 @@
 #include "net/base/backoff_entry.h"
 #include "net/traffic_annotation/network_traffic_annotation_test_helper.h"
 #include "net/url_request/url_request_test_util.h"
+#include "services/network/network_context.h"
+#include "services/network/network_service.h"
+#include "services/network/public/mojom/proxy_resolving_socket.mojom.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
 class Policy;
@@ -86,15 +91,17 @@
 // backoff policy.
 class TestConnectionFactoryImpl : public ConnectionFactoryImpl {
  public:
-  TestConnectionFactoryImpl(net::URLRequestContext* request_context,
-                            const base::Closure& finished_callback);
+  TestConnectionFactoryImpl(
+      GetProxyResolvingFactoryCallback get_socket_factory_callback,
+      const base::Closure& finished_callback);
   ~TestConnectionFactoryImpl() override;
 
   void InitializeFactory();
 
   // Overridden stubs.
   void StartConnection() override;
-  void InitHandler() override;
+  void InitHandler(mojo::ScopedDataPipeConsumerHandle receive_stream,
+                   mojo::ScopedDataPipeProducerHandle send_stream) override;
   std::unique_ptr<net::BackoffEntry> CreateBackoffEntry(
       const net::BackoffEntry::Policy* const policy) override;
   std::unique_ptr<ConnectionHandler> CreateConnectionHandler(
@@ -139,14 +146,17 @@
   FakeConnectionHandler* fake_handler_;
   // Dummy GCM Stats recorder.
   FakeGCMStatsRecorder dummy_recorder_;
+  // Dummy mojo pipes.
+  mojo::DataPipe receive_pipe_;
+  mojo::DataPipe send_pipe_;
 };
 
 TestConnectionFactoryImpl::TestConnectionFactoryImpl(
-    net::URLRequestContext* request_context,
+    GetProxyResolvingFactoryCallback get_socket_factory_callback,
     const base::Closure& finished_callback)
     : ConnectionFactoryImpl(BuildEndpoints(),
                             net::BackoffEntry::Policy(),
-                            request_context,
+                            get_socket_factory_callback,
                             &dummy_recorder_),
       connect_result_(net::ERR_UNEXPECTED),
       num_expected_attempts_(0),
@@ -169,8 +179,12 @@
   ASSERT_GT(num_expected_attempts_, 0);
   ASSERT_FALSE(GetConnectionHandler()->CanSendMessage());
   std::unique_ptr<mcs_proto::LoginRequest> request(BuildLoginRequest(0, 0, ""));
-  GetConnectionHandler()->Init(*request, TRAFFIC_ANNOTATION_FOR_TESTS, NULL);
-  OnConnectDone(connect_result_);
+  GetConnectionHandler()->Init(*request,
+                               std::move(receive_pipe_.consumer_handle),
+                               std::move(send_pipe_.producer_handle));
+  OnConnectDone(connect_result_, net::IPEndPoint(), net::IPEndPoint(),
+                mojo::ScopedDataPipeConsumerHandle(),
+                mojo::ScopedDataPipeProducerHandle());
   if (!NextRetryAttempt().is_null()) {
     // Advance the time to the next retry time.
     base::TimeDelta time_till_retry =
@@ -185,7 +199,9 @@
   }
 }
 
-void TestConnectionFactoryImpl::InitHandler() {
+void TestConnectionFactoryImpl::InitHandler(
+    mojo::ScopedDataPipeConsumerHandle receive_stream,
+    mojo::ScopedDataPipeProducerHandle send_stream) {
   EXPECT_NE(connect_result_, net::ERR_UNEXPECTED);
   if (!delay_login_)
     ConnectionHandlerCallback(net::OK);
@@ -272,31 +288,44 @@
   }
 
  private:
+  void GetProxyResolvingSocketFactory(
+      network::mojom::ProxyResolvingSocketFactoryRequest request) {
+    network_context_->CreateProxyResolvingSocketFactory(std::move(request));
+  }
   void ConnectionsComplete();
 
-  base::MessageLoop message_loop_;
-
-  // Dummy request context that is not used to make network requests, and is
-  // added to make ProxyResolvingClientSocketFactory to not DCHECK on a null
-  // context.
-  net::TestURLRequestContext request_context_;
-
+  base::test::ScopedTaskEnvironment scoped_task_environment_;
   TestConnectionFactoryImpl factory_;
   std::unique_ptr<base::RunLoop> run_loop_;
 
   GURL connected_server_;
+  std::unique_ptr<network::NetworkService> network_service_;
+  network::mojom::NetworkContextPtr network_context_ptr_;
+  std::unique_ptr<network::NetworkContext> network_context_;
 };
 
 ConnectionFactoryImplTest::ConnectionFactoryImplTest()
-    : factory_(&request_context_,
+    : scoped_task_environment_(
+          base::test::ScopedTaskEnvironment::MainThreadType::IO),
+      factory_(base::BindRepeating(
+                   &ConnectionFactoryImplTest::GetProxyResolvingSocketFactory,
+                   base::Unretained(this)),
                base::Bind(&ConnectionFactoryImplTest::ConnectionsComplete,
                           base::Unretained(this))),
-      run_loop_(new base::RunLoop()) {
+      run_loop_(new base::RunLoop()),
+      network_service_(network::NetworkService::CreateForTesting()) {
+  network::mojom::NetworkContextParamsPtr params =
+      network::mojom::NetworkContextParams::New();
+  // Use a fixed proxy config, to avoid dependencies on local network
+  // configuration.
+  params->initial_proxy_config = net::ProxyConfigWithAnnotation::CreateDirect();
+  network_context_ = std::make_unique<network::NetworkContext>(
+      network_service_.get(), mojo::MakeRequest(&network_context_ptr_),
+      std::move(params));
   factory()->SetConnectionListener(this);
-  factory()->Initialize(
-      ConnectionFactory::BuildLoginRequestCallback(),
-      ConnectionHandler::ProtoReceivedCallback(),
-      ConnectionHandler::ProtoSentCallback());
+  factory()->Initialize(ConnectionFactory::BuildLoginRequestCallback(),
+                        ConnectionHandler::ProtoReceivedCallback(),
+                        ConnectionHandler::ProtoSentCallback());
 }
 ConnectionFactoryImplTest::~ConnectionFactoryImplTest() {}
 
diff --git a/google_apis/gcm/engine/connection_handler.h b/google_apis/gcm/engine/connection_handler.h
index ee110c1..857e657 100644
--- a/google_apis/gcm/engine/connection_handler.h
+++ b/google_apis/gcm/engine/connection_handler.h
@@ -9,7 +9,7 @@
 
 #include "base/callback.h"
 #include "google_apis/gcm/base/gcm_export.h"
-#include "net/traffic_annotation/network_traffic_annotation.h"
+#include "mojo/public/cpp/system/data_pipe.h"
 
 namespace net{
 class StreamSocket;
@@ -48,8 +48,8 @@
   // Note: It is correct and expected to call Init more than once, as connection
   // issues are encountered and new connections must be made.
   virtual void Init(const mcs_proto::LoginRequest& login_request,
-                    const net::NetworkTrafficAnnotationTag& traffic_annotation,
-                    net::StreamSocket* socket) = 0;
+                    mojo::ScopedDataPipeConsumerHandle receive_stream,
+                    mojo::ScopedDataPipeProducerHandle send_stream) = 0;
 
   // Resets the handler and any internal state. Should be called any time
   // a connection reset happens externally to the handler.
diff --git a/google_apis/gcm/engine/connection_handler_impl.cc b/google_apis/gcm/engine/connection_handler_impl.cc
index 887a4be..f681298 100644
--- a/google_apis/gcm/engine/connection_handler_impl.cc
+++ b/google_apis/gcm/engine/connection_handler_impl.cc
@@ -50,7 +50,6 @@
     const ProtoSentCallback& write_callback,
     const ConnectionChangedCallback& connection_callback)
     : read_timeout_(read_timeout),
-      socket_(NULL),
       handshake_complete_(false),
       message_tag_(0),
       message_size_(0),
@@ -66,8 +65,8 @@
 
 void ConnectionHandlerImpl::Init(
     const mcs_proto::LoginRequest& login_request,
-    const net::NetworkTrafficAnnotationTag& traffic_annotation,
-    net::StreamSocket* socket) {
+    mojo::ScopedDataPipeConsumerHandle receive_stream,
+    mojo::ScopedDataPipeProducerHandle send_stream) {
   DCHECK(!read_callback_.is_null());
   DCHECK(!write_callback_.is_null());
   DCHECK(!connection_callback_.is_null());
@@ -78,9 +77,8 @@
   handshake_complete_ = false;
   message_tag_ = 0;
   message_size_ = 0;
-  socket_ = socket;
-  input_stream_.reset(new SocketInputStream(socket_));
-  output_stream_.reset(new SocketOutputStream(socket_, traffic_annotation));
+  input_stream_.reset(new SocketInputStream(std::move(receive_stream)));
+  output_stream_.reset(new SocketOutputStream(std::move(send_stream)));
 
   Login(login_request);
 }
@@ -481,9 +479,6 @@
 void ConnectionHandlerImpl::CloseConnection() {
   DVLOG(1) << "Closing connection.";
   read_timeout_timer_.Stop();
-  if (socket_)
-    socket_->Disconnect();
-  socket_ = NULL;
   handshake_complete_ = false;
   message_tag_ = 0;
   message_size_ = 0;
diff --git a/google_apis/gcm/engine/connection_handler_impl.h b/google_apis/gcm/engine/connection_handler_impl.h
index 5cd1749..e88749c 100644
--- a/google_apis/gcm/engine/connection_handler_impl.h
+++ b/google_apis/gcm/engine/connection_handler_impl.h
@@ -43,8 +43,8 @@
 
   // ConnectionHandler implementation.
   void Init(const mcs_proto::LoginRequest& login_request,
-            const net::NetworkTrafficAnnotationTag& traffic_annotation,
-            net::StreamSocket* socket) override;
+            mojo::ScopedDataPipeConsumerHandle receive_stream,
+            mojo::ScopedDataPipeProducerHandle send_stream) override;
   void Reset() override;
   bool CanSendMessage() const override;
   void SendMessage(const google::protobuf::MessageLite& message) override;
@@ -103,8 +103,7 @@
   const base::TimeDelta read_timeout_;
   base::OneShotTimer read_timeout_timer_;
 
-  // This connection's socket and the input/output streams attached to it.
-  net::StreamSocket* socket_;
+  // This connection's input/output streams.
   std::unique_ptr<SocketInputStream> input_stream_;
   std::unique_ptr<SocketOutputStream> output_stream_;
 
diff --git a/google_apis/gcm/engine/connection_handler_impl_unittest.cc b/google_apis/gcm/engine/connection_handler_impl_unittest.cc
index ff7393a..cde7645 100644
--- a/google_apis/gcm/engine/connection_handler_impl_unittest.cc
+++ b/google_apis/gcm/engine/connection_handler_impl_unittest.cc
@@ -14,6 +14,8 @@
 #include "base/message_loop/message_loop.h"
 #include "base/run_loop.h"
 #include "base/strings/string_number_conversions.h"
+#include "base/test/bind_test_util.h"
+#include "base/test/scoped_task_environment.h"
 #include "base/test/test_timeouts.h"
 #include "google/protobuf/io/coded_stream.h"
 #include "google/protobuf/io/zero_copy_stream_impl_lite.h"
@@ -28,6 +30,10 @@
 #include "net/socket/stream_socket.h"
 #include "net/test/gtest_util.h"
 #include "net/traffic_annotation/network_traffic_annotation_test_helper.h"
+#include "net/url_request/url_request_test_util.h"
+#include "services/network/network_context.h"
+#include "services/network/network_service.h"
+#include "services/network/public/mojom/proxy_resolving_socket.mojom.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
 namespace gcm {
@@ -147,19 +153,14 @@
   GCMConnectionHandlerImplTest();
   ~GCMConnectionHandlerImplTest() override;
 
-  net::StreamSocket* BuildSocket(const ReadList& read_list,
-                                 const WriteList& write_list);
+  void BuildSocket(const ReadList& read_list, const WriteList& write_list);
 
-  // Pump |message_loop_|, resetting |run_loop_| after completion.
+  // Pump |run_loop_|, and reset |run_loop_| after completion.
   void PumpLoop();
 
   ConnectionHandlerImpl* connection_handler() {
     return connection_handler_.get();
   }
-  base::MessageLoop* message_loop() { return &message_loop_; }
-  net::StaticSocketDataProvider* data_provider() {
-    return data_provider_.get();
-  }
   int last_error() const { return last_error_; }
 
   // Initialize the connection handler, setting |dst_proto| as the destination
@@ -169,15 +170,16 @@
   // Runs the message loop until a message is received.
   void WaitForMessage();
 
+  network::mojom::ProxyResolvingSocketPtr mojo_socket_ptr_;
+
  private:
   void ReadContinuation(ScopedMessage* dst_proto, ScopedMessage new_proto);
   void WriteContinuation();
   void ConnectionContinuation(int error);
 
   // SocketStreams and their data provider.
-  ReadList mock_reads_;
-  WriteList mock_writes_;
-  std::unique_ptr<net::StaticSocketDataProvider> data_provider_;
+  std::vector<std::unique_ptr<net::StaticSocketDataProvider>> data_providers_;
+  std::vector<std::unique_ptr<net::SSLSocketDataProvider>> ssl_data_providers_;
 
   // The connection handler being tested.
   std::unique_ptr<ConnectionHandlerImpl> connection_handler_;
@@ -185,45 +187,71 @@
   // The last connection error received.
   int last_error_;
 
-  // net:: components.
-  std::unique_ptr<net::StreamSocket> socket_;
-  net::MockClientSocketFactory socket_factory_;
   net::AddressList address_list_;
-
-  base::MessageLoopForIO message_loop_;
+  base::test::ScopedTaskEnvironment scoped_task_environment_;
   std::unique_ptr<base::RunLoop> run_loop_;
+  std::unique_ptr<network::NetworkService> network_service_;
+  network::mojom::NetworkContextPtr network_context_ptr_;
+  net::MockClientSocketFactory socket_factory_;
+  net::TestURLRequestContext url_request_context_;
+  std::unique_ptr<network::NetworkContext> network_context_;
+  network::mojom::ProxyResolvingSocketFactoryPtr mojo_socket_factory_ptr_;
+  mojo::ScopedDataPipeConsumerHandle receive_pipe_handle_;
+  mojo::ScopedDataPipeProducerHandle send_pipe_handle_;
 };
 
 GCMConnectionHandlerImplTest::GCMConnectionHandlerImplTest()
-  : last_error_(0) {
+    : last_error_(0),
+      scoped_task_environment_(
+          base::test::ScopedTaskEnvironment::MainThreadType::IO),
+      network_service_(network::NetworkService::CreateForTesting()),
+      url_request_context_(true /* delay_initialization */) {
   address_list_ = net::AddressList::CreateFromIPAddress(
       net::IPAddress::IPv4Localhost(), kMCSPort);
+  socket_factory_.set_enable_read_if_ready(true);
+  url_request_context_.set_client_socket_factory(&socket_factory_);
+  url_request_context_.Init();
+
+  network_context_ = std::make_unique<network::NetworkContext>(
+      network_service_.get(), mojo::MakeRequest(&network_context_ptr_),
+      &url_request_context_);
 }
 
 GCMConnectionHandlerImplTest::~GCMConnectionHandlerImplTest() {
 }
 
-net::StreamSocket* GCMConnectionHandlerImplTest::BuildSocket(
-    const ReadList& read_list,
-    const WriteList& write_list) {
-  mock_reads_ = read_list;
-  mock_writes_ = write_list;
-  data_provider_ = std::make_unique<net::StaticSocketDataProvider>(
-      mock_reads_, mock_writes_);
-  socket_factory_.AddSocketDataProvider(data_provider_.get());
+void GCMConnectionHandlerImplTest::BuildSocket(const ReadList& read_list,
+                                               const WriteList& write_list) {
+  data_providers_.push_back(
+      std::make_unique<net::StaticSocketDataProvider>(read_list, write_list));
+  socket_factory_.AddSocketDataProvider(data_providers_.back().get());
+  ssl_data_providers_.push_back(
+      std::make_unique<net::SSLSocketDataProvider>(net::SYNCHRONOUS, net::OK));
+  socket_factory_.AddSSLSocketDataProvider(ssl_data_providers_.back().get());
+
   run_loop_ = std::make_unique<base::RunLoop>();
 
-  socket_ = socket_factory_.CreateTransportClientSocket(
-      address_list_, NULL, NULL, net::NetLogSource());
-  net::TestCompletionCallback callback;
-  int rv = socket_->Connect(callback.callback());
-  EXPECT_THAT(rv, net::test::IsError(net::ERR_IO_PENDING));
-
-  rv = callback.WaitForResult();
-  EXPECT_THAT(rv, net::test::IsOk());
-
-  EXPECT_TRUE(socket_->IsConnected());
-  return socket_.get();
+  network_context_->CreateProxyResolvingSocketFactory(
+      mojo::MakeRequest(&mojo_socket_factory_ptr_));
+  base::RunLoop run_loop;
+  int net_error = net::ERR_FAILED;
+  const GURL kDestination("https://example.com");
+  mojo_socket_factory_ptr_->CreateProxyResolvingSocket(
+      kDestination, true /* use_tls */,
+      net::MutableNetworkTrafficAnnotationTag(TRAFFIC_ANNOTATION_FOR_TESTS),
+      mojo::MakeRequest(&mojo_socket_ptr_),
+      base::BindLambdaForTesting(
+          [&](int result, const base::Optional<net::IPEndPoint>& local_addr,
+              const base::Optional<net::IPEndPoint>& peer_addr,
+              mojo::ScopedDataPipeConsumerHandle receive_pipe_handle,
+              mojo::ScopedDataPipeProducerHandle send_pipe_handle) {
+            net_error = result;
+            receive_pipe_handle_ = std::move(receive_pipe_handle);
+            send_pipe_handle_ = std::move(send_pipe_handle);
+            run_loop.Quit();
+          }));
+  run_loop.Run();
+  ASSERT_EQ(net::OK, net_error);
 }
 
 void GCMConnectionHandlerImplTest::PumpLoop() {
@@ -243,7 +271,8 @@
                  base::Unretained(this)));
   EXPECT_FALSE(connection_handler()->CanSendMessage());
   connection_handler_->Init(*BuildLoginRequest(kAuthId, kAuthToken, ""),
-                            TRAFFIC_ANNOTATION_FOR_TESTS, socket_.get());
+                            std::move(receive_pipe_handle_),
+                            std::move(send_pipe_handle_));
 }
 
 void GCMConnectionHandlerImplTest::ReadContinuation(
@@ -277,9 +306,10 @@
                                          handshake_request.c_str(),
                                          handshake_request.size()));
   std::string handshake_response = EncodeHandshakeResponse();
-  ReadList read_list(1, net::MockRead(net::ASYNC,
-                                      handshake_response.c_str(),
-                                      handshake_response.size()));
+  ReadList read_list;
+  read_list.push_back(net::MockRead(net::ASYNC, handshake_response.c_str(),
+                                    handshake_response.size()));
+  read_list.push_back(net::MockRead(net::SYNCHRONOUS, net::OK) /* EOF */);
   BuildSocket(read_list, write_list);
 
   ScopedMessage received_message;
@@ -302,9 +332,10 @@
   std::string handshake_response = EncodeHandshakeResponse();
   // Overwrite the version byte.
   handshake_response[0] = 37;
-  ReadList read_list(1, net::MockRead(net::ASYNC,
-                                      handshake_response.c_str(),
-                                      handshake_response.size()));
+  ReadList read_list;
+  read_list.push_back(net::MockRead(net::ASYNC, handshake_response.c_str(),
+                                    handshake_response.size()));
+  read_list.push_back(net::MockRead(net::SYNCHRONOUS, net::OK) /* EOF */);
   BuildSocket(read_list, write_list);
 
   ScopedMessage received_message;
@@ -381,10 +412,13 @@
 
   // Build a new socket and reconnect, successfully this time.
   std::string handshake_response = EncodeHandshakeResponse();
-  read_list[0] = net::MockRead(net::ASYNC,
-                               handshake_response.c_str(),
-                               handshake_response.size());
-  BuildSocket(read_list, write_list);
+  WriteList write_list2(1, net::MockWrite(net::ASYNC, handshake_request.c_str(),
+                                          handshake_request.size()));
+  ReadList read_list2;
+  read_list2.push_back(net::MockRead(net::ASYNC, handshake_response.c_str(),
+                                     handshake_response.size()));
+  read_list2.push_back(net::MockRead(net::SYNCHRONOUS, net::OK) /* EOF */);
+  BuildSocket(read_list2, write_list2);
   Connect(&received_message);
   EXPECT_FALSE(connection_handler()->CanSendMessage());
   WaitForMessage();  // The login send.
@@ -413,6 +447,7 @@
   read_list.push_back(net::MockRead(net::ASYNC,
                                     data_message_pkt.c_str(),
                                     data_message_pkt.size()));
+  read_list.push_back(net::MockRead(net::SYNCHRONOUS, net::OK) /* EOF */);
   BuildSocket(read_list, write_list);
 
   ScopedMessage received_message;
@@ -447,6 +482,7 @@
   read_list.push_back(net::MockRead(net::SYNCHRONOUS,
                                     data_message_pkt.c_str(),
                                     data_message_pkt.size()));
+  read_list.push_back(net::MockRead(net::SYNCHRONOUS, net::OK) /* EOF */);
   BuildSocket(read_list, write_list);
 
   ScopedMessage received_message;
@@ -483,6 +519,7 @@
   read_list.push_back(net::MockRead(net::ASYNC,
                                     data_message_pkt.c_str(),
                                     data_message_pkt.size()));
+  read_list.push_back(net::MockRead(net::SYNCHRONOUS, net::OK) /* EOF */);
   BuildSocket(read_list, write_list);
 
   ScopedMessage received_message;
@@ -522,6 +559,7 @@
                                         bytes_in_first_message,
                                     data_message_pkt.size() -
                                         bytes_in_first_message));
+  read_list.push_back(net::MockRead(net::SYNCHRONOUS, net::OK) /* EOF */);
   BuildSocket(read_list, write_list);
 
   ScopedMessage received_message;
@@ -557,6 +595,7 @@
   read_list.push_back(net::MockRead(net::SYNCHRONOUS,
                                     data_message_pkt.c_str(),
                                     data_message_pkt.size()));
+  read_list.push_back(net::MockRead(net::SYNCHRONOUS, net::OK) /* EOF */);
   BuildSocket(read_list, write_list);
 
   ScopedMessage received_message;
@@ -631,6 +670,7 @@
   read_list.push_back(net::MockRead(net::ASYNC,
                                     data_message_pkt.c_str(),
                                     data_message_pkt.size()));
+  read_list.push_back(net::MockRead(net::SYNCHRONOUS, net::OK) /* EOF */);
   BuildSocket(read_list, write_list);
 
   ScopedMessage received_message;
@@ -692,14 +732,14 @@
                                     handshake_response.c_str(),
                                     handshake_response.size()));
   read_list.push_back(net::MockRead(net::SYNCHRONOUS, net::ERR_IO_PENDING));
-  net::StreamSocket* socket = BuildSocket(read_list, write_list);
+  BuildSocket(read_list, write_list);
 
   ScopedMessage received_message;
   Connect(&received_message);
   WaitForMessage();  // The login send.
   WaitForMessage();  // The login response.
   EXPECT_TRUE(connection_handler()->CanSendMessage());
-  socket->Disconnect();
+  mojo_socket_ptr_.reset();
   mcs_proto::DataMessageStanza data_message;
   data_message.set_from(kDataMsgFrom);
   data_message.set_category(kDataMsgCategory);
@@ -707,7 +747,7 @@
   EXPECT_FALSE(connection_handler()->CanSendMessage());
   WaitForMessage();  // The message send. Should result in an error
   EXPECT_FALSE(connection_handler()->CanSendMessage());
-  EXPECT_EQ(net::ERR_CONNECTION_CLOSED, last_error());
+  EXPECT_EQ(net::ERR_FAILED, last_error());
 }
 
 // Receive a message with a custom data packet that is larger than the
@@ -732,6 +772,7 @@
   read_list.push_back(net::MockRead(net::ASYNC,
                                     data_message_pkt.c_str(),
                                     data_message_pkt.size()));
+  read_list.push_back(net::MockRead(net::SYNCHRONOUS, net::OK) /* EOF */);
   BuildSocket(read_list, write_list);
 
   ScopedMessage received_message;
@@ -769,6 +810,7 @@
   read_list.push_back(net::MockRead(net::SYNCHRONOUS,
                                     data_message_pkt.c_str(),
                                     data_message_pkt.size()));
+  read_list.push_back(net::MockRead(net::SYNCHRONOUS, net::OK) /* EOF */);
   BuildSocket(read_list, write_list);
 
   ScopedMessage received_message;
@@ -805,6 +847,7 @@
   read_list.push_back(net::MockRead(net::ASYNC,
                                     invalid_message_pkt.c_str(),
                                     invalid_message_pkt.size()));
+  read_list.push_back(net::MockRead(net::SYNCHRONOUS, net::OK) /* EOF */);
   BuildSocket(read_list, write_list);
 
   ScopedMessage received_message;
@@ -842,6 +885,7 @@
   read_list.push_back(net::MockRead(net::ASYNC,
                                     data_message_pkt.c_str() + 2,
                                     data_message_pkt.size() - 2));
+  read_list.push_back(net::MockRead(net::SYNCHRONOUS, net::OK) /* EOF */);
   BuildSocket(read_list, write_list);
 
   ScopedMessage received_message;
@@ -870,6 +914,7 @@
                                     handshake_response.size()));
   read_list.push_back(net::MockRead(net::ASYNC, invalid_message_pkt.c_str(),
                                     invalid_message_pkt.size()));
+  read_list.push_back(net::MockRead(net::SYNCHRONOUS, net::OK) /* EOF */);
   BuildSocket(read_list, write_list);
 
   ScopedMessage received_message;
@@ -900,6 +945,7 @@
                                     handshake_response.size()));
   read_list.push_back(net::MockRead(net::ASYNC, invalid_message_pkt.c_str(),
                                     invalid_message_pkt.size()));
+  read_list.push_back(net::MockRead(net::SYNCHRONOUS, net::OK) /* EOF */);
   BuildSocket(read_list, write_list);
 
   ScopedMessage received_message;
diff --git a/google_apis/gcm/engine/fake_connection_factory.cc b/google_apis/gcm/engine/fake_connection_factory.cc
index 4631e57..ee6df98 100644
--- a/google_apis/gcm/engine/fake_connection_factory.cc
+++ b/google_apis/gcm/engine/fake_connection_factory.cc
@@ -6,6 +6,7 @@
 
 #include "google_apis/gcm/engine/fake_connection_handler.h"
 #include "google_apis/gcm/protocol/mcs.pb.h"
+#include "mojo/public/cpp/system/data_pipe.h"
 #include "net/socket/stream_socket.h"
 #include "net/traffic_annotation/network_traffic_annotation_test_helper.h"
 
@@ -36,7 +37,8 @@
 void FakeConnectionFactory::Connect() {
   mcs_proto::LoginRequest login_request;
   request_builder_.Run(&login_request);
-  connection_handler_->Init(login_request, TRAFFIC_ANNOTATION_FOR_TESTS, NULL);
+  connection_handler_->Init(login_request, mojo::ScopedDataPipeConsumerHandle(),
+                            mojo::ScopedDataPipeProducerHandle());
 }
 
 bool FakeConnectionFactory::IsEndpointReachable() const {
diff --git a/google_apis/gcm/engine/fake_connection_handler.cc b/google_apis/gcm/engine/fake_connection_handler.cc
index 34da9b1..4797f11 100644
--- a/google_apis/gcm/engine/fake_connection_handler.cc
+++ b/google_apis/gcm/engine/fake_connection_handler.cc
@@ -44,8 +44,8 @@
 
 void FakeConnectionHandler::Init(
     const mcs_proto::LoginRequest& login_request,
-    const net::NetworkTrafficAnnotationTag& traffic_annotation,
-    net::StreamSocket* socket) {
+    mojo::ScopedDataPipeConsumerHandle receive_stream,
+    mojo::ScopedDataPipeProducerHandle send_stream) {
   ASSERT_GE(expected_outgoing_messages_.size(), 1U);
   EXPECT_EQ(expected_outgoing_messages_.front().SerializeAsString(),
             login_request.SerializeAsString());
diff --git a/google_apis/gcm/engine/fake_connection_handler.h b/google_apis/gcm/engine/fake_connection_handler.h
index 9afee54..70afb649 100644
--- a/google_apis/gcm/engine/fake_connection_handler.h
+++ b/google_apis/gcm/engine/fake_connection_handler.h
@@ -10,7 +10,7 @@
 #include "base/macros.h"
 #include "google_apis/gcm/base/mcs_message.h"
 #include "google_apis/gcm/engine/connection_handler.h"
-#include "net/traffic_annotation/network_traffic_annotation.h"
+#include "mojo/public/cpp/system/data_pipe.h"
 
 namespace gcm {
 
@@ -25,8 +25,8 @@
 
   // ConnectionHandler implementation.
   void Init(const mcs_proto::LoginRequest& login_request,
-            const net::NetworkTrafficAnnotationTag& traffic_annotation,
-            net::StreamSocket* socket) override;
+            mojo::ScopedDataPipeConsumerHandle receive_stream,
+            mojo::ScopedDataPipeProducerHandle send_stream) override;
   void Reset() override;
   bool CanSendMessage() const override;
   void SendMessage(const google::protobuf::MessageLite& message) override;
diff --git a/google_apis/gcm/tools/mcs_probe.cc b/google_apis/gcm/tools/mcs_probe.cc
index 61f9e0ae..4b75874 100644
--- a/google_apis/gcm/tools/mcs_probe.cc
+++ b/google_apis/gcm/tools/mcs_probe.cc
@@ -59,7 +59,7 @@
 #include "services/network/network_context.h"
 #include "services/network/public/cpp/shared_url_loader_factory.h"
 #include "services/network/public/cpp/weak_wrapper_shared_url_loader_factory.h"
-#include "services/network/public/mojom/network_context.mojom.h"
+#include "services/network/public/mojom/proxy_resolving_socket.mojom.h"
 #include "services/network/public/mojom/url_loader_factory.mojom.h"
 
 #if defined(OS_MACOSX)
@@ -220,6 +220,8 @@
   uint64_t secret() const { return secret_; }
 
  private:
+  void RequestProxyResolvingSocketFactory(
+      network::mojom::ProxyResolvingSocketFactoryRequest request);
   void CheckIn();
   void InitializeNetworkState();
 
@@ -321,7 +323,9 @@
 
   connection_factory_ = std::make_unique<ConnectionFactoryImpl>(
       endpoints, kDefaultBackoffPolicy,
-      url_request_context_getter_->GetURLRequestContext(), &recorder_);
+      base::BindRepeating(&MCSProbe::RequestProxyResolvingSocketFactory,
+                          base::Unretained(this)),
+      &recorder_);
   gcm_store_ = std::make_unique<GCMStoreImpl>(
       gcm_store_path_, file_thread_.task_runner(),
       std::make_unique<FakeEncryptor>());
@@ -417,6 +421,12 @@
   LOG(INFO) << "MCS error happened";
 }
 
+void MCSProbe::RequestProxyResolvingSocketFactory(
+    network::mojom::ProxyResolvingSocketFactoryRequest request) {
+  return network_context_->CreateProxyResolvingSocketFactory(
+      std::move(request));
+}
+
 void MCSProbe::CheckIn() {
   LOG(INFO) << "Check-in request initiated.";
   checkin_proto::ChromeBuildProto chrome_build_proto;
diff --git a/ios/chrome/browser/application_context_impl.cc b/ios/chrome/browser/application_context_impl.cc
index 79acc37..6195319 100644
--- a/ios/chrome/browser/application_context_impl.cc
+++ b/ios/chrome/browser/application_context_impl.cc
@@ -57,6 +57,30 @@
 #include "services/metrics/public/cpp/ukm_recorder.h"
 #include "services/network/public/cpp/weak_wrapper_shared_url_loader_factory.h"
 
+namespace {
+
+// Requests a network::mojom::ProxyResolvingSocketFactory on the UI thread.
+// Note that this cannot be called on a thread that is not the UI thread.
+void RequestProxyResolvingSocketFactoryOnUIThread(
+    ApplicationContextImpl* app_context,
+    network::mojom::ProxyResolvingSocketFactoryRequest request) {
+  network::mojom::NetworkContext* network_context =
+      app_context->GetSystemNetworkContext();
+  network_context->CreateProxyResolvingSocketFactory(std::move(request));
+}
+
+// Wrapper on top of the method above. This does a PostTask to the UI thread.
+void RequestProxyResolvingSocketFactory(
+    ApplicationContextImpl* app_context,
+    network::mojom::ProxyResolvingSocketFactoryRequest request) {
+  web::WebThread::GetTaskRunnerForThread(web::WebThread::UI)
+      ->PostTask(FROM_HERE,
+                 base::BindOnce(&RequestProxyResolvingSocketFactoryOnUIThread,
+                                app_context, std::move(request)));
+}
+
+}  // namespace
+
 ApplicationContextImpl::ApplicationContextImpl(
     base::SequencedTaskRunner* local_state_task_runner,
     const base::CommandLine& command_line,
@@ -372,7 +396,11 @@
 
   gcm_driver_ = gcm::CreateGCMDriverDesktop(
       base::WrapUnique(new gcm::GCMClientFactory), GetLocalState(), store_path,
-      GetSystemURLRequestContext(), GetSharedURLLoaderFactory(), ::GetChannel(),
+      // Because ApplicationContextImpl is destroyed after all WebThreads have
+      // been shut down, base::Unretained() is safe here.
+      base::BindRepeating(&RequestProxyResolvingSocketFactory,
+                          base::Unretained(this)),
+      GetSharedURLLoaderFactory(), ::GetChannel(),
       IOSChromeGCMProfileServiceFactory::GetProductCategoryForSubtypes(),
       web::WebThread::GetTaskRunnerForThread(web::WebThread::UI),
       web::WebThread::GetTaskRunnerForThread(web::WebThread::IO),
diff --git a/ios/chrome/browser/gcm/ios_chrome_gcm_profile_service_factory.cc b/ios/chrome/browser/gcm/ios_chrome_gcm_profile_service_factory.cc
index c684d37..9aeed3c 100644
--- a/ios/chrome/browser/gcm/ios_chrome_gcm_profile_service_factory.cc
+++ b/ios/chrome/browser/gcm/ios_chrome_gcm_profile_service_factory.cc
@@ -20,6 +20,35 @@
 #include "ios/chrome/common/channel_info.h"
 #include "ios/web/public/web_thread.h"
 #include "services/network/public/cpp/shared_url_loader_factory.h"
+#include "services/network/public/mojom/proxy_resolving_socket.mojom.h"
+
+namespace {
+
+// Requests a ProxyResolvingSocketFactoryPtr on the UI thread. Note that a
+// WeakPtr of GCMProfileService is needed to detect when the KeyedService shuts
+// down, and avoid calling into |profile| which might have also been destroyed.
+void RequestProxyResolvingSocketFactoryOnUIThread(
+    web::BrowserState* context,
+    base::WeakPtr<gcm::GCMProfileService> service,
+    network::mojom::ProxyResolvingSocketFactoryRequest request) {
+  if (!service)
+    return;
+  context->GetProxyResolvingSocketFactory(std::move(request));
+}
+
+// A thread-safe wrapper to request a ProxyResolvingSocketFactoryPtr.
+void RequestProxyResolvingSocketFactory(
+    web::BrowserState* context,
+    base::WeakPtr<gcm::GCMProfileService> service,
+    network::mojom::ProxyResolvingSocketFactoryRequest request) {
+  web::WebThread::GetTaskRunnerForThread(web::WebThread::UI)
+      ->PostTask(
+          FROM_HERE,
+          base::BindOnce(&RequestProxyResolvingSocketFactoryOnUIThread, context,
+                         std::move(service), std::move(request)));
+}
+
+}  // namespace
 
 // static
 gcm::GCMProfileService* IOSChromeGCMProfileServiceFactory::GetForBrowserState(
@@ -68,6 +97,7 @@
   return std::make_unique<gcm::GCMProfileService>(
       browser_state->GetPrefs(), browser_state->GetStatePath(),
       browser_state->GetRequestContext(),
+      base::BindRepeating(&RequestProxyResolvingSocketFactory, context),
       browser_state->GetSharedURLLoaderFactory(), ::GetChannel(),
       GetProductCategoryForSubtypes(),
       IdentityManagerFactory::GetForBrowserState(browser_state),
diff --git a/ios/web/browser_state.mm b/ios/web/browser_state.mm
index 58d3537..bc97df0 100644
--- a/ios/web/browser_state.mm
+++ b/ios/web/browser_state.mm
@@ -168,13 +168,7 @@
 
 network::mojom::URLLoaderFactory* BrowserState::GetURLLoaderFactory() {
   if (!url_loader_factory_) {
-    DCHECK(!network_context_);
-    DCHECK(!network_context_owner_);
-
-    net::URLRequestContextGetter* request_context = GetRequestContext();
-    DCHECK(request_context);
-    network_context_owner_ = std::make_unique<NetworkContextOwner>(
-        request_context, &network_context_);
+    CreateNetworkContext();
     auto url_loader_factory_params =
         network::mojom::URLLoaderFactoryParams::New();
     url_loader_factory_params->process_id = network::mojom::kBrowserProcessId;
@@ -187,6 +181,14 @@
   return url_loader_factory_.get();
 }
 
+void BrowserState::GetProxyResolvingSocketFactory(
+    network::mojom::ProxyResolvingSocketFactoryRequest request) {
+  if (!network_context_owner_)
+    CreateNetworkContext();
+
+  network_context_->CreateProxyResolvingSocketFactory(std::move(request));
+}
+
 scoped_refptr<network::SharedURLLoaderFactory>
 BrowserState::GetSharedURLLoaderFactory() {
   return shared_url_loader_factory_;
@@ -200,6 +202,16 @@
   return url_data_manager_ios_backend_;
 }
 
+void BrowserState::CreateNetworkContext() {
+  DCHECK(!network_context_);
+  DCHECK(!network_context_owner_);
+
+  net::URLRequestContextGetter* request_context = GetRequestContext();
+  DCHECK(request_context);
+  network_context_owner_ =
+      std::make_unique<NetworkContextOwner>(request_context, &network_context_);
+}
+
 // static
 BrowserState* BrowserState::FromSupportsUserData(
     base::SupportsUserData* supports_user_data) {
diff --git a/ios/web/public/browser_state.h b/ios/web/public/browser_state.h
index b5ff40b..d00d7dd 100644
--- a/ios/web/public/browser_state.h
+++ b/ios/web/public/browser_state.h
@@ -9,6 +9,7 @@
 
 #include "base/supports_user_data.h"
 #include "services/network/public/mojom/network_service.mojom.h"
+#include "services/network/public/mojom/proxy_resolving_socket.mojom.h"
 #include "services/network/public/mojom/url_loader_factory.mojom.h"
 #include "services/service_manager/embedder/embedded_service_info.h"
 
@@ -66,6 +67,10 @@
   // Returns a URLLoaderFactory that is backed by GetRequestContext.
   network::mojom::URLLoaderFactory* GetURLLoaderFactory();
 
+  // Binds a ProxyResolvingSocketFactory request to NetworkContext.
+  void GetProxyResolvingSocketFactory(
+      network::mojom::ProxyResolvingSocketFactoryRequest request);
+
   // Like URLLoaderFactory, but wrapped inside SharedURLLoaderFactory
   scoped_refptr<network::SharedURLLoaderFactory> GetSharedURLLoaderFactory();
 
@@ -115,6 +120,8 @@
   // Not intended for usage outside of //web.
   URLDataManagerIOSBackend* GetURLDataManagerIOSBackendOnIOThread();
 
+  void CreateNetworkContext();
+
   network::mojom::URLLoaderFactoryPtr url_loader_factory_;
   scoped_refptr<network::WeakWrapperSharedURLLoaderFactory>
       shared_url_loader_factory_;
diff --git a/ios/web_view/internal/sync/web_view_gcm_profile_service_factory.mm b/ios/web_view/internal/sync/web_view_gcm_profile_service_factory.mm
index 7b4826a..297946d 100644
--- a/ios/web_view/internal/sync/web_view_gcm_profile_service_factory.mm
+++ b/ios/web_view/internal/sync/web_view_gcm_profile_service_factory.mm
@@ -19,6 +19,7 @@
 #include "ios/web_view/internal/signin/web_view_signin_manager_factory.h"
 #include "ios/web_view/internal/web_view_browser_state.h"
 #include "services/network/public/cpp/shared_url_loader_factory.h"
+#include "services/network/public/mojom/proxy_resolving_socket.mojom.h"
 
 #if !defined(__has_feature) || !__has_feature(objc_arc)
 #error "This file requires ARC support."
@@ -26,6 +27,34 @@
 
 namespace ios_web_view {
 
+namespace {
+
+// Requests a ProxyResolvingSocketFactoryPtr on the UI thread. Note that a
+// WeakPtr of GCMProfileService is needed to detect when the KeyedService shuts
+// down, and avoid calling into |profile| which might have also been destroyed.
+void RequestProxyResolvingSocketFactoryOnUIThread(
+    web::BrowserState* context,
+    base::WeakPtr<gcm::GCMProfileService> service,
+    network::mojom::ProxyResolvingSocketFactoryRequest request) {
+  if (!service)
+    return;
+  context->GetProxyResolvingSocketFactory(std::move(request));
+}
+
+// A thread-safe wrapper to request a ProxyResolvingSocketFactoryPtr.
+void RequestProxyResolvingSocketFactory(
+    web::BrowserState* context,
+    base::WeakPtr<gcm::GCMProfileService> service,
+    network::mojom::ProxyResolvingSocketFactoryRequest request) {
+  web::WebThread::GetTaskRunnerForThread(web::WebThread::UI)
+      ->PostTask(
+          FROM_HERE,
+          base::BindOnce(&RequestProxyResolvingSocketFactoryOnUIThread, context,
+                         std::move(service), std::move(request)));
+}
+
+}  // namespace
+
 // static
 gcm::GCMProfileService* WebViewGCMProfileServiceFactory::GetForBrowserState(
     WebViewBrowserState* browser_state) {
@@ -69,6 +98,7 @@
   return std::make_unique<gcm::GCMProfileService>(
       browser_state->GetPrefs(), browser_state->GetStatePath(),
       browser_state->GetRequestContext(),
+      base::BindRepeating(&RequestProxyResolvingSocketFactory, context),
       browser_state->GetSharedURLLoaderFactory(),
       version_info::Channel::UNKNOWN, GetProductCategoryForSubtypes(),
       WebViewIdentityManagerFactory::GetForBrowserState(browser_state),
diff --git a/net/socket/ssl_client_socket_impl.cc b/net/socket/ssl_client_socket_impl.cc
index 1667eee..5be7622 100644
--- a/net/socket/ssl_client_socket_impl.cc
+++ b/net/socket/ssl_client_socket_impl.cc
@@ -814,6 +814,14 @@
   return rv;
 }
 
+int SSLClientSocketImpl::CancelReadIfReady() {
+  int result = transport_->socket()->CancelReadIfReady();
+  // Cancel |user_read_callback_|, because caller does not expect the callback
+  // to be invoked after they have canceled the ReadIfReady.
+  user_read_callback_.Reset();
+  return result;
+}
+
 int SSLClientSocketImpl::Write(
     IOBuffer* buf,
     int buf_len,
diff --git a/net/socket/ssl_client_socket_impl.h b/net/socket/ssl_client_socket_impl.h
index 150a479..4648ab72 100644
--- a/net/socket/ssl_client_socket_impl.h
+++ b/net/socket/ssl_client_socket_impl.h
@@ -125,6 +125,7 @@
   int ReadIfReady(IOBuffer* buf,
                   int buf_len,
                   CompletionOnceCallback callback) override;
+  int CancelReadIfReady() override;
   int Write(IOBuffer* buf,
             int buf_len,
             CompletionOnceCallback callback,
diff --git a/services/network/proxy_resolving_socket_mojo.cc b/services/network/proxy_resolving_socket_mojo.cc
index c9c0542..8ad0b15 100644
--- a/services/network/proxy_resolving_socket_mojo.cc
+++ b/services/network/proxy_resolving_socket_mojo.cc
@@ -7,6 +7,7 @@
 #include <utility>
 
 #include "base/logging.h"
+#include "base/optional.h"
 #include "net/base/net_errors.h"
 #include "services/network/socket_data_pump.h"
 
@@ -34,18 +35,6 @@
   OnConnectCompleted(result);
 }
 
-void ProxyResolvingSocketMojo::GetPeerAddress(GetPeerAddressCallback callback) {
-  DCHECK(socket_);
-
-  net::IPEndPoint peer_addr;
-  int result = socket_->GetPeerAddress(&peer_addr);
-  if (result != net::OK) {
-    std::move(callback).Run(result, base::nullopt);
-    return;
-  }
-  std::move(callback).Run(result, peer_addr);
-}
-
 void ProxyResolvingSocketMojo::OnConnectCompleted(int result) {
   DCHECK(!connect_callback_.is_null());
   DCHECK(!socket_data_pump_);
@@ -54,9 +43,16 @@
   if (result == net::OK)
     result = socket_->GetLocalAddress(&local_addr);
 
+  net::IPEndPoint peer_addr;
+  // If |socket_| is connected through a proxy, GetPeerAddress returns
+  // net::ERR_NAME_NOT_RESOLVED.
+  bool get_peer_address_success =
+      result == net::OK && (socket_->GetPeerAddress(&peer_addr) == net::OK);
+
   if (result != net::OK) {
     std::move(connect_callback_)
-        .Run(result, base::nullopt, mojo::ScopedDataPipeConsumerHandle(),
+        .Run(result, base::nullopt, base::nullopt,
+             mojo::ScopedDataPipeConsumerHandle(),
              mojo::ScopedDataPipeProducerHandle());
     return;
   }
@@ -67,7 +63,11 @@
       std::move(receive_pipe.producer_handle),
       std::move(send_pipe.consumer_handle), traffic_annotation_);
   std::move(connect_callback_)
-      .Run(net::OK, local_addr, std::move(receive_pipe.consumer_handle),
+      .Run(net::OK, local_addr,
+           get_peer_address_success
+               ? base::make_optional<net::IPEndPoint>(peer_addr)
+               : base::nullopt,
+           std::move(receive_pipe.consumer_handle),
            std::move(send_pipe.producer_handle));
 }
 
diff --git a/services/network/proxy_resolving_socket_mojo.h b/services/network/proxy_resolving_socket_mojo.h
index 9e64383..78a0b9e 100644
--- a/services/network/proxy_resolving_socket_mojo.h
+++ b/services/network/proxy_resolving_socket_mojo.h
@@ -30,9 +30,6 @@
       mojom::ProxyResolvingSocketFactory::CreateProxyResolvingSocketCallback
           callback);
 
-  // mojom::ProxyResolvingSocket implementation.
-  void GetPeerAddress(GetPeerAddressCallback callback) override;
-
  private:
   void OnConnectCompleted(int net_result);
 
diff --git a/services/network/proxy_resolving_socket_mojo_unittest.cc b/services/network/proxy_resolving_socket_mojo_unittest.cc
index 7f6936b..9c1ee7a 100644
--- a/services/network/proxy_resolving_socket_mojo_unittest.cc
+++ b/services/network/proxy_resolving_socket_mojo_unittest.cc
@@ -99,6 +99,7 @@
 
   int CreateSocketSync(
       mojom::ProxyResolvingSocketRequest request,
+      net::IPEndPoint* peer_addr_out,
       const GURL& url,
       mojo::ScopedDataPipeConsumerHandle* receive_pipe_handle_out,
       mojo::ScopedDataPipeProducerHandle* send_pipe_handle_out) {
@@ -110,11 +111,14 @@
         std::move(request),
         base::BindLambdaForTesting(
             [&](int result, const base::Optional<net::IPEndPoint>& local_addr,
+                const base::Optional<net::IPEndPoint>& peer_addr,
                 mojo::ScopedDataPipeConsumerHandle receive_pipe_handle,
                 mojo::ScopedDataPipeProducerHandle send_pipe_handle) {
               net_error = result;
               if (net_error == net::OK)
                 EXPECT_NE(0, local_addr.value().port());
+              if (peer_addr_out && peer_addr)
+                *peer_addr_out = peer_addr.value();
               *receive_pipe_handle_out = std::move(receive_pipe_handle);
               *send_pipe_handle_out = std::move(send_pipe_handle);
               run_loop.Quit();
@@ -176,27 +180,18 @@
     mojom::ProxyResolvingSocketPtr socket;
     mojo::ScopedDataPipeConsumerHandle client_socket_receive_handle;
     mojo::ScopedDataPipeProducerHandle client_socket_send_handle;
-    EXPECT_EQ(net::OK,
-              CreateSocketSync(mojo::MakeRequest(&socket), kDestination,
-                               &client_socket_receive_handle,
-                               &client_socket_send_handle));
     net::IPEndPoint actual_remote_addr;
-    int net_error = net::ERR_FAILED;
-    base::RunLoop run_loop;
-    socket->GetPeerAddress(base::BindLambdaForTesting(
-        [&](int result, const base::Optional<net::IPEndPoint>& peer_addr) {
-          net_error = result;
-          if (net_error == net::OK)
-            actual_remote_addr = peer_addr.value();
-          run_loop.Quit();
-        }));
-    run_loop.Run();
+    EXPECT_EQ(net::OK,
+              CreateSocketSync(mojo::MakeRequest(&socket), &actual_remote_addr,
+                               kDestination, &client_socket_receive_handle,
+                               &client_socket_send_handle));
+    // Consume all read data.
+    base::RunLoop().RunUntilIdle();
     if (!is_direct) {
-      EXPECT_EQ(net::ERR_NAME_NOT_RESOLVED, net_error);
+      EXPECT_EQ(net::IPEndPoint(), actual_remote_addr);
       EXPECT_FALSE(socket_data.AllReadDataConsumed());
       EXPECT_TRUE(socket_data.AllWriteDataConsumed());
     } else {
-      EXPECT_EQ(net::OK, net_error);
       EXPECT_EQ(remote_addr.ToString(), actual_remote_addr.ToString());
       EXPECT_TRUE(socket_data.AllReadDataConsumed());
       EXPECT_FALSE(socket_data.AllWriteDataConsumed());
@@ -231,8 +226,8 @@
     mojom::ProxyResolvingSocketPtr socket;
     mojo::ScopedDataPipeConsumerHandle client_socket_receive_handle;
     mojo::ScopedDataPipeProducerHandle client_socket_send_handle;
-    int status = CreateSocketSync(mojo::MakeRequest(&socket), kDestination,
-                                  &client_socket_receive_handle,
+    int status = CreateSocketSync(mojo::MakeRequest(&socket), nullptr,
+                                  kDestination, &client_socket_receive_handle,
                                   &client_socket_send_handle);
     if (test.is_direct) {
       EXPECT_EQ(net::ERR_FAILED, status);
@@ -275,9 +270,10 @@
   mojo::ScopedDataPipeConsumerHandle client_socket_receive_handle;
   mojo::ScopedDataPipeProducerHandle client_socket_send_handle;
   const GURL kDestination("http://example.com");
-  EXPECT_EQ(net::OK, CreateSocketSync(mojo::MakeRequest(&socket), kDestination,
-                                      &client_socket_receive_handle,
-                                      &client_socket_send_handle));
+  EXPECT_EQ(net::OK,
+            CreateSocketSync(mojo::MakeRequest(&socket), nullptr, kDestination,
+                             &client_socket_receive_handle,
+                             &client_socket_send_handle));
   // Loop kNumIterations times to test that writes can follow reads, and reads
   // can follow writes.
   for (int j = 0; j < kNumIterations; ++j) {
diff --git a/services/network/public/mojom/proxy_resolving_socket.mojom b/services/network/public/mojom/proxy_resolving_socket.mojom
index 18ceb08..4adf75d 100644
--- a/services/network/public/mojom/proxy_resolving_socket.mojom
+++ b/services/network/public/mojom/proxy_resolving_socket.mojom
@@ -12,13 +12,6 @@
 // and Reads are through the data pipes supplied upon construction. Consumer
 // can close the socket by destroying the interface pointer.
 interface ProxyResolvingSocket{
-  // Gets the peer address. On success, |net_error| is
-  // net::OK and |peer_addr| contains the peer address. On failure,
-  // |peer_addr| is null and |net_error| is a net error code. If socket is
-  // connected to a proxy, |net_error| will be net::ERR_NAME_NOT_RESOLVED, and
-  // |peer_addr| will be empty.
-  GetPeerAddress() => (int32 net_error, net.interfaces.IPEndPoint? peer_addr);
-
   // TODO(xunjieli): Add methods to configure the socket connection and allow
   // consumers to specify whether they want to disconnect or return the socket
   // to socket pools.
@@ -33,7 +26,9 @@
   // connection will be established on top of a TCP connection. On success,
   // |result| is net::OK. Caller is to use |send_stream| to send data and
   // |receive_stream| to receive data over the connection. On failure, |result|
-  // is a network error code.
+  // is a network error code. |local_addr| contains the local address of the
+  // socket. |peer_addr| contains the peer address. If socket is connected to a
+  // proxy, |peer_addr| will be null.
   //
   // Any sockets that are created but are yet to be destroyed will be destroyed
   // when the implementation of this factory goes away.
@@ -42,6 +37,7 @@
       ProxyResolvingSocket& socket)
      => (int32 result,
          net.interfaces.IPEndPoint? local_addr,
+         net.interfaces.IPEndPoint? peer_addr,
          handle<data_pipe_consumer>? receive_stream,
          handle<data_pipe_producer>? send_stream);
 };