Media Remoting: Data/Control plumbing between renderer and Media Router.

chrome/browser/media/cast_remoting_sender_unittest.cc

Index: chrome/browser/media/cast_remoting_sender_unittest.cc
diff --git a/chrome/browser/media/cast_remoting_sender_unittest.cc b/chrome/browser/media/cast_remoting_sender_unittest.cc
index dd86bd6a16f5043ebdff3a62671ac6a622ec1316..fe333b8446c74a1e79c7a41374308f958eb93e7b 100644
--- a/chrome/browser/media/cast_remoting_sender_unittest.cc
+++ b/chrome/browser/media/cast_remoting_sender_unittest.cc
@@ -5,91 +5,99 @@
 #include "chrome/browser/media/cast_remoting_sender.h"
 
 #include "base/bind.h"
+#include "base/bind_helpers.h"
+#include "base/callback_helpers.h"
+#include "base/compiler_specific.h"
 #include "base/macros.h"
 #include "base/memory/ptr_util.h"
-#include "base/test/simple_test_tick_clock.h"
-#include "media/base/fake_single_thread_task_runner.h"
+#include "base/run_loop.h"
+#include "base/time/default_tick_clock.h"
+#include "content/public/test/test_browser_thread_bundle.h"
 #include "media/cast/cast_environment.h"
 #include "media/cast/constants.h"
-#include "media/cast/logging/simple_event_subscriber.h"
-#include "media/cast/net/cast_transport_config.h"
-#include "media/cast/net/cast_transport_impl.h"
-#include "media/cast/net/pacing/paced_sender.h"
+#include "media/cast/net/cast_transport.h"
 #include "media/cast/test/utility/default_config.h"
