Refactor QuicConnection to use explicit notification for getting onto

net/quic/quic_connection_test.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_connection.h"
 
 #include "base/basictypes.h"
 #include "base/bind.h"
 #include "base/stl_util.h"
 #include "net/base/net_errors.h"
@@ skipping 13 matching lines @@
 #include "net/quic/test_tools/quic_sent_packet_manager_peer.h"
 #include "net/quic/test_tools/quic_test_utils.h"
 #include "testing/gmock/include/gmock/gmock.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
 using base::StringPiece;
 using std::map;
 using std::vector;
 using testing::_;
 using testing::AnyNumber;
+using testing::AtLeast;
 using testing::ContainerEq;
 using testing::Contains;
 using testing::DoAll;
 using testing::InSequence;
 using testing::InvokeWithoutArgs;
 using testing::Ref;
 using testing::Return;
 using testing::SaveArg;
 using testing::StrictMock;
 
@@ skipping 223 matching lines @@
   MockRandom* random_generator_;
 
   DISALLOW_COPY_AND_ASSIGN(TestConnectionHelper);
 };
 
 class TestPacketWriter : public QuicPacketWriter {
  public:
   TestPacketWriter()
       : last_packet_size_(0),
         write_blocked_(false),
-        block_next_write_(false),
+        block_on_next_write_(false),
         is_write_blocked_data_buffered_(false),
         is_server_(true),
         final_bytes_of_last_packet_(0),
         final_bytes_of_previous_packet_(0),
         use_tagging_decrypter_(false),
         packets_write_attempts_(0) {
   }
 
   // QuicPacketWriter interface
   virtual WriteResult WritePacket(
@@ skipping 10 matching lines @@
              sizeof(final_bytes_of_last_packet_));
     }
 
     QuicFramer framer(QuicSupportedVersions(), QuicTime::Zero(), !is_server_);
     if (use_tagging_decrypter_) {
       framer.SetDecrypter(new TaggingDecrypter);
     }
     visitor_.Reset();
     framer.set_visitor(&visitor_);
     EXPECT_TRUE(framer.ProcessPacket(packet));
-    if (block_next_write_) {
+    if (block_on_next_write_) {
       write_blocked_ = true;
-      block_next_write_ = false;
+      block_on_next_write_ = false;
     }
     if (IsWriteBlocked()) {
       return WriteResult(WRITE_STATUS_BLOCKED, -1);
     }
     last_packet_size_ = packet.length();
     return WriteResult(WRITE_STATUS_OK, last_packet_size_);
   }
 
   virtual bool IsWriteBlockedDataBuffered() const OVERRIDE {
     return is_write_blocked_data_buffered_;
   }
 
   virtual bool IsWriteBlocked() const OVERRIDE { return write_blocked_; }
 
   virtual void SetWritable() OVERRIDE { write_blocked_ = false; }
 
-  void BlockNextWrite() { block_next_write_ = true; }
+  void BlockOnNextWrite() { block_on_next_write_ = true; }
 
   // Resets the visitor's state by clearing out the headers and frames.
   void Reset() {
     visitor_.Reset();
   }
 
   QuicPacketHeader* header() { return visitor_.header(); }
 
   size_t frame_count() const { return visitor_.frame_count(); }
 
@@ skipping 35 matching lines @@
   void use_tagging_decrypter() {
     use_tagging_decrypter_ = true;
   }
 
   uint32 packets_write_attempts() { return packets_write_attempts_; }
 
  private:
   FramerVisitorCapturingFrames visitor_;
   size_t last_packet_size_;
   bool write_blocked_;
-  bool block_next_write_;
+  bool block_on_next_write_;
   bool is_write_blocked_data_buffered_;
   bool is_server_;
   uint32 final_bytes_of_last_packet_;
   uint32 final_bytes_of_previous_packet_;
   bool use_tagging_decrypter_;
   uint32 packets_write_attempts_;
 
   DISALLOW_COPY_AND_ASSIGN(TestPacketWriter);
 };
 
