Implement PUSH_PROMISE handling in spdy_session

net/spdy/spdy_session.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_session.h"
 
 #include <algorithm>
 #include <map>
 
 #include "base/basictypes.h"
@@ skipping 217 matching lines @@
                                       NetLog::LogLevel /* log_level */) {
   base::DictionaryValue* dict = new base::DictionaryValue();
   dict->SetInteger("last_accepted_stream_id",
                    static_cast<int>(last_stream_id));
   dict->SetInteger("active_streams", active_streams);
   dict->SetInteger("unclaimed_streams", unclaimed_streams);
   dict->SetInteger("status", static_cast<int>(status));
   return dict;
 }
 
+base::Value* NetLogSpdyPushPromiseReceivedCallback(
+    const SpdyHeaderBlock* headers,
+    SpdyStreamId stream_id,
+    SpdyStreamId promised_stream_id,
+    NetLog::LogLevel log_level) {
+  base::DictionaryValue* dict = new base::DictionaryValue();
+  dict->Set("headers",
+            SpdyHeaderBlockToListValue(*headers, log_level).release());
+  dict->SetInteger("id", stream_id);
+  dict->SetInteger("promised_stream_id", promised_stream_id);
+  return dict;
+}
+
 // Helper function to return the total size of an array of objects
 // with .size() member functions.
 template <typename T, size_t N> size_t GetTotalSize(const T (&arr)[N]) {
   size_t total_size = 0;
   for (size_t i = 0; i < N; ++i) {
     total_size += arr[i].size();
   }
   return total_size;
 }
 
@@ skipping 246 matching lines @@
   net_log_ = BoundNetLog();
   callback_.Reset();
 }
 
 SpdySession::ActiveStreamInfo::ActiveStreamInfo()
     : stream(NULL),
       waiting_for_syn_reply(false) {}
 
 SpdySession::ActiveStreamInfo::ActiveStreamInfo(SpdyStream* stream)
     : stream(stream),
-      waiting_for_syn_reply(stream->type() != SPDY_PUSH_STREAM) {}
+      waiting_for_syn_reply(stream->type() != SPDY_PUSH_STREAM),
+      reserved_remote(stream->type() == SPDY_PUSH_STREAM) {
+}
 
 SpdySession::ActiveStreamInfo::~ActiveStreamInfo() {}
 
 SpdySession::PushedStreamInfo::PushedStreamInfo() : stream_id(0) {}
 
 SpdySession::PushedStreamInfo::PushedStreamInfo(
     SpdyStreamId stream_id,
     base::TimeTicks creation_time)
     : stream_id(stream_id),
       creation_time(creation_time) {}
@@ skipping 13 matching lines @@
     size_t max_concurrent_streams_limit,
     TimeFunc time_func,
     const HostPortPair& trusted_spdy_proxy,
     NetLog* net_log)
     : in_io_loop_(false),
       spdy_session_key_(spdy_session_key),
       pool_(NULL),
       http_server_properties_(http_server_properties),
       read_buffer_(new IOBuffer(kReadBufferSize)),
       stream_hi_water_mark_(kFirstStreamId),
+      reserved_remote_stream_num_(0u),
       in_flight_write_frame_type_(DATA),
       in_flight_write_frame_size_(0),
       is_secure_(false),
       certificate_error_code_(OK),
       availability_state_(STATE_AVAILABLE),
       read_state_(READ_STATE_DO_READ),
       write_state_(WRITE_STATE_IDLE),
       error_on_close_(OK),
-      max_concurrent_streams_(initial_max_concurrent_streams == 0 ?
-                              kInitialMaxConcurrentStreams :
-                              initial_max_concurrent_streams),
-      max_concurrent_streams_limit_(max_concurrent_streams_limit == 0 ?
-                                    kMaxConcurrentStreamLimit :
-                                    max_concurrent_streams_limit),
+      max_concurrent_streams_(initial_max_concurrent_streams == 0

[baranovich, 2014/06/13 11:42:46]

its not me, its git cl format=)

