admin/webrtc_m130
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement PacketReceiver::DeliverRtpPacket in FakeNetworkPipe

Browse Source 
and in DegradedCall.  In DegradedCall - ThreadPacketReceiver is no longer needed.

Implementation of DeliverRtpPacket is done in preparation of https://webrtc-review.googlesource.com/c/src/+/290540, where the parsed packet will be propagated to Call without extra parsing.

Bug: webrtc:7135, webrtc:14795
Change-Id: Ic068105d6d1f337afc6b4539b0e7184e736e7ee0
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/290704
Commit-Queue: Per Kjellander <perkj@webrtc.org>
Reviewed-by: Erik Språng <sprang@webrtc.org>
Cr-Commit-Position: refs/heads/main@{#39048}
This commit is contained in:
Per K 2023-01-09 21:44:56 +01:00 committed by WebRTC LUCI CQ
parent 353a5ce7e3
commit bc319027ae
6 changed files with 184 additions and 120 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
call/BUILD.gn
View File

@ -444,6 +444,8 @@ rtc_library("fake_network") {
"../api:sequence_checker",
"../api:simulated_network_api",
"../api:transport_api",
"../api/units:timestamp",
"../modules/rtp_rtcp:rtp_rtcp_format",
"../rtc_base:checks",
"../rtc_base:logging",
"../rtc_base:macromagic",
@ -672,6 +674,8 @@ if (rtc_include_tests) {
"../api:simulated_network_api",
"../api/units:data_rate",
"../api/units:time_delta",
"../modules/rtp_rtcp:rtp_rtcp_format",
"../rtc_base:checks",
"../system_wrappers",
"../test:test_support",
"//testing/gtest",

66
call/degraded_call.cc
View File

@ -14,6 +14,7 @@
#include <utility>
#include "absl/strings/string_view.h"
#include "api/sequence_checker.h"
#include "modules/rtp_rtcp/source/rtp_util.h"
#include "rtc_base/event.h"
@ -128,54 +129,6 @@ bool DegradedCall::FakeNetworkPipeTransportAdapter::SendRtcp(
return true;
}
DegradedCall::ThreadedPacketReceiver::ThreadedPacketReceiver(
webrtc::TaskQueueBase* worker_thread,
webrtc::TaskQueueBase* network_thread,
rtc::scoped_refptr<PendingTaskSafetyFlag> call_alive,
webrtc::PacketReceiver* receiver)
: worker_thread_(worker_thread),
network_thread_(network_thread),
call_alive_(std::move(call_alive)),
receiver_(receiver) {}
DegradedCall::ThreadedPacketReceiver::~ThreadedPacketReceiver() = default;
PacketReceiver::DeliveryStatus
DegradedCall::ThreadedPacketReceiver::DeliverPacket(
MediaType media_type,
rtc::CopyOnWriteBuffer packet,
int64_t packet_time_us) {
// `Call::DeliverPacket` expects RTCP packets to be delivered from the
// network thread and RTP packets to be delivered from the worker thread.
// Because `FakeNetworkPipe` queues packets, the thread used when this packet
// is delivered to `DegradedCall::DeliverPacket` may differ from the thread
// used when this packet is delivered to
// `ThreadedPacketReceiver::DeliverPacket`. To solve this problem, always
// make sure that packets are sent in the correct thread.
if (IsRtcpPacket(packet)) {
if (!network_thread_->IsCurrent()) {
network_thread_->PostTask(
SafeTask(call_alive_, [receiver = receiver_, media_type,
packet = std::move(packet), packet_time_us]() {
receiver->DeliverPacket(media_type, std::move(packet),
packet_time_us);
}));
return DELIVERY_OK;
}
} else {
if (!worker_thread_->IsCurrent()) {
worker_thread_->PostTask([receiver = receiver_, media_type,
packet = std::move(packet), packet_time_us]() {
receiver->DeliverPacket(media_type, std::move(packet), packet_time_us);
});
return DELIVERY_OK;
}
}
return receiver_->DeliverPacket(media_type, std::move(packet),
packet_time_us);
}
DegradedCall::DegradedCall(
std::unique_ptr<Call> call,
const std::vector<TimeScopedNetworkConfig>& send_configs,
@ -193,10 +146,7 @@ DegradedCall::DegradedCall(
receive_simulated_network_ = network.get();
receive_pipe_ =
std::make_unique<webrtc::FakeNetworkPipe>(clock_, std::move(network));
packet_receiver_ = std::make_unique<ThreadedPacketReceiver>(
call_->worker_thread(), call_->network_thread(), call_alive_,
call_->Receiver());
receive_pipe_->SetReceiver(packet_receiver_.get());
receive_pipe_->SetReceiver(call_->Receiver());
if (receive_configs_.size() > 1) {
call_->network_thread()->PostDelayedTask(
SafeTask(call_alive_, [this] { UpdateReceiveNetworkConfig(); }),
@ -400,6 +350,7 @@ PacketReceiver::DeliveryStatus DegradedCall::DeliverPacket(
MediaType media_type,
rtc::CopyOnWriteBuffer packet,
int64_t packet_time_us) {
RTC_DCHECK_RUN_ON(&received_packet_sequence_checker_);
PacketReceiver::DeliveryStatus status = receive_pipe_->DeliverPacket(
media_type, std::move(packet), packet_time_us);
// This is not optimal, but there are many places where there are thread
@ -414,7 +365,18 @@ PacketReceiver::DeliveryStatus DegradedCall::DeliverPacket(
return status;
}
void DegradedCall::DeliverRtpPacket(
MediaType media_type,
RtpPacketReceived packet,
OnUndemuxablePacketHandler undemuxable_packet_handler) {
RTC_DCHECK_RUN_ON(&received_packet_sequence_checker_);
receive_pipe_->DeliverRtpPacket(media_type, std::move(packet),
std::move(undemuxable_packet_handler));
receive_pipe_->Process();
}
void DegradedCall::DeliverRtcpPacket(rtc::CopyOnWriteBuffer packet) {
RTC_DCHECK_RUN_ON(&received_packet_sequence_checker_);
receive_pipe_->DeliverRtcpPacket(std::move(packet));
receive_pipe_->Process();
}

28
call/degraded_call.h
View File

@ -116,6 +116,10 @@ class DegradedCall : public Call, private PacketReceiver {
DeliveryStatus DeliverPacket(MediaType media_type,
rtc::CopyOnWriteBuffer packet,
int64_t packet_time_us) override;
void DeliverRtpPacket(
MediaType media_type,
RtpPacketReceived packet,
OnUndemuxablePacketHandler undemuxable_packet_handler) override;
void DeliverRtcpPacket(rtc::CopyOnWriteBuffer packet) override;
private:
@ -148,25 +152,6 @@ class DegradedCall : public Call, private PacketReceiver {
absl::optional<int64_t> next_process_ms_ RTC_GUARDED_BY(&task_queue_);
};
class ThreadedPacketReceiver : public PacketReceiver {
public:
ThreadedPacketReceiver(webrtc::TaskQueueBase* worker_thread,
webrtc::TaskQueueBase* network_thread,
rtc::scoped_refptr<PendingTaskSafetyFlag> call_alive,
PacketReceiver* receiver);
~ThreadedPacketReceiver() override;
DeliveryStatus DeliverPacket(MediaType media_type,
rtc::CopyOnWriteBuffer packet,
int64_t packet_time_us) override;
private:
webrtc::TaskQueueBase* const worker_thread_;
webrtc::TaskQueueBase* const network_thread_;
rtc::scoped_refptr<PendingTaskSafetyFlag> call_alive_;
webrtc::PacketReceiver* const receiver_;
};
// For audio/video send stream, a TransportAdapter instance is used to
// intercept packets to be sent, and put them into a common FakeNetworkPipe
// in such as way that they will eventually (unless dropped) be forwarded to
@ -212,8 +197,9 @@ class DegradedCall : public Call, private PacketReceiver {
size_t receive_config_index_;
const std::vector<TimeScopedNetworkConfig> receive_configs_;
SimulatedNetwork* receive_simulated_network_;
std::unique_ptr<FakeNetworkPipe> receive_pipe_;
std::unique_ptr<ThreadedPacketReceiver> packet_receiver_;
SequenceChecker received_packet_sequence_checker_;
std::unique_ptr<FakeNetworkPipe> receive_pipe_
RTC_GUARDED_BY(received_packet_sequence_checker_);
};
} // namespace webrtc

48
call/fake_network_pipe.cc
View File

@ -18,6 +18,8 @@
#include <vector>
#include "api/media_types.h"
#include "api/units/timestamp.h"
#include "modules/rtp_rtcp/source/rtp_packet_received.h"
#include "rtc_base/checks.h"
#include "rtc_base/logging.h"
#include "system_wrappers/include/clock.h"
@ -45,6 +47,19 @@ NetworkPacket::NetworkPacket(rtc::CopyOnWriteBuffer packet,
packet_time_us_(packet_time_us),
transport_(transport) {}
NetworkPacket::NetworkPacket(RtpPacketReceived packet_received,
MediaType media_type,
int64_t send_time,
int64_t arrival_time)
: packet_(packet_received.Buffer()),
send_time_(send_time),
arrival_time_(arrival_time),
is_rtcp_(false),
media_type_(media_type),
packet_time_us_(packet_received.arrival_time().us()),
packet_received_(std::move(packet_received)),
transport_(nullptr) {}
NetworkPacket::NetworkPacket(NetworkPacket&& o)
: packet_(std::move(o.packet_)),
send_time_(o.send_time_),
@ -53,6 +68,7 @@ NetworkPacket::NetworkPacket(NetworkPacket&& o)
is_rtcp_(o.is_rtcp_),
media_type_(o.media_type_),
packet_time_us_(o.packet_time_us_),
packet_received_(std::move(o.packet_received_)),
transport_(o.transport_) {}
NetworkPacket::~NetworkPacket() = default;
@ -65,6 +81,7 @@ NetworkPacket& NetworkPacket::operator=(NetworkPacket&& o) {
is_rtcp_ = o.is_rtcp_;
media_type_ = o.media_type_;
packet_time_us_ = o.packet_time_us_;
packet_received_ = o.packet_received_;
transport_ = o.transport_;
return *this;
@ -184,6 +201,16 @@ PacketReceiver::DeliveryStatus FakeNetworkPipe::DeliverPacket(
: PacketReceiver::DELIVERY_PACKET_ERROR;
}
void FakeNetworkPipe::DeliverRtpPacket(
MediaType media_type,
RtpPacketReceived packet,
OnUndemuxablePacketHandler undemuxable_packet_handler) {
MutexLock lock(&process_lock_);
int64_t time_now_us = clock_->TimeInMicroseconds();
EnqueuePacket(
NetworkPacket(std::move(packet), media_type, time_now_us, time_now_us));
}
void FakeNetworkPipe::DeliverRtcpPacket(rtc::CopyOnWriteBuffer packet) {
EnqueuePacket(std::move(packet), absl::nullopt, true, MediaType::ANY,
absl::nullopt);
@ -352,8 +379,25 @@ void FakeNetworkPipe::DeliverNetworkPacket(NetworkPacket* packet) {
packet_time_us += queue_time_us;
packet_time_us += (clock_offset_ms_ * 1000);
}
receiver_->DeliverPacket(packet->media_type(),
std::move(*packet->raw_packet()), packet_time_us);
if (packet->is_rtcp()) {
receiver_->DeliverRtcpPacket(std::move(*packet->raw_packet()));
} else if (packet->packet_received()) {
packet->packet_received()->set_arrival_time(
Timestamp::Micros(packet_time_us));
receiver_->DeliverRtpPacket(
packet->media_type(), *packet->packet_received(),
[](const RtpPacketReceived& packet) {
RTC_LOG(LS_WARNING)
<< "Unexpected failed demuxing packet in FakeNetworkPipe, "
"Ssrc: "
<< packet.Ssrc() << " seq : " << packet.SequenceNumber();
return false;
});
} else {
receiver_->DeliverPacket(packet->media_type(),
std::move(*packet->raw_packet()),
packet_time_us);
}
}
}

30
call/fake_network_pipe.h
View File

@ -21,8 +21,8 @@
#include "api/call/transport.h"
#include "api/test/simulated_network.h"
#include "call/call.h"
#include "call/simulated_packet_receiver.h"
#include "modules/rtp_rtcp/source/rtp_packet_received.h"
#include "rtc_base/synchronization/mutex.h"
#include "rtc_base/thread_annotations.h"
@ -43,6 +43,11 @@ class NetworkPacket {
absl::optional<int64_t> packet_time_us,
Transport* transport);
NetworkPacket(RtpPacketReceived packet,
MediaType media_type,
int64_t send_time,
int64_t arrival_time);
// Disallow copy constructor and copy assignment (no deep copies of `data_`).
NetworkPacket(const NetworkPacket&) = delete;
~NetworkPacket();
@ -65,6 +70,9 @@ class NetworkPacket {
bool is_rtcp() const { return is_rtcp_; }
MediaType media_type() const { return media_type_; }
absl::optional<int64_t> packet_time_us() const { return packet_time_us_; }
absl::optional<RtpPacketReceived> packet_received() const {
return packet_received_;
}
Transport* transport() const { return transport_; }
private:
@ -83,6 +91,7 @@ class NetworkPacket {
// network pipe.
MediaType media_type_;
absl::optional<int64_t> packet_time_us_;
absl::optional<RtpPacketReceived> packet_received_;
Transport* transport_;
};
@ -142,19 +151,20 @@ class FakeNetworkPipe : public SimulatedPacketReceiverInterface {
// Implements the PacketReceiver interface. When/if packets are delivered,
// they will be passed directly to the receiver instance given in
// SetReceiver(), without passing through a Demuxer. The receive time
// will be increased by the amount of time the packet spent in the
// fake network pipe.
// SetReceiver(). The receive time will be increased by the amount of time the
// packet spent in the fake network pipe.
void DeliverRtpPacket(
MediaType media_type,
RtpPacketReceived packet,
OnUndemuxablePacketHandler undemuxable_packet_handler) override;
void DeliverRtcpPacket(rtc::CopyOnWriteBuffer packet) override;
// TODO(perkj, https://bugs.webrtc.org/7135): Remove once implementations
// dont use it.
PacketReceiver::DeliveryStatus DeliverPacket(MediaType media_type,
rtc::CopyOnWriteBuffer packet,
int64_t packet_time_us) override;
void DeliverRtcpPacket(rtc::CopyOnWriteBuffer packet) override;
// TODO(bugs.webrtc.org/9584): Needed to inherit the alternative signature for
// this method.
using PacketReceiver::DeliverPacket;
// Processes the network queues and trigger PacketReceiver::IncomingPacket for
// packets ready to be delivered.
void Process() override;

128
call/fake_network_pipe_unittest.cc
View File

@ -14,33 +14,46 @@
#include <utility>
#include "call/simulated_network.h"
#include "modules/rtp_rtcp/include/rtp_header_extension_map.h"
#include "modules/rtp_rtcp/source/rtp_header_extensions.h"
#include "modules/rtp_rtcp/source/rtp_packet_received.h"
#include "rtc_base/checks.h"
#include "system_wrappers/include/clock.h"
#include "test/gmock.h"
#include "test/gtest.h"
using ::testing::_;
using ::testing::Property;
using ::testing::WithArg;
namespace webrtc {
class MockReceiver : public PacketReceiver {
public:
MOCK_METHOD(DeliveryStatus,
DeliverPacket,
(MediaType, rtc::CopyOnWriteBuffer, int64_t),
(override));
MOCK_METHOD(void,
DeliverRtcpPacket,
(rtc::CopyOnWriteBuffer packet),
(override));
MOCK_METHOD(void,
DeliverRtpPacket,
(MediaType media_type,
RtpPacketReceived packet,
OnUndemuxablePacketHandler undemuxable_packet_handler),
(override));
virtual ~MockReceiver() = default;
};
class ReorderTestReceiver : public MockReceiver {
public:
DeliveryStatus DeliverPacket(MediaType media_type,
rtc::CopyOnWriteBuffer packet,
int64_t /* packet_time_us */) override {
void DeliverRtpPacket(
MediaType media_type,
RtpPacketReceived packet,
OnUndemuxablePacketHandler undemuxable_packet_handler) override {
RTC_DCHECK_GE(packet.size(), sizeof(int));
int seq_num;
memcpy(&seq_num, packet.data<uint8_t>(), sizeof(int));
delivered_sequence_numbers_.push_back(seq_num);
return DeliveryStatus::DELIVERY_OK;
delivered_sequence_numbers_.push_back(packet.SequenceNumber());
}
std::vector<int> delivered_sequence_numbers_;
};
@ -52,13 +65,15 @@ class FakeNetworkPipeTest : public ::testing::Test {
protected:
void SendPackets(FakeNetworkPipe* pipe, int number_packets, int packet_size) {
RTC_DCHECK_GE(packet_size, sizeof(int));
std::unique_ptr<uint8_t[]> packet(new uint8_t[packet_size]);
for (int i = 0; i < number_packets; ++i) {
// Set a sequence number for the packets by
// using the first bytes in the packet.
memcpy(packet.get(), &i, sizeof(int));
rtc::CopyOnWriteBuffer buffer(packet.get(), packet_size);
pipe->DeliverPacket(MediaType::ANY, buffer, /* packet_time_us */ -1);
RtpPacketReceived packet;
constexpr size_t kFixedHeaderSize = 12;
packet.AllocatePayload(packet_size - kFixedHeaderSize);
packet.SetSequenceNumber(i);
packet.set_arrival_time(fake_clock_.CurrentTime());
RTC_DCHECK_EQ(packet.Buffer().size(), packet_size);
pipe->DeliverRtpPacket(MediaType::ANY, std::move(packet),
[](const RtpPacketReceived&) { return false; });
}
}
@ -90,22 +105,22 @@ TEST_F(FakeNetworkPipeTest, CapacityTest) {
PacketTimeMs(config.link_capacity_kbps, kPacketSize);
// Time haven't increased yet, so we souldn't get any packets.
EXPECT_CALL(receiver, DeliverPacket(_, _, _)).Times(0);
EXPECT_CALL(receiver, DeliverRtpPacket).Times(0);
pipe->Process();
// Advance enough time to release one packet.
fake_clock_.AdvanceTimeMilliseconds(kPacketTimeMs);
EXPECT_CALL(receiver, DeliverPacket(_, _, _)).Times(1);
EXPECT_CALL(receiver, DeliverRtpPacket).Times(1);
pipe->Process();
// Release all but one packet
fake_clock_.AdvanceTimeMilliseconds(9 * kPacketTimeMs - 1);
EXPECT_CALL(receiver, DeliverPacket(_, _, _)).Times(8);
EXPECT_CALL(receiver, DeliverRtpPacket).Times(8);
pipe->Process();
// And the last one.
fake_clock_.AdvanceTimeMilliseconds(1);
EXPECT_CALL(receiver, DeliverPacket(_, _, _)).Times(1);
EXPECT_CALL(receiver, DeliverRtpPacket).Times(1);
pipe->Process();
}
@ -130,17 +145,17 @@ TEST_F(FakeNetworkPipeTest, ExtraDelayTest) {
// Increase more than kPacketTimeMs, but not more than the extra delay.
fake_clock_.AdvanceTimeMilliseconds(kPacketTimeMs);
EXPECT_CALL(receiver, DeliverPacket(_, _, _)).Times(0);
EXPECT_CALL(receiver, DeliverRtpPacket).Times(0);
pipe->Process();
// Advance the network delay to get the first packet.
fake_clock_.AdvanceTimeMilliseconds(config.queue_delay_ms);
EXPECT_CALL(receiver, DeliverPacket(_, _, _)).Times(1);
EXPECT_CALL(receiver, DeliverRtpPacket).Times(1);
pipe->Process();
// Advance one more kPacketTimeMs to get the last packet.
fake_clock_.AdvanceTimeMilliseconds(kPacketTimeMs);
EXPECT_CALL(receiver, DeliverPacket(_, _, _)).Times(1);
EXPECT_CALL(receiver, DeliverRtpPacket).Times(1);
pipe->Process();
}
@ -165,7 +180,7 @@ TEST_F(FakeNetworkPipeTest, QueueLengthTest) {
// Increase time enough to deliver all three packets, verify only two are
// delivered.
fake_clock_.AdvanceTimeMilliseconds(3 * kPacketTimeMs);
EXPECT_CALL(receiver, DeliverPacket(_, _, _)).Times(2);
EXPECT_CALL(receiver, DeliverRtpPacket).Times(2);
pipe->Process();
}
@ -189,7 +204,7 @@ TEST_F(FakeNetworkPipeTest, StatisticsTest) {
fake_clock_.AdvanceTimeMilliseconds(3 * kPacketTimeMs +
config.queue_delay_ms);
EXPECT_CALL(receiver, DeliverPacket(_, _, _)).Times(2);
EXPECT_CALL(receiver, DeliverRtpPacket).Times(2);
pipe->Process();
// Packet 1: kPacketTimeMs + config.queue_delay_ms,
@ -222,13 +237,13 @@ TEST_F(FakeNetworkPipeTest, ChangingCapacityWithEmptyPipeTest) {
int packet_time_ms = PacketTimeMs(config.link_capacity_kbps, kPacketSize);
// Time hasn't increased yet, so we souldn't get any packets.
EXPECT_CALL(receiver, DeliverPacket(_, _, _)).Times(0);
EXPECT_CALL(receiver, DeliverRtpPacket).Times(0);
pipe->Process();
// Advance time in steps to release one packet at a time.
for (int i = 0; i < kNumPackets; ++i) {
fake_clock_.AdvanceTimeMilliseconds(packet_time_ms);
EXPECT_CALL(receiver, DeliverPacket(_, _, _)).Times(1);
EXPECT_CALL(receiver, DeliverRtpPacket).Times(1);
pipe->Process();
}
@ -244,13 +259,13 @@ TEST_F(FakeNetworkPipeTest, ChangingCapacityWithEmptyPipeTest) {
packet_time_ms = PacketTimeMs(config.link_capacity_kbps, kPacketSize);
// Time hasn't increased yet, so we souldn't get any packets.
EXPECT_CALL(receiver, DeliverPacket(_, _, _)).Times(0);
EXPECT_CALL(receiver, DeliverRtpPacket).Times(0);
pipe->Process();
// Advance time in steps to release one packet at a time.
for (int i = 0; i < kNumPackets; ++i) {
fake_clock_.AdvanceTimeMilliseconds(packet_time_ms);
EXPECT_CALL(receiver, DeliverPacket(_, _, _)).Times(1);
EXPECT_CALL(receiver, DeliverRtpPacket).Times(1);
pipe->Process();
}
@ -258,7 +273,7 @@ TEST_F(FakeNetworkPipeTest, ChangingCapacityWithEmptyPipeTest) {
EXPECT_EQ(static_cast<size_t>(2 * kNumPackets), pipe->SentPackets());
EXPECT_FALSE(pipe->TimeUntilNextProcess().has_value());
fake_clock_.AdvanceTimeMilliseconds(1000);
EXPECT_CALL(receiver, DeliverPacket(_, _, _)).Times(0);
EXPECT_CALL(receiver, DeliverRtpPacket).Times(0);
pipe->Process();
}
@ -280,14 +295,14 @@ TEST_F(FakeNetworkPipeTest, ChangingCapacityWithPacketsInPipeTest) {
SendPackets(pipe.get(), kNumPackets, kPacketSize);
// Time hasn't increased yet, so we souldn't get any packets.
EXPECT_CALL(receiver, DeliverPacket(_, _, _)).Times(0);
EXPECT_CALL(receiver, DeliverRtpPacket).Times(0);
pipe->Process();
// Advance time in steps to release half of the packets one at a time.
int step_ms = PacketTimeMs(config.link_capacity_kbps, kPacketSize);
for (int i = 0; i < kNumPackets / 2; ++i) {
fake_clock_.AdvanceTimeMilliseconds(step_ms);
EXPECT_CALL(receiver, DeliverPacket(_, _, _)).Times(1);
EXPECT_CALL(receiver, DeliverRtpPacket).Times(1);
pipe->Process();
}
@ -299,7 +314,7 @@ TEST_F(FakeNetworkPipeTest, ChangingCapacityWithPacketsInPipeTest) {
step_ms = PacketTimeMs(config.link_capacity_kbps, kPacketSize);
for (int i = 0; i < kNumPackets / 2; ++i) {
fake_clock_.AdvanceTimeMilliseconds(step_ms);
EXPECT_CALL(receiver, DeliverPacket(_, _, _)).Times(1);
EXPECT_CALL(receiver, DeliverRtpPacket).Times(1);
pipe->Process();
}
@ -307,7 +322,7 @@ TEST_F(FakeNetworkPipeTest, ChangingCapacityWithPacketsInPipeTest) {
EXPECT_EQ(static_cast<size_t>(kNumPackets), pipe->SentPackets());
EXPECT_FALSE(pipe->TimeUntilNextProcess().has_value());
fake_clock_.AdvanceTimeMilliseconds(1000);
EXPECT_CALL(receiver, DeliverPacket(_, _, _)).Times(0);
EXPECT_CALL(receiver, DeliverRtpPacket).Times(0);
pipe->Process();
}
@ -413,7 +428,7 @@ TEST_F(FakeNetworkPipeTest, SetReceiver) {
PacketTimeMs(config.link_capacity_kbps, kPacketSize);
SendPackets(pipe.get(), 1, kPacketSize);
fake_clock_.AdvanceTimeMilliseconds(kPacketTimeMs);
EXPECT_CALL(receiver, DeliverPacket(_, _, _)).Times(1);
EXPECT_CALL(receiver, DeliverRtpPacket).Times(1);
pipe->Process();
MockReceiver new_receiver;
@ -421,8 +436,51 @@ TEST_F(FakeNetworkPipeTest, SetReceiver) {
SendPackets(pipe.get(), 1, kPacketSize);
fake_clock_.AdvanceTimeMilliseconds(kPacketTimeMs);
EXPECT_CALL(receiver, DeliverPacket(_, _, _)).Times(0);
EXPECT_CALL(new_receiver, DeliverPacket(_, _, _)).Times(1);
EXPECT_CALL(receiver, DeliverRtpPacket).Times(0);
EXPECT_CALL(new_receiver, DeliverRtpPacket).Times(1);
pipe->Process();
}
TEST_F(FakeNetworkPipeTest, DeliverRtpPacketPropagatesExtensions) {
BuiltInNetworkBehaviorConfig config;
config.queue_delay_ms = 100;
MockReceiver receiver;
auto simulated_network = std::make_unique<SimulatedNetwork>(config);
std::unique_ptr<FakeNetworkPipe> pipe(new FakeNetworkPipe(
&fake_clock_, std::move(simulated_network), &receiver));
RtpHeaderExtensionMap extension_map;
extension_map.Register<TransportSequenceNumber>(/*id=*/7);
RtpPacketReceived packet(&extension_map, fake_clock_.CurrentTime());
packet.SetExtension<TransportSequenceNumber>(123);
pipe->DeliverRtpPacket(MediaType::VIDEO, std::move(packet),
[](const RtpPacketReceived&) { return false; });
// Advance the network delay to get the first packet.
fake_clock_.AdvanceTimeMilliseconds(config.queue_delay_ms);
EXPECT_CALL(receiver, DeliverRtpPacket(MediaType::VIDEO, _, _))
.WillOnce(WithArg<1>([](RtpPacketReceived packet) {
EXPECT_EQ(packet.GetExtension<TransportSequenceNumber>(), 123);
}));
pipe->Process();
}
TEST_F(FakeNetworkPipeTest, DeliverRtcpPacket) {
BuiltInNetworkBehaviorConfig config;
config.queue_delay_ms = 100;
MockReceiver receiver;
auto simulated_network = std::make_unique<SimulatedNetwork>(config);
std::unique_ptr<FakeNetworkPipe> pipe(new FakeNetworkPipe(
&fake_clock_, std::move(simulated_network), &receiver));
rtc::CopyOnWriteBuffer buffer(100);
memset(buffer.MutableData(), 0, 100);
pipe->DeliverRtcpPacket(std::move(buffer));
// Advance the network delay to get the first packet.
fake_clock_.AdvanceTimeMilliseconds(config.queue_delay_ms);
EXPECT_CALL(receiver,
DeliverRtcpPacket(Property(&rtc::CopyOnWriteBuffer::size, 100)));
pipe->Process();
}