@@ skipping 71 matching lines @@
         SendStreamDataWithString(kCryptoStreamId, "chlo", 0, !kFin, NULL);
     this->Flush();
     return consumed;
   }
 
   bool is_server() {
     return QuicConnectionPeer::IsServer(this);
   }
 
   void set_version(QuicVersion version) {
-    framer_.set_version(version);
+    QuicConnectionPeer::GetFramer(this)->set_version(version);
   }
 
   void set_is_server(bool is_server) {
     writer_->set_is_server(is_server);
     QuicPacketCreatorPeer::SetIsServer(
         QuicConnectionPeer::GetPacketCreator(this), is_server);
     QuicConnectionPeer::SetIsServer(this, is_server);
   }
 
   TestConnectionHelper::TestAlarm* GetAckAlarm() {
@@ skipping 348 matching lines @@
     // Send an erroneous packet to close the connection.
     EXPECT_CALL(visitor_,
                 OnConnectionClosed(QUIC_INVALID_PACKET_HEADER, false));
     // Call ProcessDataPacket rather than ProcessPacket, as we should not get a
     // packet call to the visitor.
     ProcessDataPacket(6000, 0, !kEntropyFlag);
     EXPECT_FALSE(
         QuicConnectionPeer::GetConnectionClosePacket(&connection_) == NULL);
   }
 
+  void BlockOnNextWrite() {
+    writer_->BlockOnNextWrite();
+    EXPECT_CALL(visitor_, OnWriteBlocked()).Times(AtLeast(1));
+  }
+
   QuicGuid guid_;
   QuicFramer framer_;
   QuicPacketCreator creator_;
   MockEntropyCalculator entropy_calculator_;
 
   MockSendAlgorithm* send_algorithm_;
   TestReceiveAlgorithm* receive_algorithm_;
   MockClock clock_;
   MockRandom random_generator_;
   scoped_ptr<TestConnectionHelper> helper_;
@@ skipping 478 matching lines @@
   // All packets carry version info till version is negotiated.
   size_t payload_length;
   connection_.options()->max_packet_length =
       GetPacketLengthForOneStream(
           connection_.version(), kIncludeVersion, PACKET_1BYTE_SEQUENCE_NUMBER,
           IN_FEC_GROUP, &payload_length);
   // And send FEC every two packets.
   connection_.options()->max_packets_per_fec_group = 2;
 
   EXPECT_EQ(0u, connection_.NumQueuedPackets());
