Implement PUSH_PROMISE handling in spdy_session

BUG=377538

R=jgraettinger@chromium.org

TEST=SpdyStream/SpdySession/SpdyNetworkTransaction push related tests.
     Re-enabled previously disabled tests:
     SpdyNetworkTransactionTest.ServerPushCrossOriginCorrectness
     SpdyNetworkTransactionTest.ServerPushInvalidAssociatedStreamID9

Review URL: https://codereview.chromium.org/331663007

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@278733 0039d316-1c4b-4281-b951-d872f2087c98

7 files changed, 342 insertions, 235 deletions

diff --git a/net/spdy/spdy_network_transaction_unittest.cc b/net/spdy/spdy_network_transaction_unittest.cc
index 5e86757..f11cb65 100644
--- a/net/spdy/spdy_network_transaction_unittest.cc
+++ b/net/spdy/spdy_network_transaction_unittest.cc
@@ -3091,10 +3091,11 @@ TEST_P(SpdyNetworkTransactionTest, ServerPushMultipleDataFrameInterrupted) {
 
 TEST_P(SpdyNetworkTransactionTest, ServerPushInvalidAssociatedStreamID0) {
   if (spdy_util_.spdy_version() == SPDY4) {
-    // TODO(jgraettinger): We don't support associated stream
-    // checks in SPDY4 yet.
+    // PUSH_PROMISE with stream id 0 is connection-level error.
+    // TODO(baranovich): Test session going away.
     return;
   }
+
   scoped_ptr<SpdyFrame> stream1_syn(
       spdy_util_.ConstructSpdyGet(NULL, 0, false, 1, LOWEST, true));
   scoped_ptr<SpdyFrame> stream1_body(
@@ -3156,11 +3157,6 @@ TEST_P(SpdyNetworkTransactionTest, ServerPushInvalidAssociatedStreamID0) {
 }
 
 TEST_P(SpdyNetworkTransactionTest, ServerPushInvalidAssociatedStreamID9) {
-  if (spdy_util_.spdy_version() == SPDY4) {
-    // TODO(jgraettinger): We don't support associated stream
-    // checks in SPDY4 yet.
-    return;
-  }
   scoped_ptr<SpdyFrame> stream1_syn(
       spdy_util_.ConstructSpdyGet(NULL, 0, false, 1, LOWEST, true));
   scoped_ptr<SpdyFrame> stream1_body(
@@ -3239,15 +3235,8 @@ TEST_P(SpdyNetworkTransactionTest, ServerPushNoURL) {
   (*incomplete_headers)["hello"] = "bye";
   (*incomplete_headers)[spdy_util_.GetStatusKey()] = "200 OK";
   (*incomplete_headers)[spdy_util_.GetVersionKey()] = "HTTP/1.1";
-  scoped_ptr<SpdyFrame> stream2_syn(
-      spdy_util_.ConstructSpdyControlFrame(incomplete_headers.Pass(),
-                                           false,
-                                           2,  // Stream ID
-                                           LOWEST,
-                                           SYN_STREAM,
-                                           CONTROL_FLAG_NONE,
-                                           // Associated stream ID
-                                           1));
+  scoped_ptr<SpdyFrame> stream2_syn(spdy_util_.ConstructInitialSpdyPushFrame(
+      incomplete_headers.Pass(), 2, 1));
   MockRead reads[] = {
     CreateMockRead(*stream1_reply, 2),
     CreateMockRead(*stream2_syn, 3),
@@ -5205,13 +5194,7 @@ TEST_P(SpdyNetworkTransactionTest, ServerPushWithHeaders) {
   spdy_util_.AddUrlToHeaderBlock(
       "http://www.google.com/foo.dat", initial_headers.get());
   scoped_ptr<SpdyFrame> stream2_syn(
-      spdy_util_.ConstructSpdyControlFrame(initial_headers.Pass(),
-                                           false,
-                                           2,
-                                           LOWEST,
-                                           SYN_STREAM,
-                                           CONTROL_FLAG_NONE,
-                                           1));
+      spdy_util_.ConstructInitialSpdyPushFrame(initial_headers.Pass(), 2, 1));
 
   scoped_ptr<SpdyHeaderBlock> late_headers(new SpdyHeaderBlock());
   (*late_headers)["hello"] = "bye";
@@ -5274,13 +5257,7 @@ TEST_P(SpdyNetworkTransactionTest, ServerPushClaimBeforeHeaders) {
   spdy_util_.AddUrlToHeaderBlock(
       "http://www.google.com/foo.dat", initial_headers.get());
   scoped_ptr<SpdyFrame> stream2_syn(
-      spdy_util_.ConstructSpdyControlFrame(initial_headers.Pass(),
-                                           false,
-                                           2,
-                                           LOWEST,
-                                           SYN_STREAM,
-                                           CONTROL_FLAG_NONE,
-                                           1));
+      spdy_util_.ConstructInitialSpdyPushFrame(initial_headers.Pass(), 2, 1));
 
   scoped_ptr<SpdyHeaderBlock> late_headers(new SpdyHeaderBlock());
   (*late_headers)["hello"] = "bye";
@@ -5397,17 +5374,14 @@ TEST_P(SpdyNetworkTransactionTest, ServerPushWithTwoHeaderFrames) {
   };
 
   scoped_ptr<SpdyHeaderBlock> initial_headers(new SpdyHeaderBlock());
-  (*initial_headers)["alpha"] = "beta";
+  if (spdy_util_.spdy_version() < SPDY4) {
+    // In SPDY4 PUSH_PROMISE headers won't show up in the response headers.
+    (*initial_headers)["alpha"] = "beta";
+  }
   spdy_util_.AddUrlToHeaderBlock(
       "http://www.google.com/foo.dat", initial_headers.get());
   scoped_ptr<SpdyFrame> stream2_syn(
-      spdy_util_.ConstructSpdyControlFrame(initial_headers.Pass(),
-                                           false,
-                                           2,
-                                           LOWEST,
-                                           SYN_STREAM,
-                                           CONTROL_FLAG_NONE,
-                                           1));
+      spdy_util_.ConstructInitialSpdyPushFrame(initial_headers.Pass(), 2, 1));
 
   scoped_ptr<SpdyHeaderBlock> middle_headers(new SpdyHeaderBlock());
   (*middle_headers)["hello"] = "bye";
@@ -5515,7 +5489,8 @@ TEST_P(SpdyNetworkTransactionTest, ServerPushWithTwoHeaderFrames) {
   EXPECT_EQ("HTTP/1.1 200 OK", response2.headers->GetStatusLine());
 
   // Verify we got all the headers from all header blocks.
-  EXPECT_TRUE(response2.headers->HasHeaderValue("alpha", "beta"));
+  if (spdy_util_.spdy_version() < SPDY4)
+    EXPECT_TRUE(response2.headers->HasHeaderValue("alpha", "beta"));
   EXPECT_TRUE(response2.headers->HasHeaderValue("hello", "bye"));
   EXPECT_TRUE(response2.headers->HasHeaderValue("status", "200"));
 
@@ -5541,13 +5516,7 @@ TEST_P(SpdyNetworkTransactionTest, ServerPushWithNoStatusHeaderFrames) {
   spdy_util_.AddUrlToHeaderBlock(
       "http://www.google.com/foo.dat", initial_headers.get());
   scoped_ptr<SpdyFrame> stream2_syn(
-      spdy_util_.ConstructSpdyControlFrame(initial_headers.Pass(),
-                                           false,
-                                           2,
-                                           LOWEST,
-                                           SYN_STREAM,
-                                           CONTROL_FLAG_NONE,
-                                           1));
+      spdy_util_.ConstructInitialSpdyPushFrame(initial_headers.Pass(), 2, 1));
 
   scoped_ptr<SpdyHeaderBlock> middle_headers(new SpdyHeaderBlock());
   (*middle_headers)["hello"] = "bye";
@@ -5744,11 +5713,6 @@ TEST_P(SpdyNetworkTransactionTest, SynReplyWithLateHeaders) {
 }
 
 TEST_P(SpdyNetworkTransactionTest, ServerPushCrossOriginCorrectness) {
-  if (spdy_util_.spdy_version() == SPDY4) {
-    // TODO(jgraettinger): We don't support associated stream
-    // checks in SPDY4 yet.
-    return;
-  }
   // In this test we want to verify that we can't accidentally push content
   // which can't be pushed by this content server.
   // This test assumes that:
diff --git a/net/spdy/spdy_session.cc b/net/spdy/spdy_session.cc
index 8362e56..7561145 100644
--- a/net/spdy/spdy_session.cc
+++ b/net/spdy/spdy_session.cc
@@ -235,6 +235,19 @@ base::Value* NetLogSpdyGoAwayCallback(SpdyStreamId last_stream_id,
   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]) {
@@ -501,7 +514,8 @@ SpdySession::ActiveStreamInfo::ActiveStreamInfo()
 
 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) {
+}
 
 SpdySession::ActiveStreamInfo::~ActiveStreamInfo() {}
 
@@ -543,12 +557,12 @@ SpdySession::SpdySession(
       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
+                                  ? 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),
@@ -565,9 +579,9 @@ SpdySession::SpdySession(
       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),
@@ -580,8 +594,7 @@ SpdySession::SpdySession(
       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) {
@@ -2084,6 +2097,12 @@ void SpdySession::OnSynStream(SpdyStreamId stream_id,
                               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_();
 
@@ -2095,122 +2114,15 @@ void SpdySession::OnSynStream(SpdyStreamId stream_id,
                    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;
-    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;
-    }
-  }
+  // Split headers to simulate push promise and response.
+  SpdyHeaderBlock request_headers;
+  SpdyHeaderBlock response_headers;
+  SplitPushedHeadersToRequestAndResponse(
+      headers, GetProtocolVersion(), &request_headers, &response_headers);
 
-  // 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());
+  if (!TryCreatePushStream(
+          stream_id, associated_stream_id, priority, request_headers))
     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()) {
@@ -2218,18 +2130,6 @@ void SpdySession::OnSynStream(SpdyStreamId stream_id,
     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) !=
-      OK)
-    return;
-
   if (OnInitialResponseHeadersReceived(response_headers,
                                        response_time,
                                        recv_first_byte_time,
@@ -2348,6 +2248,9 @@ void SpdySession::OnHeaders(SpdyStreamId 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 =
@@ -2356,12 +2259,13 @@ void SpdySession::OnHeaders(SpdyStreamId stream_id,
       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.stream->IsReservedRemote()) {
+    ignore_result(OnInitialResponseHeadersReceived(
+        headers, response_time, recv_first_byte_time, stream));
   } else {
     int rv = stream->OnAdditionalResponseHeadersReceived(headers);
     if (rv < 0) {
@@ -2521,10 +2425,172 @@ void SpdySession::OnWindowUpdate(SpdyStreamId stream_id,
   }
 }
 
+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());
+
+  ActiveStreamMap::iterator active_it = active_streams_.find(stream_id);
+  if (active_it == active_streams_.end()) {
+    NOTREACHED();
+    return false;
+  }
+
+  active_it->second.stream->OnPushPromiseHeadersReceived(headers);
+  DCHECK(active_it->second.stream->IsReservedRemote());
+  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;
+
+  base::StatsCounter push_requests("spdy.pushed_streams");
+  push_requests.Increment();
 }
 
 void SpdySession::SendStreamWindowUpdate(SpdyStreamId stream_id,
diff --git a/net/spdy/spdy_session.h b/net/spdy/spdy_session.h
index 16c17a3..b74b1ee 100644
--- a/net/spdy/spdy_session.h
+++ b/net/spdy/spdy_session.h
@@ -611,6 +611,11 @@ class NET_EXPORT SpdySession : public BufferedSpdyFramerVisitorInterface,
   // possible.
   void ProcessPendingStreamRequests();
 
+  bool TryCreatePushStream(SpdyStreamId stream_id,
+                           SpdyStreamId associated_stream_id,
+                           SpdyPriority priority,
+                           const SpdyHeaderBlock& headers);
+
   // Close the stream pointed to by the given iterator. Note that that
   // stream may hold the last reference to the session.
   void CloseActiveStreamIterator(ActiveStreamMap::iterator it, int status);
diff --git a/net/spdy/spdy_stream.cc b/net/spdy/spdy_stream.cc
index fe2f37a..40ec654b 100644
--- a/net/spdy/spdy_stream.cc
+++ b/net/spdy/spdy_stream.cc
@@ -122,10 +122,9 @@ void SpdyStream::SetDelegate(Delegate* delegate) {
   CHECK(delegate);
   delegate_ = delegate;
 
-  // TODO(baranovich): allow STATE_RESERVED_REMOTE when push promises will be
-  // implemented.
   CHECK(io_state_ == STATE_IDLE ||
-        io_state_ == STATE_HALF_CLOSED_LOCAL_UNCLAIMED);
+        io_state_ == STATE_HALF_CLOSED_LOCAL_UNCLAIMED ||
+        io_state_ == STATE_RESERVED_REMOTE);
 
   if (io_state_ == STATE_HALF_CLOSED_LOCAL_UNCLAIMED) {
     DCHECK_EQ(type_, SPDY_PUSH_STREAM);
@@ -419,13 +418,12 @@ int SpdyStream::OnInitialResponseHeadersReceived(
       // Push streams transition to a locally half-closed state upon headers.
       // We must continue to buffer data while waiting for a call to
       // SetDelegate() (which may not ever happen).
-      // TODO(baranovich): For HTTP 2 push streams, delegate may be set before
-      // receiving response headers when PUSH_PROMISE will be implemented.
-      // TODO(baranovich): In HTTP 2 additional HEADERS frames are not allowed.
-      // Set |response_headers_status_| to RESPONSE_HEADERS_ARE_COMPLETE.
       CHECK_EQ(io_state_, STATE_RESERVED_REMOTE);
-      DCHECK(!delegate_);
-      io_state_ = STATE_HALF_CLOSED_LOCAL_UNCLAIMED;
+      if (!delegate_) {
+        io_state_ = STATE_HALF_CLOSED_LOCAL_UNCLAIMED;
+      } else {
+        io_state_ = STATE_HALF_CLOSED_LOCAL;
+      }
       break;
   }
 
@@ -455,7 +453,7 @@ int SpdyStream::OnAdditionalResponseHeadersReceived(
   return MergeWithResponseHeaders(additional_response_headers);
 }
 
-int SpdyStream::OnPushPromiseHeadersReceived(const SpdyHeaderBlock& headers) {
+void SpdyStream::OnPushPromiseHeadersReceived(const SpdyHeaderBlock& headers) {
   CHECK(!request_headers_.get());
   CHECK_EQ(io_state_, STATE_IDLE);
   CHECK_EQ(type_, SPDY_PUSH_STREAM);
@@ -463,7 +461,6 @@ int SpdyStream::OnPushPromiseHeadersReceived(const SpdyHeaderBlock& headers) {
 
   io_state_ = STATE_RESERVED_REMOTE;
   request_headers_.reset(new SpdyHeaderBlock(headers));
-  return OK;
 }
 
 void SpdyStream::OnDataReceived(scoped_ptr<SpdyBuffer> buffer) {
@@ -733,6 +730,10 @@ bool SpdyStream::IsOpen() const {
   return io_state_ == STATE_OPEN;
 }
 
+bool SpdyStream::IsReservedRemote() const {
+  return io_state_ == STATE_RESERVED_REMOTE;
+}
+
 NextProto SpdyStream::GetProtocol() const {
   return session_->protocol();
 }
diff --git a/net/spdy/spdy_stream.h b/net/spdy/spdy_stream.h
index 431d26d..3a4a283 100644
--- a/net/spdy/spdy_stream.h
+++ b/net/spdy/spdy_stream.h
@@ -301,9 +301,7 @@ class NET_EXPORT_PRIVATE SpdyStream {
 
   // Called by the SpdySession when a frame carrying request headers opening a
   // push stream is received. Stream transits to STATE_RESERVED_REMOTE state.
-  // Returns a status code; if it is an error, the stream was closed by this
-  // function.
-  int OnPushPromiseHeadersReceived(const SpdyHeaderBlock& headers);
+  void OnPushPromiseHeadersReceived(const SpdyHeaderBlock& headers);
 
   // Called by the SpdySession when response data has been received
   // for this stream.  This callback may be called multiple times as
@@ -406,6 +404,11 @@ class NET_EXPORT_PRIVATE SpdyStream {
   // response headers are complete, and it is not in a half-closed state.
   bool IsOpen() const;
 
+  // Returns whether the stream is reserved by remote endpoint: server has sent
+  // intended request headers for a pushed stream, but haven't started response
+  // yet.
+  bool IsReservedRemote() const;
+
   // Returns the protocol used by this stream. Always between
   // kProtoSPDYMinimumVersion and kProtoSPDYMaximumVersion.
   NextProto GetProtocol() const;
diff --git a/net/spdy/spdy_test_util_common.cc b/net/spdy/spdy_test_util_common.cc
index b943c31..3d73c7f 100644
--- a/net/spdy/spdy_test_util_common.cc
+++ b/net/spdy/spdy_test_util_common.cc
@@ -998,21 +998,44 @@ SpdyFrame* SpdyTestUtil::ConstructSpdyPush(const char* const extra_headers[],
                                            int stream_id,
                                            int associated_stream_id,
                                            const char* url) {
-  scoped_ptr<SpdyHeaderBlock> headers(new SpdyHeaderBlock());
-  (*headers)["hello"] = "bye";
-  (*headers)[GetStatusKey()] = "200 OK";
-  if (include_version_header()) {
-    (*headers)[GetVersionKey()] = "HTTP/1.1";
+  if (spdy_version() < SPDY4) {
+    SpdySynStreamIR syn_stream(stream_id);
+    syn_stream.set_associated_to_stream_id(associated_stream_id);
+    syn_stream.SetHeader("hello", "bye");
+    syn_stream.SetHeader(GetStatusKey(), "200 OK");
+    syn_stream.SetHeader(GetVersionKey(), "HTTP/1.1");
+    AddUrlToHeaderBlock(url, syn_stream.mutable_name_value_block());
+    AppendToHeaderBlock(extra_headers,
+                        extra_header_count,
+                        syn_stream.mutable_name_value_block());
+    return CreateFramer(false)->SerializeFrame(syn_stream);
+  } else {
+    SpdyPushPromiseIR push_promise(associated_stream_id, stream_id);
+    AddUrlToHeaderBlock(url, push_promise.mutable_name_value_block());
+    scoped_ptr<SpdyFrame> push_promise_frame(
+        CreateFramer(false)->SerializeFrame(push_promise));
+
+    // Use SynStreamIR to create HEADERS+PRIORITY. Direct creation breaks
+    // framer.
+    SpdySynStreamIR headers(stream_id);
+    SetPriority(LOWEST, &headers);
+    headers.SetHeader("hello", "bye");
+    headers.SetHeader(GetStatusKey(), "200 OK");
+    AppendToHeaderBlock(
+        extra_headers, extra_header_count, headers.mutable_name_value_block());
+    scoped_ptr<SpdyFrame> headers_frame(
+        CreateFramer(false)->SerializeFrame(headers));
+
+    int joint_data_size = push_promise_frame->size() + headers_frame->size();
+    scoped_ptr<char[]> data(new char[joint_data_size]);
+    const SpdyFrame* frames[2] = {
+        push_promise_frame.get(), headers_frame.get(),
+    };
+    int combined_size =
+        CombineFrames(frames, arraysize(frames), data.get(), joint_data_size);
+    DCHECK_EQ(combined_size, joint_data_size);
+    return new SpdyFrame(data.release(), joint_data_size, true);
   }
-  AddUrlToHeaderBlock(url, headers.get());
-  AppendToHeaderBlock(extra_headers, extra_header_count, headers.get());
-  return ConstructSpdyControlFrame(headers.Pass(),
-                                   false,
-                                   stream_id,
-                                   LOWEST,
-                                   SYN_STREAM,
-                                   CONTROL_FLAG_NONE,
-                                   associated_stream_id);
 }
 
 SpdyFrame* SpdyTestUtil::ConstructSpdyPush(const char* const extra_headers[],
@@ -1022,22 +1045,63 @@ SpdyFrame* SpdyTestUtil::ConstructSpdyPush(const char* const extra_headers[],
                                            const char* url,
                                            const char* status,
                                            const char* location) {
-  scoped_ptr<SpdyHeaderBlock> headers(new SpdyHeaderBlock());
-  (*headers)["hello"] = "bye";
-  (*headers)[GetStatusKey()] = status;
-  if (include_version_header()) {
-    (*headers)[GetVersionKey()] = "HTTP/1.1";
+  if (spdy_version() < SPDY4) {
+    SpdySynStreamIR syn_stream(stream_id);
+    syn_stream.set_associated_to_stream_id(associated_stream_id);
+    syn_stream.SetHeader("hello", "bye");
+    syn_stream.SetHeader(GetStatusKey(), status);
+    syn_stream.SetHeader(GetVersionKey(), "HTTP/1.1");
+    syn_stream.SetHeader("location", location);
+    AddUrlToHeaderBlock(url, syn_stream.mutable_name_value_block());
+    AppendToHeaderBlock(extra_headers,
+                        extra_header_count,
+                        syn_stream.mutable_name_value_block());
+    return CreateFramer(false)->SerializeFrame(syn_stream);
+  } else {
+    SpdyPushPromiseIR push_promise(associated_stream_id, stream_id);
+    AddUrlToHeaderBlock(url, push_promise.mutable_name_value_block());
+    scoped_ptr<SpdyFrame> push_promise_frame(
+        CreateFramer(false)->SerializeFrame(push_promise));
+
+    // Use SynStreamIR to create HEADERS+PRIORITY. Direct creation breaks
+    // framer.
+    SpdySynStreamIR headers(stream_id);
+    SetPriority(LOWEST, &headers);
+    headers.SetHeader("hello", "bye");
+    headers.SetHeader(GetStatusKey(), status);
+    headers.SetHeader("location", location);
+    AppendToHeaderBlock(
+        extra_headers, extra_header_count, headers.mutable_name_value_block());
+    scoped_ptr<SpdyFrame> headers_frame(
+        CreateFramer(false)->SerializeFrame(headers));
+
+    int joint_data_size = push_promise_frame->size() + headers_frame->size();
+    scoped_ptr<char[]> data(new char[joint_data_size]);
+    const SpdyFrame* frames[2] = {
+        push_promise_frame.get(), headers_frame.get(),
+    };
+    int combined_size =
+        CombineFrames(frames, arraysize(frames), data.get(), joint_data_size);
+    DCHECK_EQ(combined_size, joint_data_size);
+    return new SpdyFrame(data.release(), joint_data_size, true);
+  }
+}
+
+SpdyFrame* SpdyTestUtil::ConstructInitialSpdyPushFrame(
+    scoped_ptr<SpdyHeaderBlock> headers,
+    int stream_id,
+    int associated_stream_id) {
+  if (spdy_version() < SPDY4) {
+    SpdySynStreamIR syn_stream(stream_id);
+    syn_stream.set_associated_to_stream_id(associated_stream_id);
+    SetPriority(LOWEST, &syn_stream);
+    syn_stream.set_name_value_block(*headers);
+    return CreateFramer(false)->SerializeFrame(syn_stream);
+  } else {
+    SpdyPushPromiseIR push_promise(associated_stream_id, stream_id);
+    push_promise.set_name_value_block(*headers);
+    return CreateFramer(false)->SerializeFrame(push_promise);
   }
-  (*headers)["location"] = location;
-  AddUrlToHeaderBlock(url, headers.get());
-  AppendToHeaderBlock(extra_headers, extra_header_count, headers.get());
-  return ConstructSpdyControlFrame(headers.Pass(),
-                                   false,
-                                   stream_id,
-                                   LOWEST,
-                                   SYN_STREAM,
-                                   CONTROL_FLAG_NONE,
-                                   associated_stream_id);
 }
 
 SpdyFrame* SpdyTestUtil::ConstructSpdyPushHeaders(
diff --git a/net/spdy/spdy_test_util_common.h b/net/spdy/spdy_test_util_common.h
index 3508b5d..f04ad42 100644
--- a/net/spdy/spdy_test_util_common.h
+++ b/net/spdy/spdy_test_util_common.h
@@ -457,6 +457,10 @@ class SpdyTestUtil {
                                const char* status,
                                const char* location);
 
+  SpdyFrame* ConstructInitialSpdyPushFrame(scoped_ptr<SpdyHeaderBlock> headers,
+                                           int stream_id,
+                                           int associated_stream_id);
+
   SpdyFrame* ConstructSpdyPushHeaders(int stream_id,
                                       const char* const extra_headers[],
                                       int extra_header_count);