[SPDY] Refactor SpdyStream's handling of response headers

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/strings/stringprintf.h"
 #include "base/strings/string_number_conversions.h"
 #include "base/values.h"
 #include "net/spdy/spdy_buffer_producer.h"
 #include "net/spdy/spdy_http_utils.h"
 #include "net/spdy/spdy_session.h"
@@ skipping 18 matching lines @@
     int32 delta,
     int32 window_size,
     NetLog::LogLevel /* log_level */) {
   base::DictionaryValue* dict = new base::DictionaryValue();
   dict->SetInteger("stream_id", stream_id);
   dict->SetInteger("delta", delta);
   dict->SetInteger("window_size", window_size);
   return dict;
 }
 
-bool ContainsUpperAscii(const std::string& str) {
+bool ContainsUppercaseAscii(const std::string& str) {
   for (std::string::const_iterator i(str.begin()); i != str.end(); ++i) {
     if (*i >= 'A' && *i <= 'Z') {
       return true;
     }
   }
   return false;
 }
 
 }  // namespace
 
@@ skipping 24 matching lines @@
 SpdyStream::SpdyStream(SpdyStreamType type,
                        SpdySession* session,
                        const std::string& path,
                        RequestPriority priority,
                        int32 initial_send_window_size,
                        int32 initial_recv_window_size,
                        const BoundNetLog& net_log)
     : type_(type),
       weak_ptr_factory_(this),
       in_do_loop_(false),
-      continue_buffering_data_(true),
+      continue_buffering_data_(type_ == SPDY_PUSH_STREAM),
       stream_id_(0),
       path_(path),
       priority_(priority),
       slot_(0),
       send_stalled_by_flow_control_(false),
       send_window_size_(initial_send_window_size),
       recv_window_size_(initial_recv_window_size),
       unacked_recv_window_bytes_(0),
-      response_received_(false),
       session_(session),
       delegate_(NULL),
       send_status_(
           (type_ == SPDY_PUSH_STREAM) ?
           NO_MORE_DATA_TO_SEND : MORE_DATA_TO_SEND),
       request_time_(base::Time::Now()),
-      response_(new SpdyHeaderBlock),
+      response_headers_status_(RESPONSE_HEADERS_ARE_INCOMPLETE),
       io_state_((type_ == SPDY_PUSH_STREAM) ? STATE_OPEN : STATE_NONE),
       response_status_(OK),
       net_log_(net_log),
       send_bytes_(0),
       recv_bytes_(0),
       domain_bound_cert_type_(CLIENT_CERT_INVALID_TYPE),
       just_completed_frame_type_(DATA),
       just_completed_frame_size_(0) {
   CHECK(type_ == SPDY_BIDIRECTIONAL_STREAM ||
         type_ == SPDY_REQUEST_RESPONSE_STREAM ||
         type_ == SPDY_PUSH_STREAM);
 }
 
 SpdyStream::~SpdyStream() {
   CHECK(!in_do_loop_);
   UpdateHistograms();
 }
 
 void SpdyStream::SetDelegate(Delegate* delegate) {
   CHECK(!delegate_);
   CHECK(delegate);
   delegate_ = delegate;
 
   if (type_ == SPDY_PUSH_STREAM) {
-    CHECK(response_received());
+    DCHECK(continue_buffering_data_);
     base::MessageLoop::current()->PostTask(
         FROM_HERE,
         base::Bind(&SpdyStream::PushedStreamReplayData, GetWeakPtr()));
-  } else {
-    continue_buffering_data_ = false;
   }
 }
 
