[SPDY] Replace SpdyIOBuffer with new SpdyBuffer class

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 <limits>
 
 #include "base/bind.h"
 #include "base/compiler_specific.h"
 #include "base/logging.h"
 #include "base/message_loop.h"
 #include "base/stringprintf.h"
 #include "base/values.h"
-#include "net/spdy/spdy_frame_producer.h"
+#include "net/spdy/spdy_buffer_producer.h"
 #include "net/spdy/spdy_http_utils.h"
 #include "net/spdy/spdy_session.h"
 
 namespace net {
 
 namespace {
 
 Value* NetLogSpdyStreamErrorCallback(SpdyStreamId stream_id,
                                      int status,
                                      const std::string* description,
@@ skipping 21 matching lines @@
     if (*i >= 'A' && *i <= 'Z') {
       return true;
     }
   }
   return false;
 }
 
 }  // namespace
 
 // A wrapper around a stream that calls into ProduceSynStreamFrame().
-class SpdyStream::SynStreamFrameProducer : public SpdyFrameProducer {
+class SpdyStream::SynStreamBufferProducer : public SpdyBufferProducer {
  public:
-  SynStreamFrameProducer(const base::WeakPtr<SpdyStream>& stream)
+  SynStreamBufferProducer(const base::WeakPtr<SpdyStream>& stream)
       : stream_(stream) {
     DCHECK(stream_);
   }
 
-  virtual ~SynStreamFrameProducer() {}
+  virtual ~SynStreamBufferProducer() {}
 
-  virtual scoped_ptr<SpdyFrame> ProduceFrame() OVERRIDE {
+  virtual scoped_ptr<SpdyBuffer> ProduceBuffer() OVERRIDE {
     if (!stream_) {
       NOTREACHED();
-      return scoped_ptr<SpdyFrame>();
+      return scoped_ptr<SpdyBuffer>();
     }
     DCHECK_GT(stream_->stream_id(), 0u);
-    return stream_->ProduceSynStreamFrame();
+    return scoped_ptr<SpdyBuffer>(
+        new SpdyBuffer(stream_->ProduceSynStreamFrame()));
   }
 
  private:
   const base::WeakPtr<SpdyStream> stream_;
 };
 
 // A wrapper around a stream that calls into ProduceHeaderFrame() with
 // a given header block.
-class SpdyStream::HeaderFrameProducer : public SpdyFrameProducer {
+class SpdyStream::HeaderBufferProducer : public SpdyBufferProducer {
  public:
-  HeaderFrameProducer(const base::WeakPtr<SpdyStream>& stream,
+  HeaderBufferProducer(const base::WeakPtr<SpdyStream>& stream,
                       scoped_ptr<SpdyHeaderBlock> headers)
       : stream_(stream),
         headers_(headers.Pass()) {
     DCHECK(stream_);
     DCHECK(headers_);
   }
 
-  virtual ~HeaderFrameProducer() {}
+  virtual ~HeaderBufferProducer() {}
 
-  virtual scoped_ptr<SpdyFrame> ProduceFrame() OVERRIDE {
+  virtual scoped_ptr<SpdyBuffer> ProduceBuffer() OVERRIDE {
     if (!stream_) {
       NOTREACHED();
-      return scoped_ptr<SpdyFrame>();
+      return scoped_ptr<SpdyBuffer>();
     }
     DCHECK_GT(stream_->stream_id(), 0u);
-    return stream_->ProduceHeaderFrame(headers_.Pass());
+    return scoped_ptr<SpdyBuffer>(
+        new SpdyBuffer(stream_->ProduceHeaderFrame(headers_.Pass())));
   }
 
  private:
   const base::WeakPtr<SpdyStream> stream_;
   scoped_ptr<SpdyHeaderBlock> headers_;
 };
 
 SpdyStream::SpdyStream(SpdySession* session,
                        const std::string& path,
                        RequestPriority priority,
@@ skipping 58 matching lines @@
     // HEADERS frame.  Since we don't have headers, we had better not have
     // any pending data frames.
     if (pending_buffers_.size() != 0U) {
       LogStreamError(ERR_SPDY_PROTOCOL_ERROR,
                      "HEADERS incomplete headers, but pending data frames.");
       session_->CloseStream(stream_id_, ERR_SPDY_PROTOCOL_ERROR);
     }
     return;
   }
 
-  std::vector<scoped_refptr<IOBufferWithSize> > buffers;
-  buffers.swap(pending_buffers_);
+  std::vector<SpdyBuffer*> buffers;
+  pending_buffers_.release(&buffers);
   for (size_t i = 0; i < buffers.size(); ++i) {
     // It is always possible that a callback to the delegate results in
     // the delegate no longer being available.
     if (!delegate_)
       break;
     if (buffers[i]) {
-      delegate_->OnDataReceived(buffers[i]->data(), buffers[i]->size());
+      delegate_->OnDataReceived(scoped_ptr<SpdyBuffer>(buffers[i]));
     } else {
-      delegate_->OnDataReceived(NULL, 0);
+      delegate_->OnDataReceived(scoped_ptr<SpdyBuffer>());
       session_->CloseStream(stream_id_, net::OK);
       // Note: |this| may be deleted after calling CloseStream.
       DCHECK_EQ(buffers.size() - 1, i);
     }
   }
 }
 
 scoped_ptr<SpdyFrame> SpdyStream::ProduceSynStreamFrame() {
   CHECK_EQ(io_state_, STATE_SEND_HEADERS_COMPLETE);
   CHECK(request_.get());
@@ skipping 244 matching lines @@
   if (delegate_) {
     rv = delegate_->OnResponseReceived(*response_, response_time_, rv);
     // ERR_INCOMPLETE_SPDY_HEADERS means that we are waiting for more
     // headers before the response header block is complete.
     if (rv == ERR_INCOMPLETE_SPDY_HEADERS)
       rv = OK;
   }
   return rv;
 }
 
-void SpdyStream::OnDataReceived(const char* data, size_t length) {
+void SpdyStream::OnDataReceived(scoped_ptr<SpdyBuffer> buffer) {
   DCHECK(session_->IsStreamActive(stream_id_));
-  DCHECK_LT(length, 1u << 24);
   // If we don't have a response, then the SYN_REPLY did not come through.
   // We cannot pass data up to the caller unless the reply headers have been
   // received.
   if (!response_received()) {
     LogStreamError(ERR_SYN_REPLY_NOT_RECEIVED, "Didn't receive a response.");
     session_->CloseStream(stream_id_, ERR_SYN_REPLY_NOT_RECEIVED);
     return;
   }
 
   if (!delegate_ || continue_buffering_data_) {
     // It should be valid for this to happen in the server push case.
     // We'll return received data when delegate gets attached to the stream.
-    if (length > 0) {
-      IOBufferWithSize* buf = new IOBufferWithSize(length);
-      memcpy(buf->data(), data, length);
-      pending_buffers_.push_back(make_scoped_refptr(buf));
+    if (buffer) {
+      pending_buffers_.push_back(buffer.release());
     } else {
       pending_buffers_.push_back(NULL);
       metrics_.StopStream();
       // Note: we leave the stream open in the session until the stream
       //       is claimed.
     }
     return;
   }
 
   CHECK(!closed());
 
-  // A zero-length read means that the stream is being closed.
-  if (length == 0) {
+  if (!buffer) {
     metrics_.StopStream();
     session_->CloseStream(stream_id_, net::OK);
     // Note: |this| may be deleted after calling CloseStream.
     return;
   }
 
-  if (session_->flow_control_state() >= SpdySession::FLOW_CONTROL_STREAM)
-    DecreaseRecvWindowSize(static_cast<int32>(length));
+  if (buffer) {

[Ryan Hamilton, 2013/04/16 18:44:25]

Looks like you return on line 486 if !buffer, so no need to check if (buffer)
again, right?

[akalin, 2013/04/16 23:01:46]

Done.

+    size_t length = buffer->GetRemainingSize();
+    DCHECK_LE(length, session_->GetDataFrameMaximumPayload());
+    if (session_->flow_control_state() >= SpdySession::FLOW_CONTROL_STREAM)
+      DecreaseRecvWindowSize(static_cast<int32>(length));
 
-  // Track our bandwidth.
-  metrics_.RecordBytes(length);
-  recv_bytes_ += length;
+    // Track our bandwidth.
+    metrics_.RecordBytes(length);
+    recv_bytes_ += length;
+  }
   recv_last_byte_time_ = base::TimeTicks::Now();
 
-  if (delegate_->OnDataReceived(data, length) != net::OK) {
+  if (delegate_->OnDataReceived(buffer.Pass()) != net::OK) {
     // |delegate_| rejected the data.
     LogStreamError(ERR_SPDY_PROTOCOL_ERROR, "Delegate rejected the data");
     session_->CloseStream(stream_id_, ERR_SPDY_PROTOCOL_ERROR);
     return;
   }
 }
 
 void SpdyStream::OnFrameWriteComplete(SpdyFrameType frame_type,
                                       size_t frame_size) {
   if (frame_size < session_->GetFrameMinimumSize() ||
@@ skipping 63 matching lines @@
 }
 
 void SpdyStream::QueueHeaders(scoped_ptr<SpdyHeaderBlock> headers) {
   // Until the first headers by SYN_STREAM have been completely sent, we can
   // not be sure that our stream_id is correct.
   DCHECK_GT(io_state_, STATE_SEND_HEADERS_COMPLETE);
   CHECK_GT(stream_id_, 0u);
 
   session_->EnqueueStreamWrite(
       this, HEADERS,
-      scoped_ptr<SpdyFrameProducer>(
-          new HeaderFrameProducer(
+      scoped_ptr<SpdyBufferProducer>(
+          new HeaderBufferProducer(
               weak_ptr_factory_.GetWeakPtr(), headers.Pass())));
 }
 
 void SpdyStream::QueueStreamData(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);
   CHECK_GT(stream_id_, 0u);
   CHECK(!cancelled());
 
-  scoped_ptr<SpdyFrame> data_frame(session_->CreateDataFrame(
+  scoped_ptr<SpdyBuffer> data_buffer(session_->CreateDataBuffer(
       stream_id_, data, length, flags));
-  if (!data_frame)
+  // We'll get called again by PossiblyResumeIfSendStalled().
+  if (!data_buffer)
     return;
 
   session_->EnqueueStreamWrite(
       this, DATA,
-      scoped_ptr<SpdyFrameProducer>(
-          new SimpleFrameProducer(data_frame.Pass())));
+      scoped_ptr<SpdyBufferProducer>(
+          new SimpleBufferProducer(data_buffer.Pass())));
 }
 
 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 161 matching lines @@
   DCHECK(frame);
   // TODO(akalin): Fix the following race condition:
   //
   // Since this is decoupled from sending the SYN_STREAM frame, it is
   // possible that other domain-bound cert frames will clobber ours
   // before our SYN_STREAM frame gets sent. This can be solved by
   // immediately enqueueing the SYN_STREAM frame here and adjusting
   // the state machine appropriately.
   session_->EnqueueStreamWrite(
       this, CREDENTIAL,
-      scoped_ptr<SpdyFrameProducer>(new SimpleFrameProducer(frame.Pass())));
+      scoped_ptr<SpdyBufferProducer>(
+          new SimpleBufferProducer(
+              scoped_ptr<SpdyBuffer>(new SpdyBuffer(frame.Pass())))));
   return ERR_IO_PENDING;
 }
 
 int SpdyStream::DoSendDomainBoundCertComplete() {
   DCHECK_EQ(just_completed_frame_type_, CREDENTIAL);
   io_state_ = STATE_SEND_HEADERS;
   return OK;
 }
 
 int SpdyStream::DoSendHeaders() {
   CHECK(!cancelled_);
   io_state_ = STATE_SEND_HEADERS_COMPLETE;
 
   session_->EnqueueStreamWrite(
       this, SYN_STREAM,
-      scoped_ptr<SpdyFrameProducer>(
-          new SynStreamFrameProducer(weak_ptr_factory_.GetWeakPtr())));
+      scoped_ptr<SpdyBufferProducer>(
+          new SynStreamBufferProducer(weak_ptr_factory_.GetWeakPtr())));
   return ERR_IO_PENDING;
 }
 
 int SpdyStream::DoSendHeadersComplete() {
   DCHECK_EQ(just_completed_frame_type_, SYN_STREAM);
   DCHECK_NE(stream_id_, 0u);
   if (!delegate_)
     return ERR_UNEXPECTED;
 
   io_state_ =
@@ skipping 120 matching lines @@
   }
 
   recv_window_size_ -= delta_window_size;
   net_log_.AddEvent(
       NetLog::TYPE_SPDY_STREAM_UPDATE_RECV_WINDOW,
       base::Bind(&NetLogSpdyStreamWindowUpdateCallback,
                  stream_id_, -delta_window_size, recv_window_size_));
 }
 
 }  // namespace net