Initial implementation of RtpTransportControllerReceive and related interfaces.

webrtc/call/call.cc

Index: webrtc/call/call.cc
diff --git a/webrtc/call/call.cc b/webrtc/call/call.cc
index bb83b7201671488289d4df52344cee29b5ebc931..498e3d7b8eb995d12067565d90cb36c1b5127b4d 100644
--- a/webrtc/call/call.cc
+++ b/webrtc/call/call.cc
@@ -33,6 +33,7 @@
 #include "webrtc/call/bitrate_allocator.h"
 #include "webrtc/call/call.h"
 #include "webrtc/call/flexfec_receive_stream_impl.h"
+#include "webrtc/call/rtp_transport_controller_receive.h"
 #include "webrtc/config.h"
 #include "webrtc/logging/rtc_event_log/rtc_event_log.h"
 #include "webrtc/modules/bitrate_controller/include/bitrate_controller.h"
@@ -92,6 +93,7 @@ bool UseSendSideBwe(const FlexfecReceiveStream::Config& config) {
 namespace internal {
 
 class Call : public webrtc::Call,
+             public RtpPacketObserverInterface,
              public PacketReceiver,
              public RecoveredPacketReceiver,
              public CongestionController::Observer,
@@ -162,6 +164,11 @@ class Call : public webrtc::Call,
   void OnAllocationLimitsChanged(uint32_t min_send_bitrate_bps,
                                  uint32_t max_padding_bitrate_bps) override;
 
+  // Implements RtpPacketObserverInterface.
+  void OnRtpPacket(MediaType media_type,
+                   const RtpTransportControllerReceiveInterface::Config config,
+                   const RtpPacketReceived& packet) override;
+
  private:
   DeliveryStatus DeliverRtcp(MediaType media_type, const uint8_t* packet,
                              size_t length);
@@ -176,11 +183,6 @@ class Call : public webrtc::Call,
                                  MediaType media_type)
       SHARED_LOCKS_REQUIRED(receive_crit_);
 
-  rtc::Optional<RtpPacketReceived> ParseRtpPacket(const uint8_t* packet,
-                                                  size_t length,
-                                                  const PacketTime& packet_time)
-      SHARED_LOCKS_REQUIRED(receive_crit_);
-
   void UpdateSendHistograms() EXCLUSIVE_LOCKS_REQUIRED(&bitrate_crit_);
   void UpdateReceiveHistograms();
   void UpdateHistograms();
@@ -202,45 +204,19 @@ class Call : public webrtc::Call,
   std::unique_ptr<RWLockWrapper> receive_crit_;
   // Audio, Video, and FlexFEC receive streams are owned by the client that
   // creates them.
+  std::unique_ptr<RtpTransportControllerReceiveInterface> rtp_transport_receive_
+      GUARDED_BY(receive_crit_);
+
+  // TODO(nisse): Try to eliminate these additional mappings. Two of
+  // the users are DeliverRTCP and OnRecoveredPacket.
   std::map<uint32_t, AudioReceiveStream*> audio_receive_ssrcs_
       GUARDED_BY(receive_crit_);
   std::map<uint32_t, VideoReceiveStream*> video_receive_ssrcs_
       GUARDED_BY(receive_crit_);
   std::set<VideoReceiveStream*> video_receive_streams_
       GUARDED_BY(receive_crit_);
-  // Each media stream could conceivably be protected by multiple FlexFEC
-  // streams.
-  std::multimap<uint32_t, FlexfecReceiveStreamImpl*>
-      flexfec_receive_ssrcs_media_ GUARDED_BY(receive_crit_);
-  std::map<uint32_t, FlexfecReceiveStreamImpl*>
-      flexfec_receive_ssrcs_protection_ GUARDED_BY(receive_crit_);
-  std::set<FlexfecReceiveStreamImpl*> flexfec_receive_streams_
-      GUARDED_BY(receive_crit_);
-  std::map<std::string, AudioReceiveStream*> sync_stream_mapping_
-      GUARDED_BY(receive_crit_);
 
-  // This extra map is used for receive processing which is
-  // independent of media type.
-
-  // TODO(nisse): In the RTP transport refactoring, we should have a
-  // single mapping from ssrc to a more abstract receive stream, with
-  // accessor methods for all configuration we need at this level.
-  struct ReceiveRtpConfig {
-    ReceiveRtpConfig() = default;  // Needed by std::map
-    ReceiveRtpConfig(const std::vector<RtpExtension>& extensions,
-                     bool use_send_side_bwe)
-        : extensions(extensions), use_send_side_bwe(use_send_side_bwe) {}
-
-    // Registered RTP header extensions for each stream. Note that RTP header
-    // extensions are negotiated per track ("m= line") in the SDP, but we have
-    // no notion of tracks at the Call level. We therefore store the RTP header
-    // extensions per SSRC instead, which leads to some storage overhead.
-    RtpHeaderExtensionMap extensions;
-    // Set if both RTP extension the RTCP feedback message needed for
-    // send side BWE are negotiated.
-    bool use_send_side_bwe = false;
-  };
-  std::map<uint32_t, ReceiveRtpConfig> receive_rtp_config_
+  std::map<std::string, AudioReceiveStream*> sync_stream_mapping_
       GUARDED_BY(receive_crit_);
 
   std::unique_ptr<RWLockWrapper> send_crit_;
@@ -317,6 +293,8 @@ Call::Call(const Call::Config& config)
       audio_network_state_(kNetworkDown),
       video_network_state_(kNetworkDown),
       receive_crit_(RWLockWrapper::CreateRWLock()),
+      rtp_transport_receive_(
+          RtpTransportControllerReceiveInterface::Create(this)),
       send_crit_(RWLockWrapper::CreateRWLock()),
       event_log_(config.event_log),
       first_packet_sent_ms_(-1),
@@ -372,9 +350,6 @@ Call::~Call() {
   RTC_CHECK(audio_send_ssrcs_.empty());
   RTC_CHECK(video_send_ssrcs_.empty());
   RTC_CHECK(video_send_streams_.empty());
-  RTC_CHECK(audio_receive_ssrcs_.empty());
-  RTC_CHECK(video_receive_ssrcs_.empty());
-  RTC_CHECK(video_receive_streams_.empty());
 
   pacer_thread_->Stop();
   pacer_thread_->DeRegisterModule(congestion_controller_->pacer());
@@ -397,29 +372,6 @@ Call::~Call() {
   Trace::ReturnTrace();
 }
 
-rtc::Optional<RtpPacketReceived> Call::ParseRtpPacket(
-    const uint8_t* packet,
-    size_t length,
-    const PacketTime& packet_time) {
-  RtpPacketReceived parsed_packet;
-  if (!parsed_packet.Parse(packet, length))
-    return rtc::Optional<RtpPacketReceived>();
-
-  auto it = receive_rtp_config_.find(parsed_packet.Ssrc());
-  if (it != receive_rtp_config_.end())
-    parsed_packet.IdentifyExtensions(it->second.extensions);
-
-  int64_t arrival_time_ms;
-  if (packet_time.timestamp != -1) {
-    arrival_time_ms = (packet_time.timestamp + 500) / 1000;
-  } else {
-    arrival_time_ms = clock_->TimeInMilliseconds();
-  }
-  parsed_packet.set_arrival_time_ms(arrival_time_ms);
-
-  return rtc::Optional<RtpPacketReceived>(std::move(parsed_packet));
-}
-
 void Call::UpdateHistograms() {
   RTC_HISTOGRAM_COUNTS_100000(
       "WebRTC.Call.LifetimeInSeconds",
@@ -558,14 +510,17 @@ webrtc::AudioReceiveStream* Call::CreateAudioReceiveStream(
   AudioReceiveStream* receive_stream = new AudioReceiveStream(
       &packet_router_, config,
       config_.audio_state, event_log_);
+
+  RtpTransportControllerReceiveInterface::Config receive_config;
+  receive_config.use_send_side_bwe = UseSendSideBwe(config);
+
   {
     WriteLockScoped write_lock(*receive_crit_);
-    RTC_DCHECK(audio_receive_ssrcs_.find(config.rtp.remote_ssrc) ==
-               audio_receive_ssrcs_.end());
+    bool success = rtp_transport_receive_->AddReceiver(
+        config.rtp.remote_ssrc, MediaType::AUDIO, receive_config,
+        receive_stream);
+    RTC_DCHECK(success);
     audio_receive_ssrcs_[config.rtp.remote_ssrc] = receive_stream;
-    receive_rtp_config_[config.rtp.remote_ssrc] =
-        ReceiveRtpConfig(config.rtp.extensions, UseSendSideBwe(config));
-
     ConfigureSync(config.sync_group);
   }
   {
@@ -589,6 +544,9 @@ void Call::DestroyAudioReceiveStream(
       static_cast<webrtc::internal::AudioReceiveStream*>(receive_stream);
   {
     WriteLockScoped write_lock(*receive_crit_);
+    bool success = rtp_transport_receive_->RemoveReceiver(audio_receive_stream);
+    RTC_DCHECK(success);
+
     const AudioReceiveStream::Config& config = audio_receive_stream->config();
     uint32_t ssrc = config.rtp.remote_ssrc;
     congestion_controller_->GetRemoteBitrateEstimator(UseSendSideBwe(config))
@@ -602,7 +560,6 @@ void Call::DestroyAudioReceiveStream(
       sync_stream_mapping_.erase(it);
       ConfigureSync(sync_group);
     }
-    receive_rtp_config_.erase(ssrc);
   }
   UpdateAggregateNetworkState();
   delete audio_receive_stream;
@@ -687,22 +644,26 @@ webrtc::VideoReceiveStream* Call::CreateVideoReceiveStream(
       module_process_thread_.get(), call_stats_.get(), &remb_);
 
   const webrtc::VideoReceiveStream::Config& config = receive_stream->config();
-  ReceiveRtpConfig receive_config(config.rtp.extensions,
-                                  UseSendSideBwe(config));
+  RtpTransportControllerReceiveInterface::Config receive_config;
+  receive_config.use_send_side_bwe = UseSendSideBwe(config);
+
   {
     WriteLockScoped write_lock(*receive_crit_);
-    RTC_DCHECK(video_receive_ssrcs_.find(config.rtp.remote_ssrc) ==
-               video_receive_ssrcs_.end());
-    video_receive_ssrcs_[config.rtp.remote_ssrc] = receive_stream;
+    bool success = rtp_transport_receive_->AddReceiver(
+        config.rtp.remote_ssrc, MediaType::VIDEO, receive_config,
+        receive_stream);
+    RTC_DCHECK(success);
     if (config.rtp.rtx_ssrc) {
-      video_receive_ssrcs_[config.rtp.rtx_ssrc] = receive_stream;
       // We record identical config for the rtx stream as for the main
       // stream. Since the transport_cc negotiation is per payload
       // type, we may get an incorrect value for the rtx stream, but
       // that is unlikely to matter in practice.
-      receive_rtp_config_[config.rtp.rtx_ssrc] = receive_config;
+      bool success = rtp_transport_receive_->AddReceiver(
+          config.rtp.remote_ssrc, MediaType::VIDEO, receive_config,
+          receive_stream);
+
+      RTC_DCHECK(success);
     }
-    receive_rtp_config_[config.rtp.remote_ssrc] = receive_config;
     video_receive_streams_.insert(receive_stream);
     ConfigureSync(config.sync_group);
   }
@@ -728,7 +689,6 @@ void Call::DestroyVideoReceiveStream(
         if (receive_stream_impl != nullptr)
           RTC_DCHECK(receive_stream_impl == it->second);
         receive_stream_impl = it->second;
-        receive_rtp_config_.erase(it->first);
         it = video_receive_ssrcs_.erase(it);
       } else {
         ++it;
@@ -737,6 +697,7 @@ void Call::DestroyVideoReceiveStream(
     video_receive_streams_.erase(receive_stream_impl);
     RTC_CHECK(receive_stream_impl != nullptr);
     ConfigureSync(receive_stream_impl->config().sync_group);
+    rtp_transport_receive_->RemoveReceiver(receive_stream_impl);
   }
   const VideoReceiveStream::Config& config = receive_stream_impl->config();
 
@@ -759,22 +720,13 @@ FlexfecReceiveStream* Call::CreateFlexfecReceiveStream(
 
   {
     WriteLockScoped write_lock(*receive_crit_);
+    rtp_transport_receive_->AddReceiver(
+        config.remote_ssrc, MediaType::VIDEO,
+        RtpTransportControllerReceiveInterface::Config(), receive_stream);
 
-    RTC_DCHECK(flexfec_receive_streams_.find(receive_stream) ==
-               flexfec_receive_streams_.end());
-    flexfec_receive_streams_.insert(receive_stream);
-
-    for (auto ssrc : config.protected_media_ssrcs)
-      flexfec_receive_ssrcs_media_.insert(std::make_pair(ssrc, receive_stream));
-
-    RTC_DCHECK(flexfec_receive_ssrcs_protection_.find(config.remote_ssrc) ==
-               flexfec_receive_ssrcs_protection_.end());
-    flexfec_receive_ssrcs_protection_[config.remote_ssrc] = receive_stream;
-
-    RTC_DCHECK(receive_rtp_config_.find(config.remote_ssrc) ==
-               receive_rtp_config_.end());
-    receive_rtp_config_[config.remote_ssrc] =
-        ReceiveRtpConfig(config.rtp_header_extensions, UseSendSideBwe(config));
+    for (auto ssrc : config.protected_media_ssrcs) {
+      rtp_transport_receive_->AddSink(ssrc, MediaType::VIDEO, receive_stream);
+    }
   }
 
   // TODO(brandtr): Store config in RtcEventLog here.
@@ -793,33 +745,14 @@ void Call::DestroyFlexfecReceiveStream(FlexfecReceiveStream* receive_stream) {
       static_cast<FlexfecReceiveStreamImpl*>(receive_stream);
   {
     WriteLockScoped write_lock(*receive_crit_);
-
+    rtp_transport_receive_->RemoveSink(receive_stream_impl);
+    rtp_transport_receive_->RemoveReceiver(receive_stream_impl);
     const FlexfecReceiveStream::Config& config =
         receive_stream_impl->GetConfig();
     uint32_t ssrc = config.remote_ssrc;
-    receive_rtp_config_.erase(ssrc);
-
-    // Remove all SSRCs pointing to the FlexfecReceiveStreamImpl to be
-    // destroyed.
-    auto prot_it = flexfec_receive_ssrcs_protection_.begin();
-    while (prot_it != flexfec_receive_ssrcs_protection_.end()) {
-      if (prot_it->second == receive_stream_impl)
-        prot_it = flexfec_receive_ssrcs_protection_.erase(prot_it);
-      else
-        ++prot_it;
-    }
-    auto media_it = flexfec_receive_ssrcs_media_.begin();
-    while (media_it != flexfec_receive_ssrcs_media_.end()) {
-      if (media_it->second == receive_stream_impl)
-        media_it = flexfec_receive_ssrcs_media_.erase(media_it);
-      else
-        ++media_it;
-    }
 
     congestion_controller_->GetRemoteBitrateEstimator(UseSendSideBwe(config))
         ->RemoveStream(ssrc);
-
-    flexfec_receive_streams_.erase(receive_stream_impl);
   }
 
   delete receive_stream_impl;
@@ -1174,57 +1107,18 @@ PacketReceiver::DeliveryStatus Call::DeliverRtp(MediaType media_type,
                                                 const PacketTime& packet_time) {
   TRACE_EVENT0("webrtc", "Call::DeliverRtp");
 
-  ReadLockScoped read_lock(*receive_crit_);
-  // TODO(nisse): We should parse the RTP header only here, and pass
-  // on parsed_packet to the receive streams.
-  rtc::Optional<RtpPacketReceived> parsed_packet =
-      ParseRtpPacket(packet, length, packet_time);
-
-  if (!parsed_packet)
-    return DELIVERY_PACKET_ERROR;
-
-  NotifyBweOfReceivedPacket(*parsed_packet, media_type);
+  int64_t arrival_time_ms;
+  if (packet_time.timestamp != -1) {
+    arrival_time_ms = (packet_time.timestamp + 500) / 1000;
+  } else {
+    arrival_time_ms = clock_->TimeInMilliseconds();
+  }
 
-  uint32_t ssrc = parsed_packet->Ssrc();
+  ReadLockScoped read_lock(*receive_crit_);
 
-  if (media_type == MediaType::ANY || media_type == MediaType::AUDIO) {
-    auto it = audio_receive_ssrcs_.find(ssrc);
-    if (it != audio_receive_ssrcs_.end()) {
-      received_bytes_per_second_counter_.Add(static_cast<int>(length));
-      received_audio_bytes_per_second_counter_.Add(static_cast<int>(length));
-      it->second->OnRtpPacket(*parsed_packet);
-      event_log_->LogRtpHeader(kIncomingPacket, media_type, packet, length);
-      return DELIVERY_OK;
-    }
-  }
-  if (media_type == MediaType::ANY || media_type == MediaType::VIDEO) {
-    auto it = video_receive_ssrcs_.find(ssrc);
-    if (it != video_receive_ssrcs_.end()) {
-      received_bytes_per_second_counter_.Add(static_cast<int>(length));
-      received_video_bytes_per_second_counter_.Add(static_cast<int>(length));
-      it->second->OnRtpPacket(*parsed_packet);
-
-      // Deliver media packets to FlexFEC subsystem.
-      auto it_bounds = flexfec_receive_ssrcs_media_.equal_range(ssrc);
-      for (auto it = it_bounds.first; it != it_bounds.second; ++it)
-        it->second->OnRtpPacket(*parsed_packet);
-
-      event_log_->LogRtpHeader(kIncomingPacket, media_type, packet, length);
-      return DELIVERY_OK;
-    }
-  }
-  if (media_type == MediaType::ANY || media_type == MediaType::VIDEO) {
-    received_bytes_per_second_counter_.Add(static_cast<int>(length));
-    // TODO(brandtr): Update here when FlexFEC supports protecting audio.
-    received_video_bytes_per_second_counter_.Add(static_cast<int>(length));
-    auto it = flexfec_receive_ssrcs_protection_.find(ssrc);
-    if (it != flexfec_receive_ssrcs_protection_.end()) {
-      it->second->OnRtpPacket(*parsed_packet);
-      event_log_->LogRtpHeader(kIncomingPacket, media_type, packet, length);
-      return DELIVERY_OK;
-    }
-  }
-  return DELIVERY_UNKNOWN_SSRC;
+  return rtp_transport_receive_->OnRtpPacket(
+      media_type, arrival_time_ms,
+      rtc::ArrayView<const uint8_t>(packet, length));
 }
 
 PacketReceiver::DeliveryStatus Call::DeliverPacket(
@@ -1244,6 +1138,9 @@ PacketReceiver::DeliveryStatus Call::DeliverPacket(
 
 // TODO(brandtr): Update this member function when we support protecting
 // audio packets with FlexFEC.
+
+// TODO(nisse): Add a recovered flag to RtpParsedPacket, if needed for stats,
+// and demux recovered packets in the same way as ordinary packets.
 bool Call::OnRecoveredPacket(const uint8_t* packet, size_t length) {
   uint32_t ssrc = ByteReader<uint32_t>::ReadBigEndian(&packet[8]);
   ReadLockScoped read_lock(*receive_crit_);
@@ -1253,16 +1150,15 @@ bool Call::OnRecoveredPacket(const uint8_t* packet, size_t length) {
   return it->second->OnRecoveredPacket(packet, length);
 }
 
-void Call::NotifyBweOfReceivedPacket(const RtpPacketReceived& packet,
-                                     MediaType media_type) {
-  auto it = receive_rtp_config_.find(packet.Ssrc());
-  bool use_send_side_bwe =
-      (it != receive_rtp_config_.end()) && it->second.use_send_side_bwe;
-
+void Call::OnRtpPacket(
+    MediaType media_type,
+    const RtpTransportControllerReceiveInterface::Config config,
+    const RtpPacketReceived& packet) {
   RTPHeader header;
   packet.GetHeader(&header);
 
-  if (!use_send_side_bwe && header.extension.hasTransportSequenceNumber) {
+  if (!config.use_send_side_bwe &&
+      header.extension.hasTransportSequenceNumber) {
     // Inconsistent configuration of send side BWE. Do nothing.
     // TODO(nisse): Without this check, we may produce RTCP feedback
     // packets even when not negotiated. But it would be cleaner to
@@ -1277,7 +1173,8 @@ void Call::NotifyBweOfReceivedPacket(const RtpPacketReceived& packet,
   // FakeNetworkPipe::Process. We need to treat that as video. Tests
   // should be fixed to use the same MediaType as the production code.
   if (media_type != MediaType::AUDIO ||
-      (use_send_side_bwe && header.extension.hasTransportSequenceNumber)) {
+      (config.use_send_side_bwe &&
+       header.extension.hasTransportSequenceNumber)) {
     congestion_controller_->OnReceivedPacket(
         packet.arrival_time_ms(), packet.payload_size() + packet.padding_size(),
         header);