Protected streams report RTP messages directly to the FlexFec streams

webrtc/video/rtp_video_stream_receiver.cc

 /*
  *  Copyright (c) 2012 The WebRTC project authors. All Rights Reserved.
  *
  *  Use of this source code is governed by a BSD-style license
  *  that can be found in the LICENSE file in the root of the source
  *  tree. An additional intellectual property rights grant can be found
  *  in the file PATENTS.  All contributing project authors may
  *  be found in the AUTHORS file in the root of the source tree.
  */
 
 #include "webrtc/video/rtp_video_stream_receiver.h"
 
+#include <algorithm>
 #include <utility>
 #include <vector>
 
 #include "webrtc/common_types.h"
 #include "webrtc/config.h"
 #include "webrtc/media/base/mediaconstants.h"
 #include "webrtc/modules/pacing/packet_router.h"
 #include "webrtc/modules/remote_bitrate_estimator/include/remote_bitrate_estimator.h"
 #include "webrtc/modules/rtp_rtcp/include/receive_statistics.h"
 #include "webrtc/modules/rtp_rtcp/include/rtp_cvo.h"
@@ skipping 161 matching lines @@
         new NackModule(clock_, nack_sender, keyframe_request_sender));
     process_thread_->RegisterModule(nack_module_.get(), RTC_FROM_HERE);
   }
 
   packet_buffer_ = video_coding::PacketBuffer::Create(
       clock_, kPacketBufferStartSize, kPacketBufferMaxSixe, this);
   reference_finder_.reset(new video_coding::RtpFrameReferenceFinder(this));
 }
 
 RtpVideoStreamReceiver::~RtpVideoStreamReceiver() {
+  RTC_DCHECK(secondary_sinks_.empty());
+
   if (nack_module_) {
     process_thread_->DeRegisterModule(nack_module_.get());
   }
 
   process_thread_->DeRegisterModule(rtp_rtcp_.get());
 
   packet_router_->RemoveReceiveRtpModule(rtp_rtcp_.get());
   UpdateHistograms();
 }
 
@@ skipping 97 matching lines @@
     const int frequency,
     const size_t channels,
     const uint32_t rate) {
   RTC_NOTREACHED();
   return 0;
 }
 
 // This method handles both regular RTP packets and packets recovered
 // via FlexFEC.
 void RtpVideoStreamReceiver::OnRtpPacket(const RtpPacketReceived& packet) {
+  RTC_DCHECK_RUN_ON(&worker_thread_checker_);

[eladalon, 2017/08/01 08:19:38]

This will not work for video_loopback, because DirectTransport creates its own
thread, "NetworkProcess". I don't want to put a lock in actual code just to get
rid of a problem in tests, and I don't want to remove this check, either.
Refactoring DirectTransport would likely not be an easy task. Any suggestions?
The simplest I can think of is checking for (ON-WORKER-THREAD ||
IN-SPECIFIC-BUILDS-LIST), where the latter would whitelist the loopback,
unit-tests, etc. Wdyt?
+danilchap@, +stefan@, +solenberg@

[brandtr, 2017/08/01 08:34:15]

Nice! This addition makes the code faster to read.

+
   {
     rtc::CritScope lock(&receive_cs_);
     if (!receiving_) {
       return;
     }
 
     if (!packet.recovered()) {
       int64_t now_ms = clock_->TimeInMilliseconds();
 
       // Periodically log the RTP header of incoming packets.
@@ skipping 33 matching lines @@
   ReceivePacket(packet.data(), packet.size(), header, in_order);
   // Update receive statistics after ReceivePacket.
   // Receive statistics will be reset if the payload type changes (make sure
   // that the first packet is included in the stats).
   if (!packet.recovered()) {
     // TODO(nisse): We should pass a recovered flag to stats, to aid
     // fixing bug bugs.webrtc.org/6339.
     rtp_receive_statistics_->IncomingPacket(
         header, packet.size(), IsPacketRetransmitted(header, in_order));
   }
+
+  for (auto* secondary_sink : secondary_sinks_) {

[brandtr, 2017/08/01 08:34:15]

RtpPacketSinkInterface* ?

[eladalon, 2017/08/01 09:30:28]

Done.

+    secondary_sink->OnRtpPacket(packet);
+  }
 }
 
 int32_t RtpVideoStreamReceiver::RequestKeyFrame() {
   return rtp_rtcp_->RequestKeyFrame();
 }
 
 bool RtpVideoStreamReceiver::IsUlpfecEnabled() const {
   return config_.rtp.ulpfec.ulpfec_payload_type != -1;
 }
 
@@ skipping 47 matching lines @@
 
 rtc::Optional<int64_t> RtpVideoStreamReceiver::LastReceivedPacketMs() const {
   return packet_buffer_->LastReceivedPacketMs();
 }
 
 rtc::Optional<int64_t> RtpVideoStreamReceiver::LastReceivedKeyframePacketMs()
     const {
   return packet_buffer_->LastReceivedKeyframePacketMs();
 }
 
+void RtpVideoStreamReceiver::AddSecondarySink(RtpPacketSinkInterface* sink) {
+  RTC_DCHECK_RUN_ON(&worker_thread_checker_);
+  RTC_DCHECK(std::find(secondary_sinks_.cbegin(), secondary_sinks_.cend(),
+                       sink) == secondary_sinks_.cend());
+  secondary_sinks_.push_back(sink);
+}
+
+void RtpVideoStreamReceiver::RemoveSecondarySink(
+    const RtpPacketSinkInterface* sink) {
+  RTC_DCHECK_RUN_ON(&worker_thread_checker_);
+  auto it = std::find(secondary_sinks_.begin(), secondary_sinks_.end(), sink);
+  if (it == secondary_sinks_.end()) {
+    // We might be rolling-back a call whose setup failed mid-way. In such a
+    // case, it's simpler to remove "everything" rather than remember what
+    // has already been added.
+    LOG(LS_WARNING) << "Removal of unknown sink.";
+    return;
+  }
+  secondary_sinks_.erase(it);
+}
+
 void RtpVideoStreamReceiver::ReceivePacket(const uint8_t* packet,
                                            size_t packet_length,
                                            const RTPHeader& header,
                                            bool in_order) {
   if (rtp_payload_registry_.IsEncapsulated(header)) {
     ParseAndHandleEncapsulatingHeader(packet, packet_length, header);
     return;
   }
   const uint8_t* payload = packet + header.headerLength;
   assert(packet_length >= header.headerLength);
@@ skipping 239 matching lines @@
     return;
 
   if (!sprop_decoder.DecodeSprop(sprop_base64_it->second.c_str()))
     return;
 
   tracker_.InsertSpsPpsNalus(sprop_decoder.sps_nalu(),
                              sprop_decoder.pps_nalu());
 }
 
 }  // namespace webrtc