-  writer_->BlockNextWrite();
+  BlockOnNextWrite();
   const string payload(payload_length, 'a');
   connection_.SendStreamDataWithString(1, payload, 0, !kFin, NULL);
   EXPECT_FALSE(creator_.ShouldSendFec(true));
   // Expect the first data packet and the fec packet to be queued.
   EXPECT_EQ(2u, connection_.NumQueuedPackets());
 }
 
 TEST_F(QuicConnectionTest, AbandonFECFromCongestionWindow) {
   connection_.options()->max_packets_per_fec_group = 1;
   // 1 Data and 1 FEC packet.
@@ skipping 249 matching lines @@
   EXPECT_EQ(1u, frame.stream_id);
   EXPECT_EQ("ABCD", string(static_cast<char*>
                            (frame.data.iovec()[0].iov_base),
                            (frame.data.iovec()[0].iov_len)));
 }
 
 TEST_F(QuicConnectionTest, FramePackingSendvQueued) {
   // Try to send two stream frames in 1 packet by using writev.
   EXPECT_CALL(*send_algorithm_, OnPacketSent(_, _, _, NOT_RETRANSMISSION, _));
 
-  writer_->BlockNextWrite();
+  BlockOnNextWrite();
   char data[] = "ABCD";
   IOVector data_iov;
   data_iov.AppendNoCoalesce(data, 2);
   data_iov.AppendNoCoalesce(data + 2, 2);
   connection_.SendStreamData(1, data_iov, 0, !kFin, NULL);
 
   EXPECT_EQ(1u, connection_.NumQueuedPackets());
   EXPECT_TRUE(connection_.HasQueuedData());
 
   // Unblock the writes and actually send.
@@ skipping 98 matching lines @@
   ProcessAckPacket(&ack_one);
   ProcessAckPacket(&ack_one);
   ProcessAckPacket(&ack_one);
 
   // Peer acks up to 3 with two explicitly missing.  Two nacks should cause no
   // change.
   QuicAckFrame nack_two = InitAckFrame(3, 0);
   NackPacket(2, &nack_two);
   // The first nack should trigger a fast retransmission, but we'll be
   // write blocked, so the packet will be queued.
-  writer_->BlockNextWrite();
+  BlockOnNextWrite();
   EXPECT_CALL(*send_algorithm_, UpdateRtt(_));
   EXPECT_CALL(*send_algorithm_, OnPacketAcked(_, _));
-
   ProcessAckPacket(&nack_two);
   EXPECT_EQ(1u, connection_.NumQueuedPackets());
 
   // Now, ack the previous transmission.
   QuicAckFrame ack_all = InitAckFrame(3, 0);
   ProcessAckPacket(&ack_all);
 
   // Unblock the socket and attempt to send the queued packets.  However,
   // since the previous transmission has been acked, we will not
   // send the retransmission.
@@ skipping 26 matching lines @@
   ProcessAckPacket(&frame);
 }
 
 TEST_F(QuicConnectionTest, QueueAfterTwoRTOs) {
   for (int i = 0; i < 10; ++i) {
     EXPECT_CALL(*send_algorithm_, OnPacketSent(_, _, _, _, _)).Times(1);
     connection_.SendStreamDataWithString(3, "foo", i * 3, !kFin, NULL);
   }
 
   // Block the congestion window and ensure they're queued.
-  writer_->BlockNextWrite();
+  BlockOnNextWrite();
   clock_.AdvanceTime(DefaultRetransmissionTime());
   // Only one packet should be retransmitted.
   EXPECT_CALL(*send_algorithm_, OnRetransmissionTimeout(true));
   connection_.GetRetransmissionAlarm()->Fire();
   EXPECT_TRUE(connection_.HasQueuedData());
 
   // Unblock the congestion window.
   writer_->SetWritable();
   clock_.AdvanceTime(QuicTime::Delta::FromMicroseconds(
       2 * DefaultRetransmissionTime().ToMicroseconds()));
   // Retransmit already retransmitted packets event though the sequence number
   // greater than the largest observed.
   EXPECT_CALL(*send_algorithm_, OnPacketSent(_, _, _, _, _)).Times(10);
   connection_.GetRetransmissionAlarm()->Fire();
   connection_.OnCanWrite();
 }
 
 TEST_F(QuicConnectionTest, WriteBlockedThenSent) {
-  writer_->BlockNextWrite();
+  BlockOnNextWrite();
   writer_->set_is_write_blocked_data_buffered(true);
-
   connection_.SendStreamDataWithString(1, "foo", 0, !kFin, NULL);
   EXPECT_FALSE(connection_.GetRetransmissionAlarm()->IsSet());
 
   EXPECT_CALL(*send_algorithm_, OnPacketSent(_, _, _, _, _)).Times(1);
   connection_.OnPacketSent(WriteResult(WRITE_STATUS_OK, 0));
   EXPECT_TRUE(connection_.GetRetransmissionAlarm()->IsSet());
 }
 
 TEST_F(QuicConnectionTest, WriteBlockedAckedThenSent) {
   EXPECT_CALL(visitor_, OnSuccessfulVersionNegotiation(_));
-  writer_->BlockNextWrite();
-
+  BlockOnNextWrite();
   writer_->set_is_write_blocked_data_buffered(true);
   connection_.SendStreamDataWithString(1, "foo", 0, !kFin, NULL);
   EXPECT_FALSE(connection_.GetRetransmissionAlarm()->IsSet());
 
   // Ack the sent packet before the callback returns, which happens in
   // rare circumstances with write blocked sockets.
   QuicAckFrame ack = InitAckFrame(1, 0);
   ProcessAckPacket(&ack);
 
   EXPECT_CALL(*send_algorithm_, OnPacketSent(_, _, _, _, _)).Times(0);
   connection_.OnPacketSent(WriteResult(WRITE_STATUS_OK, 0));
   EXPECT_FALSE(connection_.GetRetransmissionAlarm()->IsSet());
 }
 
 TEST_F(QuicConnectionTest, RetransmitWriteBlockedAckedOriginalThenSent) {
   EXPECT_CALL(visitor_, OnSuccessfulVersionNegotiation(_));
   connection_.SendStreamDataWithString(3, "foo", 0, !kFin, NULL);
   EXPECT_TRUE(connection_.GetRetransmissionAlarm()->IsSet());
 
-  writer_->BlockNextWrite();
+  BlockOnNextWrite();
   writer_->set_is_write_blocked_data_buffered(true);
-
   // Simulate the retransmission alarm firing.
   EXPECT_CALL(*send_algorithm_, OnRetransmissionTimeout(_));
   clock_.AdvanceTime(DefaultRetransmissionTime());
   connection_.GetRetransmissionAlarm()->Fire();
 
   // Ack the sent packet before the callback returns, which happens in
   // rare circumstances with write blocked sockets.
   EXPECT_CALL(*send_algorithm_, UpdateRtt(_));
   QuicAckFrame ack = InitAckFrame(1, 0);
   ProcessAckPacket(&ack);
 
   connection_.OnPacketSent(WriteResult(WRITE_STATUS_OK, 0));
   EXPECT_TRUE(connection_.GetRetransmissionAlarm()->IsSet());
 }
 
 TEST_F(QuicConnectionTest, ResumptionAlarmWhenWriteBlocked) {
   // Block the connection.
-  writer_->BlockNextWrite();
+  BlockOnNextWrite();
   connection_.SendStreamDataWithString(3, "foo", 0, !kFin, NULL);
   EXPECT_EQ(1u, writer_->packets_write_attempts());
   EXPECT_TRUE(writer_->IsWriteBlocked());
 
   // Set the send and resumption alarms. Fire the alarms and ensure they don't
   // attempt to write.
   connection_.GetResumeWritesAlarm()->Set(clock_.ApproximateNow());
   connection_.GetSendAlarm()->Set(clock_.ApproximateNow());
   connection_.GetResumeWritesAlarm()->Fire();
   connection_.GetSendAlarm()->Fire();
@@ skipping 382 matching lines @@
   ASSERT_NE(0u, nack_sequence_number);
   EXPECT_FALSE(QuicConnectionPeer::IsSavedForRetransmission(
       &connection_, rto_sequence_number));
   ASSERT_TRUE(QuicConnectionPeer::IsSavedForRetransmission(
       &connection_, nack_sequence_number));
   EXPECT_TRUE(QuicConnectionPeer::IsRetransmission(
       &connection_, nack_sequence_number));
 }
 
 TEST_F(QuicConnectionTest, SetRTOAfterWritingToSocket) {
-  writer_->BlockNextWrite();
+  BlockOnNextWrite();
   connection_.SendStreamDataWithString(1, "foo", 0, !kFin, NULL);
   // Make sure that RTO is not started when the packet is queued.
   EXPECT_FALSE(connection_.GetRetransmissionAlarm()->IsSet());
 
   // Test that RTO is started once we write to the socket.
   writer_->SetWritable();
   connection_.OnCanWrite();
   EXPECT_TRUE(connection_.GetRetransmissionAlarm()->IsSet());
 }
 
