Ensure Call is notified of un demuxable packets

With this cl, packets that are discarded in RtpTransport now notifies Call, so that
they can be part of BWE even if they are dropped.
These packets have been recevied on the transport, and has bin decrypted
and parsed and thus can be accounted for.

The un demuxable packets are forwarded to Call similarly how RTCP packets are forwarded.

Bug: webrtc:14928
Change-Id: Ia53349c7b316c4442a3c7aac085a85ec4f4ab9ae
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/299262
Reviewed-by: Henrik Boström <hbos@webrtc.org>
Commit-Queue: Per Kjellander <perkj@webrtc.org>
Cr-Commit-Position: refs/heads/main@{#39727}

pc/BUILD.gn

@@ -309,6 +309,7 @@ rtc_source_set("jsep_transport_controller") {
   ]
   absl_deps = [
     "//third_party/abseil-cpp/absl/algorithm:container",
+    "//third_party/abseil-cpp/absl/functional:any_invocable",
     "//third_party/abseil-cpp/absl/types:optional",
   ]
 }

pc/jsep_transport_controller.cc

@@ -53,7 +53,7 @@ JsepTransportController::JsepTransportController(
             RTC_DCHECK_RUN_ON(network_thread_);
             UpdateAggregateStates_n();
           }),
-      config_(config),
+      config_(std::move(config)),
       active_reset_srtp_params_(config.active_reset_srtp_params),
       bundles_(config.bundle_policy) {
   // The `transport_observer` is assumed to be non-null.
@@ -1090,6 +1090,8 @@ RTCError JsepTransportController::MaybeCreateJsepTransport(
 
   jsep_transport->rtp_transport()->SignalRtcpPacketReceived.connect(
       this, &JsepTransportController::OnRtcpPacketReceived_n);
+  jsep_transport->rtp_transport()->SignalUnDemuxableRtpPacketReceived.connect(
+      this, &JsepTransportController::OnUnDemuxableRtpPacketReceived_n);
 
   transports_.RegisterTransport(content_info.name, std::move(jsep_transport));
   UpdateAggregateStates_n();
@@ -1410,6 +1412,12 @@ void JsepTransportController::OnRtcpPacketReceived_n(
   config_.rtcp_handler(*packet, packet_time_us);
 }
 
+void JsepTransportController::OnUnDemuxableRtpPacketReceived_n(
+    const webrtc::RtpPacketReceived& packet) {
+  RTC_DCHECK(config_.un_demuxable_packet_handler);
+  config_.un_demuxable_packet_handler(packet);
+}
+
 void JsepTransportController::OnDtlsHandshakeError(
     rtc::SSLHandshakeError error) {
   config_.on_dtls_handshake_error_(error);

pc/jsep_transport_controller.h

@@ -21,6 +21,7 @@
 #include <utility>
 #include <vector>
 
+#include "absl/functional/any_invocable.h"
 #include "absl/types/optional.h"
 #include "api/async_dns_resolver.h"
 #include "api/candidate.h"
@@ -124,9 +125,11 @@ class JsepTransportController : public sigslot::has_slots<> {
     Observer* transport_observer = nullptr;
     // Must be provided and valid for the lifetime of the
     // JsepTransportController instance.
-    std::function<void(const rtc::CopyOnWriteBuffer& packet,
-                       int64_t packet_time_us)>
+    absl::AnyInvocable<void(const rtc::CopyOnWriteBuffer& packet,
+                            int64_t packet_time_us) const>
         rtcp_handler;
+    absl::AnyInvocable<void(const RtpPacketReceived& parsed_packet) const>
+        un_demuxable_packet_handler;
     // Initial value for whether DtlsTransport reset causes a reset
     // of SRTP parameters.
     bool active_reset_srtp_params = false;
@@ -450,6 +453,8 @@ class JsepTransportController : public sigslot::has_slots<> {
   void OnRtcpPacketReceived_n(rtc::CopyOnWriteBuffer* packet,
                               int64_t packet_time_us)
       RTC_RUN_ON(network_thread_);
+  void OnUnDemuxableRtpPacketReceived_n(const webrtc::RtpPacketReceived& packet)
+      RTC_RUN_ON(network_thread_);
 
   void OnDtlsHandshakeError(rtc::SSLHandshakeError error);
 

pc/jsep_transport_controller_unittest.cc

@@ -103,7 +103,7 @@ class JsepTransportControllerTest : public JsepTransportController::Observer,
     config.field_trials = &field_trials_;
     transport_controller_ = std::make_unique<JsepTransportController>(
         network_thread, port_allocator, nullptr /* async_resolver_factory */,
-        config);
+        std::move(config));
     SendTask(network_thread, [&] { ConnectTransportControllerSignals(); });
   }
 
@@ -400,7 +400,7 @@ TEST_F(JsepTransportControllerTest, GetRtpTransport) {
 TEST_F(JsepTransportControllerTest, GetDtlsTransport) {
   JsepTransportController::Config config;
   config.rtcp_mux_policy = PeerConnectionInterface::kRtcpMuxPolicyNegotiate;
-  CreateJsepTransportController(config);
+  CreateJsepTransportController(std::move(config));
   auto description = CreateSessionDescriptionWithoutBundle();
   EXPECT_TRUE(transport_controller_
                   ->SetLocalDescription(SdpType::kOffer, description.get())
@@ -435,7 +435,7 @@ TEST_F(JsepTransportControllerTest, GetDtlsTransport) {
 TEST_F(JsepTransportControllerTest, GetDtlsTransportWithRtcpMux) {
   JsepTransportController::Config config;
   config.rtcp_mux_policy = PeerConnectionInterface::kRtcpMuxPolicyRequire;
-  CreateJsepTransportController(config);
+  CreateJsepTransportController(std::move(config));
   auto description = CreateSessionDescriptionWithoutBundle();
   EXPECT_TRUE(transport_controller_
                   ->SetLocalDescription(SdpType::kOffer, description.get())
@@ -987,7 +987,7 @@ TEST_F(JsepTransportControllerTest, IceRoleNotRedetermined) {
   JsepTransportController::Config config;
   config.redetermine_role_on_ice_restart = false;
 
-  CreateJsepTransportController(config);
+  CreateJsepTransportController(std::move(config));
   // Let the `transport_controller_` be the controlled side initially.
   auto remote_offer = std::make_unique<cricket::SessionDescription>();
   AddAudioSection(remote_offer.get(), kAudioMid1, kIceUfrag1, kIcePwd1,
@@ -2317,7 +2317,7 @@ TEST_F(JsepTransportControllerTest, RejectFirstContentInBundleGroup) {
 TEST_F(JsepTransportControllerTest, ApplyNonRtcpMuxOfferWhenMuxingRequired) {
   JsepTransportController::Config config;
   config.rtcp_mux_policy = PeerConnectionInterface::kRtcpMuxPolicyRequire;
-  CreateJsepTransportController(config);
+  CreateJsepTransportController(std::move(config));
   auto local_offer = std::make_unique<cricket::SessionDescription>();
   AddAudioSection(local_offer.get(), kAudioMid1, kIceUfrag1, kIcePwd1,
                   cricket::ICEMODE_FULL, cricket::CONNECTIONROLE_ACTPASS,
@@ -2335,7 +2335,7 @@ TEST_F(JsepTransportControllerTest, ApplyNonRtcpMuxOfferWhenMuxingRequired) {
 TEST_F(JsepTransportControllerTest, ApplyNonRtcpMuxAnswerWhenMuxingRequired) {
   JsepTransportController::Config config;
   config.rtcp_mux_policy = PeerConnectionInterface::kRtcpMuxPolicyRequire;
-  CreateJsepTransportController(config);
+  CreateJsepTransportController(std::move(config));
   auto local_offer = std::make_unique<cricket::SessionDescription>();
   AddAudioSection(local_offer.get(), kAudioMid1, kIceUfrag1, kIcePwd1,
                   cricket::ICEMODE_FULL, cricket::CONNECTIONROLE_ACTPASS,

pc/peer_connection.cc

@@ -24,6 +24,7 @@
 #include "absl/strings/string_view.h"
 #include "absl/types/optional.h"
 #include "api/jsep_ice_candidate.h"
+#include "api/media_types.h"
 #include "api/rtp_parameters.h"
 #include "api/rtp_transceiver_direction.h"
 #include "api/uma_metrics.h"
@@ -699,6 +700,7 @@ JsepTransportController* PeerConnection::InitializeTransportController_n(
                               : options_.crypto_options;
   config.transport_observer = this;
   config.rtcp_handler = InitializeRtcpCallback();
+  config.un_demuxable_packet_handler = InitializeUnDemuxablePacketHandler();
   config.event_log = event_log_ptr_;
 #if defined(ENABLE_EXTERNAL_AUTH)
   config.enable_external_auth = true;
@@ -720,9 +722,9 @@ JsepTransportController* PeerConnection::InitializeTransportController_n(
 
   config.field_trials = trials_.get();
 
-  transport_controller_.reset(
-      new JsepTransportController(network_thread(), port_allocator_.get(),
-                                  async_dns_resolver_factory_.get(), config));
+  transport_controller_.reset(new JsepTransportController(
+      network_thread(), port_allocator_.get(),
+      async_dns_resolver_factory_.get(), std::move(config)));
 
   transport_controller_->SubscribeIceConnectionState(
       [this](cricket::IceConnectionState s) {
@@ -2991,4 +2993,21 @@ PeerConnection::InitializeRtcpCallback() {
   };
 }
 
+std::function<void(const RtpPacketReceived& parsed_packet)>
+PeerConnection::InitializeUnDemuxablePacketHandler() {
+  RTC_DCHECK_RUN_ON(network_thread());
+  return [this](const RtpPacketReceived& parsed_packet) {
+    worker_thread()->PostTask(
+        SafeTask(worker_thread_safety_, [this, parsed_packet]() {
+          // Deliver the packet anyway to Call to allow Call to do BWE.
+          // Even if there is no media receiver, the packet has still
+          // been received on the network and has been correcly parsed.
+          call_ptr_->Receiver()->DeliverRtpPacket(
+              MediaType::ANY, parsed_packet,
+              /*undemuxable_packet_handler=*/
+              [](const RtpPacketReceived& packet) { return false; });
+        }));
+  };
+}
+
 }  // namespace webrtc

pc/peer_connection.h

@@ -51,6 +51,7 @@
 #include "api/transport/enums.h"
 #include "api/turn_customizer.h"
 #include "call/call.h"
+#include "modules/rtp_rtcp/source/rtp_packet_received.h"
 #include "p2p/base/ice_transport_internal.h"
 #include "p2p/base/port.h"
 #include "p2p/base/port_allocator.h"
@@ -600,6 +601,9 @@ class PeerConnection : public PeerConnectionInternal,
                      int64_t packet_time_us)>
   InitializeRtcpCallback();
 
+  std::function<void(const RtpPacketReceived& parsed_packet)>
+  InitializeUnDemuxablePacketHandler();
+
   const rtc::scoped_refptr<ConnectionContext> context_;
   // Field trials active for this PeerConnection is the first of:
   // a) Specified in PeerConnectionDependencies (owned).

pc/rtp_transport.cc

@@ -193,8 +193,9 @@ void RtpTransport::DemuxPacket(rtc::CopyOnWriteBuffer packet,
   }
 
   if (!rtp_demuxer_.OnRtpPacket(parsed_packet)) {
-    RTC_LOG(LS_WARNING) << "Failed to demux RTP packet: "
+    RTC_LOG(LS_VERBOSE) << "Failed to demux RTP packet: "
                         << RtpDemuxer::DescribePacket(parsed_packet);
+    SignalUnDemuxableRtpPacketReceived(parsed_packet);
   }
 }
 

pc/rtp_transport_internal.h

@@ -53,10 +53,15 @@ class RtpTransportInternal : public sigslot::has_slots<> {
   sigslot::signal1<bool> SignalReadyToSend;
 
   // Called whenever an RTCP packet is received. There is no equivalent signal
-  // for RTP packets because they would be forwarded to the BaseChannel through
-  // the RtpDemuxer callback.
+  // for demuxable RTP packets because they would be forwarded to the
+  // BaseChannel through the RtpDemuxer callback.
   sigslot::signal2<rtc::CopyOnWriteBuffer*, int64_t> SignalRtcpPacketReceived;
 
+  // Called whenever a RTP packet that can not be demuxed by the transport is
+  // received.
+  sigslot::signal<const webrtc::RtpPacketReceived&>
+      SignalUnDemuxableRtpPacketReceived;
+
   // Called whenever the network route of the P2P layer transport changes.
   // The argument is an optional network route.
   sigslot::signal1<absl::optional<rtc::NetworkRoute>> SignalNetworkRouteChanged;

pc/rtp_transport_unittest.cc

@@ -288,6 +288,7 @@ TEST(RtpTransportTest, SignalHandledRtpPayloadType) {
   rtc::Buffer rtp_data(kRtpData, kRtpLen);
   fake_rtp.SendPacket(rtp_data.data<char>(), kRtpLen, options, flags);
   EXPECT_EQ(1, observer.rtp_count());
+  EXPECT_EQ(0, observer.un_demuxable_rtp_count());
   EXPECT_EQ(0, observer.rtcp_count());
   // Remove the sink before destroying the transport.
   transport.UnregisterRtpDemuxerSink(&observer);
@@ -311,6 +312,7 @@ TEST(RtpTransportTest, DontSignalUnhandledRtpPayloadType) {
   rtc::Buffer rtp_data(kRtpData, kRtpLen);
   fake_rtp.SendPacket(rtp_data.data<char>(), kRtpLen, options, flags);
   EXPECT_EQ(0, observer.rtp_count());
+  EXPECT_EQ(1, observer.un_demuxable_rtp_count());
   EXPECT_EQ(0, observer.rtcp_count());
   // Remove the sink before destroying the transport.
   transport.UnregisterRtpDemuxerSink(&observer);

pc/test/rtp_transport_test_util.h

@@ -30,6 +30,8 @@ class TransportObserver : public RtpPacketSinkInterface,
         this, &TransportObserver::OnRtcpPacketReceived);
     rtp_transport->SignalReadyToSend.connect(this,
                                              &TransportObserver::OnReadyToSend);
+    rtp_transport->SignalUnDemuxableRtpPacketReceived.connect(
+        this, &TransportObserver::OnUndemuxableRtpPacket);
   }
 
   // RtpPacketInterface override.
@@ -38,6 +40,10 @@ class TransportObserver : public RtpPacketSinkInterface,
     last_recv_rtp_packet_ = packet.Buffer();
   }
 
+  void OnUndemuxableRtpPacket(const RtpPacketReceived& packet) {
+    un_demuxable_rtp_count_++;
+  }
+
   void OnRtcpPacketReceived(rtc::CopyOnWriteBuffer* packet,
                             int64_t packet_time_us) {
     rtcp_count_++;
@@ -45,6 +51,7 @@ class TransportObserver : public RtpPacketSinkInterface,
   }
 
   int rtp_count() const { return rtp_count_; }
+  int un_demuxable_rtp_count() const { return un_demuxable_rtp_count_; }
   int rtcp_count() const { return rtcp_count_; }
 
   rtc::CopyOnWriteBuffer last_recv_rtp_packet() {
@@ -67,6 +74,7 @@ class TransportObserver : public RtpPacketSinkInterface,
  private:
   bool ready_to_send_ = false;
   int rtp_count_ = 0;
+  int un_demuxable_rtp_count_ = 0;
   int rtcp_count_ = 0;
   int ready_to_send_signal_count_ = 0;
   rtc::CopyOnWriteBuffer last_recv_rtp_packet_;