[Johnny, 2014/06/17 04:01:33]

Yep, not a problem.

+                                  ? kInitialMaxConcurrentStreams
+                                  : initial_max_concurrent_streams),
+      max_concurrent_streams_limit_(max_concurrent_streams_limit == 0
+                                        ? kMaxConcurrentStreamLimit
+                                        : max_concurrent_streams_limit),
       streams_initiated_count_(0),
       streams_pushed_count_(0),
       streams_pushed_and_claimed_count_(0),
       streams_abandoned_count_(0),
       total_bytes_received_(0),
       sent_settings_(false),
       received_settings_(false),
       stalled_streams_(0),
       pings_in_flight_(0),
       next_ping_id_(1),
       last_activity_time_(time_func()),
       last_compressed_frame_len_(0),
       check_ping_status_pending_(false),
       send_connection_header_prefix_(false),
       flow_control_state_(FLOW_CONTROL_NONE),
       stream_initial_send_window_size_(kSpdyStreamInitialWindowSize),
-      stream_initial_recv_window_size_(stream_initial_recv_window_size == 0 ?
-                                       kDefaultInitialRecvWindowSize :
-                                       stream_initial_recv_window_size),
+      stream_initial_recv_window_size_(stream_initial_recv_window_size == 0
+                                           ? kDefaultInitialRecvWindowSize
+                                           : stream_initial_recv_window_size),
       session_send_window_size_(0),
       session_recv_window_size_(0),
       session_unacked_recv_window_bytes_(0),
       net_log_(BoundNetLog::Make(net_log, NetLog::SOURCE_SPDY_SESSION)),
       verify_domain_authentication_(verify_domain_authentication),
       enable_sending_initial_data_(enable_sending_initial_data),
       enable_compression_(enable_compression),
       enable_ping_based_connection_checking_(
           enable_ping_based_connection_checking),
       protocol_(default_protocol),
       connection_at_risk_of_loss_time_(
           base::TimeDelta::FromSeconds(kDefaultConnectionAtRiskOfLossSeconds)),
-      hung_interval_(
-          base::TimeDelta::FromSeconds(kHungIntervalSeconds)),
+      hung_interval_(base::TimeDelta::FromSeconds(kHungIntervalSeconds)),
       trusted_spdy_proxy_(trusted_spdy_proxy),
       time_func_(time_func),
       weak_factory_(this) {
   DCHECK_GE(protocol_, kProtoSPDYMinimumVersion);
   DCHECK_LE(protocol_, kProtoSPDYMaximumVersion);
   DCHECK(HttpStreamFactory::spdy_enabled());
   net_log_.BeginEvent(
       NetLog::TYPE_SPDY_SESSION,
       base::Bind(&NetLogSpdySessionCallback, &host_port_proxy_pair()));
   next_unclaimed_push_stream_sweep_time_ = time_func_() +
@@ skipping 162 matching lines @@
     return ERR_FAILED;
 
   if (availability_state_ == STATE_DRAINING)
     return ERR_CONNECTION_CLOSED;
 
   Error err = TryAccessStream(request->url());
   if (err != OK)
     return err;
 
   if (!max_concurrent_streams_ ||
-      (active_streams_.size() + created_streams_.size() <
+      (active_streams_.size() + created_streams_.size() -
+           reserved_remote_stream_num_ <

[baranovich, 2014/06/13 11:42:46]

This behavior should be tested. But it is incorrect from the start, since we
count both client and push streams towards a single limit
|max_concurrent_streams_| which is set from server settings and should be a
limit of client initiated stream only. This is not only about SPDY4. Adding a
pushed stream to SpdyNetworkTransactionTest.*MaxConcurrent*  will break them all
=) So, I could:
1) write a test on this incorrect behavior
2) leave this code untested for now, and later implement correct stream limits
in separate CL
3) write that code after correct stream limits implementation.
What is better?

