Instead of enqueueing SPDY frames, enqueue SPDY streams that are ready to produce data.

net/spdy/spdy_stream.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/spdy/spdy_stream.h"
 
 #include "base/bind.h"
 #include "base/logging.h"
 #include "base/message_loop.h"
 #include "base/stringprintf.h"
@@ skipping 32 matching lines @@
     if (*i >= 'A' && *i <= 'Z') {
       return true;
     }
   }
   return false;
 }
 
 }  // namespace
 
 SpdyStream::SpdyStream(SpdySession* session,
-                       SpdyStreamId stream_id,
                        bool pushed,
                        const BoundNetLog& net_log)
     : continue_buffering_data_(true),
-      stream_id_(stream_id),
+      stream_id_(0),
       priority_(HIGHEST),
       slot_(0),
       stalled_by_flow_control_(false),
       send_window_size_(kSpdyStreamInitialWindowSize),
       recv_window_size_(kSpdyStreamInitialWindowSize),
       unacked_recv_window_bytes_(0),
       pushed_(pushed),
       response_received_(false),
       session_(session),
       delegate_(NULL),
       request_time_(base::Time::Now()),
       response_(new SpdyHeaderBlock),
       io_state_(STATE_NONE),
       response_status_(OK),
       cancelled_(false),
       has_upload_data_(false),
       net_log_(net_log),
       send_bytes_(0),
       recv_bytes_(0),
       domain_bound_cert_type_(CLIENT_CERT_INVALID_TYPE) {
 }
 
