QUIC WIP

net/tools/naive/naive_connection.cc

@@ -36,14 +36,16 @@ constexpr int kMaxPaddingSize = 255;
 
 NaiveConnection::NaiveConnection(
     unsigned int id,
+    Protocol protocol,
+    bool use_proxy,
     Direction pad_direction,
     std::unique_ptr<StreamSocket> accepted_socket,
-    Delegate* delegate,
     const NetworkTrafficAnnotationTag& traffic_annotation)
     : id_(id),
+      protocol_(protocol),
+      use_proxy_(use_proxy),
       pad_direction_(pad_direction),
       next_state_(STATE_NONE),
-      delegate_(delegate),
       client_socket_(std::move(accepted_socket)),
       server_socket_handle_(std::make_unique<ClientSocketHandle>()),
       sockets_{client_socket_.get(), nullptr},
@@ -62,19 +64,51 @@ NaiveConnection::NaiveConnection(
                                      weak_ptr_factory_.GetWeakPtr());
 }
 
+NaiveConnection::NaiveConnection(
+    unsigned int id,
+    Direction pad_direction,
+    quic::QuicNaiveServerStream* quic_stream,
+    const quic::QuicHeaderList& quic_headers,
+    const NetworkTrafficAnnotationTag& traffic_annotation)
+    : id_(id),
+      protocol_(kQuic),
+      use_proxy_(false),
+      pad_direction_(kNone),
+      next_state_(STATE_NONE),
+      client_socket_(nullptr),
+      client_quic_stream_(stream),
+      client_quic_headers_(client_quic_headers),
+      server_socket_handle_(std::make_unique<ClientSocketHandle>()),
+      sockets_{nullptr, nullptr},
+      errors_{OK, OK},
+      write_pending_{false, false},
+      early_pull_pending_(false),
+      can_push_to_server_(false),
+      early_pull_result_(ERR_IO_PENDING),
+      num_paddings_{0, 0},
+      read_padding_state_(STATE_READ_PAYLOAD_LENGTH_1),
+      full_duplex_(false),
+      time_func_(&base::TimeTicks::Now),
+      traffic_annotation_(traffic_annotation),
+      weak_ptr_factory_(this) {}
+
 NaiveConnection::~NaiveConnection() {
   Disconnect();
 }
 
 int NaiveConnection::Connect(CompletionOnceCallback callback) {
-  DCHECK(client_socket_);
+  DCHECK(client_socket_ || protocol == kQuic);
   DCHECK_EQ(next_state_, STATE_NONE);
   DCHECK(!connect_callback_);
 
   if (full_duplex_)
     return OK;
 
+  if (protocol_ != kQuic) {
     next_state_ = STATE_CONNECT_CLIENT;
+  } else {
+    next_state_ = STATE_CONNECT_SERVER;
+  }
 
   int rv = DoLoop(OK);
   if (rv == ERR_IO_PENDING) {
@@ -83,12 +117,30 @@ int NaiveConnection::Connect(CompletionOnceCallback callback) {
   return rv;
 }
 
+void NaiveConnection::OnReadData() {
+  DCHECK_EQ(protocol_, kQuic);
+  if (!client_quic_stream_)
+    return;
+
+  
+}
+
+void NaiveConnection::OnDeleteStream() {
+  DCHECK_EQ(protocol_, kQuic);
+  client_quic_stream_ = nullptr;
+}
+
 void NaiveConnection::Disconnect() {
   full_duplex_ = false;
   // Closes server side first because latency is higher.
   if (server_socket_handle_->socket())
     server_socket_handle_->socket()->Disconnect();
+  if (protocol_ != kQuic) {
     client_socket_->Disconnect();
+  } else if (client_quic_stream_) {
+    client_quic_stream_->Reset(quic::QUIC_STREAM_NO_ERROR);
+    client_quic_stream_ = nullptr;
+  }
 
   next_state_ = STATE_NONE;
   connect_callback_.Reset();
@@ -166,11 +218,78 @@ int NaiveConnection::DoConnectClientComplete(int result) {
 }
 
 int NaiveConnection::DoConnectServer() {
-  DCHECK(delegate_);
   next_state_ = STATE_CONNECT_SERVER_COMPLETE;
 
-  return delegate_->OnConnectServer(id_, client_socket_.get(),
+  // Ignores socket limit set by socket pool for this type of socket.
-                                    server_socket_handle_.get(), io_callback_);
+  constexpr int request_load_flags = LOAD_IGNORE_LIMITS;
+  constexpr RequestPriority request_priority = MAXIMUM_PRIORITY;
+
+  ProxyInfo proxy_info;
+  SSLConfig server_ssl_config;
+  SSLConfig proxy_ssl_config;
+
+  if (use_proxy_) {
+    const auto& proxy_config = session_->proxy_resolution_service()->config();
+    DCHECK(proxy_config);
+    const ProxyList& proxy_list =
+        proxy_config.value().value().proxy_rules().single_proxies;
+    if (proxy_list.IsEmpty())
+      return ERR_MANDATORY_PROXY_CONFIGURATION_FAILED;
+    proxy_info.UseProxyList(proxy_list);
+    proxy_info.set_traffic_annotation(
+        net::MutableNetworkTrafficAnnotationTag(traffic_annotation_));
+
+    HttpRequestInfo req_info;
+    session_->GetSSLConfig(req_info, &server_ssl_config, &proxy_ssl_config);
+    proxy_ssl_config.disable_cert_verification_network_fetches = true;
+  } else {
+    proxy_info.UseDirect();
+  }
+
+  HostPortPair request_endpoint;
+  if (protocol_ == kSocks5) {
+    const auto* socket = static_cast<const Socks5ServerSocket*>(client_socket_.get());
+    request_endpoint = socket->request_endpoint();
+  } else if (protocol_ == kHttp) {
+    const auto* socket = static_cast<const HttpProxySocket*>(client_socket_.get());
+    request_endpoint = socket->request_endpoint();
+  } else if (protocol_ == kQuic) {
+    // client_quic_headers is cleared after OnReadHeaders.
+    // This function is synchronous inside NaiveProxy::OnReadHeaders so
+    // client_quic_headers is ok to be a reference.
+    for (const auto& p : client_quic_headers) {
+      const auto& name = p.first;
+      const auto& value = p.second;
+      if (name == ":method" && value != "CONNECT") {
+        LOG(ERROR) << "Connection " << id_ << " method not supported " << value;
+        return ERR_METHOD_NOT_SUPPORTED;
+      }
+      if (name == ":authority") {
+        request_endpoint = HostPortPair::FromString(value);
+      }
+    }
+    spdy::SpdyHeaderBlock headers;
+    headers[":status"] = "200";
+    client_quic_stream_->WriteHeaders(std::move(headers), /*fin=*/false, nullptr);
+  }
+
+  if (request_endpoint.IsEmpty()) {
+    LOG(ERROR) << "Connection " << id_ << " to invalid origin";
+    return ERR_ADDRESS_INVALID;
+  }
+
+  LOG(INFO) << "Connection " << id_ << " to "
+            << request_endpoint.ToString();
+
+  auto quic_version = quic::QUIC_VERSION_UNSUPPORTED;
+  if (proxy_info.is_quic()) {
+    quic_version = quic::QUIC_VERSION_43;
+  }
+
+  return InitSocketHandleForRawConnect2(
+      request_endpoint, session_, request_load_flags, request_priority,
+      proxy_info, quic_version, server_ssl_config, proxy_ssl_config,
+      PRIVACY_MODE_DISABLED, net_log_, server_socket_handle_.get(), callback);
 }
 
 int NaiveConnection::DoConnectServerComplete(int result) {
@@ -186,7 +305,7 @@ int NaiveConnection::DoConnectServerComplete(int result) {
 }
 
 int NaiveConnection::Run(CompletionOnceCallback callback) {
-  DCHECK(sockets_[kClient]);
+  DCHECK(sockets_[kClient] || protocol_ == kQuic);
   DCHECK(sockets_[kServer]);
   DCHECK_EQ(next_state_, STATE_NONE);
   DCHECK(!connect_callback_);
@@ -207,7 +326,7 @@ int NaiveConnection::Run(CompletionOnceCallback callback) {
   yield_after_time_[kServer] = yield_after_time_[kClient];
 
   can_push_to_server_ = true;
-  if (!early_pull_pending_) {
+  if (!early_pull_pending_ && protocol_ != kQuic) {
     DCHECK_GT(early_pull_result_, 0);
     Push(kClient, kServer, early_pull_result_);
   }

net/tools/naive/naive_connection.h

@@ -16,6 +16,11 @@
 #include "net/base/completion_once_callback.h"
 #include "net/base/completion_repeating_callback.h"
 
+namespace quic {
+class QuicNaiveServerStream;
+class QuicHeaderList;
+}  // namespace quic
+
 namespace net {
 
 class ClientSocketHandle;
@@ -28,6 +33,12 @@ class NaiveConnection {
  public:
   using TimeFunc = base::TimeTicks (*)();
 
+  enum Protocol {
+    kSocks5,
+    kHttp,
+    kQuic,
+  };
+
   // From this direction.
   enum Direction {
     kClient = 0,
@@ -36,24 +47,15 @@ class NaiveConnection {
     kNone = 2,
   };
 
-  class Delegate {
-   public:
-    Delegate() {}
-    virtual ~Delegate() {}
-
-    virtual int OnConnectServer(unsigned int connection_id,
-                                const StreamSocket* accepted_socket,
-                                ClientSocketHandle* server_socket,
-                                CompletionRepeatingCallback callback) = 0;
-
-   private:
-    DISALLOW_COPY_AND_ASSIGN(Delegate);
-  };
-
   NaiveConnection(unsigned int id,
+                  Protocol protocol,
+                  bool use_proxy,
                   Direction pad_direction,
                   std::unique_ptr<StreamSocket> accepted_socket,
-                  Delegate* delegate,
+                  const NetworkTrafficAnnotationTag& traffic_annotation);
+  NaiveConnection(unsigned int id,
+                  quic::QuicNaiveServerStream* stream,
+                  const quic::QuicHeaderList& header_list,
                   const NetworkTrafficAnnotationTag& traffic_annotation);
   ~NaiveConnection();
 
@@ -97,6 +99,8 @@ class NaiveConnection {
   void OnPushComplete(Direction from, Direction to, int result);
 
   unsigned int id_;
+  Protocol protocol_;
+  bool use_proxy_;
   Direction pad_direction_;
 
   CompletionRepeatingCallback io_callback_;

net/tools/naive/naive_proxy.cc

@@ -20,14 +20,17 @@
 #include "net/socket/client_socket_pool_manager.h"
 #include "net/socket/server_socket.h"
 #include "net/socket/stream_socket.h"
+#include "net/third_party/quic/core/http/quic_header_list.h"
 #include "net/third_party/quic/core/quic_versions.h"
+#include "net/third_party/quic/tools/quic_naive_server_stream.h"
+#include "net/third_party/spdy/core/spdy_header_block.h"
 #include "net/tools/naive/http_proxy_socket.h"
 #include "net/tools/naive/socks5_server_socket.h"
 
 namespace net {
 
 NaiveProxy::NaiveProxy(std::unique_ptr<ServerSocket> listen_socket,
-                       Protocol protocol,
+                       NaiveConnection::Protocol protocol,
                        bool use_proxy,
                        HttpNetworkSession* session,
                        const NetworkTrafficAnnotationTag& traffic_annotation)
@@ -41,15 +44,69 @@ NaiveProxy::NaiveProxy(std::unique_ptr<ServerSocket> listen_socket,
       traffic_annotation_(traffic_annotation),
       weak_ptr_factory_(this) {
   DCHECK(listen_socket_);
-  // Start accepting connections in next run loop in case when delegate is not
+  // Start accepting connections in next run loop in case when delegate is
-  // ready to get callbacks.
+  // not ready to get callbacks.
   base::ThreadTaskRunnerHandle::Get()->PostTask(
       FROM_HERE, base::BindOnce(&NaiveProxy::DoAcceptLoop,
                                 weak_ptr_factory_.GetWeakPtr()));
 }
+NaiveProxy::NaiveProxy(HttpNetworkSession* session,
+                       const NetworkTrafficAnnotationTag& traffic_annotation)
+    : listen_socket_(std::move(listen_socket)),
+      protocol_(NaiveConnection::kQuic),
+      use_proxy_(false),
+      session_(session),
+      net_log_(
+          NetLogWithSource::Make(session->net_log(), NetLogSourceType::NONE)),
+      last_id_(0),
+      traffic_annotation_(traffic_annotation),
+      weak_ptr_factory_(this) {}
 
 NaiveProxy::~NaiveProxy() = default;
 
+bool NaiveProxy::InitializeBackend(const std::string& backend_url) {
+  return true;
+}
+
+bool NaiveProxy::IsBackendInitialized() const {
+  return true;
+}
+
+void NaiveProxy::FetchResponseFromBackend(
+    const spdy::SpdyHeaderBlock& request_headers,
+    const std::string& incoming_body,
+    QuicSimpleServerBackend::RequestHandler* quic_stream) {}
+
+void NaiveProxy::CloseBackendResponseStream(
+    QuicSimpleServerBackend::RequestHandler* quic_stream) {}
+
+void NaiveProxy::OnReadHeaders(quic::QuicNaiveServerStream* stream,
+                               const quic::QuicHeaderList& header_list) {
+  constexpr auto kPadDirection = NaiveConnection::kNone;
+  auto connection_ptr = std::make_unique<NaiveConnection>(
+      ++last_id_, kPadDirection, stream, header_list, traffic_annotation_);
+  auto* connection = connection_ptr.get();
+  stream->set_naive_id(connection->id());
+  connection_by_id_[connection->id()] = std::move(connection_ptr);
+  int result = connection->Connect(
+      base::BindRepeating(&NaiveProxy::OnConnectComplete,
+                          weak_ptr_factory_.GetWeakPtr(), connection->id()));
+  if (result == ERR_IO_PENDING)
+    return;
+  HandleConnectResult(connection, result);
+}
+
+void NaiveProxy::OnReadData(quic::QuicNaiveServerStream* stream) {
+  auto* connection = FindConnection(stream->naive_id());
+  if (connection) {
+    connection->OnReadData();
+  }
+}
+
+void NaiveProxy::OnDeleteStream(quic::QuicNaiveServerStream* stream) {
+  Close(stream->naive_id(), OK);
+}
+
 void NaiveProxy::DoAcceptLoop() {
   int result;
   do {
@@ -92,7 +149,7 @@ void NaiveProxy::DoConnect() {
   }
   pad_direction = NaiveConnection::kNone;
   auto connection_ptr = std::make_unique<NaiveConnection>(
-      ++last_id_, pad_direction, std::move(socket), this, traffic_annotation_);
+      ++last_id_, protocol_, use_proxy_, pad_direction, std::move(socket), traffic_annotation_);
   auto* connection = connection_ptr.get();
   connection_by_id_[connection->id()] = std::move(connection_ptr);
   int result = connection->Connect(
@@ -103,65 +160,8 @@ void NaiveProxy::DoConnect() {
   HandleConnectResult(connection, result);
 }
 
-int NaiveProxy::OnConnectServer(unsigned int connection_id,
+void NaiveProxy::OnConnectComplete(unsigned int connection_id, int result) {
-                                const StreamSocket* client_socket,
+  auto* connection = FindConnection(connection_id);
-                                ClientSocketHandle* server_socket,
-                                CompletionRepeatingCallback callback) {
-  // Ignores socket limit set by socket pool for this type of socket.
-  constexpr int request_load_flags = LOAD_IGNORE_LIMITS;
-  constexpr RequestPriority request_priority = MAXIMUM_PRIORITY;
-
-  ProxyInfo proxy_info;
-  SSLConfig server_ssl_config;
-  SSLConfig proxy_ssl_config;
-
-  if (use_proxy_) {
-    const auto& proxy_config = session_->proxy_resolution_service()->config();
-    DCHECK(proxy_config);
-    const ProxyList& proxy_list =
-        proxy_config.value().value().proxy_rules().single_proxies;
-    if (proxy_list.IsEmpty())
-      return ERR_MANDATORY_PROXY_CONFIGURATION_FAILED;
-    proxy_info.UseProxyList(proxy_list);
-    proxy_info.set_traffic_annotation(
-        net::MutableNetworkTrafficAnnotationTag(traffic_annotation_));
-
-    HttpRequestInfo req_info;
-    session_->GetSSLConfig(req_info, &server_ssl_config, &proxy_ssl_config);
-    proxy_ssl_config.disable_cert_verification_network_fetches = true;
-  } else {
-    proxy_info.UseDirect();
-  }
-
-  HostPortPair request_endpoint;
-  if (protocol_ == kSocks5) {
-    const auto* socket = static_cast<const Socks5ServerSocket*>(client_socket);
-    request_endpoint = socket->request_endpoint();
-  } else if (protocol_ == kHttp) {
-    const auto* socket = static_cast<const HttpProxySocket*>(client_socket);
-    request_endpoint = socket->request_endpoint();
-  }
-  if (request_endpoint.IsEmpty()) {
-    LOG(ERROR) << "Connection " << connection_id << " to invalid origin";
-    return ERR_ADDRESS_INVALID;
-  }
-
-  LOG(INFO) << "Connection " << connection_id << " to "
-            << request_endpoint.ToString();
-
-  auto quic_version = quic::QUIC_VERSION_UNSUPPORTED;
-  if (proxy_info.is_quic()) {
-    quic_version = quic::QUIC_VERSION_43;
-  }
-
-  return InitSocketHandleForRawConnect2(
-      request_endpoint, session_, request_load_flags, request_priority,
-      proxy_info, quic_version, server_ssl_config, proxy_ssl_config,
-      PRIVACY_MODE_DISABLED, net_log_, server_socket, callback);
-}
-
-void NaiveProxy::OnConnectComplete(int connection_id, int result) {
-  NaiveConnection* connection = FindConnection(connection_id);
   if (!connection)
     return;
   HandleConnectResult(connection, result);
@@ -184,8 +184,8 @@ void NaiveProxy::DoRun(NaiveConnection* connection) {
   HandleRunResult(connection, result);
 }
 
-void NaiveProxy::OnRunComplete(int connection_id, int result) {
+void NaiveProxy::OnRunComplete(unsigned int connection_id, int result) {
-  NaiveConnection* connection = FindConnection(connection_id);
+  auto* connection = FindConnection(connection_id);
   if (!connection)
     return;
   HandleRunResult(connection, result);
@@ -195,7 +195,7 @@ void NaiveProxy::HandleRunResult(NaiveConnection* connection, int result) {
   Close(connection->id(), result);
 }
 
-void NaiveProxy::Close(int connection_id, int reason) {
+void NaiveProxy::Close(unsigned int connection_id, int reason) {
   auto it = connection_by_id_.find(connection_id);
   if (it == connection_by_id_.end())
     return;
@@ -203,6 +203,11 @@ void NaiveProxy::Close(int connection_id, int reason) {
   LOG(INFO) << "Connection " << connection_id
             << " closed: " << ErrorToShortString(reason);
 
+  // QUIC stream must be deleted immediately.
+  if (protocol_ == NaiveConnection::kQuic) {
+    it->second->OnDeleteStream();
+  }
+
   // The call stack might have callbacks which still have the pointer of
   // connection. Instead of referencing connection with ID all the time,
   // destroys the connection in next run loop to make sure any pending
@@ -212,19 +217,11 @@ void NaiveProxy::Close(int connection_id, int reason) {
   connection_by_id_.erase(it);
 }
 
-NaiveConnection* NaiveProxy::FindConnection(int connection_id) {
+NaiveConnection* NaiveProxy::FindConnection(unsigned int connection_id) {
   auto it = connection_by_id_.find(connection_id);
   if (it == connection_by_id_.end())
     return nullptr;
   return it->second.get();
 }
 
-// This is called after any delegate callbacks are called to check if Close()
-// has been called during callback processing. Using the pointer of connection,
-// |connection| is safe here because Close() deletes the connection in next run
-// loop.
-bool NaiveProxy::HasClosedConnection(NaiveConnection* connection) {
-  return FindConnection(connection->id()) != connection;
-}
-
 }  // namespace net

net/tools/naive/naive_proxy.h

@@ -6,15 +6,27 @@
 #ifndef NET_TOOLS_NAIVE_NAIVE_PROXY_H_
 #define NET_TOOLS_NAIVE_NAIVE_PROXY_H_
 
+#include <cstddef>
 #include <map>
 #include <memory>
+#include <string>
 
 #include "base/macros.h"
 #include "base/memory/weak_ptr.h"
 #include "net/base/completion_repeating_callback.h"
 #include "net/log/net_log_with_source.h"
+#include "net/third_party/quic/tools/quic_simple_server_backend.h"
 #include "net/tools/naive/naive_connection.h"
 
+namespace spdy {
+class SpdyHeaderBlock;
+}  // namespace spdy
+
+namespace quic {
+class QuicNaiveServerStream;
+class QuicHeaderList;
+}  // namespace quic
+
 namespace net {
 
 class ClientSocketHandle;
@@ -24,24 +36,29 @@ class ServerSocket;
 class StreamSocket;
 struct NetworkTrafficAnnotationTag;
 
-class NaiveProxy : public NaiveConnection::Delegate {
+class NaiveProxy : public quic::QuicSimpleServerBackend {
  public:
-  enum Protocol {
-    kSocks5,
-    kHttp,
-  };
-
   NaiveProxy(std::unique_ptr<ServerSocket> server_socket,
-             Protocol protocol,
+             NaiveConnection::Protocol protocol,
              bool use_proxy,
              HttpNetworkSession* session,
              const NetworkTrafficAnnotationTag& traffic_annotation);
   ~NaiveProxy() override;
 
-  int OnConnectServer(unsigned int connection_id,
+  // Implements quic::QuicSimpleServerBackend
-                      const StreamSocket* accepted_socket,
+  bool InitializeBackend(const std::string& backend_url) override;
-                      ClientSocketHandle* server_socket,
+  bool IsBackendInitialized() const override;
-                      CompletionRepeatingCallback callback) override;
+  void FetchResponseFromBackend(
+      const spdy::SpdyHeaderBlock& request_headers,
+      const std::string& incoming_body,
+      quic::QuicSimpleServerBackend::RequestHandler* quic_stream) override;
+  void CloseBackendResponseStream(
+      quic::QuicSimpleServerBackend::RequestHandler* quic_stream) override;
+
+  void OnReadHeaders(quic::QuicNaiveServerStream* stream,
+                     const quic::QuicHeaderList& header_list) override;
+  void OnReadData(quic::QuicNaiveServerStream* stream) override;
+  void OnDeleteStream(quic::QuicNaiveServerStream* stream) override;
 
  private:
   void DoAcceptLoop();
@@ -49,20 +66,19 @@ class NaiveProxy : public NaiveConnection::Delegate {
   void HandleAcceptResult(int result);
 
   void DoConnect();
-  void OnConnectComplete(int connection_id, int result);
+  void OnConnectComplete(unsigned int connection_id, int result);
   void HandleConnectResult(NaiveConnection* connection, int result);
 
   void DoRun(NaiveConnection* connection);
-  void OnRunComplete(int connection_id, int result);
+  void OnRunComplete(unsigned int connection_id, int result);
   void HandleRunResult(NaiveConnection* connection, int result);
 
-  void Close(int connection_id, int reason);
+  void Close(unsigned int connection_id, int reason);
 
   NaiveConnection* FindConnection(int connection_id);
-  bool HasClosedConnection(NaiveConnection* connection);
 
   std::unique_ptr<ServerSocket> listen_socket_;
-  Protocol protocol_;
+  NaiveConnection::Protocol protocol_;
   bool use_proxy_;
   HttpNetworkSession* session_;
   NetLogWithSource net_log_;

net/tools/naive/quic_proxy_backend.cc

@@ -37,6 +37,7 @@ void QuicProxyBackend::CloseBackendResponseStream(
 
 void QuicProxyBackend::OnReadHeaders(quic::QuicSpdyStream* stream,
                                      const quic::QuicHeaderList& header_list) {
+  HostPortPair request_endpoint;
   for (const auto& p : header_list) {
     const auto& name = p.first;
     const auto& value = p.second;
@@ -47,12 +48,91 @@ void QuicProxyBackend::OnReadHeaders(quic::QuicSpdyStream* stream,
       return;
     }
     if (name == ":authority") {
+      request_endpoint = HostPortPair::FromString(value);
     }
   }
 
-  LOG(INFO) << "OnReadHeaders " << stream;
+  if (request_endpoint.IsEmpty()) {
+    spdy::SpdyHeaderBlock headers;
+    headers[":status"] = "400";
+    stream->WriteHeaders(std::move(headers), /*fin=*/true, nullptr);
+    LOG(ERROR) << "Invalid origin";
+    return;
+  }
+
+  
+  auto connection_ptr = std::make_unique<QuicConnection>(++last_id_, stream, this, traffic_annotation_);
+  auto* connection = connection_ptr.get();
+  connection_by_id_[connection->id()] = std::move(connection_ptr);
+  int result = connection->Connect(
+      base::BindRepeating(&QuicProxyBackend::OnConnectComplete,
+                          weak_ptr_factory_.GetWeakPtr(), connection->id()));
+  if (result == ERR_IO_PENDING)
+    return;
+  HandleConnectResult(connection, result);
 }
 
+void QuicProxyBackend::OnConnectComplete(int connection_id, int result) {
+  auto* connection = FindConnection(connection_id);
+  if (!connection)
+    return;
+  HandleConnectResult(connection, result);
+}
+
+void QuicProxyBackend::HandleConnectResult(QuicConnection* connection, int result) {
+  if (result != OK) {
+    Close(connection->id(), result);
+    return;
+  }
+  DoRun(connection);
+}
+
+void QuicProxyBackend::DoRun(QuicConnection* connection) {
+  int result = connection->Run(
+      base::BindRepeating(&QuicProxyBackend::OnRunComplete,
+                          weak_ptr_factory_.GetWeakPtr(), connection->id()));
+  if (result == ERR_IO_PENDING)
+    return;
+  HandleRunResult(connection, result);
+}
+
+void QuicProxyBackend::OnRunComplete(int connection_id, int result) {
+  auto* connection = FindConnection(connection_id);
+  if (!connection)
+    return;
+  HandleRunResult(connection, result);
+}
+
+void QuicProxyBackend::HandleRunResult(QuicConnection* connection, int result) {
+  Close(connection->id(), result);
+}
+
+void QuicProxyBackend::Close(int connection_id, int reason) {
+  auto it = connection_by_id_.find(connection_id);
+  if (it == connection_by_id_.end())
+    return;
+
+  LOG(INFO) << "Connection " << connection_id
+            << " closed: " << ErrorToShortString(reason);
+
+  it->second->Close();
+  // The call stack might have callbacks which still have the pointer of
+  // connection. Instead of referencing connection with ID all the time,
+  // destroys the connection in next run loop to make sure any pending
+  // callbacks in the call stack return.
+  base::ThreadTaskRunnerHandle::Get()->DeleteSoon(FROM_HERE,
+                                                  std::move(it->second));
+  connection_by_id_.erase(it);
+}
+
+QuicConnection* QuicProxyBackend::FindConnection(int connection_id) {
+  auto it = connection_by_id_.find(connection_id);
+  if (it == connection_by_id_.end())
+    return nullptr;
+  return it->second.get();
+}
+
+
 void QuicProxyBackend::OnReadData(quic::QuicSpdyStream* stream,
                                   void* data,
                                   size_t len) {