Create RtcpDemuxer

webrtc/call/rtp_demuxer_unittest.cc

Index: webrtc/call/rtp_demuxer_unittest.cc
diff --git a/webrtc/call/rtp_demuxer_unittest.cc b/webrtc/call/rtp_demuxer_unittest.cc
index 8ca827c94bd78d3e242d8d04c53ffa389eab7246..4d22c51d374406387f909f784ba12d8d1cc71f9f 100644
--- a/webrtc/call/rtp_demuxer_unittest.cc
+++ b/webrtc/call/rtp_demuxer_unittest.cc
@@ -16,7 +16,9 @@
 #include "webrtc/base/arraysize.h"
 #include "webrtc/base/checks.h"
 #include "webrtc/base/ptr_util.h"
+#include "webrtc/call/rsid_resolution_observer.h"
 #include "webrtc/call/rtp_packet_sink_interface.h"
+#include "webrtc/common_types.h"
 #include "webrtc/modules/rtp_rtcp/include/rtp_header_extension_map.h"
 #include "webrtc/modules/rtp_rtcp/source/rtp_header_extensions.h"
 #include "webrtc/modules/rtp_rtcp/source/rtp_packet_received.h"
@@ -37,6 +39,11 @@ class MockRtpPacketSink : public RtpPacketSinkInterface {
   MOCK_METHOD1(OnRtpPacket, void(const RtpPacketReceived&));
 };
 