+SpdyStream::Delegate* SpdyStream::GetDelegate() {
+  return delegate_;
+}
+
 void SpdyStream::PushedStreamReplayData() {
+  DCHECK_EQ(type_, SPDY_PUSH_STREAM);
   DCHECK_NE(stream_id_, 0u);
-
-  if (!delegate_)
-    return;
+  DCHECK(continue_buffering_data_);
 
   continue_buffering_data_ = false;
 
-  // TODO(akalin): This call may delete this object. Figure out what
-  // to do in that case.
-  int rv = delegate_->OnResponseHeadersReceived(*response_, response_time_, OK);
-  if (rv == ERR_INCOMPLETE_SPDY_HEADERS) {
-    // We don't have complete headers.  Assume we're waiting for another
-    // HEADERS frame.  Since we don't have headers, we had better not have
-    // any pending data frames.
-    if (pending_buffers_.size() != 0U) {
+  // The delegate methods called below may delete |this|, so use
+  // |weak_this| to detect that.
+  base::WeakPtr<SpdyStream> weak_this = GetWeakPtr();
+
+  CHECK(delegate_);
+  SpdyResponseHeadersStatus status =
+      delegate_->OnResponseHeadersUpdated(response_headers_);
+  if (status == RESPONSE_HEADERS_ARE_INCOMPLETE) {
+    // Since RESPONSE_HEADERS_ARE_INCOMPLETE was returned, we must not
+    // have been closed. Since we don't have complete headers, assume
+    // we're waiting for another HEADERS frame, and we had better not
+    // have any pending data frames.
+    CHECK(weak_this);
+    if (!pending_buffers_.empty()) {
       LogStreamError(ERR_SPDY_PROTOCOL_ERROR,
-                     "HEADERS incomplete headers, but pending data frames.");
+                     "Data received with incomplete headers.");
       session_->CloseActiveStream(stream_id_, ERR_SPDY_PROTOCOL_ERROR);
     }
     return;
   }
 
-  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_)
+  // OnResponseHeadersUpdated() may have closed |this|.
+  if (!weak_this)
+    return;
+
+  response_headers_status_ = RESPONSE_HEADERS_ARE_COMPLETE;
+
+  while (!pending_buffers_.empty()) {
+    // Take ownership of the first element of |pending_buffers_|.
+    scoped_ptr<SpdyBuffer> buffer(pending_buffers_.front());
+    pending_buffers_.weak_erase(pending_buffers_.begin());
+
+    bool eof = (buffer == NULL);
+
+    CHECK(delegate_);
+    delegate_->OnDataReceived(buffer.Pass());
+
+    // OnDataReceived() may have closed |this|.
+    if (!weak_this)
+      return;
+
+    if (eof) {
+      DCHECK(pending_buffers_.empty());
+      session_->CloseActiveStream(stream_id_, OK);
+      DCHECK(!weak_this);
+      // |pending_buffers_| is invalid at this point.
       break;
-    if (buffers[i]) {
-      delegate_->OnDataReceived(scoped_ptr<SpdyBuffer>(buffers[i]));
-    } else {
-      delegate_->OnDataReceived(scoped_ptr<SpdyBuffer>());
-      session_->CloseActiveStream(stream_id_, OK);
-      // Note: |this| may be deleted after calling CloseActiveStream.
-      DCHECK_EQ(buffers.size() - 1, i);
     }
   }
 }
 
 scoped_ptr<SpdyFrame> SpdyStream::ProduceSynStreamFrame() {
   CHECK_EQ(io_state_, STATE_SEND_REQUEST_HEADERS_COMPLETE);
-  CHECK(request_);
+  CHECK(request_headers_);
   CHECK_GT(stream_id_, 0u);
 
   SpdyControlFlags flags =
       (send_status_ == NO_MORE_DATA_TO_SEND) ?
       CONTROL_FLAG_FIN : CONTROL_FLAG_NONE;
   scoped_ptr<SpdyFrame> frame(session_->CreateSynStream(
-      stream_id_, priority_, slot_, flags, *request_));
+      stream_id_, priority_, slot_, flags, *request_headers_));
   send_time_ = base::TimeTicks::Now();
   return frame.Pass();
 }
 
 void SpdyStream::DetachDelegate() {
   CHECK(!in_do_loop_);
   DCHECK(!closed());
   delegate_ = NULL;
   Cancel();
 }
@@ skipping 163 matching lines @@
 }
 
 base::Time SpdyStream::GetRequestTime() const {
   return request_time_;
 }
 
 void SpdyStream::SetRequestTime(base::Time t) {
   request_time_ = t;
 }
 