-#include "testing/gmock/include/gmock/gmock.h"
+#include "media/mojo/interfaces/remoting.mojom.h"
+#include "mojo/public/cpp/system/data_pipe.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
 namespace cast {
 
 namespace {
 
-// Number of bytes in each faked frame.
-constexpr int kFrameSize = 5000;
+// Data pipe capacity is 1KB.
+constexpr int kDataPipeCapacity = 1024;
 
-class TestPacketSender : public media::cast::PacketTransport {
+// Implements the CastTransport interface to capture output from the
+// CastRemotingSender.
+class FakeTransport : public media::cast::CastTransport {
  public:
-  TestPacketSender()
-      : number_of_rtp_packets_(0), number_of_rtcp_packets_(0), paused_(false) {}
-
-  // A singular packet implies a RTCP packet.
-  bool SendPacket(media::cast::PacketRef packet,
-                  const base::Closure& cb) final {
-    if (paused_) {
-      stored_packet_ = packet;
-      callback_ = cb;
-      return false;
-    }
-    if (media::cast::IsRtcpPacket(&packet->data[0], packet->data.size())) {
-      ++number_of_rtcp_packets_;
-    } else {
-      ++number_of_rtp_packets_;
-    }
-    return true;
-  }
+  FakeTransport() {}
+  ~FakeTransport() final {}
 
-  int64_t GetBytesSent() final { return 0; }
+  void TakeSentFrames(std::vector<media::cast::EncodedFrame>* frames) {
+    frames->swap(sent_frames_);
+    sent_frames_.clear();
+  }
 
-  void StartReceiving(const media::cast::PacketReceiverCallbackWithStatus&
-                          packet_receiver) final {}
+  void TakeCanceledFrameIds(std::vector<media::cast::FrameId>* frame_ids) {
+    frame_ids->swap(canceled_frame_ids_);
+    canceled_frame_ids_.clear();
+  }
 
-  void StopReceiving() final {}
+  media::cast::FrameId WaitForKickstart() {
+    base::RunLoop run_loop;
+    kickstarted_callback_ = run_loop.QuitClosure();
+    run_loop.Run();
+    return kickstarted_frame_id_;
+  }
 
-  int number_of_rtp_packets() const { return number_of_rtp_packets_; }
+ protected:
+  void InsertFrame(uint32_t ssrc,
+                   const media::cast::EncodedFrame& frame) final {
+    sent_frames_.push_back(frame);
+  }
 
-  int number_of_rtcp_packets() const { return number_of_rtcp_packets_; }
+  void CancelSendingFrames(
+      uint32_t ssrc, const std::vector<media::cast::FrameId>& frame_ids) final {
+    for (media::cast::FrameId frame_id : frame_ids)
+      canceled_frame_ids_.push_back(frame_id);
+  }
 
-  void SetPause(bool paused) {
-    paused_ = paused;
-    if (!paused && stored_packet_.get()) {
-      SendPacket(stored_packet_, callback_);
-      callback_.Run();
-    }
+  void ResendFrameForKickstart(uint32_t ssrc,
+                               media::cast::FrameId frame_id) final {
+    kickstarted_frame_id_ = frame_id;
+    if (!kickstarted_callback_.is_null())
+      base::ResetAndReturn(&kickstarted_callback_).Run();
   }
 
+  // The rest of the interface is not used for these tests.
+  void SendSenderReport(
+      uint32_t ssrc, base::TimeTicks current_time,
+      media::cast::RtpTimeTicks current_time_as_rtp_timestamp) final {}
+  void AddValidRtpReceiver(uint32_t rtp_sender_ssrc,
+                           uint32_t rtp_receiver_ssrc) final {}
+  void InitializeRtpReceiverRtcpBuilder(
+      uint32_t rtp_receiver_ssrc,
+      const media::cast::RtcpTimeData& time_data) final {}
+  void AddCastFeedback(const media::cast::RtcpCastMessage& cast_message,
+                       base::TimeDelta target_delay) final {}
+  void AddPli(const media::cast::RtcpPliMessage& pli_message) final {}
+  void AddRtcpEvents(
+      const media::cast::ReceiverRtcpEventSubscriber::RtcpEvents& e) final {}
+  void AddRtpReceiverReport(const media::cast::RtcpReportBlock& b) final {}
+  void SendRtcpFromRtpReceiver() final {}
+  void SetOptions(const base::DictionaryValue& options) final {}
+
  private:
-  int number_of_rtp_packets_;
-  int number_of_rtcp_packets_;
-  bool paused_;
-  base::Closure callback_;
-  media::cast::PacketRef stored_packet_;
+  const scoped_refptr<media::cast::CastEnvironment> cast_environment_;
+  std::vector<media::cast::EncodedFrame> sent_frames_;
+  std::vector<media::cast::FrameId> canceled_frame_ids_;
 
-  DISALLOW_COPY_AND_ASSIGN(TestPacketSender);
-};
+  base::Closure kickstarted_callback_;
+  media::cast::FrameId kickstarted_frame_id_;
 
-class DummyClient : public media::cast::CastTransport::Client {
- public:
-  DummyClient() {}
-
-  void OnStatusChanged(media::cast::CastTransportStatus status) final {
-    EXPECT_EQ(media::cast::TRANSPORT_STREAM_INITIALIZED, status);
-  };
-  void OnLoggingEventsReceived(
-      std::unique_ptr<std::vector<media::cast::FrameEvent>> frame_events,
-      std::unique_ptr<std::vector<media::cast::PacketEvent>> packet_events)
-      final {}
-  void ProcessRtpPacket(std::unique_ptr<media::cast::Packet> packet) final {}
-
-  DISALLOW_COPY_AND_ASSIGN(DummyClient);
+  DISALLOW_COPY_AND_ASSIGN(FakeTransport);
 };
 
 }  // namespace
@@ -97,107 +105,493 @@ class DummyClient : public media::cast::CastTransport::Client {
 class CastRemotingSenderTest : public ::testing::Test {
  protected:
   CastRemotingSenderTest()
-      : testing_clock_(new base::SimpleTestTickClock()),
-        task_runner_(new media::FakeSingleThreadTaskRunner(testing_clock_)),
-        cast_environment_(
-            new media::cast::CastEnvironment(base::WrapUnique(testing_clock_),
-                                             task_runner_,
-                                             task_runner_,
-                                             task_runner_)) {
-    testing_clock_->Advance(base::TimeTicks::Now() - base::TimeTicks());
-    transport_ = new TestPacketSender();
-    transport_sender_.reset(new media::cast::CastTransportImpl(
-        testing_clock_, base::TimeDelta(), base::MakeUnique<DummyClient>(),
-        base::WrapUnique(transport_), task_runner_));
+      : thread_bundle_(content::TestBrowserThreadBundle::IO_MAINLOOP),
+        cast_environment_(new media::cast::CastEnvironment(
+            base::WrapUnique(new base::DefaultTickClock()),
+            base::ThreadTaskRunnerHandle::Get(), nullptr, nullptr)),
+        expecting_error_callback_run_(false) {
+
     media::cast::FrameSenderConfig video_config =
         media::cast::GetDefaultVideoSenderConfig();
     video_config.rtp_payload_type = media::cast::RtpPayloadType::REMOTE_VIDEO;
-    media::cast::CastTransportRtpConfig transport_config;
-    transport_config.ssrc = video_config.sender_ssrc;
-    transport_config.rtp_stream_id = 5;
-    transport_config.feedback_ssrc = video_config.receiver_ssrc;
-    transport_config.rtp_payload_type = video_config.rtp_payload_type;
-    transport_config.aes_key = video_config.aes_key;
-    transport_config.aes_iv_mask = video_config.aes_iv_mask;
+    transport_config_.ssrc = video_config.sender_ssrc;
+    transport_config_.rtp_stream_id = 5;
+    transport_config_.feedback_ssrc = video_config.receiver_ssrc;
+    transport_config_.rtp_payload_type = video_config.rtp_payload_type;
+    transport_config_.aes_key = video_config.aes_key;
+    transport_config_.aes_iv_mask = video_config.aes_iv_mask;
     remoting_sender_.reset(new CastRemotingSender(
-        cast_environment_, transport_sender_.get(), transport_config));
-    task_runner_->RunTasks();
+        cast_environment_, &transport_, transport_config_));
+    // Give CastRemotingSender a small RTT measurement to prevent kickstart
+    // testing from taking too long.
+    remoting_sender_->OnReceivedRtt(base::TimeDelta::FromMilliseconds(1));
+
+    const MojoCreateDataPipeOptions data_pipe_options{
+        sizeof(MojoCreateDataPipeOptions),
+        MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE, 1, kDataPipeCapacity };
+    mojo::ScopedDataPipeConsumerHandle consumer_end;
+    CHECK_EQ(MOJO_RESULT_OK,
+             mojo::CreateDataPipe(&data_pipe_options, &producer_end_,
+                                  &consumer_end));
+
+    CastRemotingSender::FindAndBind(
+        transport_config_.rtp_stream_id, std::move(consumer_end),
+        GetProxy(&sender_),
+        base::Bind(&CastRemotingSenderTest::OnError, base::Unretained(this)));
+
+    RunPendingTasks();
   }
 
   ~CastRemotingSenderTest() override {}
 
   void TearDown() final {
     remoting_sender_.reset();
-    task_runner_->RunTasks();
+    // Allow any pending tasks to run before destruction.
+    RunPendingTasks();
   }
 
-  void InsertFrame() {
-    remoting_sender_->next_frame_data_.resize(kFrameSize);
-    remoting_sender_->SendFrame();
+  // Allow pending tasks, such as Mojo method calls, to execute.
+  static void RunPendingTasks() {
+    base::RunLoop().RunUntilIdle();
   }
 
-  void OnReceivedCastMessage(
-      const media::cast::RtcpCastMessage& cast_feedback) {
-    remoting_sender_->OnReceivedCastMessage(cast_feedback);
+ protected:
+  media::cast::FrameId latest_acked_frame_id() const {
+    return remoting_sender_->latest_acked_frame_id_;
   }
 
   int NumberOfFramesInFlight() {
     return remoting_sender_->NumberOfFramesInFlight();
   }
 
-  void CancelFramesInFlight() { remoting_sender_->CancelFramesInFlight(); }
+  size_t GetSizeOfNextFrameData() {
+    return remoting_sender_->next_frame_data_.size();
+  }
+
+  bool IsFlowControlInStartingState() const {
+    return remoting_sender_->flow_control_state_ ==
+        CastRemotingSender::STARTING;
+  }
+  bool IsFlowControlInConsumingState() const {
+    return remoting_sender_->flow_control_state_ ==
+        CastRemotingSender::CONSUMING;
+  }
+  bool IsFlowControlInPausedState() const {
+    return remoting_sender_->flow_control_state_ ==
+        CastRemotingSender::CONSUME_PAUSED;
+  }
+
+  bool ProduceDataChunk(size_t offset, size_t size) WARN_UNUSED_RESULT {
+    std::vector<uint8_t> fake_chunk(size);
+    for (size_t i = 0; i < size; ++i)
+      fake_chunk[i] = static_cast<uint8_t>(offset + i);
+    uint32_t num_bytes = fake_chunk.size();
+    return mojo::WriteDataRaw(
+        producer_end_.get(), fake_chunk.data(), &num_bytes,
+        MOJO_WRITE_DATA_FLAG_ALL_OR_NONE) == MOJO_RESULT_OK;
+  }
+
+  void PostMojoCallTask_ConsumeDataChunk(size_t offset, size_t size,
+                                         size_t total_payload_size) {
+    sender_->ConsumeDataChunk(offset, size, total_payload_size);

[dcheng, 2016/09/20 08:20:13]

It might make things slightly easier to just make sender_ protected: and have
things call this directly? I'm not quite sure I understand the significance of
these helpers otherwise.

[miu, 2016/09/21 03:15:50]

Required by the style guide:
https://google.github.io/styleguide/cppguide.html#Inheritance (note in the
"Decision" subsection that it explicitly states data members must be private).

Another reason was in the naming here: I wanted to call-out in the test code
that we are only "posting tasks" that make mojo method calls, which means the
task runner has to be spun up for those tasks to actually execute.

[dcheng, 2016/09/22 02:29:15]

From https://google.github.io/styleguide/cppguide.html#Access_Control
Make data members private, unless they are static const (and follow the naming
convention for constants). For technical reasons, we allow data members of a
test fixture class to be protected when using Google Test).
I think that the async part is just part of the contract that individual test
cases need to know about anyway, but this is fine too.

[miu, 2016/09/22 08:39:14]

Ah, okay. Good to know.
Ok. IMHO the naming is likely help future code maintainers understand this
subtlety as they read the test code, so I'll leave it as-is.

+  }
+
+  void PostMojoCallTask_SendFrame() {
+    sender_->SendFrame();
+  }
+
+  void PostMojoCallTask_CancelInFlightData() { sender_->CancelInFlightData(); }
+
+  void TakeSentFrames(std::vector<media::cast::EncodedFrame>* frames) {
+    transport_.TakeSentFrames(frames);
+  }
+
+  bool ExpectOneFrameWasSent(size_t expected_payload_size) {
+    std::vector<media::cast::EncodedFrame> frames;
+    transport_.TakeSentFrames(&frames);
+    EXPECT_EQ(1u, frames.size());
+    if (frames.empty())
+      return false;
+    return ExpectCorrectFrameData(expected_payload_size, frames.front());
+  }
+
+  void AckUpToAndIncluding(media::cast::FrameId frame_id) {
+    media::cast::RtcpCastMessage cast_feedback(transport_config_.feedback_ssrc);
+    cast_feedback.ack_frame_id = frame_id;
+    remoting_sender_->OnReceivedCastMessage(cast_feedback);
+  }
+
+  void AckOldestInFlightFrames(int count) {
+    AckUpToAndIncluding(latest_acked_frame_id() + count);
+  }
+
+  // Blocks the caller indefinitely, until a kickstart frame is sent, and then
+  // returns the FrameId of the kickstarted-frame.
+  media::cast::FrameId WaitForKickstart() {
+    return transport_.WaitForKickstart();
+  }
+
+  bool ExpectNoFramesCanceled() {
+    std::vector<media::cast::FrameId> frame_ids;
+    transport_.TakeCanceledFrameIds(&frame_ids);
+    return frame_ids.empty();
+  }
+
+  bool ExpectFramesCanceled(media::cast::FrameId first_frame_id,
+                            media::cast::FrameId last_frame_id) {
+    std::vector<media::cast::FrameId> frame_ids;
+    transport_.TakeCanceledFrameIds(&frame_ids);
+    auto begin = frame_ids.begin();
+    auto end = frame_ids.end();
+    for (auto fid = first_frame_id; fid <= last_frame_id; ++fid) {
+      auto new_end = std::remove(begin, end, fid);
+      if (new_end == end)
+        return false;
+      end = new_end;
+    }
+    return begin == end;
+  }
 
-  base::SimpleTestTickClock* const testing_clock_;  // Owned by CastEnvironment.
-  const scoped_refptr<media::FakeSingleThreadTaskRunner> task_runner_;
+  static bool ExpectCorrectFrameData(size_t expected_payload_size,
+                                     const media::cast::EncodedFrame& frame) {
+    if (expected_payload_size != frame.data.size()) {
+      ADD_FAILURE() << "Expected frame data size != frame.data.size(): "
+                    << expected_payload_size << " vs " << frame.data.size();
+      return false;
+    }
+    for (size_t i = 0; i < expected_payload_size; ++i) {
+      if (static_cast<uint8_t>(frame.data[i]) != static_cast<uint8_t>(i)) {
+        ADD_FAILURE() << "Frame data byte mismatch at offset " << i;
+        return false;
+      }
+    }
+    return true;
+  }
+
+ private:
+  void OnError() {
+    CHECK(expecting_error_callback_run_);
+  }
+
+  content::TestBrowserThreadBundle thread_bundle_;
   const scoped_refptr<media::cast::CastEnvironment> cast_environment_;
-  TestPacketSender* transport_;  // Owned by CastTransport.
-  std::unique_ptr<media::cast::CastTransportImpl> transport_sender_;
+  media::cast::CastTransportRtpConfig transport_config_;
+  FakeTransport transport_;
   std::unique_ptr<CastRemotingSender> remoting_sender_;
+  media::mojom::RemotingDataStreamSenderPtr sender_;
+  mojo::ScopedDataPipeProducerHandle producer_end_;
+  bool expecting_error_callback_run_;
 
  private:
   DISALLOW_COPY_AND_ASSIGN(CastRemotingSenderTest);
 };
 
-TEST_F(CastRemotingSenderTest, SendAndAckEncodedFrame) {
-  // Send a fake video frame that will be decomposed into 4 packets.
-  InsertFrame();
-  task_runner_->RunTasks();
-  EXPECT_EQ(4, transport_->number_of_rtp_packets());
-  EXPECT_EQ(1, NumberOfFramesInFlight());
-
-  // Ack the frame.
-  media::cast::RtcpCastMessage cast_feedback(11);
-  cast_feedback.ack_frame_id = media::cast::FrameId::first();
-  OnReceivedCastMessage(cast_feedback);
-  EXPECT_EQ(0, NumberOfFramesInFlight());
+TEST_F(CastRemotingSenderTest, SendsFramesViaMojoInterface) {
+  // One 256-byte chunk pushed through the data pipe to make one frame.
+  ASSERT_TRUE(ProduceDataChunk(0, 256));
+  PostMojoCallTask_ConsumeDataChunk(0, 256, 256);
+  PostMojoCallTask_SendFrame();
+  RunPendingTasks();
+  EXPECT_TRUE(ExpectOneFrameWasSent(256));
+  AckOldestInFlightFrames(1);
+  EXPECT_EQ(media::cast::FrameId::first(), latest_acked_frame_id());
+
+  // Four 256-byte chunks pushed through the data pipe to make one frame.
+  for (int i = 0; i < 4; ++i) {
+    ASSERT_TRUE(ProduceDataChunk(i * 256, 256));
+    PostMojoCallTask_ConsumeDataChunk(i * 256, 256, 1024);
+  }
+  PostMojoCallTask_SendFrame();
+  RunPendingTasks();
+  EXPECT_TRUE(ExpectOneFrameWasSent(1024));
+  AckOldestInFlightFrames(1);
+  EXPECT_EQ(media::cast::FrameId::first() + 1, latest_acked_frame_id());
+
+  // 10 differently-sized chunks pushed through the data pipe to make one frame
+  // that is larger than the data pipe's total capacity.
+  size_t offset = 0;
+  for (int i = 0; i < 10; ++i) {
+    const size_t chunk_size = 500 + i * 37;
+    ASSERT_TRUE(ProduceDataChunk(offset, chunk_size));
+    PostMojoCallTask_ConsumeDataChunk(offset, chunk_size, 6665);
+    RunPendingTasks();
+    offset += chunk_size;
+  }
+  PostMojoCallTask_SendFrame();
+  RunPendingTasks();
+  EXPECT_TRUE(ExpectOneFrameWasSent(6665));
+  AckOldestInFlightFrames(1);
+  EXPECT_EQ(media::cast::FrameId::first() + 2, latest_acked_frame_id());
+}
 
+TEST_F(CastRemotingSenderTest, SendsMultipleFramesWithDelayedAcks) {
   // Send 4 frames.
-  for (int i = 0; i < 4; ++i)
-    InsertFrame();
+  for (int i = 0; i < 4; ++i) {
+    ASSERT_TRUE(ProduceDataChunk(0, 16));
+    PostMojoCallTask_ConsumeDataChunk(0, 16, 16);
+    PostMojoCallTask_SendFrame();
+  }
+  RunPendingTasks();
   EXPECT_EQ(4, NumberOfFramesInFlight());
+  EXPECT_TRUE(ExpectNoFramesCanceled());
+
   // Ack one frame.
-  cast_feedback.ack_frame_id = media::cast::FrameId::first() + 1;
-  OnReceivedCastMessage(cast_feedback);
+  AckOldestInFlightFrames(1);
   EXPECT_EQ(3, NumberOfFramesInFlight());
+  EXPECT_TRUE(ExpectFramesCanceled(media::cast::FrameId::first(),
+                                   media::cast::FrameId::first()));
+
   // Ack all.
-  cast_feedback.received_later_frames.clear();
-  cast_feedback.ack_frame_id = media::cast::FrameId::first() + 4;
-  OnReceivedCastMessage(cast_feedback);
+  AckOldestInFlightFrames(3);
   EXPECT_EQ(0, NumberOfFramesInFlight());
+  EXPECT_TRUE(ExpectFramesCanceled(media::cast::FrameId::first() + 1,
+                                   media::cast::FrameId::first() + 3));
+}
+
+TEST_F(CastRemotingSenderTest, KickstartsIfAckNotTimely) {
+  // Send first frame and don't Ack it. Expect the first frame to be
+  // kickstarted.
+  ASSERT_TRUE(ProduceDataChunk(0, 16));
+  PostMojoCallTask_ConsumeDataChunk(0, 16, 16);
+  PostMojoCallTask_SendFrame();
+  EXPECT_EQ(media::cast::FrameId::first(), WaitForKickstart());
+  EXPECT_EQ(1, NumberOfFramesInFlight());
+
+  // Send 3 more frames and don't Ack them either. Expect the 4th frame to be
+  // kickstarted.
+  for (int i = 0; i < 3; ++i) {
+    ASSERT_TRUE(ProduceDataChunk(0, 16));
+    PostMojoCallTask_ConsumeDataChunk(0, 16, 16);
+    PostMojoCallTask_SendFrame();
+  }
+  EXPECT_EQ(media::cast::FrameId::first() + 3, WaitForKickstart());
+  EXPECT_EQ(4, NumberOfFramesInFlight());
+
+  // Ack the first two frames and wait for another kickstart (for the 4th frame
+  // again).
+  AckOldestInFlightFrames(2);
+  EXPECT_EQ(2, NumberOfFramesInFlight());
+  EXPECT_EQ(media::cast::FrameId::first() + 3, WaitForKickstart());
+}
+
+TEST_F(CastRemotingSenderTest, CancelsUnsentFrame) {
+  EXPECT_EQ(0u, GetSizeOfNextFrameData());
+  ASSERT_TRUE(ProduceDataChunk(0, 16));
+  PostMojoCallTask_ConsumeDataChunk(0, 16, 16);
+  RunPendingTasks();
+  EXPECT_EQ(16u, GetSizeOfNextFrameData());
+  PostMojoCallTask_CancelInFlightData();
+  RunPendingTasks();
+  EXPECT_EQ(0u, GetSizeOfNextFrameData());
+
+  // Since no frames were sent, none should have been passed to the
+  // CastTransport, and none should have been canceled.
+  std::vector<media::cast::EncodedFrame> frames;
+  TakeSentFrames(&frames);
+  EXPECT_TRUE(frames.empty());
+  EXPECT_TRUE(ExpectNoFramesCanceled());
 }
 
-TEST_F(CastRemotingSenderTest, CancelFramesInFlight) {
-  for (int i = 0; i < 10; ++i)
-    InsertFrame();
+// http://crbug.com/647423
+#define MAYBE_CancelsFramesInFlight DISABLED_CancelsFramesInFlight
+TEST_F(CastRemotingSenderTest, MAYBE_CancelsFramesInFlight) {
+  EXPECT_TRUE(IsFlowControlInStartingState());
+
+  // Send 10 frames.
+  for (int i = 0; i < 10; ++i) {
+    ASSERT_TRUE(ProduceDataChunk(0, 16));
+    PostMojoCallTask_ConsumeDataChunk(0, 16, 16);
+    PostMojoCallTask_SendFrame();
+  }
+  RunPendingTasks();
+  EXPECT_TRUE(IsFlowControlInConsumingState());
   EXPECT_EQ(10, NumberOfFramesInFlight());
 
-  media::cast::RtcpCastMessage cast_feedback(11);
-  cast_feedback.ack_frame_id = media::cast::FrameId::first();
-  OnReceivedCastMessage(cast_feedback);
+  // Ack the first frame.
+  AckOldestInFlightFrames(1);
+  EXPECT_TRUE(IsFlowControlInConsumingState());
   EXPECT_EQ(9, NumberOfFramesInFlight());
+  EXPECT_TRUE(ExpectFramesCanceled(media::cast::FrameId::first(),
+                                   media::cast::FrameId::first()));
+
+  // Cancel all in-flight data. This should cause the remaining 9 frames to be
+  // canceled.
+  PostMojoCallTask_CancelInFlightData();
+  RunPendingTasks();
+  EXPECT_TRUE(IsFlowControlInStartingState());
+  EXPECT_EQ(0, NumberOfFramesInFlight());
+  EXPECT_TRUE(ExpectFramesCanceled(media::cast::FrameId::first() + 1,
+                                   media::cast::FrameId::first() + 9));
+
+  // Send one more frame and ack it.
+  ASSERT_TRUE(ProduceDataChunk(0, 16));
+  PostMojoCallTask_ConsumeDataChunk(0, 16, 16);
+  PostMojoCallTask_SendFrame();
+  RunPendingTasks();
+  EXPECT_TRUE(IsFlowControlInConsumingState());
+  EXPECT_EQ(1, NumberOfFramesInFlight());
+  AckOldestInFlightFrames(1);
+  EXPECT_EQ(0, NumberOfFramesInFlight());
+
+  // Check that the dependency metadata was set correctly to indicate a frame
+  // that immediately follows a CancelInFlightData() operation.
+  std::vector<media::cast::EncodedFrame> frames;
+  TakeSentFrames(&frames);
+  ASSERT_EQ(11u, frames.size());
+  for (size_t i = 0; i < 11; ++i) {
+    const media::cast::EncodedFrame& frame = frames[i];
+    EXPECT_EQ(media::cast::FrameId::first() + i, frame.frame_id);
+    if (i == 0 || i == 10)
+      EXPECT_EQ(media::cast::EncodedFrame::KEY, frame.dependency);
+    else
+      EXPECT_EQ(media::cast::EncodedFrame::DEPENDENT, frame.dependency);
+  }
+}
+
+TEST_F(CastRemotingSenderTest, ManagesFlowControl) {
+  EXPECT_TRUE(IsFlowControlInStartingState());
 
-  CancelFramesInFlight();
+  // Send out a the maximum possible number of unacked frames, but don't ack any
+  // yet.
+  for (int i = 0; i < media::cast::kMaxUnackedFrames; ++i) {
+    ASSERT_TRUE(ProduceDataChunk(0, 4));
+    PostMojoCallTask_ConsumeDataChunk(0, 4, 4);
+    PostMojoCallTask_SendFrame();
+  }
+  RunPendingTasks();
+  EXPECT_TRUE(IsFlowControlInConsumingState());
+  EXPECT_EQ(media::cast::kMaxUnackedFrames, NumberOfFramesInFlight());
+  // Note: All frames should have been sent to the Transport, and so
+  // CastRemotingSender's single-frame data buffer should be empty.
+  EXPECT_EQ(0u, GetSizeOfNextFrameData());
+
+  // Attempting to send the next frame will place the CastRemotingSenderTest in
+  // a "paused flow" state.
+  ASSERT_TRUE(ProduceDataChunk(0, 4));
+  PostMojoCallTask_ConsumeDataChunk(0, 4, 4);
+  PostMojoCallTask_SendFrame();
+  RunPendingTasks();
+  EXPECT_TRUE(IsFlowControlInPausedState());
+  EXPECT_EQ(media::cast::kMaxUnackedFrames, NumberOfFramesInFlight());
+  // Note: The unsent frame resides in CastRemotingSender's single-frame data
+  // buffer.
+  EXPECT_EQ(4u, GetSizeOfNextFrameData());
+
+  // Ack the the first frame and expect sending to resume, starting with the
+  // frame that triggered the pause.
+  AckOldestInFlightFrames(1);
+  EXPECT_TRUE(IsFlowControlInConsumingState());
+  EXPECT_EQ(media::cast::kMaxUnackedFrames, NumberOfFramesInFlight());
+  // Note: The frame causing the earlier pause should have been sent, and so
+  // CastRemotingSender's single-frame data buffer should now be empty.
+  EXPECT_EQ(0u, GetSizeOfNextFrameData());
+
+  // Attempting to send another frame will once again place the
+  // CastRemotingSender in a "paused flow" state.
+  ASSERT_TRUE(ProduceDataChunk(0, 4));
+  PostMojoCallTask_ConsumeDataChunk(0, 4, 4);
+  PostMojoCallTask_SendFrame();
+  RunPendingTasks();
+  EXPECT_TRUE(IsFlowControlInPausedState());
+  EXPECT_EQ(media::cast::kMaxUnackedFrames, NumberOfFramesInFlight());
+  // Note: Once again, CastRemotingSender's single-frame data buffer contains an
+  // unsent frame.
+  EXPECT_EQ(4u, GetSizeOfNextFrameData());
+
+  // Send more frames: Some number of frames will queue-up inside the Mojo data
+  // pipe (the exact number depends on the data pipe's capacity, and how Mojo
+  // manages memory internally). At some point, attempting to produce and push
+  // another frame will fail because the data pipe is full.
+  int num_frames_in_data_pipe = 0;
+  while (ProduceDataChunk(0, 768)) {
+    ++num_frames_in_data_pipe;
+    PostMojoCallTask_ConsumeDataChunk(0, 768, 768);
+    PostMojoCallTask_SendFrame();
+    RunPendingTasks();
+    EXPECT_TRUE(IsFlowControlInPausedState());
+    EXPECT_EQ(media::cast::kMaxUnackedFrames, NumberOfFramesInFlight());
+    // Note: CastRemotingSender's single-frame data buffer should still contain
+    // the unsent 4-byte frame.
+    EXPECT_EQ(4u, GetSizeOfNextFrameData());
+  }
+  EXPECT_LT(0, num_frames_in_data_pipe);
+
+  // Ack one frame at a time until the backlog in the Mojo data pipe has
+  // cleared. In each iteration, the flow will resume, but then it will become
+  // paused again when Mojo method calls are resumed and cause
+  // CastRemotingSender to pull another frame out of the data pipe.
+  int remaining_frames_in_data_pipe = num_frames_in_data_pipe;
+  while (remaining_frames_in_data_pipe > 0) {
+    AckOldestInFlightFrames(1);
+    EXPECT_TRUE(IsFlowControlInConsumingState());
+    EXPECT_EQ(0u, GetSizeOfNextFrameData());
+    // The following will execute one ConsumeDataChunk() and one SendFrame()
+    // Mojo method call.
+    RunPendingTasks();
+    --remaining_frames_in_data_pipe;
+    EXPECT_TRUE(IsFlowControlInPausedState());
+    EXPECT_EQ(media::cast::kMaxUnackedFrames, NumberOfFramesInFlight());
+    EXPECT_EQ(768u, GetSizeOfNextFrameData());
+  }
+
+  // Ack one more frame. This should cause the flow to resume since there are no
+  // longer any frames backlogged on the input side of things.
+  AckOldestInFlightFrames(1);
+  EXPECT_TRUE(IsFlowControlInConsumingState());
+  RunPendingTasks();  // No additional Mojo method calls should be made here.
+  EXPECT_TRUE(IsFlowControlInConsumingState());
+  EXPECT_EQ(media::cast::kMaxUnackedFrames, NumberOfFramesInFlight());
+  EXPECT_EQ(0u, GetSizeOfNextFrameData());
+
+  // Ack all but one frame.
+  AckOldestInFlightFrames(NumberOfFramesInFlight() - 1);
+  EXPECT_TRUE(IsFlowControlInConsumingState());
+  EXPECT_EQ(1, NumberOfFramesInFlight());
+  // ..and one more frame can be sent without pausing.
+  ASSERT_TRUE(ProduceDataChunk(0, 4));
+  PostMojoCallTask_ConsumeDataChunk(0, 4, 4);
+  PostMojoCallTask_SendFrame();
+  RunPendingTasks();
+  EXPECT_TRUE(IsFlowControlInConsumingState());
+  EXPECT_EQ(2, NumberOfFramesInFlight());
+  // ...and ack these last two frames.
+  AckOldestInFlightFrames(2);
+  EXPECT_TRUE(IsFlowControlInConsumingState());
   EXPECT_EQ(0, NumberOfFramesInFlight());
+
+  // Finally, examine all frames that were sent to the CastTransport, and
+  // confirm their metadata and data is valid.
+  std::vector<media::cast::EncodedFrame> frames;
+  TakeSentFrames(&frames);
+  const size_t total_frames_sent =
+      media::cast::kMaxUnackedFrames + 2 + num_frames_in_data_pipe + 1;
+  ASSERT_EQ(total_frames_sent, frames.size());
+  media::cast::RtpTimeTicks last_rtp_timestamp =
+      media::cast::RtpTimeTicks() - media::cast::RtpTimeDelta::FromTicks(1);
+  for (size_t i = 0; i < total_frames_sent; ++i) {
+    const media::cast::EncodedFrame& frame = frames[i];
+    EXPECT_EQ(media::cast::FrameId::first() + i, frame.frame_id);
+    if (i == 0) {
+      EXPECT_EQ(media::cast::EncodedFrame::KEY, frame.dependency);
+      EXPECT_EQ(media::cast::FrameId::first() + i, frame.referenced_frame_id);
+    } else {
+      EXPECT_EQ(media::cast::EncodedFrame::DEPENDENT, frame.dependency);
+      EXPECT_EQ(media::cast::FrameId::first() + i - 1,
+                frame.referenced_frame_id);
+    }
+
+    // RTP timestamp must be monotonically increasing.
+    EXPECT_GT(frame.rtp_timestamp, last_rtp_timestamp);
+    last_rtp_timestamp = frame.rtp_timestamp;
+
+    size_t expected_frame_size = 4;
+    if ((i >= static_cast<size_t>(media::cast::kMaxUnackedFrames + 2)) &&

[dcheng, 2016/09/20 08:20:13]

These casts might be avoidable by writing media::cast::kMaxUnackedFrames + 2U

[miu, 2016/09/21 03:15:50]

Done. Good idea.

+        (i < static_cast<size_t>(media::cast::kMaxUnackedFrames + 2 +
+                                 num_frames_in_data_pipe))) {
+      expected_frame_size = 768;
+    }
+    EXPECT_TRUE(ExpectCorrectFrameData(expected_frame_size, frame));
+  }
 }
 
 }  // namespace cast