+class MockRsidResolutionObserver : public RsidResolutionObserver {
+ public:
+  MOCK_METHOD2(OnRsidResolved, void(const std::string& rsid, uint32_t ssrc));
+};
+
 MATCHER_P(SamePacketAs, other, "") {
   return arg.Ssrc() == other.Ssrc() &&
          arg.SequenceNumber() == other.SequenceNumber();
@@ -119,20 +126,18 @@ TEST(RtpDemuxerTest, OnRtpPacketCalledOnCorrectSinkByRsid) {
 TEST(RtpDemuxerTest, PacketsDeliveredInRightOrder) {
   RtpDemuxer demuxer;
 
-  constexpr uint32_t ssrcs[] = {101, 202, 303};
-  MockRtpPacketSink sinks[arraysize(ssrcs)];
-  for (size_t i = 0; i < arraysize(ssrcs); i++) {
-    demuxer.AddSink(ssrcs[i], &sinks[i]);
-  }
+  constexpr uint32_t ssrc = 101;
+  MockRtpPacketSink sink;
+  demuxer.AddSink(ssrc, &sink);
 
   std::unique_ptr<RtpPacketReceived> packets[5];
   for (size_t i = 0; i < arraysize(packets); i++) {
-    packets[i] = CreateRtpPacketReceived(ssrcs[0], i);
+    packets[i] = CreateRtpPacketReceived(ssrc, i);
   }
 
   InSequence sequence;
   for (const auto& packet : packets) {
-    EXPECT_CALL(sinks[0], OnRtpPacket(SamePacketAs(*packet))).Times(1);
+    EXPECT_CALL(sink, OnRtpPacket(SamePacketAs(*packet))).Times(1);
   }
 
   for (const auto& packet : packets) {
@@ -140,9 +145,7 @@ TEST(RtpDemuxerTest, PacketsDeliveredInRightOrder) {
   }
 
   // Test tear-down
-  for (const auto& sink : sinks) {
-    demuxer.RemoveSink(&sink);
-  }
+  demuxer.RemoveSink(&sink);
 }
 
 TEST(RtpDemuxerTest, MultipleSinksMappedToSameSsrc) {
@@ -424,6 +427,31 @@ TEST(RtpDemuxerTest, MultipleRsidsOnSameSink) {
   demuxer.RemoveSink(&sink);
 }
 
+TEST(RtpDemuxerTest, RsidUsedByMultipleSinks) {
+  RtpDemuxer demuxer;
+
+  MockRtpPacketSink sinks[3];
+  const std::string shared_rsid = "a";
+
+  for (MockRtpPacketSink& sink : sinks) {
+    demuxer.AddSink(shared_rsid, &sink);
+  }
+
+  constexpr uint32_t shared_ssrc = 888;
+  auto packet = CreateRtpPacketReceivedWithRsid(shared_rsid, shared_ssrc);
+
+  for (auto& sink : sinks) {
+    EXPECT_CALL(sink, OnRtpPacket(SamePacketAs(*packet))).Times(1);
+  }
+
+  EXPECT_TRUE(demuxer.OnRtpPacket(*packet));
+
+  // Test tear-down
+  for (MockRtpPacketSink& sink : sinks) {
+    demuxer.RemoveSink(&sink);
+  }
+}
+
 TEST(RtpDemuxerTest, SinkWithBothRsidAndSsrcAssociations) {
   RtpDemuxer demuxer;
 
@@ -468,6 +496,88 @@ TEST(RtpDemuxerTest, AssociatingByRsidAndBySsrcCannotTriggerDoubleCall) {
   demuxer.RemoveSink(&sink);
 }
 
+TEST(RtpDemuxerTest, RsidObserversInformedOfResolutions) {
+  RtpDemuxer demuxer;
+
+  constexpr uint32_t ssrc = 111;
+  const std::string rsid = "a";
+
+  MockRsidResolutionObserver rsid_resolution_observers[3];
+  for (auto& observer : rsid_resolution_observers) {
+    demuxer.RegisterRsidResolutionObserver(&observer);
+    EXPECT_CALL(observer, OnRsidResolved(rsid, ssrc)).Times(1);
+  }
+
+  // The expected calls to OnRsidResolved() will be triggered by this.
+  demuxer.OnRtpPacket(*CreateRtpPacketReceivedWithRsid(rsid, ssrc));
+
+  // Test tear-down
+  for (auto& observer : rsid_resolution_observers) {
+    demuxer.DeregisterRsidResolutionObserver(&observer);
+  }
+}
+
+// Normally, we only produce one notification per resolution (though no such
+// guarantee is made), but when a new observer is added, we reset
+// this suppression - we "re-resolve" associations for the benefit of the
+// new observer..
+TEST(RtpDemuxerTest, NotificationSuppressionResetWhenNewObserverAdded) {

[danilchap, 2017/06/19 13:31:56]

do you have any (practical) scenario where you want to rely on this feature?

[eladalon, 2017/06/19 14:35:17]

The line between practical and theoretical is blurry. Here are some
possibilities:
1. RTP packets arriving before negotiation is complete? A callee might be
sending RTP messages right after it's produced the ANSWER, before the caller has
managed to get it (maybe it got lost, maybe delayed). 
2. The Proxy-like class we've discussed for PlType-based demuxing.

Besides, I feel things are more correct if we choose either:
* Disallow further observer registration after RTP demuxing has started.
* Give new observers the full service they expect.
The second seems both cleaner as well as more useful.

+  RtpDemuxer demuxer;
+
+  constexpr uint32_t ssrc = 111;
+  const std::string rsid = "a";
+
+  // First observer registered, then gets a notification.
+  NiceMock<MockRsidResolutionObserver> first_observer;
+  demuxer.RegisterRsidResolutionObserver(&first_observer);
+  demuxer.OnRtpPacket(*CreateRtpPacketReceivedWithRsid(rsid, ssrc));
+
+  // Second observer registered, then gets a notification. No guarantee is made
+  // about whether the first observer would get an additional notification.
+  MockRsidResolutionObserver second_observer;
+  demuxer.RegisterRsidResolutionObserver(&second_observer);
+  EXPECT_CALL(first_observer, OnRsidResolved(rsid, ssrc)).Times(AtLeast(0));
+  EXPECT_CALL(second_observer, OnRsidResolved(rsid, ssrc)).Times(1);
+  demuxer.OnRtpPacket(*CreateRtpPacketReceivedWithRsid(rsid, ssrc));
+
+  // Test tear-down
+  demuxer.DeregisterRsidResolutionObserver(&first_observer);
+  demuxer.DeregisterRsidResolutionObserver(&second_observer);
+}
+
+TEST(RtpDemuxerTest, DeregisteredRsidObserversNotInformedOfResolutions) {
+  RtpDemuxer demuxer;
+
+  constexpr uint32_t ssrc = 111;
+  const std::string rsid = "a";
+  NiceMock<MockRtpPacketSink> sink;
+  demuxer.AddSink(rsid, &sink);
+
+  // Register several, then deregister only one, to show that not all of the
+  // observers had been forgotten when one was removed.
+  MockRsidResolutionObserver observer_1;
+  MockRsidResolutionObserver observer_2_removed;
+  MockRsidResolutionObserver observer_3;
+
+  demuxer.RegisterRsidResolutionObserver(&observer_1);
+  demuxer.RegisterRsidResolutionObserver(&observer_2_removed);
+  demuxer.RegisterRsidResolutionObserver(&observer_3);
+
+  demuxer.DeregisterRsidResolutionObserver(&observer_2_removed);
+
+  EXPECT_CALL(observer_1, OnRsidResolved(rsid, ssrc)).Times(1);
+  EXPECT_CALL(observer_2_removed, OnRsidResolved(_, _)).Times(0);
+  EXPECT_CALL(observer_3, OnRsidResolved(rsid, ssrc)).Times(1);
+
+  // The expected calls to OnRsidResolved() will be triggered by this.
+  demuxer.OnRtpPacket(*CreateRtpPacketReceivedWithRsid(rsid, ssrc));
+
+  // Test tear-down
+  demuxer.RemoveSink(&sink);
+  demuxer.DeregisterRsidResolutionObserver(&observer_1);
+  demuxer.DeregisterRsidResolutionObserver(&observer_3);
+}
+
 #if RTC_DCHECK_IS_ON && GTEST_HAS_DEATH_TEST && !defined(WEBRTC_ANDROID)
 TEST(RtpDemuxerTest, RsidMustBeNonEmpty) {
   RtpDemuxer demuxer;
@@ -493,7 +603,25 @@ TEST(RtpDemuxerTest, RepeatedRsidAssociationsDisallowed) {
   MockRtpPacketSink sink;
   demuxer.AddSink("a", &sink);
   EXPECT_DEATH(demuxer.AddSink("a", &sink), "");
+  demuxer.RemoveSink(&sink);
+}
+
+TEST(RtpDemuxerTest,
+     DoubleRegisterationOfNeverRegisteredRsidResolutionObserverDisallowed) {
+  RtpDemuxer demuxer;
+  MockRsidResolutionObserver observer;
+  demuxer.RegisterRsidResolutionObserver(&observer);
+  EXPECT_DEATH(demuxer.RegisterRsidResolutionObserver(&observer), "");
+  demuxer.DeregisterRsidResolutionObserver(&observer);
 }
+
+TEST(RtpDemuxerTest,
+     DregisterationOfNeverRegisteredRsidResolutionObserverDisallowed) {
+  RtpDemuxer demuxer;
+  MockRsidResolutionObserver observer;
+  EXPECT_DEATH(demuxer.DeregisterRsidResolutionObserver(&observer), "");
+}
+
 #endif
 
 }  // namespace