[Johnny, 2014/06/17 04:01:33]

Agreed with your assessment. The current behavior is incorrect, but it's been
consistently incorrect for some time now: I think I prefer option 3), and
leaving as the status quo with a TODO until it can be done properly.

[baranovich, 2014/06/17 21:33:43]

Done. Left this code intact

        max_concurrent_streams_)) {
     return CreateStream(*request, stream);
   }
 
   stalled_streams_++;
   net_log().AddEvent(NetLog::TYPE_SPDY_SESSION_STALLED_MAX_STREAMS);
   RequestPriority priority = request->priority();
   CHECK_GE(priority, MINIMUM_PRIORITY);
   CHECK_LE(priority, MAXIMUM_PRIORITY);
   pending_create_stream_queues_[priority].push_back(request);
@@ skipping 1299 matching lines @@
 }
 
 void SpdySession::OnSynStream(SpdyStreamId stream_id,
                               SpdyStreamId associated_stream_id,
                               SpdyPriority priority,
                               bool fin,
                               bool unidirectional,
                               const SpdyHeaderBlock& headers) {
   CHECK(in_io_loop_);
 
+  if (GetProtocolVersion() >= SPDY4) {
+    DCHECK_EQ(0u, associated_stream_id);
+    OnHeaders(stream_id, fin, headers);
+    return;
+  }
+
   base::Time response_time = base::Time::Now();
   base::TimeTicks recv_first_byte_time = time_func_();
 
   if (net_log_.IsLogging()) {
     net_log_.AddEvent(
         NetLog::TYPE_SPDY_SESSION_PUSHED_SYN_STREAM,
         base::Bind(&NetLogSpdySynStreamReceivedCallback,
                    &headers, fin, unidirectional, priority,
                    stream_id, associated_stream_id));
   }
 
-  // Server-initiated streams should have even sequence numbers.
-  if ((stream_id & 0x1) != 0) {
-    LOG(WARNING) << "Received invalid OnSyn stream id " << stream_id;
+  if (!TryCreatePushStream(stream_id, associated_stream_id, priority, headers))
     return;
-  }
-
-  if (IsStreamActive(stream_id)) {
-    LOG(WARNING) << "Received OnSyn for active stream " << stream_id;
-    return;
-  }
-
-  RequestPriority request_priority =
-      ConvertSpdyPriorityToRequestPriority(priority, GetProtocolVersion());
-
-  if (availability_state_ == STATE_GOING_AWAY) {
-    // TODO(akalin): This behavior isn't in the SPDY spec, although it
-    // probably should be.
-    EnqueueResetStreamFrame(stream_id, request_priority,
-                            RST_STREAM_REFUSED_STREAM,
-                            "OnSyn received when going away");
-    return;
-  }
-
-  // TODO(jgraettinger): SpdyFramer simulates OnSynStream() from HEADERS
-  // frames, which don't convey associated stream ID. Disable this check
-  // for now, and re-enable when PUSH_PROMISE is implemented properly.
-  if (associated_stream_id == 0 && GetProtocolVersion() < SPDY4) {
-    std::string description = base::StringPrintf(
-        "Received invalid OnSyn associated stream id %d for stream %d",
-        associated_stream_id, stream_id);
-    EnqueueResetStreamFrame(stream_id, request_priority,
-                            RST_STREAM_REFUSED_STREAM, description);
-    return;
-  }
-
-  streams_pushed_count_++;
-
-  // TODO(mbelshe): DCHECK that this is a GET method?
-
-  // Verify that the response had a URL for us.
-  GURL gurl = GetUrlFromHeaderBlock(headers, GetProtocolVersion(), true);
-  if (!gurl.is_valid()) {
-    EnqueueResetStreamFrame(
-        stream_id, request_priority, RST_STREAM_PROTOCOL_ERROR,
-        "Pushed stream url was invalid: " + gurl.spec());
-    return;
-  }
-
-  // Verify we have a valid stream association.
-  ActiveStreamMap::iterator associated_it =
-      active_streams_.find(associated_stream_id);
-  // TODO(jgraettinger): (See PUSH_PROMISE comment above).
-  if (GetProtocolVersion() < SPDY4 && associated_it == active_streams_.end()) {
-    EnqueueResetStreamFrame(
-        stream_id, request_priority, RST_STREAM_INVALID_STREAM,
-        base::StringPrintf(
-            "Received OnSyn with inactive associated stream %d",
-            associated_stream_id));
-    return;
-  }
-
-  // Check that the SYN advertises the same origin as its associated stream.
-  // Bypass this check if and only if this session is with a SPDY proxy that
-  // is trusted explicitly via the --trusted-spdy-proxy switch.
-  if (trusted_spdy_proxy_.Equals(host_port_pair())) {
-    // Disallow pushing of HTTPS content.
-    if (gurl.SchemeIs("https")) {
-      EnqueueResetStreamFrame(
-          stream_id, request_priority, RST_STREAM_REFUSED_STREAM,
-          base::StringPrintf(
-              "Rejected push of Cross Origin HTTPS content %d",
-              associated_stream_id));
-    }
-  } else if (GetProtocolVersion() < SPDY4) {
-    // TODO(jgraettinger): (See PUSH_PROMISE comment above).
-    GURL associated_url(associated_it->second.stream->GetUrlFromHeaders());
-    if (associated_url.GetOrigin() != gurl.GetOrigin()) {
-      EnqueueResetStreamFrame(
-          stream_id, request_priority, RST_STREAM_REFUSED_STREAM,
-          base::StringPrintf(
-              "Rejected Cross Origin Push Stream %d",
-              associated_stream_id));
-      return;
-    }
-  }
-
-  // There should not be an existing pushed stream with the same path.
-  PushedStreamMap::iterator pushed_it =
-      unclaimed_pushed_streams_.lower_bound(gurl);
-  if (pushed_it != unclaimed_pushed_streams_.end() &&
-      pushed_it->first == gurl) {
-    EnqueueResetStreamFrame(
-        stream_id, request_priority, RST_STREAM_PROTOCOL_ERROR,
-        "Received duplicate pushed stream with url: " +
-        gurl.spec());
-    return;
-  }
-
-  scoped_ptr<SpdyStream> stream(
-      new SpdyStream(SPDY_PUSH_STREAM, GetWeakPtr(), gurl,
-                     request_priority,
-                     stream_initial_send_window_size_,
-                     stream_initial_recv_window_size_,
-                     net_log_));
-  stream->set_stream_id(stream_id);
-  stream->IncrementRawReceivedBytes(last_compressed_frame_len_);
-  last_compressed_frame_len_ = 0;
-
-  DeleteExpiredPushedStreams();
-  PushedStreamMap::iterator inserted_pushed_it =
-      unclaimed_pushed_streams_.insert(
-          pushed_it,
-          std::make_pair(gurl, PushedStreamInfo(stream_id, time_func_())));
-  DCHECK(inserted_pushed_it != pushed_it);
-
-  InsertActivatedStream(stream.Pass());
 
   ActiveStreamMap::iterator active_it = active_streams_.find(stream_id);
   if (active_it == active_streams_.end()) {
     NOTREACHED();
     return;
   }
 
   // Parse the headers.
 
   // Split headers to simulate push promise and response.
   SpdyHeaderBlock request_headers;
   SpdyHeaderBlock response_headers;
   SplitPushedHeadersToRequestAndResponse(
       headers, GetProtocolVersion(), &request_headers, &response_headers);
 
   if (active_it->second.stream->OnPushPromiseHeadersReceived(request_headers) !=

[Johnny, 2014/06/17 04:01:33]

It seems like TryCreatePushStream() should receive |request_headers|, and should
make the call to OnPushPromiseHeadersReceived().

Also, it should abort adding to the stream to |active_streams_| if
OnPushPromiseHeadersReceived() returns != OK (and also probably call
DoDrainSession()),  or OnPushPromiseHeadersReceived() should just return void
(as it currently does nothing but return OK).

[baranovich, 2014/06/17 21:33:43]

Done.

       OK)
     return;
 
+  DCHECK(active_it->second.reserved_remote);
+  ActivateReservedStreamIterator(active_it);
+
   if (OnInitialResponseHeadersReceived(response_headers,
                                        response_time,
                                        recv_first_byte_time,
                                        active_it->second.stream) != OK)
     return;
 
   base::StatsCounter push_requests("spdy.pushed_streams");
   push_requests.Increment();
 }
 
