Add QuicStream and friends to QUIC code.

net/quic/quic_session.cc

Index: net/quic/quic_session.cc
diff --git a/net/quic/quic_session.cc b/net/quic/quic_session.cc
new file mode 100644
index 0000000000000000000000000000000000000000..5860771a3b034286547f383806633718faf2fd0c
--- /dev/null
+++ b/net/quic/quic_session.cc
@@ -0,0 +1,195 @@
+// 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 "net/quic/quic_connection.h"
+
+using base::StringPiece;
+using base::hash_map;
+using base::hash_set;
+using std::vector;
+
+namespace net {
+
+QuicSession::QuicSession(QuicConnection* connection, bool is_server)
+    : connection_(connection),
+      max_open_streams_(kDefaultMaxStreamsPerConnection),
+      next_stream_id_(is_server ? 2 : 3),
+      is_server_(is_server),
+      largest_peer_created_stream_id_(0) {
+  connection_->set_visitor(this);
+}
+
+QuicSession::~QuicSession() {
+}
+
+bool QuicSession::OnPacket(const IPEndPoint& self_address,
+                           const IPEndPoint& peer_address,
+                           const QuicPacketHeader& header,
+                           const vector<QuicStreamFrame>& frames) {
+  if (header.guid != connection()->guid()) {
+    DLOG(INFO) << "Got packet header for invalid GUID: " << header.guid;
+    return false;
+  }
+  for (size_t i = 0; i < frames.size(); ++i) {
+    // TODO(rch) deal with the error case of stream id 0
+    if (IsClosedStream(frames[i].stream_id)) continue;
+
+    ReliableQuicStream* stream = GetStream(frames[i].stream_id);
+    if (stream == NULL) return false;
+    if (!stream->WillAcceptStreamFrame(frames[i])) return false;
+
+    // TODO(alyssar) check against existing connection address: if changed, make
+    // sure we update the connection.
+  }
+
+  for (size_t i = 0; i < frames.size(); ++i) {
+    ReliableQuicStream* stream = GetStream(frames[i].stream_id);
+    if (stream) {
+      stream->OnStreamFrame(frames[i]);
+    }
+  }
+  return true;
+}
+
+void QuicSession::OnRstStream(const QuicRstStreamFrame& frame) {
+  ReliableQuicStream* stream = GetStream(frame.stream_id);
+  if (!stream) {
+    return;  // Errors are handled by GetStream.
+  }
+  stream->OnStreamReset(frame.error_code, frame.offset);
+}
+
+void QuicSession::ConnectionClose(QuicErrorCode error, bool from_peer) {
+  while (stream_map_.size() != 0) {
+    ReliableStreamMap::iterator it = stream_map_.begin();
+    QuicStreamId id = it->first;
+    it->second->ConnectionClose(error, from_peer);
+    // The stream should call CloseStream as part of ConnectionClose.
+    if (stream_map_.find(id) != stream_map_.end()) {
+      LOG(DFATAL) << "Stream failed to close under ConnectionClose";
+      CloseStream(id);
+    }
+  }
+}
+
+ssize_t QuicSession::WriteData(QuicStreamId id, StringPiece data,
+                               QuicStreamOffset offset, bool fin) {
+  return connection_->SendStreamData(id, data, offset, fin, NULL);
+}
+
+void QuicSession::SendRstStream(QuicStreamId id,
+                                QuicErrorCode error,
+                                QuicStreamOffset offset) {
+  connection_->SendRstStream(id, error, offset);
+  CloseStream(id);
+}
+
+void QuicSession::CloseStream(QuicStreamId stream_id) {
+  DLOG(INFO) << "Closing stream " << stream_id;
+
+  ReliableStreamMap::iterator it = stream_map_.find(stream_id);
+  if (it == stream_map_.end()) {
+    DLOG(INFO) << "Stream is already closed: " << stream_id;
+    return;
+  }
+  stream_map_.erase(it);
+}
+
+bool QuicSession::IsHandshakeComplete() {
+  return GetCryptoStream()->handshake_complete();
+}
+
+void QuicSession::ActivateStream(ReliableQuicStream* stream) {
+  LOG(INFO) << "num_streams: " << stream_map_.size()
+            << ". activating " << stream->id();
+  DCHECK(stream_map_.count(stream->id()) == 0);
+  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();
+  }
+
+  ReliableStreamMap::iterator it = stream_map_.find(stream_id);
+  if (it != stream_map_.end()) {
+    return it->second;
+  }
+
+  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(
+    QuicStreamId stream_id) {
+  if (IsClosedStream(stream_id)) {
+    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) {
+      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);
+  if (stream == NULL) {
+    connection()->SendConnectionClose(QUIC_TOO_MANY_OPEN_STREAMS);
+    return NULL;
+  }
+  ActivateStream(stream);
+  return stream;
+}
+
+bool QuicSession::IsClosedStream(QuicStreamId id) {
+  DCHECK_NE(0u, id);
+  if (id == kCryptoStreamId) {
+    return false;
+  }
+  if (stream_map_.count(id) != 0) {
+    // Stream is active
+    return false;
+  }
+  if (id % 2 == next_stream_id_ % 2) {
+    // If the stream was locally initiated we strictly in-order creation.

[jar (doing other things), 2012/10/31 22:37:37]

nit: typo?  "...we strictly in-order creation"

[Ryan Hamilton, 2012/11/01 22:52:20]

Fixed upstream.

+    // If the id is in the range of created streams and it's not active, it
+    // must have been closed.
+    return id < next_stream_id_;
+  } else {

[jar (doing other things), 2012/10/31 22:37:37]

nit: with return on line 182, else and curlies are not needed.

[Ryan Hamilton, 2012/11/01 22:52:20]

Fixed upstream.

+    // For peer created streams, we also need to consider
+    // implicitly created streams.
+    return id <= largest_peer_created_stream_id_ &&
+        implicitly_created_streams_.count(id) == 0;

[jar (doing other things), 2012/10/31 22:37:37]

I don't think there are every any entries in implicitly_created_streams_ that
are larger than the largest_peer_created_stream_id.

Why are we checking in line 187?

[Ryan Hamilton, 2012/11/01 22:52:20]

We need IsClosed() to return true *only* when a stream has been closed.  So if
IsClosed() is called with a stream id *larger* than the largest peer created
stream id, we need IsClosed() to return false. 

The check on 186 will cause us to return false if the stream id is larger than
the largest peer created stream id.

The check on 187 will cause us to return false if the stream id has been
implicitly created.

Only if it has actually been closed, will these two checks allows us to return
true.  (FWIW, there is some pretty rigorous testing (I hope :>) in
quic_session_test)

Added a more explicit comment in the .h file. (upstream)

+  }
+}
+
+size_t QuicSession::GetNumOpenStreams() {
+  return stream_map_.size() + implicitly_created_streams_.size();
+}
+
+}  // namespace net