Set up Receiver and Sender for remoting

This patch updates the Sender and Receiver classes to be usable for
remoting.

Bug: b/175606988
Change-Id: Ia515f211ab26362933746c660848bfd6474c959f
Reviewed-on: https://chromium-review.googlesource.com/c/openscreen/+/2611785
Reviewed-by: Yuri Wiitala <miu@chromium.org>
Commit-Queue: Jordan Bayles <jophba@chromium.org>

10 files changed, 89 insertions, 25 deletions

diff --git a/cast/streaming/receiver.cc b/cast/streaming/receiver.cc
index d62d6251..0d3358b4 100644
--- a/cast/streaming/receiver.cc
+++ b/cast/streaming/receiver.cc
@@ -40,6 +40,7 @@ Receiver::Receiver(Environment* environment,
       rtp_parser_(config.sender_ssrc),
       rtp_timebase_(config.rtp_timebase),
       crypto_(config.aes_secret_key, config.aes_iv_mask),
+      is_pli_enabled_(config.is_pli_enabled),
       rtcp_buffer_capacity_(environment->GetMaxPacketSize()),
       rtcp_buffer_(new uint8_t[rtcp_buffer_capacity_]),
       rtcp_alarm_(environment->now_function(), environment->task_runner()),
@@ -72,7 +73,10 @@ void Receiver::SetPlayerProcessingTime(Clock::duration needed_time) {
 }
 
 void Receiver::RequestKeyFrame() {
-  if (!last_key_frame_received_.is_null() &&
+  // If we don't have picture loss indication enabled, we should not request
+  // any key frames.
+  OSP_DCHECK(is_pli_enabled_) << "PLI is not enabled.";
+  if (is_pli_enabled_ && !last_key_frame_received_.is_null() &&
       last_frame_consumed_ >= last_key_frame_received_ &&
       !rtcp_builder_.is_picture_loss_indicator_set()) {
     rtcp_builder_.SetPictureLossIndicator(true);
diff --git a/cast/streaming/receiver.h b/cast/streaming/receiver.h
index abdc1852..057c56d8 100644
--- a/cast/streaming/receiver.h
+++ b/cast/streaming/receiver.h
@@ -173,6 +173,12 @@ class Receiver {
   // portion of the buffer that was populated.
   EncodedFrame ConsumeNextFrame(absl::Span<uint8_t> buffer);
 
+  // Allows setting picture loss indication for testing. In production, this
+  // should be done using the config.
+  void SetPliEnabledForTesting(bool is_pli_enabled) {
+    is_pli_enabled_ = is_pli_enabled;
+  }
+
   // The default "player processing time" amount. See SetPlayerProcessingTime().
   static constexpr std::chrono::milliseconds kDefaultPlayerProcessingTime{5};
 
@@ -266,6 +272,7 @@ class Receiver {
   RtpPacketParser rtp_parser_;
   const int rtp_timebase_;    // RTP timestamp ticks per second.
   const FrameCrypto crypto_;  // Decrypts assembled frames.
+  bool is_pli_enabled_;       // Whether picture loss indication is enabled.
 
   // Buffer for serializing/sending RTCP packets.
   const int rtcp_buffer_capacity_;
diff --git a/cast/streaming/receiver_session.cc b/cast/streaming/receiver_session.cc
index 7e4c70d3..70abe7aa 100644
--- a/cast/streaming/receiver_session.cc
+++ b/cast/streaming/receiver_session.cc
@@ -157,10 +157,11 @@ void ReceiverSession::OnOffer(SenderMessage message) {
 
 std::unique_ptr<Receiver> ReceiverSession::ConstructReceiver(
     const Stream& stream) {
+  // Session config is currently only for mirroring.
   SessionConfig config = {stream.ssrc,         stream.ssrc + 1,
                           stream.rtp_timebase, stream.channels,
                           stream.target_delay, stream.aes_key,
-                          stream.aes_iv_mask};
+                          stream.aes_iv_mask,  /* is_pli_enabled */ true};
   return std::make_unique<Receiver>(environment_, &packet_router_,
                                     std::move(config));
 }
diff --git a/cast/streaming/receiver_unittest.cc b/cast/streaming/receiver_unittest.cc
index d3c77178..0a6817e6 100644
--- a/cast/streaming/receiver_unittest.cc
+++ b/cast/streaming/receiver_unittest.cc
@@ -275,7 +275,8 @@ class ReceiverTest : public testing::Test {
                    /* .channels = */ 2,
                    /* .target_playout_delay = */ kTargetPlayoutDelay,
                    /* .aes_secret_key = */ kAesKey,
-                   /* .aes_iv_mask = */ kCastIvMask}),
+                   /* .aes_iv_mask = */ kCastIvMask,
+                   /* .is_pli_enabled = */ true}),
         sender_(&task_runner_, &env_) {
     env_.set_socket_error_handler(
         [](Error error) { ASSERT_TRUE(error.ok()) << error; });
@@ -662,6 +663,19 @@ TEST_F(ReceiverTest, RequestsKeyFrameToRectifyPictureLoss) {
   testing::Mock::VerifyAndClearExpectations(sender());
 }
 
+TEST_F(ReceiverTest, PLICanBeDisabled) {
+  receiver()->SetPliEnabledForTesting(false);
+
+#if OSP_DCHECK_IS_ON()
+  EXPECT_DEATH(receiver()->RequestKeyFrame(), ".*PLI is not enabled.*");
+#else
+  EXPECT_CALL(*sender(), OnReceiverIndicatesPictureLoss()).Times(0);
+  receiver()->RequestKeyFrame();
+  AdvanceClockAndRunTasks(kOneWayNetworkDelay);
+  testing::Mock::VerifyAndClearExpectations(sender());
+#endif
+}
+
 // Tests that the Receiver will start dropping packets once its frame queue is
 // full (i.e., when the consumer is not pulling them out of the queue). Since
 // the Receiver will stop ACK'ing frames, the Sender will become stalled.
diff --git a/cast/streaming/sender.cc b/cast/streaming/sender.cc
index dcb2f04b..ba42bcb4 100644
--- a/cast/streaming/sender.cc
+++ b/cast/streaming/sender.cc
@@ -170,6 +170,13 @@ Sender::EnqueueFrameResult Sender::EnqueueFrame(const EncodedFrame& frame) {
   return OK;
 }
 
+void Sender::CancelInFlightData() {
+  while (checkpoint_frame_id_ <= last_enqueued_frame_id_) {
+    ++checkpoint_frame_id_;
+    CancelPendingFrame(checkpoint_frame_id_);
+  }
+}
+
 void Sender::OnReceivedRtcpPacket(Clock::time_point arrival_time,
                                   absl::Span<const uint8_t> packet) {
   rtcp_packet_arrival_time_ = arrival_time;
diff --git a/cast/streaming/sender.h b/cast/streaming/sender.h
index 7e827117..ed28d3e6 100644
--- a/cast/streaming/sender.h
+++ b/cast/streaming/sender.h
@@ -168,6 +168,10 @@ class Sender final : public SenderPacketRouter::Sender,
   // prior frame; and the frame's |data| pointer must be set.
   [[nodiscard]] EnqueueFrameResult EnqueueFrame(const EncodedFrame& frame);
 
+  // Causes all pending operations to discard data when they are processed
+  // later.
+  void CancelInFlightData();
+
  private:
   // Tracking/Storage for frames that are ready-to-send, and until they are
   // fully received at the other end.
diff --git a/cast/streaming/sender_session.cc b/cast/streaming/sender_session.cc
index 1a2e8cc4..7a54c7e1 100644
--- a/cast/streaming/sender_session.cc
+++ b/cast/streaming/sender_session.cc
@@ -215,9 +215,15 @@ void SenderSession::OnAnswer(ReceiverMessage message) {
 std::unique_ptr<Sender> SenderSession::CreateSender(Ssrc receiver_ssrc,
                                                     const Stream& stream,
                                                     RtpPayloadType type) {
-  SessionConfig config{
-      stream.ssrc,         receiver_ssrc,  stream.rtp_timebase, stream.channels,
-      stream.target_delay, stream.aes_key, stream.aes_iv_mask};
+  // Session config is currently only for mirroring.
+  SessionConfig config{stream.ssrc,
+                       receiver_ssrc,
+                       stream.rtp_timebase,
+                       stream.channels,
+                       stream.target_delay,
+                       stream.aes_key,
+                       stream.aes_iv_mask,
+                       /* is_pli_enabled*/ true};
 
   return std::make_unique<Sender>(environment_, &packet_router_,
                                   std::move(config), type);
diff --git a/cast/streaming/sender_unittest.cc b/cast/streaming/sender_unittest.cc
index 7b162161..1296e4a1 100644
--- a/cast/streaming/sender_unittest.cc
+++ b/cast/streaming/sender_unittest.cc
@@ -101,6 +101,11 @@ constexpr milliseconds kCaptureDelay{11};
     EXPECT_GE((duration_a), (duration_b) - (epsilon));       \
   }
 
+void OverrideRtpTimestamp(int frame_count, EncodedFrame* frame, int fps) {
+  const int ticks = frame_count * kRtpTimebase / fps;
+  frame->rtp_timestamp = RtpTimeTicks() + RtpTimeDelta::FromTicks(ticks);
+}
+
 // Simulates UDP/IPv6 traffic in one direction (from Sender→Receiver, or
 // Receiver→Sender), with a settable amount of delay.
 class SimulatedNetworkPipe {
@@ -339,7 +344,8 @@ class SenderTest : public testing::Test {
                  /* .channels = */ 2,
                  /* .target_playout_delay = */ kTargetPlayoutDelay,
                  /* .aes_secret_key = */ kAesKey,
-                 /* .aes_iv_mask = */ kCastIvMask},
+                 /* .aes_iv_mask = */ kCastIvMask,
+                 /* .is_pli_enabled = */ true},
                 kRtpPayloadType),
         receiver_to_sender_pipe_(&task_runner_, &sender_packet_router_),
         receiver_(&receiver_to_sender_pipe_),
@@ -615,18 +621,13 @@ TEST_F(SenderTest, RejectsEnqueuingBeforeProtocolDesignLimit) {
   constexpr int kFramesPerSecond = 1000;
   constexpr milliseconds kFrameDuration{1};
 
-  const auto OverrideRtpTimestamp = [](int frame_count, EncodedFrame* frame) {
-    const int ticks = frame_count * kRtpTimebase / kFramesPerSecond;
-    frame->rtp_timestamp = RtpTimeTicks() + RtpTimeDelta::FromTicks(ticks);
-  };
-
   // Send the absolute design-limit maximum number of frames.
   int frame_count = 0;
   for (; frame_count < kMaxUnackedFrames; ++frame_count) {
     EncodedFrameWithBuffer frame;
     PopulateFrameWithDefaults(sender()->GetNextFrameId(), FakeClock::now(), 0,
                               13 /* bytes */, &frame);
-    OverrideRtpTimestamp(frame_count, &frame);
+    OverrideRtpTimestamp(frame_count, &frame, kFramesPerSecond);
     ASSERT_EQ(Sender::OK, sender()->EnqueueFrame(frame));
     SimulateExecution(kFrameDuration);
   }
@@ -635,7 +636,7 @@ TEST_F(SenderTest, RejectsEnqueuingBeforeProtocolDesignLimit) {
   EncodedFrameWithBuffer one_frame_too_much;
   PopulateFrameWithDefaults(sender()->GetNextFrameId(), FakeClock::now(), 0,
                             13 /* bytes */, &one_frame_too_much);
-  OverrideRtpTimestamp(frame_count++, &one_frame_too_much);
+  OverrideRtpTimestamp(frame_count++, &one_frame_too_much, kFramesPerSecond);
   EXPECT_EQ(Sender::REACHED_ID_SPAN_LIMIT,
             sender()->EnqueueFrame(one_frame_too_much));
   SimulateExecution(kFrameDuration);
@@ -652,12 +653,31 @@ TEST_F(SenderTest, RejectsEnqueuingBeforeProtocolDesignLimit) {
   EncodedFrameWithBuffer another_frame_too_much;
   PopulateFrameWithDefaults(sender()->GetNextFrameId(), FakeClock::now(), 0,
                             13 /* bytes */, &another_frame_too_much);
-  OverrideRtpTimestamp(frame_count++, &another_frame_too_much);
+  OverrideRtpTimestamp(frame_count++, &another_frame_too_much,
+                       kFramesPerSecond);
   EXPECT_EQ(Sender::REACHED_ID_SPAN_LIMIT,
             sender()->EnqueueFrame(another_frame_too_much));
   SimulateExecution(kFrameDuration);
 }
 
+TEST_F(SenderTest, CanCancelAllInFlightFrames) {
+  NiceMock<MockObserver> observer;
+  sender()->SetObserver(&observer);
+
+  // Send the absolute design-limit maximum number of frames.
+  for (int i = 0; i < kMaxUnackedFrames; ++i) {
+    EncodedFrameWithBuffer frame;
+    PopulateFrameWithDefaults(sender()->GetNextFrameId(), FakeClock::now(), 0,
+                              13 /* bytes */, &frame);
+    OverrideRtpTimestamp(i, &frame, 1000 /* fps */);
+    ASSERT_EQ(Sender::OK, sender()->EnqueueFrame(frame));
+    SimulateExecution(kFrameDuration);
+  }
+
+  EXPECT_CALL(observer, OnFrameCanceled(_)).Times(kMaxUnackedFrames);
+  sender()->CancelInFlightData();
+}
+
 // Tests that the Sender rejects frames if too-long a media duration is
 // in-flight. This is the Sender's primary flow control mechanism.
 TEST_F(SenderTest, RejectsEnqueuingIfTooLongMediaDurationIsInFlight) {
@@ -666,11 +686,6 @@ TEST_F(SenderTest, RejectsEnqueuingIfTooLongMediaDurationIsInFlight) {
   constexpr int kFramesPerSecond = 20;
   constexpr milliseconds kFrameDuration{50};
 
-  const auto OverrideRtpTimestamp = [](int frame_count, EncodedFrame* frame) {
-    const int ticks = frame_count * kRtpTimebase / kFramesPerSecond;
-    frame->rtp_timestamp = RtpTimeTicks() + RtpTimeDelta::FromTicks(ticks);
-  };
-
   // Enqueue frames until one is rejected because the in-flight duration would
   // be too high.
   EncodedFrameWithBuffer frame;
@@ -678,7 +693,7 @@ TEST_F(SenderTest, RejectsEnqueuingIfTooLongMediaDurationIsInFlight) {
   for (; frame_count < kMaxUnackedFrames; ++frame_count) {
     PopulateFrameWithDefaults(sender()->GetNextFrameId(), FakeClock::now(), 0,
                               13 /* bytes */, &frame);
-    OverrideRtpTimestamp(frame_count, &frame);
+    OverrideRtpTimestamp(frame_count, &frame, kFramesPerSecond);
     const auto result = sender()->EnqueueFrame(frame);
     SimulateExecution(kFrameDuration);
     if (result == Sender::MAX_DURATION_IN_FLIGHT) {
@@ -699,7 +714,7 @@ TEST_F(SenderTest, RejectsEnqueuingIfTooLongMediaDurationIsInFlight) {
   EncodedFrameWithBuffer one_frame_too_much;
   PopulateFrameWithDefaults(sender()->GetNextFrameId(), FakeClock::now(), 0,
                             13 /* bytes */, &one_frame_too_much);
-  OverrideRtpTimestamp(++frame_count, &one_frame_too_much);
+  OverrideRtpTimestamp(++frame_count, &one_frame_too_much, kFramesPerSecond);
   EXPECT_EQ(Sender::MAX_DURATION_IN_FLIGHT,
             sender()->EnqueueFrame(one_frame_too_much));
   SimulateExecution(kFrameDuration);
diff --git a/cast/streaming/session_config.cc b/cast/streaming/session_config.cc
index f6f4aade..226629d8 100644
--- a/cast/streaming/session_config.cc
+++ b/cast/streaming/session_config.cc
@@ -15,14 +15,16 @@ SessionConfig::SessionConfig(Ssrc sender_ssrc,
                              int channels,
                              std::chrono::milliseconds target_playout_delay,
                              std::array<uint8_t, 16> aes_secret_key,
-                             std::array<uint8_t, 16> aes_iv_mask)
+                             std::array<uint8_t, 16> aes_iv_mask,
+                             bool is_pli_enabled)
     : sender_ssrc(sender_ssrc),
       receiver_ssrc(receiver_ssrc),
       rtp_timebase(rtp_timebase),
       channels(channels),
       target_playout_delay(target_playout_delay),
       aes_secret_key(std::move(aes_secret_key)),
-      aes_iv_mask(std::move(aes_iv_mask)) {}
+      aes_iv_mask(std::move(aes_iv_mask)),
+      is_pli_enabled(is_pli_enabled) {}
 
 SessionConfig::SessionConfig(const SessionConfig& other) = default;
 SessionConfig::SessionConfig(SessionConfig&& other) noexcept = default;
diff --git a/cast/streaming/session_config.h b/cast/streaming/session_config.h
index cf87667e..15a7251b 100644
--- a/cast/streaming/session_config.h
+++ b/cast/streaming/session_config.h
@@ -24,7 +24,8 @@ struct SessionConfig final {
                 int channels,
                 std::chrono::milliseconds target_playout_delay,
                 std::array<uint8_t, 16> aes_secret_key,
-                std::array<uint8_t, 16> aes_iv_mask);
+                std::array<uint8_t, 16> aes_iv_mask,
+                bool is_pli_enabled);
   SessionConfig(const SessionConfig& other);
   SessionConfig(SessionConfig&& other) noexcept;
   SessionConfig& operator=(const SessionConfig& other);
@@ -50,6 +51,9 @@ struct SessionConfig final {
   // The AES-128 crypto key and initialization vector.
   std::array<uint8_t, 16> aes_secret_key{};
   std::array<uint8_t, 16> aes_iv_mask{};
+
+  // Whether picture loss indication (PLI) should be used for this session.
+  bool is_pli_enabled = false;
 };
 
 }  // namespace cast