+void SpdySession::ActivateReservedStreamIterator(ActiveStreamMap::iterator it) {
+  DCHECK_GT(reserved_remote_stream_num_, 0u);

[Johnny, 2014/06/17 04:01:33]

I'm thinking this ought to be folded into OnResponseHeadersReceived(), as a
check of "IsReservedRemote()" prior to calling
stream->OnResponseHeadersReceived().

[baranovich, 2014/06/17 21:33:43]

Done. This method in not necessary unless we have reserved stream counter

+  DCHECK_LE(reserved_remote_stream_num_, active_streams_.size());
+  reserved_remote_stream_num_--;
+  it->second.reserved_remote = false;
+}
+
 void SpdySession::DeleteExpiredPushedStreams() {
   if (unclaimed_pushed_streams_.empty())
     return;
 
   // Check that adequate time has elapsed since the last sweep.
   if (time_func_() < next_unclaimed_push_stream_sweep_time_)
     return;
 
   // Gather old streams to delete.
   base::TimeTicks minimum_freshness = time_func_() -
@@ skipping 88 matching lines @@
     LOG(WARNING) << "Received HEADERS for invalid stream " << stream_id;
     return;
   }
 
   SpdyStream* stream = it->second.stream;
   CHECK_EQ(stream->stream_id(), stream_id);
 
   stream->IncrementRawReceivedBytes(last_compressed_frame_len_);
   last_compressed_frame_len_ = 0;
 
