Break out the basic reliable QUIC stream functionality from the

net/quic/quic_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/quic/quic_session.h"
 
 #include "base/stl_util.h"
 #include "net/quic/crypto/proof_verifier.h"
 #include "net/quic/quic_connection.h"
 #include "net/ssl/ssl_info.h"
@@ skipping 119 matching lines @@
   }
 
   for (size_t i = 0; i < frames.size(); ++i) {
     QuicStreamId stream_id = frames[i].stream_id;
     ReliableQuicStream* stream = GetStream(stream_id);
     if (!stream) {
       continue;
     }
     stream->OnStreamFrame(frames[i]);
 
-    // If the stream had been prematurely closed, and the
+    // If the stream is a data stream had been prematurely closed, and the
     // headers are now decompressed, then we are finally finished
     // with this stream.
     if (ContainsKey(zombie_streams_, stream_id) &&
-        stream->headers_decompressed()) {
+        static_cast<QuicDataStream*>(stream)->headers_decompressed()) {
       CloseZombieStream(stream_id);
     }
   }
 
   while (!decompression_blocked_streams_.empty()) {
     QuicHeaderId header_id = decompression_blocked_streams_.begin()->first;
     if (header_id != decompressor_.current_header_id()) {
       break;
     }
     QuicStreamId stream_id = decompression_blocked_streams_.begin()->second;
     decompression_blocked_streams_.erase(header_id);
-    ReliableQuicStream* stream = GetStream(stream_id);
+    QuicDataStream* stream = GetDataStream(stream_id);
     if (!stream) {
       connection()->SendConnectionClose(
           QUIC_STREAM_RST_BEFORE_HEADERS_DECOMPRESSED);
       return false;
     }
     stream->OnDecompressorAvailable();
   }
   return true;
 }
 
 void QuicSession::OnRstStream(const QuicRstStreamFrame& frame) {
-  ReliableQuicStream* stream = GetStream(frame.stream_id);
+  if (frame.stream_id == kCryptoStreamId) {
+    connection()->SendConnectionCloseWithDetails(
+        QUIC_INVALID_STREAM_ID,
+        "Attempt to reset the crypto stream");
+    return;
+  }
+  QuicDataStream* stream = GetDataStream(frame.stream_id);
   if (!stream) {
     return;  // Errors are handled by GetStream.
   }
   if (ContainsKey(zombie_streams_, stream->id())) {
     // If this was a zombie stream then we close it out now.
     CloseZombieStream(stream->id());
     // However, since the headers still have not been decompressed, we want to
     // mark it a prematurely closed so that if we ever receive frames
     // for this stream we can close the connection.
     DCHECK(!stream->headers_decompressed());
@@ skipping 12 matching lines @@
   goaway_received_ = true;
 }
 
 void QuicSession::OnConnectionClosed(QuicErrorCode error, bool from_peer) {
   DCHECK(!connection_->connected());
   if (error_ == QUIC_NO_ERROR) {
     error_ = error;
   }
 
   while (!stream_map_.empty()) {
-    ReliableStreamMap::iterator it = stream_map_.begin();
+    DataStreamMap::iterator it = stream_map_.begin();
     QuicStreamId id = it->first;
     it->second->OnConnectionClosed(error, from_peer);
     // The stream should call CloseStream as part of OnConnectionClosed.
     if (stream_map_.find(id) != stream_map_.end()) {
       LOG(DFATAL) << ENDPOINT
                   << "Stream failed to close under OnConnectionClosed";
       CloseStream(id);
     }
   }
 }
@@ skipping 57 matching lines @@
 }
 
 void QuicSession::CloseStream(QuicStreamId stream_id) {
   CloseStreamInner(stream_id, false);
 }
 
 void QuicSession::CloseStreamInner(QuicStreamId stream_id,
                                    bool locally_reset) {
   DVLOG(1) << ENDPOINT << "Closing stream " << stream_id;
 
-  ReliableStreamMap::iterator it = stream_map_.find(stream_id);
+  DataStreamMap::iterator it = stream_map_.find(stream_id);
   if (it == stream_map_.end()) {
     DVLOG(1) << ENDPOINT << "Stream is already closed: " << stream_id;
     return;
   }
-  ReliableQuicStream* stream = it->second;
+  QuicDataStream* stream = it->second;
   if (connection_->connected() && !stream->headers_decompressed()) {
     // If the stream is being closed locally (for example a client cancelling
     // a request before receiving the response) then we need to make sure that
     // we keep the stream alive long enough to process any response or
     // RST_STREAM frames.
     if (locally_reset && !is_server_) {
       AddZombieStream(stream_id);
       return;
     }
 
@@ skipping 22 matching lines @@
     // mark it a prematurely closed so that if we ever receive frames
     // for this stream we can close the connection.
     AddPrematurelyClosedStream(oldest_zombie_stream_id);
   }
   zombie_streams_.insert(make_pair(stream_id, true));
 }
 
 void QuicSession::CloseZombieStream(QuicStreamId stream_id) {
   DCHECK(ContainsKey(zombie_streams_, stream_id));
   zombie_streams_.erase(stream_id);
-  ReliableQuicStream* stream = GetStream(stream_id);
+  QuicDataStream* stream = GetDataStream(stream_id);
   if (!stream) {
     return;
   }
   stream_map_.erase(stream_id);
   stream->OnClose();
   closed_streams_.push_back(stream);
 }
 
 void QuicSession::AddPrematurelyClosedStream(QuicStreamId stream_id) {
   if (prematurely_closed_streams_.size() ==
@@ skipping 45 matching lines @@
 }
 
 void QuicSession::OnCryptoHandshakeMessageReceived(
     const CryptoHandshakeMessage& message) {
 }
 
 QuicConfig* QuicSession::config() {
   return &config_;
 }
 
-void QuicSession::ActivateStream(ReliableQuicStream* stream) {
+void QuicSession::ActivateStream(QuicDataStream* stream) {
   DVLOG(1) << ENDPOINT << "num_streams: " << stream_map_.size()
              << ". activating " << stream->id();
   DCHECK_EQ(stream_map_.count(stream->id()), 0u);
   stream_map_[stream->id()] = stream;
 }
 
 QuicStreamId QuicSession::GetNextStreamId() {
   QuicStreamId id = next_stream_id_;
   next_stream_id_ += 2;
   return id;
 }
 
 ReliableQuicStream* QuicSession::GetStream(const QuicStreamId stream_id) {
   if (stream_id == kCryptoStreamId) {
     return GetCryptoStream();
   }
+  return GetDataStream(stream_id);
+}
 
-  ReliableStreamMap::iterator it = stream_map_.find(stream_id);
+QuicDataStream* QuicSession::GetDataStream(const QuicStreamId stream_id) {
+  if (stream_id == kCryptoStreamId) {
+    DLOG(FATAL) << "Attempt to call GetDataStream with the crypto stream id";
+    return NULL;
+  }
+
+  DataStreamMap::iterator it = stream_map_.find(stream_id);
   if (it != stream_map_.end()) {
     return it->second;
   }
 
   if (IsClosedStream(stream_id)) {
     return NULL;
   }
 
   if (stream_id % 2 == next_stream_id_ % 2) {
     // We've received a frame for a locally-created stream that is not
     // currently active.  This is an error.
     connection()->SendConnectionClose(QUIC_PACKET_FOR_NONEXISTENT_STREAM);
     return NULL;
   }
 
   return GetIncomingReliableStream(stream_id);
 }
 
-ReliableQuicStream* QuicSession::GetIncomingReliableStream(
+QuicDataStream* QuicSession::GetIncomingReliableStream(
     QuicStreamId stream_id) {
   if (IsClosedStream(stream_id)) {
     return NULL;
   }
 
   if (goaway_sent_) {
     // We've already sent a GoAway
     SendRstStream(stream_id, QUIC_STREAM_PEER_GOING_AWAY);
     return NULL;
   }
 
   implicitly_created_streams_.erase(stream_id);
   if (stream_id > largest_peer_created_stream_id_) {
     // TODO(rch) add unit test for this
     if (stream_id - largest_peer_created_stream_id_ > kMaxStreamIdDelta) {
       connection()->SendConnectionClose(QUIC_INVALID_STREAM_ID);
       return NULL;
     }
     if (largest_peer_created_stream_id_ == 0) {
       largest_peer_created_stream_id_= 1;
     }
     for (QuicStreamId id = largest_peer_created_stream_id_ + 2;
          id < stream_id;
          id += 2) {
       implicitly_created_streams_.insert(id);
     }
     largest_peer_created_stream_id_ = stream_id;
   }
-  ReliableQuicStream* stream = CreateIncomingReliableStream(stream_id);
+  QuicDataStream* stream = CreateIncomingDataStream(stream_id);
   if (stream == NULL) {
     return NULL;
   }
   ActivateStream(stream);
   return stream;
 }
 
 bool QuicSession::IsClosedStream(QuicStreamId id) {
   DCHECK_NE(0u, id);
   if (id == kCryptoStreamId) {
@@ skipping 48 matching lines @@
   NOTIMPLEMENTED();
   return false;
 }
 
 void QuicSession::PostProcessAfterData() {
   STLDeleteElements(&closed_streams_);
   closed_streams_.clear();
 }
 
 }  // namespace net