Add RSID-based demuxing to RtpDemuxer

Make RtpDemuxer able to demux RTP packets according to RSID (RTP Stream ID), as well as the (pre-existing) ability to demux according to SSRC.

BUG=None

Review-Url: https://codereview.webrtc.org/2920993002
Cr-Commit-Position: refs/heads/master@{#18495}

webrtc/call/rtp_demuxer.cc

@@ -11,12 +11,15 @@
 #include "webrtc/base/checks.h"
 #include "webrtc/call/rtp_demuxer.h"
 #include "webrtc/call/rtp_packet_sink_interface.h"
+#include "webrtc/modules/rtp_rtcp/source/rtp_header_extensions.h"
 #include "webrtc/modules/rtp_rtcp/source/rtp_packet_received.h"
 
 namespace webrtc {
 
 namespace {
 
+constexpr size_t kMaxProcessedSsrcs = 1000;  // Prevent memory overuse.
+
 template <typename Key, typename Value>
 bool MultimapAssociationExists(const std::multimap<Key, Value>& multimap,
                                Key key,
@@ -27,6 +30,21 @@ bool MultimapAssociationExists(const std::multimap<Key, Value>& multimap,
                      [val](Reference elem) { return elem.second == val; });
 }
 
+template <typename Key, typename Value>
+size_t RemoveFromMultimapByValue(std::multimap<Key, Value*>* multimap,
+                                 const Value* value) {
+  size_t count = 0;
+  for (auto it = multimap->begin(); it != multimap->end();) {
+    if (it->second == value) {
+      it = multimap->erase(it);
+      ++count;
+    } else {
+      ++it;
+    }
+  }
+  return count;
+}
+
 }  // namespace
 
 RtpDemuxer::RtpDemuxer() {}
@@ -36,33 +54,71 @@ RtpDemuxer::~RtpDemuxer() {
 }
 
 void RtpDemuxer::AddSink(uint32_t ssrc, RtpPacketSinkInterface* sink) {
-  RTC_DCHECK(sink);
-  RTC_DCHECK(!MultimapAssociationExists(sinks_, ssrc, sink));
-  sinks_.emplace(ssrc, sink);
+  RecordSsrcToSinkAssociation(ssrc, sink);
 }
 
-size_t RtpDemuxer::RemoveSink(const RtpPacketSinkInterface* sink) {
+void RtpDemuxer::AddSink(const std::string& rsid,
+                         RtpPacketSinkInterface* sink) {
+  RTC_DCHECK(StreamId::IsLegalName(rsid));
   RTC_DCHECK(sink);
-  size_t count = 0;
-  for (auto it = sinks_.begin(); it != sinks_.end(); ) {
-    if (it->second == sink) {
-      it = sinks_.erase(it);
-      ++count;
-    } else {
-      ++it;
-    }
+  RTC_DCHECK(!MultimapAssociationExists(rsid_sinks_, rsid, sink));
+
+  rsid_sinks_.emplace(rsid, sink);
+
+  // This RSID might now map to an SSRC which we saw earlier.
+  processed_ssrcs_.clear();
+}
+
+bool RtpDemuxer::RemoveSink(const RtpPacketSinkInterface* sink) {
+  RTC_DCHECK(sink);
+  return (RemoveFromMultimapByValue(&sinks_, sink) +
+          RemoveFromMultimapByValue(&rsid_sinks_, sink)) > 0;
+}
+
+void RtpDemuxer::RecordSsrcToSinkAssociation(uint32_t ssrc,
+                                             RtpPacketSinkInterface* sink) {
+  RTC_DCHECK(sink);
+  // The association might already have been set by a different
+  // configuration source.
+  if (!MultimapAssociationExists(sinks_, ssrc, sink)) {
+    sinks_.emplace(ssrc, sink);
   }
-  return count;
 }
 
 bool RtpDemuxer::OnRtpPacket(const RtpPacketReceived& packet) {
-  bool found = false;
+  FindSsrcAssociations(packet);
   auto it_range = sinks_.equal_range(packet.Ssrc());
   for (auto it = it_range.first; it != it_range.second; ++it) {
-    found = true;
     it->second->OnRtpPacket(packet);
   }
-  return found;
+  return it_range.first != it_range.second;
+}
+
+void RtpDemuxer::FindSsrcAssociations(const RtpPacketReceived& packet) {
+  // Avoid expensive string comparisons for RSID by looking the sinks up only
+  // by SSRC whenever possible.
+  if (processed_ssrcs_.find(packet.Ssrc()) != processed_ssrcs_.cend()) {
+    return;
+  }
+
+  // RSID-based associations:
+  std::string rsid;
+  if (packet.GetExtension<RtpStreamId>(&rsid)) {
+    // All streams associated with this RSID need to be marked as associated
+    // with this SSRC (if they aren't already).
+    auto it_range = rsid_sinks_.equal_range(rsid);
+    for (auto it = it_range.first; it != it_range.second; ++it) {
+      RecordSsrcToSinkAssociation(packet.Ssrc(), it->second);
+    }
+
+    // To prevent memory-overuse attacks, forget this RSID. Future packets
+    // with this RSID, but a different SSRC, will not spawn new associations.
+    rsid_sinks_.erase(it_range.first, it_range.second);
+  }
+
+  if (processed_ssrcs_.size() < kMaxProcessedSsrcs) {  // Prevent memory overuse
+    processed_ssrcs_.insert(packet.Ssrc());  // Avoid re-examining in-depth.
+  }
 }
 
 }  // namespace webrtc

webrtc/call/rtp_demuxer.h

@@ -7,10 +7,13 @@
  *  in the file PATENTS.  All contributing project authors may
  *  be found in the AUTHORS file in the root of the source tree.
  */
+
 #ifndef WEBRTC_CALL_RTP_DEMUXER_H_
 #define WEBRTC_CALL_RTP_DEMUXER_H_
 
 #include <map>
+#include <set>
+#include <string>
 
 namespace webrtc {
 
@@ -30,15 +33,43 @@ class RtpDemuxer {
   // Registers a sink. The same sink can be registered for multiple ssrcs, and
   // the same ssrc can have multiple sinks. Null pointer is not allowed.
   void AddSink(uint32_t ssrc, RtpPacketSinkInterface* sink);
-  // Removes a sink. Returns deletion count (a sink may be registered
-  // for multiple ssrcs). Null pointer is not allowed.
-  size_t RemoveSink(const RtpPacketSinkInterface* sink);
+
+  // Registers a sink's association to an RSID. Null pointer is not allowed.
+  void AddSink(const std::string& rsid, RtpPacketSinkInterface* sink);
+
+  // Removes a sink. Return value reports if anything was actually removed.
+  // Null pointer is not allowed.
+  bool RemoveSink(const RtpPacketSinkInterface* sink);
 
   // Returns true if at least one matching sink was found, otherwise false.
   bool OnRtpPacket(const RtpPacketReceived& packet);
 
  private:
+  // Records a sink<->SSRC association. This can happen by explicit
+  // configuration by AddSink(ssrc...), or by inferred configuration from an
+  // RSID-based configuration which is resolved to an SSRC upon
+  // packet reception.
+  void RecordSsrcToSinkAssociation(uint32_t ssrc, RtpPacketSinkInterface* sink);
+
+  // When a new packet arrives, we attempt to resolve extra associations,
+  // such as which RSIDs are associated with which SSRCs.
+  void FindSsrcAssociations(const RtpPacketReceived& packet);
+
+  // This records the association SSRCs to sinks. Other associations, such
+  // as by RSID, also end up here once the RSID, etc., is resolved to an SSRC.
   std::multimap<uint32_t, RtpPacketSinkInterface*> sinks_;
+
+  // A sink may be associated with an RSID - RTP Stream ID. This tag has a
+  // one-to-one association with an SSRC, but that SSRC is not yet known.
+  // When it becomes known, the association of the sink to the RSID is deleted
+  // from this container, and moved into |sinks_|.
+  std::multimap<std::string, RtpPacketSinkInterface*> rsid_sinks_;
+
+  // Iterating over |rsid_sinks_| for each incoming and performing multiple
+  // string comparisons is of non-trivial cost. To avoid this cost, we only
+  // check RSIDs for the first packet on each incoming SSRC stream.
+  // (If RSID associations are added later, we check again.)
+  std::set<uint32_t> processed_ssrcs_;
 };
 
 }  // namespace webrtc

webrtc/call/rtp_demuxer_unittest.cc

@@ -11,11 +11,14 @@
 #include "webrtc/call/rtp_demuxer.h"
 
 #include <memory>
+#include <string>
 
 #include "webrtc/base/arraysize.h"
 #include "webrtc/base/checks.h"
 #include "webrtc/base/ptr_util.h"
 #include "webrtc/call/rtp_packet_sink_interface.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"
 #include "webrtc/test/gmock.h"
 #include "webrtc/test/gtest.h"
@@ -25,122 +28,471 @@ namespace webrtc {
 namespace {
 
 using ::testing::_;
+using ::testing::AtLeast;
+using ::testing::InSequence;
+using ::testing::NiceMock;
 
 class MockRtpPacketSink : public RtpPacketSinkInterface {
  public:
   MOCK_METHOD1(OnRtpPacket, void(const RtpPacketReceived&));
 };
 
-constexpr uint32_t kSsrcs[] = {101, 202, 303};
-
-MATCHER_P(SsrcSameAsIn, other, "") {
-  return arg.Ssrc() == other.Ssrc();
+MATCHER_P(SamePacketAs, other, "") {
+  return arg.Ssrc() == other.Ssrc() &&
+         arg.SequenceNumber() == other.SequenceNumber();
 }
 
-std::unique_ptr<RtpPacketReceived> CreateRtpPacketReceived(uint32_t ssrc) {
+std::unique_ptr<RtpPacketReceived> CreateRtpPacketReceived(
+    uint32_t ssrc,
+    size_t sequence_number = 0) {
+  // |sequence_number| is declared |size_t| to prevent ugly casts when calling
+  // the function, but should in reality always be a |uint16_t|.
+  EXPECT_LT(sequence_number, 1u << 16);
+
   auto packet = rtc::MakeUnique<RtpPacketReceived>();
   packet->SetSsrc(ssrc);
+  packet->SetSequenceNumber(static_cast<uint16_t>(sequence_number));
   return packet;
 }
 
-class RtpDemuxerTest : public ::testing::Test {
- protected:
-  RtpDemuxerTest() {
-    for (size_t i = 0; i < arraysize(sinks); i++) {
-      demuxer.AddSink(kSsrcs[i], &sinks[i]);
-    }
-  }
+std::unique_ptr<RtpPacketReceived> CreateRtpPacketReceivedWithRsid(
+    const std::string& rsid,
+    uint32_t ssrc,
+    size_t sequence_number = 0) {
+  // |sequence_number| is declared |size_t| to prevent ugly casts when calling
+  // the function, but should in reality always be a |uint16_t|.
+  EXPECT_LT(sequence_number, 1u << 16);
 
-  ~RtpDemuxerTest() override {
-    for (auto& sink : sinks) {
-      EXPECT_EQ(demuxer.RemoveSink(&sink), 1u);
-    }
-  }
+  const int rsid_extension_id = 6;
+  RtpPacketReceived::ExtensionManager extension_manager;
+  extension_manager.Register<RtpStreamId>(rsid_extension_id);
+  auto packet = rtc::MakeUnique<RtpPacketReceived>(&extension_manager);
+  packet->SetExtension<RtpStreamId>(rsid);
+  packet->SetSsrc(ssrc);
+  packet->SetSequenceNumber(static_cast<uint16_t>(sequence_number));
+  return packet;
+}
 
+TEST(RtpDemuxerTest, OnRtpPacketCalledOnCorrectSinkBySsrc) {
   RtpDemuxer demuxer;
-  MockRtpPacketSink sinks[arraysize(kSsrcs)];
-};
 
-TEST_F(RtpDemuxerTest, OnRtpPacketCalledOnCorrectSink) {
-  for (size_t i = 0; i < arraysize(sinks); i++) {
-    auto packet = CreateRtpPacketReceived(kSsrcs[i]);
-    EXPECT_CALL(sinks[i], OnRtpPacket(SsrcSameAsIn(*packet)));
-    demuxer.OnRtpPacket(*packet);
+  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]);
+  }
+
+  for (size_t i = 0; i < arraysize(ssrcs); i++) {
+    auto packet = CreateRtpPacketReceived(ssrcs[i]);
+    EXPECT_CALL(sinks[i], OnRtpPacket(SamePacketAs(*packet))).Times(1);
+    EXPECT_TRUE(demuxer.OnRtpPacket(*packet));
+  }
+
+  // Test tear-down
+  for (const auto& sink : sinks) {
+    demuxer.RemoveSink(&sink);
   }
 }
 
-TEST_F(RtpDemuxerTest, MultipleSinksMappedToSameSsrc) {
-  // |sinks| associated with different SSRCs each. Add a few additional sinks
-  // that are all associated with one new, distinct SSRC.
-  MockRtpPacketSink same_ssrc_sinks[arraysize(sinks)];
-  constexpr uint32_t kSharedSsrc = 404;
-  for (auto& sink : same_ssrc_sinks) {
-    demuxer.AddSink(kSharedSsrc, &sink);
+TEST(RtpDemuxerTest, OnRtpPacketCalledOnCorrectSinkByRsid) {
+  RtpDemuxer demuxer;
+
+  const std::string rsids[] = {"a", "b", "c"};
+  MockRtpPacketSink sinks[arraysize(rsids)];
+  for (size_t i = 0; i < arraysize(rsids); i++) {
+    demuxer.AddSink(rsids[i], &sinks[i]);
+  }
+
+  for (size_t i = 0; i < arraysize(rsids); i++) {
+    auto packet =
+        CreateRtpPacketReceivedWithRsid(rsids[i], static_cast<uint32_t>(i), i);
+    EXPECT_CALL(sinks[i], OnRtpPacket(SamePacketAs(*packet))).Times(1);
+    EXPECT_TRUE(demuxer.OnRtpPacket(*packet));
+  }
+
+  // Test tear-down
+  for (const auto& sink : sinks) {
+    demuxer.RemoveSink(&sink);
+  }
+}
+
+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]);
+  }
+
+  std::unique_ptr<RtpPacketReceived> packets[5];
+  for (size_t i = 0; i < arraysize(packets); i++) {
+    packets[i] = CreateRtpPacketReceived(ssrcs[0], i);
+  }
+
+  InSequence sequence;
+  for (const auto& packet : packets) {
+    EXPECT_CALL(sinks[0], OnRtpPacket(SamePacketAs(*packet))).Times(1);
+  }
+
+  for (const auto& packet : packets) {
+    EXPECT_TRUE(demuxer.OnRtpPacket(*packet));
+  }
+
+  // Test tear-down
+  for (const auto& sink : sinks) {
+    demuxer.RemoveSink(&sink);
+  }
+}
+
+TEST(RtpDemuxerTest, MultipleSinksMappedToSameSsrc) {
+  RtpDemuxer demuxer;
+
+  MockRtpPacketSink sinks[3];
+  constexpr uint32_t ssrc = 404;
+  for (auto& sink : sinks) {
+    demuxer.AddSink(ssrc, &sink);
   }
 
   // Reception of an RTP packet associated with the shared SSRC triggers the
-  // callback on all of the interfaces associated with it.
-  auto packet = CreateRtpPacketReceived(kSharedSsrc);
-  for (auto& sink : same_ssrc_sinks) {
-    EXPECT_CALL(sink, OnRtpPacket(SsrcSameAsIn(*packet)));
+  // callback on all of the sinks associated with it.
+  auto packet = CreateRtpPacketReceived(ssrc);
+  for (auto& sink : sinks) {
+    EXPECT_CALL(sink, OnRtpPacket(SamePacketAs(*packet)));
   }
-  demuxer.OnRtpPacket(*packet);
+  EXPECT_TRUE(demuxer.OnRtpPacket(*packet));
 
-  // Test-specific tear-down
-  for (auto& sink : same_ssrc_sinks) {
-    EXPECT_EQ(demuxer.RemoveSink(&sink), 1u);
+  // Test tear-down
+  for (const auto& sink : sinks) {
+    demuxer.RemoveSink(&sink);
   }
 }
 
-TEST_F(RtpDemuxerTest, SinkMappedToMultipleSsrcs) {
-  // |sinks| associated with different SSRCs each. We set one of them to also
-  // be mapped to additional SSRCs.
-  constexpr uint32_t kSsrcsOfMultiSsrcSink[] = {404, 505, 606};
-  MockRtpPacketSink multi_ssrc_sink;
-  for (uint32_t ssrc : kSsrcsOfMultiSsrcSink) {
-    demuxer.AddSink(ssrc, &multi_ssrc_sink);
+TEST(RtpDemuxerTest, SinkMappedToMultipleSsrcs) {
+  RtpDemuxer demuxer;
+
+  constexpr uint32_t ssrcs[] = {404, 505, 606};
+  MockRtpPacketSink sink;
+  for (uint32_t ssrc : ssrcs) {
+    demuxer.AddSink(ssrc, &sink);
   }
 
   // The sink which is associated with multiple SSRCs gets the callback
   // triggered for each of those SSRCs.
-  for (uint32_t ssrc : kSsrcsOfMultiSsrcSink) {
+  for (uint32_t ssrc : ssrcs) {
     auto packet = CreateRtpPacketReceived(ssrc);
-    EXPECT_CALL(multi_ssrc_sink, OnRtpPacket(SsrcSameAsIn(*packet)));
-    demuxer.OnRtpPacket(*packet);
+    EXPECT_CALL(sink, OnRtpPacket(SamePacketAs(*packet)));
+    EXPECT_TRUE(demuxer.OnRtpPacket(*packet));
   }
 
-  // Test-specific tear-down
-  EXPECT_EQ(demuxer.RemoveSink(&multi_ssrc_sink),
-            arraysize(kSsrcsOfMultiSsrcSink));
+  // Test tear-down
+  demuxer.RemoveSink(&sink);
 }
 
-TEST_F(RtpDemuxerTest, OnRtpPacketNotCalledOnRemovedSinks) {
-  // |sinks| associated with different SSRCs each. We set one of them to also
-  // be mapped to additional SSRCs.
-  constexpr uint32_t kSsrcsOfMultiSsrcSink[] = {404, 505, 606};
-  MockRtpPacketSink multi_ssrc_sink;
-  for (uint32_t ssrc : kSsrcsOfMultiSsrcSink) {
-    demuxer.AddSink(ssrc, &multi_ssrc_sink);
+TEST(RtpDemuxerTest, NoCallbackOnSsrcSinkRemovedBeforeFirstPacket) {
+  RtpDemuxer demuxer;
+
+  constexpr uint32_t ssrc = 404;
+  MockRtpPacketSink sink;
+  demuxer.AddSink(ssrc, &sink);
+
+  ASSERT_TRUE(demuxer.RemoveSink(&sink));
+
+  // The removed sink does not get callbacks.
+  auto packet = CreateRtpPacketReceived(ssrc);
+  EXPECT_CALL(sink, OnRtpPacket(_)).Times(0);  // Not called.
+  EXPECT_FALSE(demuxer.OnRtpPacket(*packet));
+}
+
+TEST(RtpDemuxerTest, NoCallbackOnSsrcSinkRemovedAfterFirstPacket) {
+  RtpDemuxer demuxer;
+
+  constexpr uint32_t ssrc = 404;
+  NiceMock<MockRtpPacketSink> sink;
+  demuxer.AddSink(ssrc, &sink);
+
+  InSequence sequence;
+  uint16_t seq_num;
+  for (seq_num = 0; seq_num < 10; seq_num++) {
+    ASSERT_TRUE(demuxer.OnRtpPacket(*CreateRtpPacketReceived(ssrc, seq_num)));
   }
 
-  // Remove the sink.
-  EXPECT_EQ(demuxer.RemoveSink(&multi_ssrc_sink),
-            arraysize(kSsrcsOfMultiSsrcSink));
+  ASSERT_TRUE(demuxer.RemoveSink(&sink));
 
-  // The removed sink does not get callbacks triggered for any of the SSRCs
-  // with which it was previously associated.
-  EXPECT_CALL(multi_ssrc_sink, OnRtpPacket(_)).Times(0);
-  for (uint32_t ssrc : kSsrcsOfMultiSsrcSink) {
-    auto packet = CreateRtpPacketReceived(ssrc);
-    demuxer.OnRtpPacket(*packet);
+  // The removed sink does not get callbacks.
+  auto packet = CreateRtpPacketReceived(ssrc, seq_num);
+  EXPECT_CALL(sink, OnRtpPacket(_)).Times(0);  // Not called.
+  EXPECT_FALSE(demuxer.OnRtpPacket(*packet));
+}
+
+TEST(RtpDemuxerTest, RepeatedSsrcAssociationsDoNotTriggerRepeatedCallbacks) {
+  RtpDemuxer demuxer;
+
+  constexpr uint32_t ssrc = 111;
+  MockRtpPacketSink sink;
+
+  demuxer.AddSink(ssrc, &sink);
+  demuxer.AddSink(ssrc, &sink);
+
+  auto packet = CreateRtpPacketReceived(ssrc);
+  EXPECT_CALL(sink, OnRtpPacket(SamePacketAs(*packet))).Times(1);
+  EXPECT_TRUE(demuxer.OnRtpPacket(*packet));
+
+  // Test tear-down
+  demuxer.RemoveSink(&sink);
+}
+
+TEST(RtpDemuxerTest, RemoveSinkReturnsFalseForNeverAddedSink) {
+  RtpDemuxer demuxer;
+  MockRtpPacketSink sink;
+
+  EXPECT_FALSE(demuxer.RemoveSink(&sink));
+}
+
+TEST(RtpDemuxerTest, RemoveSinkReturnsTrueForPreviouslyAddedSsrcSink) {
+  RtpDemuxer demuxer;
+
+  constexpr uint32_t ssrc = 101;
+  MockRtpPacketSink sink;
+  demuxer.AddSink(ssrc, &sink);
+
+  EXPECT_TRUE(demuxer.RemoveSink(&sink));
+}
+
+TEST(RtpDemuxerTest,
+     RemoveSinkReturnsTrueForUnresolvedPreviouslyAddedRsidSink) {
+  RtpDemuxer demuxer;
+
+  const std::string rsid = "a";
+  MockRtpPacketSink sink;
+  demuxer.AddSink(rsid, &sink);
+
+  EXPECT_TRUE(demuxer.RemoveSink(&sink));
+}
+
+TEST(RtpDemuxerTest, RemoveSinkReturnsTrueForResolvedPreviouslyAddedRsidSink) {
+  RtpDemuxer demuxer;
+
+  const std::string rsid = "a";
+  constexpr uint32_t ssrc = 101;
+  NiceMock<MockRtpPacketSink> sink;
+  demuxer.AddSink(rsid, &sink);
+  ASSERT_TRUE(
+      demuxer.OnRtpPacket(*CreateRtpPacketReceivedWithRsid(rsid, ssrc)));
+
+  EXPECT_TRUE(demuxer.RemoveSink(&sink));
+}
+
+TEST(RtpDemuxerTest, OnRtpPacketCalledForRsidSink) {
+  RtpDemuxer demuxer;
+
+  MockRtpPacketSink sink;
+  const std::string rsid = "a";
+  demuxer.AddSink(rsid, &sink);
+
+  // Create a sequence of RTP packets, where only the first one actually
+  // mentions the RSID.
+  std::unique_ptr<RtpPacketReceived> packets[5];
+  constexpr uint32_t rsid_ssrc = 111;
+  packets[0] = CreateRtpPacketReceivedWithRsid(rsid, rsid_ssrc);
+  for (size_t i = 1; i < arraysize(packets); i++) {
+    packets[i] = CreateRtpPacketReceived(rsid_ssrc, i);
   }
+
+  // The first packet associates the RSID with the SSRC, thereby allowing the
+  // demuxer to correctly demux all of the packets.
+  InSequence sequence;
+  for (const auto& packet : packets) {
+    EXPECT_CALL(sink, OnRtpPacket(SamePacketAs(*packet))).Times(1);
+  }
+  for (const auto& packet : packets) {
+    EXPECT_TRUE(demuxer.OnRtpPacket(*packet));
+  }
+
+  // Test tear-down
+  demuxer.RemoveSink(&sink);
+}
+
+TEST(RtpDemuxerTest, NoCallbackOnRsidSinkRemovedBeforeFirstPacket) {
+  RtpDemuxer demuxer;
+
+  MockRtpPacketSink sink;
+  const std::string rsid = "a";
+  demuxer.AddSink(rsid, &sink);
+
+  // Sink removed - it won't get triggers even if packets with its RSID arrive.
+  ASSERT_TRUE(demuxer.RemoveSink(&sink));
+
+  constexpr uint32_t ssrc = 111;
+  auto packet = CreateRtpPacketReceivedWithRsid(rsid, ssrc);
+  EXPECT_CALL(sink, OnRtpPacket(_)).Times(0);  // Not called.
+  EXPECT_FALSE(demuxer.OnRtpPacket(*packet));
+}
+
+TEST(RtpDemuxerTest, NoCallbackOnRsidSinkRemovedAfterFirstPacket) {
+  RtpDemuxer demuxer;
+
+  NiceMock<MockRtpPacketSink> sink;
+  const std::string rsid = "a";
+  demuxer.AddSink(rsid, &sink);
+
+  InSequence sequence;
+  constexpr uint32_t ssrc = 111;
+  uint16_t seq_num;
+  for (seq_num = 0; seq_num < 10; seq_num++) {
+    auto packet = CreateRtpPacketReceivedWithRsid(rsid, ssrc, seq_num);
+    ASSERT_TRUE(demuxer.OnRtpPacket(*packet));
+  }
+
+  // Sink removed - it won't get triggers even if packets with its RSID arrive.
+  ASSERT_TRUE(demuxer.RemoveSink(&sink));
+
+  auto packet = CreateRtpPacketReceivedWithRsid(rsid, ssrc, seq_num);
+  EXPECT_CALL(sink, OnRtpPacket(_)).Times(0);  // Not called.
+  EXPECT_FALSE(demuxer.OnRtpPacket(*packet));
+}
+
+// The RSID to SSRC mapping should be one-to-one. If we end up receiving
+// two (or more) packets with the same SSRC, but different RSIDs, we guarantee
+// remembering the first one; no guarantees are made about further associations.
+TEST(RtpDemuxerTest, FirstSsrcAssociatedWithAnRsidIsNotForgotten) {
+  RtpDemuxer demuxer;
+
+  // Each sink has a distinct RSID.
+  MockRtpPacketSink sink_a;
+  const std::string rsid_a = "a";
+  demuxer.AddSink(rsid_a, &sink_a);
+
+  MockRtpPacketSink sink_b;
+  const std::string rsid_b = "b";
+  demuxer.AddSink(rsid_b, &sink_b);
+
+  InSequence sequence;  // Verify that the order of delivery is unchanged.
+
+  constexpr uint32_t shared_ssrc = 100;
+
+  // First a packet with |rsid_a| is received, and |sink_a| is associated with
+  // its SSRC.
+  auto packet_a = CreateRtpPacketReceivedWithRsid(rsid_a, shared_ssrc, 10);
+  EXPECT_CALL(sink_a, OnRtpPacket(SamePacketAs(*packet_a))).Times(1);
+  EXPECT_TRUE(demuxer.OnRtpPacket(*packet_a));
+
+  // Second, a packet with |rsid_b| is received. Its RSID is ignored.
+  auto packet_b = CreateRtpPacketReceivedWithRsid(rsid_b, shared_ssrc, 20);
+  EXPECT_CALL(sink_a, OnRtpPacket(SamePacketAs(*packet_b))).Times(1);
+  EXPECT_TRUE(demuxer.OnRtpPacket(*packet_b));
+
+  // Known edge-case; adding a new RSID association makes us re-examine all
+  // SSRCs. |sink_b| may or may not be associated with the SSRC now; we make
+  // no promises on that. We do however still guarantee that |sink_a| still
+  // receives the new packets.
+  MockRtpPacketSink sink_ignored;
+  demuxer.AddSink("ignored", &sink_ignored);
+  auto packet_c = CreateRtpPacketReceivedWithRsid(rsid_b, shared_ssrc, 30);
+  EXPECT_CALL(sink_a, OnRtpPacket(SamePacketAs(*packet_c))).Times(1);
+  EXPECT_CALL(sink_b, OnRtpPacket(SamePacketAs(*packet_c))).Times(AtLeast(0));
+  EXPECT_TRUE(demuxer.OnRtpPacket(*packet_c));
+
+  // Test tear-down
+  demuxer.RemoveSink(&sink_a);
+  demuxer.RemoveSink(&sink_b);
+  demuxer.RemoveSink(&sink_ignored);
+}
+
+TEST(RtpDemuxerTest, MultipleRsidsOnSameSink) {
+  RtpDemuxer demuxer;
+
+  MockRtpPacketSink sink;
+  const std::string rsids[] = {"a", "b", "c"};
+
+  for (const std::string& rsid : rsids) {
+    demuxer.AddSink(rsid, &sink);
+  }
+
+  InSequence sequence;
+  for (size_t i = 0; i < arraysize(rsids); i++) {
+    // Assign different SSRCs and sequence numbers to all packets.
+    const uint32_t ssrc = 1000 + static_cast<uint32_t>(i);
+    const uint16_t sequence_number = 50 + static_cast<uint16_t>(i);
+    auto packet =
+        CreateRtpPacketReceivedWithRsid(rsids[i], ssrc, sequence_number);
+    EXPECT_CALL(sink, OnRtpPacket(SamePacketAs(*packet))).Times(1);
+    EXPECT_TRUE(demuxer.OnRtpPacket(*packet));
+  }
+
+  // Test tear-down
+  demuxer.RemoveSink(&sink);
+}
+
+TEST(RtpDemuxerTest, SinkWithBothRsidAndSsrcAssociations) {
+  RtpDemuxer demuxer;
+
+  MockRtpPacketSink sink;
+  constexpr uint32_t standalone_ssrc = 10101;
+  constexpr uint32_t rsid_ssrc = 20202;
+  const std::string rsid = "a";
+
+  demuxer.AddSink(standalone_ssrc, &sink);
+  demuxer.AddSink(rsid, &sink);
+
+  InSequence sequence;
+
+  auto ssrc_packet = CreateRtpPacketReceived(standalone_ssrc, 11);
+  EXPECT_CALL(sink, OnRtpPacket(SamePacketAs(*ssrc_packet))).Times(1);
+  EXPECT_TRUE(demuxer.OnRtpPacket(*ssrc_packet));
+
+  auto rsid_packet = CreateRtpPacketReceivedWithRsid(rsid, rsid_ssrc, 22);
+  EXPECT_CALL(sink, OnRtpPacket(SamePacketAs(*rsid_packet))).Times(1);
+  EXPECT_TRUE(demuxer.OnRtpPacket(*rsid_packet));
+
+  // Test tear-down
+  demuxer.RemoveSink(&sink);
+}
+
+TEST(RtpDemuxerTest, AssociatingByRsidAndBySsrcCannotTriggerDoubleCall) {
+  RtpDemuxer demuxer;
+  MockRtpPacketSink sink;
+
+  constexpr uint32_t ssrc = 10101;
+  demuxer.AddSink(ssrc, &sink);
+
+  const std::string rsid = "a";
+  demuxer.AddSink(rsid, &sink);
+
+  constexpr uint16_t seq_num = 999;
+  auto packet = CreateRtpPacketReceivedWithRsid(rsid, ssrc, seq_num);
+  EXPECT_CALL(sink, OnRtpPacket(SamePacketAs(*packet))).Times(1);
+  EXPECT_TRUE(demuxer.OnRtpPacket(*packet));
+
+  // Test tear-down
+  demuxer.RemoveSink(&sink);
 }
 
 #if RTC_DCHECK_IS_ON && GTEST_HAS_DEATH_TEST && !defined(WEBRTC_ANDROID)
-TEST_F(RtpDemuxerTest, RepeatedAssociationsForbidden) {
-  // Set-up already associated sinks[0] with kSsrcs[0]. Repeating the
-  // association is an error.
-  EXPECT_DEATH(demuxer.AddSink(kSsrcs[0], &sinks[0]), "");
+TEST(RtpDemuxerTest, RsidMustBeNonEmpty) {
+  RtpDemuxer demuxer;
+  MockRtpPacketSink sink;
+  EXPECT_DEATH(demuxer.AddSink("", &sink), "");
+}
+
+TEST(RtpDemuxerTest, RsidMustBeAlphaNumeric) {
+  RtpDemuxer demuxer;
+  MockRtpPacketSink sink;
+  EXPECT_DEATH(demuxer.AddSink("a_3", &sink), "");
+}
+
+TEST(RtpDemuxerTest, RsidMustNotExceedMaximumLength) {
+  RtpDemuxer demuxer;
+  MockRtpPacketSink sink;
+  std::string rsid(StreamId::kMaxSize + 1, 'a');
+  EXPECT_DEATH(demuxer.AddSink(rsid, &sink), "");
+}
+
+TEST(RtpDemuxerTest, RepeatedRsidAssociationsDisallowed) {
+  RtpDemuxer demuxer;
+  MockRtpPacketSink sink;
+  demuxer.AddSink("a", &sink);
+  EXPECT_DEATH(demuxer.AddSink("a", &sink), "");
 }
 #endif
 

webrtc/common_types.cc

@@ -11,6 +11,7 @@
 #include "webrtc/common_types.h"
 
 #include <string.h>
+#include <algorithm>
 #include <limits>
 #include <type_traits>
 
@@ -23,6 +24,11 @@ StreamDataCounters::StreamDataCounters() : first_packet_time_ms(-1) {}
 
 constexpr size_t StreamId::kMaxSize;
 
+bool StreamId::IsLegalName(rtc::ArrayView<const char> name) {
+  return (name.size() <= kMaxSize && name.size() > 0 &&
+          std::all_of(name.data(), name.data() + name.size(), isalnum));
+}
+
 void StreamId::Set(const char* data, size_t size) {
   // If |data| contains \0, the stream id size might become less than |size|.
   RTC_CHECK_LE(size, kMaxSize);

webrtc/common_types.h

@@ -707,6 +707,8 @@ class StreamId {
   // that can be encoded with one-byte header extensions.
   static constexpr size_t kMaxSize = 16;
 
+  static bool IsLegalName(rtc::ArrayView<const char> name);
+
   StreamId() { value_[0] = 0; }
   explicit StreamId(rtc::ArrayView<const char> value) {
     Set(value.data(), value.size());