+  base::Time response_time = base::Time::Now();
+  base::TimeTicks recv_first_byte_time = time_func_();
+
   if (it->second.waiting_for_syn_reply) {
     if (GetProtocolVersion() < SPDY4) {
       const std::string& error =
           "Was expecting SYN_REPLY, not HEADERS.";
       stream->LogStreamError(ERR_SPDY_PROTOCOL_ERROR, error);
       ResetStreamIterator(it, RST_STREAM_PROTOCOL_ERROR, error);
       return;
     }
-    base::Time response_time = base::Time::Now();
-    base::TimeTicks recv_first_byte_time = time_func_();
 
     it->second.waiting_for_syn_reply = false;
     ignore_result(OnInitialResponseHeadersReceived(
         headers, response_time, recv_first_byte_time, stream));
+  } else if (it->second.reserved_remote) {
+    ActivateReservedStreamIterator(it);
+    ignore_result(OnInitialResponseHeadersReceived(
+        headers, response_time, recv_first_byte_time, stream));
   } else {
     int rv = stream->OnAdditionalResponseHeadersReceived(headers);
     if (rv < 0) {
       DCHECK_NE(rv, ERR_IO_PENDING);
       DCHECK(active_streams_.find(stream_id) == active_streams_.end());
     }
   }
 }
 
 void SpdySession::OnRstStream(SpdyStreamId stream_id,
@@ skipping 139 matching lines @@
                               "delta_window_size %ud", delta_window_size));
       return;
     }
 
     CHECK_EQ(it->second.stream->stream_id(), stream_id);
     it->second.stream->IncreaseSendWindowSize(
         static_cast<int32>(delta_window_size));
   }
 }
 
