Introduce RtpStreamReceiver and RtpStreamReceiverControllerInterface.

webrtc/call/call.cc

Index: webrtc/call/call.cc
diff --git a/webrtc/call/call.cc b/webrtc/call/call.cc
index 802778eb4d6eda4cf55c642fb6f35aaa85074de5..526a5f081aa1c10dc29a0be5cc25d7595ee701b2 100644
--- a/webrtc/call/call.cc
+++ b/webrtc/call/call.cc
@@ -34,7 +34,7 @@
 #include "webrtc/call/bitrate_allocator.h"
 #include "webrtc/call/call.h"
 #include "webrtc/call/flexfec_receive_stream_impl.h"
-#include "webrtc/call/rtp_demuxer.h"
+#include "webrtc/call/rtp_transport_controller_receive.h"
 #include "webrtc/call/rtp_transport_controller_send.h"
 #include "webrtc/config.h"
 #include "webrtc/logging/rtc_event_log/rtc_event_log.h"
@@ -213,10 +213,12 @@ class Call : public webrtc::Call,
   std::map<std::string, AudioReceiveStream*> sync_stream_mapping_
       GUARDED_BY(receive_crit_);
 
-  // TODO(nisse): Should eventually be part of injected
-  // RtpTransportControllerReceive, with a single demuxer in the bundled case.
-  RtpDemuxer audio_rtp_demuxer_ GUARDED_BY(receive_crit_);
-  RtpDemuxer video_rtp_demuxer_ GUARDED_BY(receive_crit_);
+  // TODO(nisse): Should eventually be injected at creation,
+  // with a single object in the bundled case.
+  RtpTransportControllerReceive audio_transport_receive_
+      GUARDED_BY(receive_crit_);
+  RtpTransportControllerReceive video_transport_receive_
+      GUARDED_BY(receive_crit_);
 
   // This extra map is used for receive processing which is
   // independent of media type.
