Implement QUIC-based net::BidirectionalStream

net/quic/quic_stream_factory.cc

 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "net/quic/quic_stream_factory.h"
 
 #include <algorithm>
 #include <set>
 #include <utility>
 
@@ skipping 41 matching lines @@
 #include "base/win/windows_version.h"
 #endif
 
 #if defined(USE_OPENSSL)
 #include <openssl/aead.h>
 #include "crypto/openssl_util.h"
 #else
 #include "base/cpu.h"
 #endif
 
+#if BUILDFLAG(ENABLE_BIDIRECTIONAL_STREAM)
+#include "net/http/bidirectional_stream_job.h"
+#include "net/quic/bidirectional_stream_quic_impl.h"
+#endif
+
 using std::min;
 using NetworkHandle = net::NetworkChangeNotifier::NetworkHandle;
 
 namespace net {
 
 namespace {
 
 enum CreateSessionFailure {
   CREATION_ERROR_CONNECTING_SOCKET,
   CREATION_ERROR_SETTING_RECEIVE_BUFFER,
@@ skipping 417 matching lines @@
     session_ = nullptr;
     return OK;
   }
 
   factory_->ActivateSession(server_id_, session_);
 
   return OK;
 }
 
 QuicStreamRequest::QuicStreamRequest(QuicStreamFactory* factory)
-    : factory_(factory) {}
+    : factory_(factory), for_bidirectional_(false) {}
 
 QuicStreamRequest::~QuicStreamRequest() {
   if (factory_ && !callback_.is_null())
     factory_->CancelRequest(this);
 }
 
 int QuicStreamRequest::Request(const HostPortPair& host_port_pair,
                                PrivacyMode privacy_mode,
                                int cert_verify_flags,
                                base::StringPiece origin_host,
                                base::StringPiece method,
                                const BoundNetLog& net_log,
-                               const CompletionCallback& callback) {
+                               const CompletionCallback& callback,
+                               bool for_bidirectional) {
   DCHECK(!stream_);
+#if BUILDFLAG(ENABLE_BIDIRECTIONAL_STREAM)
+  DCHECK(!bidirectional_stream_job_);
+#endif
   DCHECK(callback_.is_null());
   DCHECK(factory_);
   origin_host_ = origin_host.as_string();
   privacy_mode_ = privacy_mode;
+  for_bidirectional_ = for_bidirectional;
+
   int rv = factory_->Create(host_port_pair, privacy_mode, cert_verify_flags,
                             origin_host, method, net_log, this);
   if (rv == ERR_IO_PENDING) {
     host_port_pair_ = host_port_pair;
     net_log_ = net_log;
     callback_ = callback;
   } else {
     factory_ = nullptr;
   }
-  if (rv == OK)
+  if (rv == OK) {
+#if BUILDFLAG(ENABLE_BIDIRECTIONAL_STREAM)
+    DCHECK(stream_ || (for_bidirectional_ && bidirectional_stream_job_));
+#else
     DCHECK(stream_);
+#endif
+  }
   return rv;
 }
 
-void QuicStreamRequest::set_stream(scoped_ptr<QuicHttpStream> stream) {
-  DCHECK(stream);
-  stream_ = std::move(stream);
+void QuicStreamRequest::InitializeStreamFromSession(
+    QuicChromiumClientSession* session) {
+  DCHECK(session);
+  if (for_bidirectional_) {
+#if BUILDFLAG(ENABLE_BIDIRECTIONAL_STREAM)
+    bidirectional_stream_job_.reset(
+        new BidirectionalStreamQuicImpl(session->GetWeakPtr()));
+#else
+    DCHECK(false);
+#endif
+  } else {
+    stream_.reset(new QuicHttpStream(session->GetWeakPtr()));
+  }
 }
 
 void QuicStreamRequest::OnRequestComplete(int rv) {
   factory_ = nullptr;
   callback_.Run(rv);
 }
 
 base::TimeDelta QuicStreamRequest::GetTimeDelayForWaitingJob() const {
   if (!factory_)
     return base::TimeDelta();
   return factory_->GetTimeDelayForWaitingJob(
       QuicServerId(host_port_pair_, privacy_mode_));
 }
 
 scoped_ptr<QuicHttpStream> QuicStreamRequest::ReleaseStream() {
   DCHECK(stream_);
+  DCHECK(!for_bidirectional_);
   return std::move(stream_);
 }
 
+#if BUILDFLAG(ENABLE_BIDIRECTIONAL_STREAM)
+scoped_ptr<BidirectionalStreamJob>
+QuicStreamRequest::ReleaseBidirectionalStreamJob() {
+  DCHECK(bidirectional_stream_job_);
+  DCHECK(for_bidirectional_);
+
+  return std::move(bidirectional_stream_job_);
+}
+#endif
+
 QuicStreamFactory::QuicStreamFactory(
     HostResolver* host_resolver,
     ClientSocketFactory* client_socket_factory,
     base::WeakPtr<HttpServerProperties> http_server_properties,
     CertVerifier* cert_verifier,
     CTPolicyEnforcer* ct_policy_enforcer,
     ChannelIDService* channel_id_service,
     TransportSecurityState* transport_security_state,
     CTVerifier* cert_transparency_verifier,
     SocketPerformanceWatcherFactory* socket_performance_watcher_factory,
@@ skipping 201 matching lines @@
                               const BoundNetLog& net_log,
                               QuicStreamRequest* request) {
   QuicServerId server_id(host_port_pair, privacy_mode);
   // TODO(rtenneti): crbug.com/498823 - delete active_sessions_.empty() checks.
   if (!active_sessions_.empty()) {
     SessionMap::iterator it = active_sessions_.find(server_id);
     if (it != active_sessions_.end()) {
       QuicChromiumClientSession* session = it->second;
       if (!session->CanPool(origin_host.as_string(), privacy_mode))
         return ERR_ALTERNATIVE_CERT_NOT_VALID_FOR_ORIGIN;
-      request->set_stream(CreateFromSession(session));
+      request->InitializeStreamFromSession(session);
       return OK;
     }
   }
 
   if (HasActiveJob(server_id)) {
     active_requests_[request] = server_id;
     job_requests_map_[server_id].insert(request);
     return ERR_IO_PENDING;
   }
 
@@ skipping 35 matching lines @@
     // related changes.
     if (active_sessions_.empty())
       return ERR_QUIC_PROTOCOL_ERROR;
     SessionMap::iterator it = active_sessions_.find(server_id);
     DCHECK(it != active_sessions_.end());
     if (it == active_sessions_.end())
       return ERR_QUIC_PROTOCOL_ERROR;
     QuicChromiumClientSession* session = it->second;
     if (!session->CanPool(origin_host.as_string(), privacy_mode))
       return ERR_ALTERNATIVE_CERT_NOT_VALID_FOR_ORIGIN;
-    request->set_stream(CreateFromSession(session));
+    request->InitializeStreamFromSession(session);
   }
   return rv;
 }
 
 void QuicStreamFactory::CreateAuxilaryJob(const QuicServerId server_id,
                                           int cert_verify_flags,
                                           bool server_and_origin_have_same_host,
                                           bool is_post,
                                           const BoundNetLog& net_log) {
   Job* aux_job = new Job(
@@ skipping 57 matching lines @@
         RequestSet::iterator old_request_it = request_it;
         ++request_it;
         // Remove request from containers so that OnRequestComplete() is not
         // called later again on the same request.
         job_requests_map_[server_id].erase(old_request_it);
         active_requests_.erase(request);
         // Notify request of certificate error.
         request->OnRequestComplete(ERR_ALTERNATIVE_CERT_NOT_VALID_FOR_ORIGIN);
         continue;
       }
-      request->set_stream(CreateFromSession(session));
+      request->InitializeStreamFromSession(session);
       ++request_it;
     }
   }
 
   while (!job_requests_map_[server_id].empty()) {
     RequestSet::iterator it = job_requests_map_[server_id].begin();
     QuicStreamRequest* request = *it;
     job_requests_map_[server_id].erase(it);
     active_requests_.erase(request);
     // Even though we're invoking callbacks here, we don't need to worry
@@ skipping 725 matching lines @@
   // Since the session was active, there's no longer an
   // HttpStreamFactoryImpl::Job running which can mark it broken, unless the TCP
   // job also fails. So to avoid not using QUIC when we otherwise could, we mark
   // it as recently broken, which means that 0-RTT will be disabled but we'll
   // still race.
   http_server_properties_->MarkAlternativeServiceRecentlyBroken(
       alternative_service);
 }
 
 }  // namespace net