-int SpdyStream::OnResponseHeadersReceived(const SpdyHeaderBlock& response) {
-  int rv = OK;
-
-  metrics_.StartStream();
-
-  // TODO(akalin): This should be handled as a protocol error.
-  DCHECK(response_->empty());
-  *response_ = response;  // TODO(ukai): avoid copy.
-
-  recv_first_byte_time_ = base::TimeTicks::Now();
-  response_time_ = base::Time::Now();
+int SpdyStream::OnInitialResponseHeadersReceived(
+    const SpdyHeaderBlock& initial_response_headers,
+    base::Time response_time,
+    base::TimeTicks recv_first_byte_time) {
+  // SpdySession guarantees that this is called at most once.
+  CHECK(response_headers_.empty());
 
   // Check to make sure that we don't receive the response headers
   // before we're ready for it.
   switch (type_) {
     case SPDY_BIDIRECTIONAL_STREAM:
       // For a bidirectional stream, we're ready for the response
       // headers once we've finished sending the request headers.
       if (io_state_ < STATE_OPEN)
         return ERR_SPDY_PROTOCOL_ERROR;
       break;
 
     case SPDY_REQUEST_RESPONSE_STREAM:
       // For a request/response stream, we're ready for the response
       // headers once we've finished sending the request headers and
       // the request body (if we have one).
       if ((io_state_ < STATE_OPEN) || (send_status_ == MORE_DATA_TO_SEND) ||
           pending_send_data_.get())
         return ERR_SPDY_PROTOCOL_ERROR;
       break;
 
     case SPDY_PUSH_STREAM:
       // For a push stream, we're ready immediately.
       DCHECK_EQ(send_status_, NO_MORE_DATA_TO_SEND);
       DCHECK_EQ(io_state_, STATE_OPEN);
       break;
   }
 
+  metrics_.StartStream();
+
   DCHECK_EQ(io_state_, STATE_OPEN);
 
-  // TODO(akalin): Merge the code below with the code in OnHeaders().
-
-  // Append all the headers into the response header block.
-  for (SpdyHeaderBlock::const_iterator it = response.begin();
-       it != response.end(); ++it) {
-    // Disallow uppercase headers.
-    if (ContainsUpperAscii(it->first)) {
-      session_->ResetStream(stream_id_, priority_, RST_STREAM_PROTOCOL_ERROR,
-                            "Upper case characters in header: " + it->first);
-      return ERR_SPDY_PROTOCOL_ERROR;
-    }
-  }
-
-  if ((*response_).find("transfer-encoding") != (*response_).end()) {
-    session_->ResetStream(stream_id_, priority_, RST_STREAM_PROTOCOL_ERROR,
-                         "Received transfer-encoding header");
-    return ERR_SPDY_PROTOCOL_ERROR;
-  }
-
-  if (delegate_) {
-    // May delete this object.
-    rv = delegate_->OnResponseHeadersReceived(*response_, response_time_, rv);
-  }
-  // If delegate_ is not yet attached, we'll call
-  // OnResponseHeadersReceived after the delegate gets attached to the
-  // stream.
-
-  return rv;
+  response_time_ = response_time;
+  recv_first_byte_time_ = recv_first_byte_time;
+  return MergeWithResponseHeaders(initial_response_headers);
 }
 
-int SpdyStream::OnHeaders(const SpdyHeaderBlock& headers) {
-  DCHECK(!response_->empty());
-
-  // Append all the headers into the response header block.
-  for (SpdyHeaderBlock::const_iterator it = headers.begin();
-      it != headers.end(); ++it) {
-    // Disallow duplicate headers.  This is just to be conservative.
-    if ((*response_).find(it->first) != (*response_).end()) {
-      LogStreamError(ERR_SPDY_PROTOCOL_ERROR, "HEADERS duplicate header");
-      response_status_ = ERR_SPDY_PROTOCOL_ERROR;
-      return ERR_SPDY_PROTOCOL_ERROR;
-    }
-
-    // Disallow uppercase headers.
-    if (ContainsUpperAscii(it->first)) {
-      session_->ResetStream(stream_id_, priority_, RST_STREAM_PROTOCOL_ERROR,
-                            "Upper case characters in header: " + it->first);
-      return ERR_SPDY_PROTOCOL_ERROR;
-    }
-
-    (*response_)[it->first] = it->second;
+int SpdyStream::OnAdditionalResponseHeadersReceived(
+    const SpdyHeaderBlock& additional_response_headers) {
+  if (type_ == SPDY_REQUEST_RESPONSE_STREAM) {
+    LOG(WARNING) << "Additional headers received for request/response stream";
+    return OK;
+  } else if (type_ == SPDY_PUSH_STREAM &&
+             response_headers_status_ == RESPONSE_HEADERS_ARE_COMPLETE) {
+    LOG(WARNING) << "Additional headers received for push stream";
+    return OK;
   }
-
-  if ((*response_).find("transfer-encoding") != (*response_).end()) {
-    session_->ResetStream(stream_id_, priority_, RST_STREAM_PROTOCOL_ERROR,
-                         "Received transfer-encoding header");
-    return ERR_SPDY_PROTOCOL_ERROR;
-  }
-
-  int rv = OK;
-  if (delegate_) {
-    // May delete this object.
-    rv = delegate_->OnResponseHeadersReceived(*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;
+  return MergeWithResponseHeaders(additional_response_headers);
 }
 
 void SpdyStream::OnDataReceived(scoped_ptr<SpdyBuffer> buffer) {
   DCHECK(session_->IsStreamActive(stream_id_));
-  // 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_->CloseActiveStream(stream_id_, ERR_SYN_REPLY_NOT_RECEIVED);
-    return;
-  }
 
+  // If we're still buffering data for a push stream, we will do the
+  // check for data received with incomplete headers in
+  // PushedStreamReplayData().
   if (!delegate_ || continue_buffering_data_) {
+    DCHECK_EQ(type_, SPDY_PUSH_STREAM);
     // 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 (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;
   }
 
+  // If we have response headers but the delegate has indicated that
+  // it's still incomplete, then that's a protocol error.
+  if (response_headers_status_ == RESPONSE_HEADERS_ARE_INCOMPLETE) {
+    LogStreamError(ERR_SPDY_PROTOCOL_ERROR,
+                   "Data received with incomplete headers.");
+    session_->CloseActiveStream(stream_id_, ERR_SPDY_PROTOCOL_ERROR);
+    return;
+  }
+
   CHECK(!closed());
 
   if (!buffer) {
     metrics_.StopStream();
+    // Deletes |this|.
     session_->CloseActiveStream(stream_id_, OK);
-    // Note: |this| may be deleted after calling CloseActiveStream.
     return;
   }
 
   size_t length = buffer->GetRemainingSize();
   DCHECK_LE(length, session_->GetDataFrameMaximumPayload());
   if (session_->flow_control_state() >= SpdySession::FLOW_CONTROL_STREAM) {
     DecreaseRecvWindowSize(static_cast<int32>(length));
     buffer->AddConsumeCallback(
         base::Bind(&SpdyStream::OnReadBufferConsumed, GetWeakPtr()));
   }
 
   // Track our bandwidth.
   metrics_.RecordBytes(length);
   recv_bytes_ += length;
   recv_last_byte_time_ = base::TimeTicks::Now();
 
-  if (delegate_->OnDataReceived(buffer.Pass()) != OK) {
-    // |delegate_| rejected the data.
-    LogStreamError(ERR_SPDY_PROTOCOL_ERROR, "Delegate rejected the data");
-    session_->CloseActiveStream(stream_id_, ERR_SPDY_PROTOCOL_ERROR);
-    return;
-  }
+  // May close |this|.
+  delegate_->OnDataReceived(buffer.Pass());
 }
 
 void SpdyStream::OnFrameWriteComplete(SpdyFrameType frame_type,
                                       size_t frame_size) {
   if (frame_size < session_->GetFrameMinimumSize() ||
       frame_size > session_->GetFrameMaximumSize()) {
     NOTREACHED();
     return;
   }
   if (closed())
@@ skipping 35 matching lines @@
 
 void SpdyStream::Close() {
   CHECK(!in_do_loop_);
   if (stream_id_ != 0) {
     session_->CloseActiveStream(stream_id_, OK);
   } else {
     session_->CloseCreatedStream(GetWeakPtr(), OK);
   }
 }
 
-int SpdyStream::SendRequestHeaders(scoped_ptr<SpdyHeaderBlock> headers,
+int SpdyStream::SendRequestHeaders(scoped_ptr<SpdyHeaderBlock> request_headers,
                                    SpdySendStatus send_status) {
   CHECK_NE(type_, SPDY_PUSH_STREAM);
   CHECK_EQ(send_status_, MORE_DATA_TO_SEND);
-  CHECK(!request_);
+  CHECK(!request_headers_);
   CHECK(!pending_send_data_.get());
   CHECK_EQ(io_state_, STATE_NONE);
-  request_ = headers.Pass();
+  request_headers_ = request_headers.Pass();
   send_status_ = send_status;
   io_state_ = STATE_GET_DOMAIN_BOUND_CERT;
   return DoLoop(OK);
 }
 
 void SpdyStream::SendData(IOBuffer* data,
                           int length,
                           SpdySendStatus send_status) {
   CHECK_NE(type_, SPDY_PUSH_STREAM);
   CHECK_EQ(send_status_, MORE_DATA_TO_SEND);
@@ skipping 25 matching lines @@
         NetLog::IntegerCallback("stream_id", stream_id_));
     send_stalled_by_flow_control_ = false;
     QueueNextDataFrame();
   }
 }
 
 base::WeakPtr<SpdyStream> SpdyStream::GetWeakPtr() {
   return weak_ptr_factory_.GetWeakPtr();
 }
 
-bool SpdyStream::HasUrl() const {
-  if (type_ == SPDY_PUSH_STREAM)
-    return response_received();
-  return request_ != NULL;
-}
-
-GURL SpdyStream::GetUrl() const {
-  DCHECK(HasUrl());
-
-  const SpdyHeaderBlock& headers =
-      (type_ == SPDY_PUSH_STREAM) ? *response_ : *request_;
-  return GetUrlFromHeaderBlock(headers, GetProtocolVersion(),
-                               type_ == SPDY_PUSH_STREAM);
-}
-
 bool SpdyStream::GetLoadTimingInfo(LoadTimingInfo* load_timing_info) const {
   if (stream_id_ == 0)
     return false;
 
   return session_->GetLoadTimingInfo(stream_id_, load_timing_info);
 }
 
+GURL SpdyStream::GetUrl() const {
+  if (type_ != SPDY_PUSH_STREAM && !request_headers_)
+    return GURL();
+
+  const SpdyHeaderBlock& headers =
+      (type_ == SPDY_PUSH_STREAM) ? response_headers_ : *request_headers_;
+  return GetUrlFromHeaderBlock(headers, GetProtocolVersion(),
+                               type_ == SPDY_PUSH_STREAM);
+}
+
+bool SpdyStream::HasUrl() const {
+  return !GetUrl().is_empty();
+}
+
 void SpdyStream::OnGetDomainBoundCertComplete(int result) {
   DCHECK_EQ(io_state_, STATE_GET_DOMAIN_BOUND_CERT_COMPLETE);
   DoLoop(result);
 }
 
 int SpdyStream::DoLoop(int result) {
   CHECK(!in_do_loop_);
   in_do_loop_ = true;
 
   do {
@@ skipping 47 matching lines @@
   } while (result != ERR_IO_PENDING && io_state_ != STATE_NONE &&
            io_state_ != STATE_OPEN);
 
   CHECK(in_do_loop_);
   in_do_loop_ = false;
 
   return result;
 }
 
 int SpdyStream::DoGetDomainBoundCert() {
-  CHECK(request_);
+  CHECK(request_headers_);
   DCHECK_NE(type_, SPDY_PUSH_STREAM);
   GURL url = GetUrl();
   if (!session_->NeedsCredentials() || !url.SchemeIs("https")) {
     // Proceed directly to sending the request headers
     io_state_ = STATE_SEND_REQUEST_HEADERS;
     return OK;
   }
 
   slot_ = session_->credential_state()->FindCredentialSlot(GetUrl());
   if (slot_ != SpdyCredentialState::kNoEntry) {
@@ skipping 19 matching lines @@
   DCHECK_NE(type_, SPDY_PUSH_STREAM);
   if (result != OK)
     return result;
 
   io_state_ = STATE_SEND_DOMAIN_BOUND_CERT;
   slot_ =  session_->credential_state()->SetHasCredential(GetUrl());
   return OK;
 }
 
 int SpdyStream::DoSendDomainBoundCert() {
-  CHECK(request_);
+  CHECK(request_headers_);
   DCHECK_NE(type_, SPDY_PUSH_STREAM);
   io_state_ = STATE_SEND_DOMAIN_BOUND_CERT_COMPLETE;
 
   std::string origin = GetUrl().GetOrigin().spec();
   DCHECK(origin[origin.length() - 1] == '/');
   origin.erase(origin.length() - 1);  // Trim trailing slash.
   scoped_ptr<SpdyFrame> frame;
   int rv = session_->CreateCredentialFrame(
       origin, domain_bound_cert_type_, domain_bound_private_key_,
       domain_bound_cert_, priority_, &frame);
@@ skipping 71 matching lines @@
 
 }  // namespace
 
 int SpdyStream::DoSendRequestHeadersComplete() {
   DCHECK_NE(type_, SPDY_PUSH_STREAM);
   DCHECK_EQ(just_completed_frame_type_, SYN_STREAM);
   DCHECK_NE(stream_id_, 0u);
 
   io_state_ = STATE_OPEN;
 
-  // Do this before calling into the |delegate_| as that call may
-  // delete us.
-  int result = GetOpenStateResult(type_, send_status_);
-
   CHECK(delegate_);
+  // Must not close |this|; if it does, it will trigger the |in_do_loop_|
+  // check in the destructor.
   delegate_->OnRequestHeadersSent();
 
-  return result;
+  return GetOpenStateResult(type_, send_status_);
 }
 
 int SpdyStream::DoOpen() {
   DCHECK_NE(type_, SPDY_PUSH_STREAM);
 
   if (just_completed_frame_type_ != DATA) {
     NOTREACHED();
     return ERR_UNEXPECTED;
   }
 
@@ skipping 15 matching lines @@
   send_bytes_ += frame_payload_size;
 
   pending_send_data_->DidConsume(frame_payload_size);
   if (pending_send_data_->BytesRemaining() > 0) {
     QueueNextDataFrame();
     return ERR_IO_PENDING;
   }
 
   pending_send_data_ = NULL;
 
-  // Do this before calling into the |delegate_| as that call may
-  // delete us.
-  int result = GetOpenStateResult(type_, send_status_);
-
   CHECK(delegate_);
+  // Must not close |this|; if it does, it will trigger the
+  // |in_do_loop_| check in the destructor.
   delegate_->OnDataSent();
 
-  return result;
+  return GetOpenStateResult(type_, send_status_);
 }
 
 void SpdyStream::UpdateHistograms() {
   // We need at least the receive timers to be filled in, as otherwise
   // metrics can be bogus.
   if (recv_first_byte_time_.is_null() || recv_last_byte_time_.is_null())
     return;
 
   base::TimeTicks effective_send_time;
   if (type_ == SPDY_PUSH_STREAM) {
@@ skipping 53 matching lines @@
         base::Bind(&SpdyStream::OnWriteBufferConsumed,
                    GetWeakPtr(), payload_size));
   }
 
   session_->EnqueueStreamWrite(
       GetWeakPtr(), DATA,
       scoped_ptr<SpdyBufferProducer>(
           new SimpleBufferProducer(data_buffer.Pass())));
 }
 
+int SpdyStream::MergeWithResponseHeaders(
+    const SpdyHeaderBlock& new_response_headers) {
+  if (new_response_headers.find("transfer-encoding") !=
+      new_response_headers.end()) {
+    session_->ResetStream(stream_id_, priority_, RST_STREAM_PROTOCOL_ERROR,
+                         "Received transfer-encoding header");
+    return ERR_SPDY_PROTOCOL_ERROR;
+  }
+
+  for (SpdyHeaderBlock::const_iterator it = new_response_headers.begin();
+      it != new_response_headers.end(); ++it) {
+    // Disallow uppercase headers.
+    if (ContainsUppercaseAscii(it->first)) {
+      session_->ResetStream(stream_id_, priority_, RST_STREAM_PROTOCOL_ERROR,
+                            "Upper case characters in header: " + it->first);
+      return ERR_SPDY_PROTOCOL_ERROR;
+    }
+
+    SpdyHeaderBlock::iterator it2 = response_headers_.lower_bound(it->first);
+    // Disallow duplicate headers.  This is just to be conservative.
+    if (it2 != response_headers_.end() && it2->first == it->first) {
+      session_->ResetStream(stream_id_, priority_, RST_STREAM_PROTOCOL_ERROR,
+                            "Duplicate header: " + it->first);
+      return ERR_SPDY_PROTOCOL_ERROR;
+    }
+
+    response_headers_.insert(it2, *it);
+  }
+
+  // If delegate_ is not yet attached, we'll call
+  // OnResponseHeadersUpdated() after the delegate gets attached to
+  // the stream.
+  if (delegate_) {
+    // The call to OnResponseHeadersUpdated() below may delete |this|,
+    // so use |weak_this| to detect that.
+    base::WeakPtr<SpdyStream> weak_this = GetWeakPtr();
+
+    SpdyResponseHeadersStatus status =
+        delegate_->OnResponseHeadersUpdated(response_headers_);
+    if (status == RESPONSE_HEADERS_ARE_INCOMPLETE) {
+      // Since RESPONSE_HEADERS_ARE_INCOMPLETE was returned, we must not
+      // have been closed.
+      CHECK(weak_this);
+      // Incomplete headers are OK only for push streams.
+      if (type_ != SPDY_PUSH_STREAM)
+        return ERR_INCOMPLETE_SPDY_HEADERS;
+    } else if (weak_this) {
+      response_headers_status_ = RESPONSE_HEADERS_ARE_COMPLETE;
+    }
+  }
+
+  return OK;
+}
+
 }  // namespace net