@@ -559,12 +561,17 @@ webrtc::AudioReceiveStream* Call::CreateAudioReceiveStream(
   TRACE_EVENT0("webrtc", "Call::CreateAudioReceiveStream");
   RTC_DCHECK(configuration_thread_checker_.CalledOnValidThread());
   event_log_->LogAudioReceiveStreamConfig(config);
-  AudioReceiveStream* receive_stream =
-      new AudioReceiveStream(transport_send_->packet_router(), config,
-                             config_.audio_state, event_log_);
+  AudioReceiveStream* receive_stream;
   {
     WriteLockScoped write_lock(*receive_crit_);
-    audio_rtp_demuxer_.AddSink(config.rtp.remote_ssrc, receive_stream);
+    // Constructor is expected to register with audio_transport_receive_,
+    // and hence needs to be called while holding the lock.
+
+    // TODO(nisse): Make RtpTransportControllerReceive thread safe. But it's
+    // unfortunate if we need additional locks for each packet processed.
+    receive_stream = new AudioReceiveStream(
+        &audio_transport_receive_, transport_send_->packet_router(), config,
+        config_.audio_state, event_log_);
     receive_rtp_config_[config.rtp.remote_ssrc] =
         ReceiveRtpConfig(config.rtp.extensions, UseSendSideBwe(config));
     audio_receive_streams_.insert(receive_stream);
@@ -596,8 +603,6 @@ void Call::DestroyAudioReceiveStream(
     uint32_t ssrc = config.rtp.remote_ssrc;
     receive_side_cc_.GetRemoteBitrateEstimator(UseSendSideBwe(config))
         ->RemoveStream(ssrc);
-    size_t num_deleted = audio_rtp_demuxer_.RemoveSink(audio_receive_stream);
-    RTC_DCHECK(num_deleted == 1);
     audio_receive_streams_.erase(audio_receive_stream);
     const std::string& sync_group = audio_receive_stream->config().sync_group;
     const auto it = sync_stream_mapping_.find(sync_group);
@@ -607,9 +612,11 @@ void Call::DestroyAudioReceiveStream(
       ConfigureSync(sync_group);
     }
     receive_rtp_config_.erase(ssrc);
+    // Destructor is expected to register with audio_transport_receive_,
+    // and hence needs to be called while holding the lock.
+    delete audio_receive_stream;
   }
   UpdateAggregateNetworkState();
-  delete audio_receive_stream;
 }
 
 webrtc::VideoSendStream* Call::CreateVideoSendStream(
@@ -685,32 +692,34 @@ webrtc::VideoReceiveStream* Call::CreateVideoReceiveStream(
   TRACE_EVENT0("webrtc", "Call::CreateVideoReceiveStream");
   RTC_DCHECK(configuration_thread_checker_.CalledOnValidThread());
 
-  VideoReceiveStream* receive_stream =
-      new VideoReceiveStream(num_cpu_cores_, transport_send_->packet_router(),
-                             std::move(configuration),
-                             module_process_thread_.get(), call_stats_.get());
-
-  const webrtc::VideoReceiveStream::Config& config = receive_stream->config();
-  ReceiveRtpConfig receive_config(config.rtp.extensions,
-                                  UseSendSideBwe(config));
+  VideoReceiveStream* receive_stream;
+  const webrtc::VideoReceiveStream::Config* config;
   {
     WriteLockScoped write_lock(*receive_crit_);
-    video_rtp_demuxer_.AddSink(config.rtp.remote_ssrc, receive_stream);
-    if (config.rtp.rtx_ssrc) {
-      video_rtp_demuxer_.AddSink(config.rtp.rtx_ssrc, receive_stream);
+    // Constructor is expected to register with audio_transport_receive_,
+    // and hence needs to be called while holding the lock.
+    receive_stream = new VideoReceiveStream(
+        &video_transport_receive_, num_cpu_cores_,
+        transport_send_->packet_router(), std::move(configuration),
+        module_process_thread_.get(), call_stats_.get());
+
+    config = &receive_stream->config();
+    ReceiveRtpConfig receive_config(config->rtp.extensions,
+                                    UseSendSideBwe(*config));
+    if (config->rtp.rtx_ssrc) {
       // We record identical config for the rtx stream as for the main
       // stream. Since the transport_send_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;
+      receive_rtp_config_[config->rtp.rtx_ssrc] = receive_config;
     }
-    receive_rtp_config_[config.rtp.remote_ssrc] = receive_config;
+    receive_rtp_config_[config->rtp.remote_ssrc] = receive_config;
     video_receive_streams_.insert(receive_stream);
-    ConfigureSync(config.sync_group);
+    ConfigureSync(config->sync_group);
   }
   receive_stream->SignalNetworkState(video_network_state_);
   UpdateAggregateNetworkState();
-  event_log_->LogVideoReceiveStreamConfig(config);
+  event_log_->LogVideoReceiveStreamConfig(*config);
   return receive_stream;
 }
 
@@ -726,21 +735,21 @@ void Call::DestroyVideoReceiveStream(
     WriteLockScoped write_lock(*receive_crit_);
     // Remove all ssrcs pointing to a receive stream. As RTX retransmits on a
     // separate SSRC there can be either one or two.
-    size_t num_deleted = video_rtp_demuxer_.RemoveSink(receive_stream_impl);
-    RTC_DCHECK_GE(num_deleted, 1);
     receive_rtp_config_.erase(config.rtp.remote_ssrc);
     if (config.rtp.rtx_ssrc) {
       receive_rtp_config_.erase(config.rtp.rtx_ssrc);
     }
     video_receive_streams_.erase(receive_stream_impl);
     ConfigureSync(config.sync_group);
-  }
 
-  receive_side_cc_.GetRemoteBitrateEstimator(UseSendSideBwe(config))
-      ->RemoveStream(config.rtp.remote_ssrc);
+    receive_side_cc_.GetRemoteBitrateEstimator(UseSendSideBwe(config))
+        ->RemoveStream(config.rtp.remote_ssrc);
 
+    // Destructor is expected to register with audio_transport_receive_,
+    // and hence needs to be called while holding the lock.
+    delete receive_stream_impl;
+  }
   UpdateAggregateNetworkState();
-  delete receive_stream_impl;
 }
 
 FlexfecReceiveStream* Call::CreateFlexfecReceiveStream(
@@ -749,16 +758,15 @@ FlexfecReceiveStream* Call::CreateFlexfecReceiveStream(
   RTC_DCHECK(configuration_thread_checker_.CalledOnValidThread());
 
   RecoveredPacketReceiver* recovered_packet_receiver = this;
-  FlexfecReceiveStreamImpl* receive_stream = new FlexfecReceiveStreamImpl(
-      config, recovered_packet_receiver, call_stats_->rtcp_rtt_stats(),
-      module_process_thread_.get());
 
+  FlexfecReceiveStreamImpl* receive_stream;
   {
     WriteLockScoped write_lock(*receive_crit_);
-    video_rtp_demuxer_.AddSink(config.remote_ssrc, receive_stream);
-
-    for (auto ssrc : config.protected_media_ssrcs)
-      video_rtp_demuxer_.AddSink(ssrc, receive_stream);
+    // Constructor is expected to register with audio_transport_receive_,
+    // and hence needs to be called while holding the lock.
+    receive_stream = new FlexfecReceiveStreamImpl(
+        &video_transport_receive_, config, recovered_packet_receiver,
+        call_stats_->rtcp_rtt_stats(), module_process_thread_.get());
 
     RTC_DCHECK(receive_rtp_config_.find(config.remote_ssrc) ==
                receive_rtp_config_.end());
@@ -790,12 +798,13 @@ void Call::DestroyFlexfecReceiveStream(FlexfecReceiveStream* receive_stream) {
 
     // Remove all SSRCs pointing to the FlexfecReceiveStreamImpl to be
     // destroyed.
-    video_rtp_demuxer_.RemoveSink(receive_stream_impl);
     receive_side_cc_.GetRemoteBitrateEstimator(UseSendSideBwe(config))
         ->RemoveStream(ssrc);
-  }
 
-  delete receive_stream_impl;
+    // Destructor is expected to register with audio_transport_receive_,
+    // and hence needs to be called while holding the lock.
+    delete receive_stream_impl;
+  }
 }
 
 Call::Stats Call::GetStats() const {
@@ -1162,14 +1171,14 @@ PacketReceiver::DeliveryStatus Call::DeliverRtp(MediaType media_type,
   NotifyBweOfReceivedPacket(*parsed_packet, media_type);
 
   if (media_type == MediaType::AUDIO) {
-    if (audio_rtp_demuxer_.OnRtpPacket(*parsed_packet)) {
+    if (audio_transport_receive_.OnRtpPacket(*parsed_packet)) {
       received_bytes_per_second_counter_.Add(static_cast<int>(length));
       received_audio_bytes_per_second_counter_.Add(static_cast<int>(length));
       event_log_->LogRtpHeader(kIncomingPacket, media_type, packet, length);
       return DELIVERY_OK;
     }
   } else if (media_type == MediaType::VIDEO) {
-    if (video_rtp_demuxer_.OnRtpPacket(*parsed_packet)) {
+    if (video_transport_receive_.OnRtpPacket(*parsed_packet)) {
       received_bytes_per_second_counter_.Add(static_cast<int>(length));
       received_video_bytes_per_second_counter_.Add(static_cast<int>(length));
       event_log_->LogRtpHeader(kIncomingPacket, media_type, packet, length);
@@ -1205,7 +1214,7 @@ void Call::OnRecoveredPacket(const uint8_t* packet, size_t length) {
 
   parsed_packet->set_recovered(true);
 
-  video_rtp_demuxer_.OnRtpPacket(*parsed_packet);
+  video_transport_receive_.OnRtpPacket(*parsed_packet);

[pthatcher1, 2017/05/17 23:24:53]

What is the advantage of working with the wrapper around the demuxer rather than
just working with the demuxer?

[nisse-webrtc, 2017/05/18 08:45:43]

At the moment, little. Responsibility for parsing should be moved to this
wrapper, but before that we need to think about responsibility for the
extensions mapping.

It would also make sense to let the wrapper object do the needed locking (while
the demuxer object itself should b a simple class unaware of multithreading
issues). Do you agree? 

[danilchap, 2017/05/19 15:37:43]

using ReceiverController feels preliminary: it should be able handle recovering
packets itself.

[nisse-webrtc, 2017/05/22 07:09:37]

Agreed. I think we should aim to move all per-packet processing out of Call, and
connect RtpTransportController directly to the packet transport. But there's
many steps to get there.

 }
 
 void Call::NotifyBweOfReceivedPacket(const RtpPacketReceived& packet,