+class SpdyStream::SpdyStreamIOBufferProducer
+    : public SpdySession::SpdyIOBufferProducer {
+ public:
+  SpdyStreamIOBufferProducer(SpdyStream* stream) : stream_(stream) {}
+
+  // SpdyFrameProducer
+  virtual RequestPriority GetPriority() const OVERRIDE {
+    return stream_->priority();
+  }
+
+  virtual SpdyIOBuffer* ProduceNextBuffer(SpdySession* session) OVERRIDE {
+    if (stream_->stream_id() == 0)
+      SpdySession::SpdyIOBufferProducer::ActivateStream(session, stream_);
+    frame_.reset(stream_->ProduceNextFrame());
+    return frame_ == NULL ? NULL :
+        SpdySession::SpdyIOBufferProducer::CreateIOBuffer(
+            frame_.get(), GetPriority(), stream_);
+  }
+
+ private:
+  scoped_refptr<SpdyStream> stream_;
+  scoped_ptr<SpdyFrame> frame_;
+};
+
+void SpdyStream::SetHasWriteAvailable() {
+  session_->SetStreamHasWriteAvailable(this,
+                                       new SpdyStreamIOBufferProducer(this));
+}
+
+SpdyFrame* SpdyStream::ProduceNextFrame() {
+  if (io_state_ == STATE_SEND_DOMAIN_BOUND_CERT_COMPLETE) {
+    CHECK(request_.get());
+    CHECK_GT(stream_id_, 0u);
+
+    std::string origin = GetUrl().GetOrigin().spec();
+    DCHECK(origin[origin.length() - 1] == '/');
+    origin.erase(origin.length() - 1);  // Trim trailing slash.
+    SpdyCredentialControlFrame* frame = session_->CreateCredentialFrame(
+        origin, domain_bound_cert_type_, domain_bound_private_key_,
+        domain_bound_cert_, priority_);
+    return frame;
+  } else if (io_state_ == STATE_SEND_HEADERS_COMPLETE) {
+    CHECK(request_.get());
+    CHECK_GT(stream_id_, 0u);
+
+    SpdyControlFlags flags =
+        has_upload_data_ ? CONTROL_FLAG_NONE : CONTROL_FLAG_FIN;
+    SpdySynStreamControlFrame* frame = session_->CreateSynStream(
+        stream_id_, priority_, slot_, flags, *request_);
+    send_time_ = base::TimeTicks::Now();
+    return frame;
+  } else {
+    CHECK(!cancelled());
+    // We must need to write stream data.
+    // Until the headers have been completely sent, we can not be sure
+    // that our stream_id is correct.
+    DCHECK_GT(io_state_, STATE_SEND_HEADERS_COMPLETE);
+    DCHECK_GT(stream_id_, 0u);
+    DCHECK(!pending_data_frames_.empty());
+    SpdyFrame* frame = pending_data_frames_.front();
+    pending_data_frames_.pop_front();
+    return frame;
+  }
+}
+
 SpdyStream::~SpdyStream() {
   UpdateHistograms();
+  while (!pending_data_frames_.empty()) {
+    SpdyFrame* frame = pending_data_frames_.back();
+    pending_data_frames_.pop_back();
+    delete frame;
+  }
 }
 
 void SpdyStream::SetDelegate(Delegate* delegate) {
   CHECK(delegate);
   delegate_ = delegate;
 
   if (pushed_) {
     CHECK(response_received());
     MessageLoop::current()->PostTask(
         FROM_HERE, base::Bind(&SpdyStream::PushedStreamReplayData, this));
@@ skipping 361 matching lines @@
 void SpdyStream::Cancel() {
   if (cancelled())
     return;
 
   cancelled_ = true;
   if (session_->IsStreamActive(stream_id_))
     session_->ResetStream(stream_id_, CANCEL, "");
 }
 
 void SpdyStream::Close() {
-  session_->CloseStream(stream_id_, net::OK);
+  if (stream_id_ != 0)
+    session_->CloseStream(stream_id_, net::OK);
+  else
+    session_->CloseCreatedStream(this, OK);
 }
 
 int SpdyStream::SendRequest(bool has_upload_data) {
   // Pushed streams do not send any data, and should always be in STATE_OPEN or
   // STATE_DONE. However, we still want to return IO_PENDING to mimic non-push
   // behavior.
   has_upload_data_ = has_upload_data;
   if (pushed_) {
     send_time_ = base::TimeTicks::Now();
     DCHECK(!has_upload_data_);
     DCHECK(response_received());
     return ERR_IO_PENDING;
   }
   CHECK_EQ(STATE_NONE, io_state_);
   io_state_ = STATE_GET_DOMAIN_BOUND_CERT;
   return DoLoop(OK);
 }
 
 int SpdyStream::WriteStreamData(IOBuffer* data, int length,
                                 SpdyDataFlags flags) {
   // Until the headers have been completely sent, we can not be sure
   // that our stream_id is correct.
   DCHECK_GT(io_state_, STATE_SEND_HEADERS_COMPLETE);
-  return session_->WriteStreamData(stream_id_, data, length, flags);
+  CHECK_GT(stream_id_, 0u);
+
+  pending_data_frames_.push_back(
+      session_->CreateDataFrame(stream_id_, data, length, flags));
+
+  SetHasWriteAvailable();
+  return ERR_IO_PENDING;
 }
 
 bool SpdyStream::GetSSLInfo(SSLInfo* ssl_info,
                             bool* was_npn_negotiated,
                             NextProto* protocol_negotiated) {
   return session_->GetSSLInfo(
       ssl_info, was_npn_negotiated, protocol_negotiated);
 }
 
 bool SpdyStream::GetSSLCertRequestInfo(SSLCertRequestInfo* cert_request_info) {
@@ skipping 123 matching lines @@
     return result;
 
   io_state_ = STATE_SEND_DOMAIN_BOUND_CERT;
   slot_ =  session_->credential_state()->SetHasCredential(GetUrl());
   return OK;
 }
 
 int SpdyStream::DoSendDomainBoundCert() {
   io_state_ = STATE_SEND_DOMAIN_BOUND_CERT_COMPLETE;
   CHECK(request_.get());
-  std::string origin = GetUrl().GetOrigin().spec();
-  origin.erase(origin.length() - 1);  // trim trailing slash
-  int rv =  session_->WriteCredentialFrame(
-      origin, domain_bound_cert_type_, domain_bound_private_key_,
-      domain_bound_cert_, priority_);
-  if (rv != ERR_IO_PENDING)
-    return rv;
-  return OK;
+  SetHasWriteAvailable();
+  return ERR_IO_PENDING;
 }
 
 int SpdyStream::DoSendDomainBoundCertComplete(int result) {
   if (result < 0)
     return result;
 
   io_state_ = STATE_SEND_HEADERS;
   return OK;
 }
 
 int SpdyStream::DoSendHeaders() {
   CHECK(!cancelled_);
 
-  SpdyControlFlags flags = CONTROL_FLAG_NONE;
-  if (!has_upload_data_)
-    flags = CONTROL_FLAG_FIN;
-
-  CHECK(request_.get());
-  int result = session_->WriteSynStream(
-      stream_id_, priority_, slot_, flags,
-      *request_);
-  if (result != ERR_IO_PENDING)
-    return result;
-
-  send_time_ = base::TimeTicks::Now();
+  SetHasWriteAvailable();
   io_state_ = STATE_SEND_HEADERS_COMPLETE;
   return ERR_IO_PENDING;
 }
 
 int SpdyStream::DoSendHeadersComplete(int result) {
   if (result < 0)
     return result;
 
   CHECK_GT(result, 0);
 
@@ skipping 59 matching lines @@
   UMA_HISTOGRAM_TIMES("Net.SpdyStreamDownloadTime",
       recv_last_byte_time_ - recv_first_byte_time_);
   UMA_HISTOGRAM_TIMES("Net.SpdyStreamTime",
       recv_last_byte_time_ - send_time_);
 
   UMA_HISTOGRAM_COUNTS("Net.SpdySendBytes", send_bytes_);
   UMA_HISTOGRAM_COUNTS("Net.SpdyRecvBytes", recv_bytes_);
 }
 
 }  // namespace net