Added a net::BidirectionalStream to expose a bidirectional streaming interface

net/http/http_stream_factory_impl_request.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/http/http_stream_factory_impl_request.h"
 
 #include "base/callback.h"
 #include "base/logging.h"
 #include "base/stl_util.h"
+#include "net/http/bidirectional_stream.h"
 #include "net/http/http_stream_factory_impl_job.h"
 #include "net/spdy/spdy_http_stream.h"
 #include "net/spdy/spdy_session.h"
 
 namespace net {
 
 HttpStreamFactoryImpl::Request::Request(
     const GURL& url,
     HttpStreamFactoryImpl* factory,
     HttpStreamRequest::Delegate* delegate,
     WebSocketHandshakeStreamBase::CreateHelper*
         websocket_handshake_stream_create_helper,
     const BoundNetLog& net_log)
+    : Request(url,
+              factory,
+              delegate,
+              websocket_handshake_stream_create_helper,
+              net_log,
+              false) {}
+
+HttpStreamFactoryImpl::Request::Request(

[mef, 2015/09/30 16:18:16]

why do we need to duplicate constructors? Can we add parameter to existing one?

[xunjieli, 2015/10/01 18:41:16]

Done.

+    const GURL& url,
+    HttpStreamFactoryImpl* factory,
+    HttpStreamRequest::Delegate* delegate,
+    WebSocketHandshakeStreamBase::CreateHelper*
+        websocket_handshake_stream_create_helper,
+    const BoundNetLog& net_log,
+    bool for_bidirectional)
     : url_(url),
       factory_(factory),
       websocket_handshake_stream_create_helper_(
           websocket_handshake_stream_create_helper),
       delegate_(delegate),
       net_log_(net_log),
       completed_(false),
       was_npn_negotiated_(false),
       protocol_negotiated_(kProtoUnknown),
-      using_spdy_(false) {
+      using_spdy_(false),
+      for_bidirectional_(for_bidirectional) {
   DCHECK(factory_);
   DCHECK(delegate_);
 
   net_log_.BeginEvent(NetLog::TYPE_HTTP_STREAM_REQUEST);
 }
 
 HttpStreamFactoryImpl::Request::~Request() {
   if (bound_job_.get())
     DCHECK(jobs_.empty());
 
@@ skipping 27 matching lines @@
   protocol_negotiated_ = protocol_negotiated;
   using_spdy_ = using_spdy;
 }
 
 void HttpStreamFactoryImpl::Request::OnStreamReady(
     Job* job,
     const SSLConfig& used_ssl_config,
     const ProxyInfo& used_proxy_info,
     HttpStream* stream) {
   DCHECK(!factory_->for_websockets_);
+  DCHECK(!for_bidirectional_);
   DCHECK(stream);
   DCHECK(completed_);
 
   OnJobSucceeded(job);
   delegate_->OnStreamReady(used_ssl_config, used_proxy_info, stream);
 }
 
+void HttpStreamFactoryImpl::Request::OnBidirectionalStreamReady(
+    Job* job,
+    const SSLConfig& used_ssl_config,
+    const ProxyInfo& used_proxy_info,
+    BidirectionalStream* stream) {
+  DCHECK(!factory_->for_websockets_);
+  DCHECK(for_bidirectional_);
+  DCHECK(stream);
+  DCHECK(completed_);
+
+  OnJobSucceeded(job);
+  delegate_->OnBidirectionalStreamReady(used_ssl_config, used_proxy_info,
+                                        stream);
+}
+
 void HttpStreamFactoryImpl::Request::OnWebSocketHandshakeStreamReady(
     Job* job,
     const SSLConfig& used_ssl_config,
     const ProxyInfo& used_proxy_info,
     WebSocketHandshakeStreamBase* stream) {
   DCHECK(factory_->for_websockets_);
+  DCHECK(!for_bidirectional_);
   DCHECK(stream);
   DCHECK(completed_);
 
   OnJobSucceeded(job);
   delegate_->OnWebSocketHandshakeStreamReady(
       used_ssl_config, used_proxy_info, stream);
 }
 
 void HttpStreamFactoryImpl::Request::OnStreamFailed(
     Job* job,
@@ skipping 142 matching lines @@
 // TODO(jgraettinger): Currently, HttpStreamFactoryImpl::Job notifies a
 // Request that the session is ready, which in turn notifies it's delegate,
 // and then it notifies HttpStreamFactoryImpl so that /other/ requests may
 // be woken, but only if the spdy_session is still okay. This is tough to grok.
 // Instead, see if Job can notify HttpStreamFactoryImpl only, and have one
 // path for notifying any requests waiting for the session (including the
 // request which spawned it).
 void HttpStreamFactoryImpl::Request::OnNewSpdySessionReady(
     Job* job,
     scoped_ptr<HttpStream> stream,
+    scoped_ptr<BidirectionalStream> bidirectional_stream,
     const base::WeakPtr<SpdySession>& spdy_session,
     bool direct) {
   DCHECK(job);
   DCHECK(job->using_spdy());
 
   // Note: |spdy_session| may be NULL. In that case, |delegate_| should still
   // receive |stream| so the error propogates up correctly, however there is no
   // point in broadcasting |spdy_session| to other requests.
 
   // The first case is the usual case.
@@ skipping 14 matching lines @@
   const BoundNetLog net_log = job->net_log();
 
   Complete(was_npn_negotiated, protocol_negotiated, using_spdy);
 
   // Cache this so we can still use it if the request is deleted.
   HttpStreamFactoryImpl* factory = factory_;
   if (factory->for_websockets_) {
     // TODO(ricea): Re-instate this code when WebSockets over SPDY is
     // implemented.
     NOTREACHED();
+  } else if (for_bidirectional_) {

[mef, 2015/09/30 16:18:16]

Does this only work for new spdy session?

[xunjieli, 2015/10/01 18:41:16]

It also works for existing spdy session if HttpStreamFactoryImplJob finds one
(see http_stream_factory_impl_job.cc). 

+    delegate_->OnBidirectionalStreamReady(job->server_ssl_config(),
+                                          job->proxy_info(),
+                                          bidirectional_stream.release());
   } else {
     delegate_->OnStreamReady(job->server_ssl_config(), job->proxy_info(),
                              stream.release());
   }
   // |this| may be deleted after this point.
   if (spdy_session && spdy_session->IsAvailable()) {
     factory->OnNewSpdySessionReady(spdy_session,
                                    direct,
                                    used_ssl_config,
                                    used_proxy_info,
@@ skipping 76 matching lines @@
     }
     // We may have other jobs in |jobs_|. For example, if we start multiple jobs
     // for Alternate-Protocol.
     BindJob(job);
     return;
   }
   DCHECK(jobs_.empty());
 }
 
 }  // namespace net