+bool SpdySession::TryCreatePushStream(SpdyStreamId stream_id,
+                                      SpdyStreamId associated_stream_id,
+                                      SpdyPriority priority,
+                                      const SpdyHeaderBlock& headers) {
+  // Server-initiated streams should have even sequence numbers.
+  if ((stream_id & 0x1) != 0) {
+    LOG(WARNING) << "Received invalid push stream id " << stream_id;
+    return false;
+  }
+
+  if (IsStreamActive(stream_id)) {
+    LOG(WARNING) << "Received push for active stream " << stream_id;
+    return false;
+  }
+
+  RequestPriority request_priority =
+      ConvertSpdyPriorityToRequestPriority(priority, GetProtocolVersion());
+
+  if (availability_state_ == STATE_GOING_AWAY) {
+    // TODO(akalin): This behavior isn't in the SPDY spec, although it
+    // probably should be.
+    EnqueueResetStreamFrame(stream_id,
+                            request_priority,
+                            RST_STREAM_REFUSED_STREAM,
+                            "push stream request received when going away");
+    return false;
+  }
+
+  if (associated_stream_id == 0) {
+    // In SPDY4 0 stream id in PUSH_PROMISE frame leads to framer error and
+    // session going away. We should never get here.
+    CHECK_GT(SPDY4, GetProtocolVersion());
+    std::string description = base::StringPrintf(
+        "Received invalid associated stream id %d for pushed stream %d",
+        associated_stream_id,
+        stream_id);
+    EnqueueResetStreamFrame(
+        stream_id, request_priority, RST_STREAM_REFUSED_STREAM, description);
+    return false;
+  }
+
+  streams_pushed_count_++;
+
+  // TODO(mbelshe): DCHECK that this is a GET method?
+
+  // Verify that the response had a URL for us.
+  GURL gurl = GetUrlFromHeaderBlock(headers, GetProtocolVersion(), true);
+  if (!gurl.is_valid()) {
+    EnqueueResetStreamFrame(stream_id,
+                            request_priority,
+                            RST_STREAM_PROTOCOL_ERROR,
+                            "Pushed stream url was invalid: " + gurl.spec());
+    return false;
+  }
+
+  // Verify we have a valid stream association.
+  ActiveStreamMap::iterator associated_it =
+      active_streams_.find(associated_stream_id);
+  if (associated_it == active_streams_.end()) {
+    EnqueueResetStreamFrame(
+        stream_id,
+        request_priority,
+        RST_STREAM_INVALID_STREAM,
+        base::StringPrintf("Received push for inactive associated stream %d",
+                           associated_stream_id));
+    return false;
+  }
+
+  // Check that the pushed stream advertises the same origin as its associated
+  // stream. Bypass this check if and only if this session is with a SPDY proxy
+  // that is trusted explicitly via the --trusted-spdy-proxy switch.
+  if (trusted_spdy_proxy_.Equals(host_port_pair())) {
+    // Disallow pushing of HTTPS content.
+    if (gurl.SchemeIs("https")) {
+      EnqueueResetStreamFrame(
+          stream_id,
+          request_priority,
+          RST_STREAM_REFUSED_STREAM,
+          base::StringPrintf("Rejected push of Cross Origin HTTPS content %d",
+                             associated_stream_id));
+    }
+  } else {
+    GURL associated_url(associated_it->second.stream->GetUrlFromHeaders());
+    if (associated_url.GetOrigin() != gurl.GetOrigin()) {
+      EnqueueResetStreamFrame(
+          stream_id,
+          request_priority,
+          RST_STREAM_REFUSED_STREAM,
+          base::StringPrintf("Rejected Cross Origin Push Stream %d",
+                             associated_stream_id));
+      return false;
+    }
+  }
+
+  // There should not be an existing pushed stream with the same path.
+  PushedStreamMap::iterator pushed_it =
+      unclaimed_pushed_streams_.lower_bound(gurl);
+  if (pushed_it != unclaimed_pushed_streams_.end() &&
+      pushed_it->first == gurl) {
+    EnqueueResetStreamFrame(
+        stream_id,
+        request_priority,
+        RST_STREAM_PROTOCOL_ERROR,
+        "Received duplicate pushed stream with url: " + gurl.spec());
+    return false;
+  }
+
+  scoped_ptr<SpdyStream> stream(new SpdyStream(SPDY_PUSH_STREAM,
+                                               GetWeakPtr(),
+                                               gurl,
+                                               request_priority,
+                                               stream_initial_send_window_size_,
+                                               stream_initial_recv_window_size_,
+                                               net_log_));
+  stream->set_stream_id(stream_id);
+
+  // In spdy4/http2 PUSH_PROMISE arrives on associated stream.
+  if (associated_it != active_streams_.end() && GetProtocolVersion() >= SPDY4) {
+    associated_it->second.stream->IncrementRawReceivedBytes(
+        last_compressed_frame_len_);
+  } else {
+    stream->IncrementRawReceivedBytes(last_compressed_frame_len_);
+  }
+
+  last_compressed_frame_len_ = 0;
+
+  DeleteExpiredPushedStreams();
+  PushedStreamMap::iterator inserted_pushed_it =
+      unclaimed_pushed_streams_.insert(
+          pushed_it,
+          std::make_pair(gurl, PushedStreamInfo(stream_id, time_func_())));
+  DCHECK(inserted_pushed_it != pushed_it);
+
+  InsertActivatedStream(stream.Pass());
+
+  reserved_remote_stream_num_++;
+  DCHECK_GT(reserved_remote_stream_num_, 0u);
+  DCHECK_LE(reserved_remote_stream_num_, active_streams_.size());
+
+  return true;
+}
+
 void SpdySession::OnPushPromise(SpdyStreamId stream_id,
                                 SpdyStreamId promised_stream_id,
                                 const SpdyHeaderBlock& headers) {
-  // TODO(akalin): Handle PUSH_PROMISE frames.
+  CHECK(in_io_loop_);
+
+  if (net_log_.IsLogging()) {
+    net_log_.AddEvent(NetLog::TYPE_SPDY_SESSION_RECV_PUSH_PROMISE,
+                      base::Bind(&NetLogSpdyPushPromiseReceivedCallback,
+                                 &headers,
+                                 stream_id,
+                                 promised_stream_id));
+  }
+
+  // Any priority will do.
+  // TODO(baranovich): pass parent stream id priority?
+  if (!TryCreatePushStream(promised_stream_id, stream_id, 0, headers))
+    return;
+
+  ActiveStreamMap::iterator active_it =
+      active_streams_.find(promised_stream_id);
+  if (active_it == active_streams_.end()) {
+    NOTREACHED();
+    return;
+  }
+  DCHECK(active_it->second.reserved_remote);
+
+  // Parse the headers.
+  if (active_it->second.stream->OnPushPromiseHeadersReceived(headers) != OK)
+    return;
+
+  base::StatsCounter push_requests("spdy.pushed_streams");
+  push_requests.Increment();
 }
 
 void SpdySession::SendStreamWindowUpdate(SpdyStreamId stream_id,
                                          uint32 delta_window_size) {
   CHECK_GE(flow_control_state_, FLOW_CONTROL_STREAM);
   ActiveStreamMap::const_iterator it = active_streams_.find(stream_id);
   CHECK(it != active_streams_.end());
   CHECK_EQ(it->second.stream->stream_id(), stream_id);
   SendWindowUpdateFrame(
       stream_id, delta_window_size, it->second.stream->priority());
@@ skipping 513 matching lines @@
     if (!queue->empty()) {
       SpdyStreamId stream_id = queue->front();
       queue->pop_front();
       return stream_id;
     }
   }
   return 0;
 }
 
 }  // namespace net