@@ skipping 30 matching lines @@
   // than previously.
   EXPECT_TRUE(retransmission_alarm->IsSet());
   QuicTime next_rto_time = retransmission_alarm->deadline();
   QuicTime expected_rto_time =
       connection_.sent_packet_manager().GetRetransmissionTime();
   EXPECT_EQ(next_rto_time, expected_rto_time);
 }
 
 TEST_F(QuicConnectionTest, TestQueued) {
   EXPECT_EQ(0u, connection_.NumQueuedPackets());
-  writer_->BlockNextWrite();
+  BlockOnNextWrite();
   connection_.SendStreamDataWithString(1, "foo", 0, !kFin, NULL);
   EXPECT_EQ(1u, connection_.NumQueuedPackets());
 
   // Unblock the writes and actually send.
   writer_->SetWritable();
   EXPECT_TRUE(connection_.OnCanWrite());
   EXPECT_EQ(0u, connection_.NumQueuedPackets());
 }
 
 TEST_F(QuicConnectionTest, CloseFecGroup) {
@@ skipping 134 matching lines @@
               TimeUntilSend(_, NACK_RETRANSMISSION, _, _)).Times(0);
   EXPECT_CALL(*send_algorithm_, OnPacketSent(_, _, _, _, _));
   connection_.SendPacket(
       ENCRYPTION_NONE, 1, packet, kTestEntropyHash, HAS_RETRANSMITTABLE_DATA);
   // XXX: fixme.  was:  connection_.SendPacket(1, packet, kForce);
   EXPECT_EQ(0u, connection_.NumQueuedPackets());
 }
 
 TEST_F(QuicConnectionTest, SendSchedulerEAGAIN) {
   QuicPacket* packet = ConstructDataPacket(1, 0, !kEntropyFlag);
-  writer_->BlockNextWrite();
+  BlockOnNextWrite();
   EXPECT_CALL(*send_algorithm_,
               TimeUntilSend(_, NOT_RETRANSMISSION, _, _)).WillOnce(
                   testing::Return(QuicTime::Delta::Zero()));
   EXPECT_CALL(*send_algorithm_, OnPacketSent(_, 1, _, _, _)).Times(0);
   connection_.SendPacket(
       ENCRYPTION_NONE, 1, packet, kTestEntropyHash, HAS_RETRANSMITTABLE_DATA);
   EXPECT_EQ(1u, connection_.NumQueuedPackets());
 }
 
 TEST_F(QuicConnectionTest, SendSchedulerDelayThenSend) {
@@ skipping 454 matching lines @@
   QuicFrames frames;
   QuicFrame frame(&frame1_);
   frames.push_back(frame);
   scoped_ptr<QuicPacket> packet(
       framer_.BuildUnsizedDataPacket(header, frames).packet);
   scoped_ptr<QuicEncryptedPacket> encrypted(
       framer_.EncryptPacket(ENCRYPTION_NONE, 12, *packet));
 
   framer_.set_version(QuicVersionMax());
   connection_.set_is_server(true);
-  writer_->BlockNextWrite();
+  BlockOnNextWrite();
   connection_.ProcessUdpPacket(IPEndPoint(), IPEndPoint(), *encrypted);
   EXPECT_EQ(0u, writer_->last_packet_size());
   EXPECT_TRUE(connection_.HasQueuedData());
 
   writer_->SetWritable();
   connection_.OnCanWrite();
   EXPECT_TRUE(writer_->version_negotiation_packet() != NULL);
 
   size_t num_versions = arraysize(kSupportedQuicVersions);
   EXPECT_EQ(num_versions,
@@ skipping 23 matching lines @@
   QuicFrames frames;
   QuicFrame frame(&frame1_);
   frames.push_back(frame);
   scoped_ptr<QuicPacket> packet(
       framer_.BuildUnsizedDataPacket(header, frames).packet);
   scoped_ptr<QuicEncryptedPacket> encrypted(
       framer_.EncryptPacket(ENCRYPTION_NONE, 12, *packet));
 
   framer_.set_version(QuicVersionMax());
   connection_.set_is_server(true);
-  writer_->BlockNextWrite();
+  BlockOnNextWrite();
   writer_->set_is_write_blocked_data_buffered(true);
   connection_.ProcessUdpPacket(IPEndPoint(), IPEndPoint(), *encrypted);
   EXPECT_EQ(0u, writer_->last_packet_size());
   EXPECT_FALSE(connection_.HasQueuedData());
 }
 
 TEST_F(QuicConnectionTest, ClientHandlesVersionNegotiation) {
   // Start out with some unsupported version.
   QuicConnectionPeer::GetFramer(&connection_)->set_version_for_tests(
       QUIC_VERSION_UNSUPPORTED);
@@ skipping 257 matching lines @@
   // Send a packet.
   connection_.SendStreamDataWithString(1, "foo", 0, !kFin, NULL);
   EXPECT_EQ(0u, connection_.NumQueuedPackets());
   EXPECT_EQ(1u, writer_->packets_write_attempts());
 
   TriggerConnectionClose();
   EXPECT_EQ(2u, writer_->packets_write_attempts());
 }
 
 TEST_F(QuicConnectionTest, ConnectionCloseGettingWriteBlocked) {
-  writer_->BlockNextWrite();
+  BlockOnNextWrite();
   TriggerConnectionClose();
   EXPECT_EQ(1u, writer_->packets_write_attempts());
   EXPECT_TRUE(writer_->IsWriteBlocked());
 }
 
 TEST_F(QuicConnectionTest, ConnectionCloseWhenWriteBlocked) {
-  writer_->BlockNextWrite();
+  BlockOnNextWrite();
   connection_.SendStreamDataWithString(1, "foo", 0, !kFin, NULL);
   EXPECT_EQ(1u, connection_.NumQueuedPackets());
   EXPECT_EQ(1u, writer_->packets_write_attempts());
   EXPECT_TRUE(writer_->IsWriteBlocked());
   TriggerConnectionClose();
   EXPECT_EQ(1u, writer_->packets_write_attempts());
 }
 
 TEST_F(QuicConnectionTest, AckNotifierTriggerCallback) {
   EXPECT_CALL(visitor_, OnSuccessfulVersionNegotiation(_));
@@ skipping 184 matching lines @@
                         true);
   TestConnection client(guid_, IPEndPoint(), helper_.get(), writer_.get(),
                         false);
   EXPECT_TRUE(client.sent_packet_manager().using_pacing());
   EXPECT_FALSE(server.sent_packet_manager().using_pacing());
 }
 
 }  // namespace
 }  // namespace test
 }  // namespace net