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

Introduce RtpStreamReceiverInterface and RtpStreamReceiverControllerInterface.

Browse Source 
And implementation class RtpStreamReceiverController.
It's responsible for demuxing, and acts as factory for
RtpStreamReceiverInterface.

BUG=webrtc:7135

Review-Url: https://codereview.webrtc.org/2886993005
Cr-Commit-Position: refs/heads/master@{#18696}
This commit is contained in:
nisse 2017-06-21 01:05:22 -07:00 committed by Commit Bot
parent 130ca7e783
commit 0f15f926e3
18 changed files with 331 additions and 73 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
webrtc/audio/BUILD.gn
View File

@ -79,6 +79,7 @@ if (rtc_include_tests) {
"../api:mock_audio_mixer",
"../base:rtc_base_approved",
"../base:rtc_task_queue",
"../call:rtp_receiver",
"../modules/audio_device:mock_audio_device",
"../modules/audio_mixer:audio_mixer_impl",
"../modules/congestion_controller:congestion_controller",

10
webrtc/audio/audio_receive_stream.cc
View File

@ -20,6 +20,7 @@
#include "webrtc/base/checks.h"
#include "webrtc/base/logging.h"
#include "webrtc/base/timeutils.h"
#include "webrtc/call/rtp_stream_receiver_controller_interface.h"
#include "webrtc/modules/remote_bitrate_estimator/include/remote_bitrate_estimator.h"
#include "webrtc/modules/rtp_rtcp/include/rtp_receiver.h"
#include "webrtc/modules/rtp_rtcp/include/rtp_rtcp.h"
@ -62,12 +63,12 @@ std::string AudioReceiveStream::Config::ToString() const {
namespace internal {
AudioReceiveStream::AudioReceiveStream(
RtpStreamReceiverControllerInterface* receiver_controller,
PacketRouter* packet_router,
const webrtc::AudioReceiveStream::Config& config,
const rtc::scoped_refptr<webrtc::AudioState>& audio_state,
webrtc::RtcEventLog* event_log)
: config_(config),
audio_state_(audio_state) {
: config_(config), audio_state_(audio_state) {
LOG(LS_INFO) << "AudioReceiveStream: " << config_.ToString();
RTC_DCHECK_NE(config_.voe_channel_id, -1);
RTC_DCHECK(audio_state_.get());
@ -107,6 +108,11 @@ AudioReceiveStream::AudioReceiveStream(
}
// Configure bandwidth estimation.
channel_proxy_->RegisterReceiverCongestionControlObjects(packet_router);
// Register with transport.
rtp_stream_receiver_ =
receiver_controller->CreateReceiver(config_.rtp.remote_ssrc,
channel_proxy_.get());
}
AudioReceiveStream::~AudioReceiveStream() {

17
webrtc/audio/audio_receive_stream.h
View File

@ -26,6 +26,8 @@ namespace webrtc {
class PacketRouter;
class RtcEventLog;
class RtpPacketReceived;
class RtpStreamReceiverControllerInterface;
class RtpStreamReceiverInterface;
namespace voe {
class ChannelProxy;
@ -36,10 +38,10 @@ class AudioSendStream;
class AudioReceiveStream final : public webrtc::AudioReceiveStream,
public AudioMixer::Source,
public Syncable,
public RtpPacketSinkInterface {
public Syncable {
public:
AudioReceiveStream(PacketRouter* packet_router,
AudioReceiveStream(RtpStreamReceiverControllerInterface* receiver_controller,
PacketRouter* packet_router,
const webrtc::AudioReceiveStream::Config& config,
const rtc::scoped_refptr<webrtc::AudioState>& audio_state,
webrtc::RtcEventLog* event_log);
@ -54,8 +56,11 @@ class AudioReceiveStream final : public webrtc::AudioReceiveStream,
void SetGain(float gain) override;
std::vector<webrtc::RtpSource> GetSources() const override;
// RtpPacketSinkInterface.
void OnRtpPacket(const RtpPacketReceived& packet) override;
// TODO(nisse): We don't formally implement RtpPacketSinkInterface, and this
// method shouldn't be needed. But it's currently used by the
// AudioReceiveStreamTest.ReceiveRtpPacket unittest. Figure out if that test
// shuld be refactored or deleted, and then delete this method.
void OnRtpPacket(const RtpPacketReceived& packet);
// AudioMixer::Source
AudioFrameInfo GetAudioFrameWithInfo(int sample_rate_hz,
@ -87,6 +92,8 @@ class AudioReceiveStream final : public webrtc::AudioReceiveStream,
bool playing_ ACCESS_ON(worker_thread_checker_) = false;
std::unique_ptr<RtpStreamReceiverInterface> rtp_stream_receiver_;
RTC_DISALLOW_IMPLICIT_CONSTRUCTORS(AudioReceiveStream);
};
} // namespace internal

12
webrtc/audio/audio_receive_stream_unittest.cc
View File

@ -15,6 +15,7 @@
#include "webrtc/api/test/mock_audio_mixer.h"
#include "webrtc/audio/audio_receive_stream.h"
#include "webrtc/audio/conversion.h"
#include "webrtc/call/rtp_stream_receiver_controller.h"
#include "webrtc/logging/rtc_event_log/mock/mock_rtc_event_log.h"
#include "webrtc/modules/bitrate_controller/include/mock/mock_bitrate_controller.h"
#include "webrtc/modules/pacing/packet_router.h"
@ -137,6 +138,9 @@ struct ConfigHelper {
rtc::scoped_refptr<MockAudioMixer> audio_mixer() { return audio_mixer_; }
MockVoiceEngine& voice_engine() { return voice_engine_; }
MockVoEChannelProxy* channel_proxy() { return channel_proxy_; }
RtpStreamReceiverControllerInterface* rtp_stream_receiver_controller() {
return &rtp_stream_receiver_controller_;
}
void SetupMockForGetStats() {
using testing::DoAll;
@ -166,6 +170,7 @@ struct ConfigHelper {
rtc::scoped_refptr<MockAudioMixer> audio_mixer_;
AudioReceiveStream::Config stream_config_;
testing::StrictMock<MockVoEChannelProxy>* channel_proxy_ = nullptr;
RtpStreamReceiverController rtp_stream_receiver_controller_;
};
void BuildOneByteExtension(std::vector<uint8_t>::iterator it,
@ -238,6 +243,7 @@ TEST(AudioReceiveStreamTest, ConfigToString) {
TEST(AudioReceiveStreamTest, ConstructDestruct) {
ConfigHelper helper;
internal::AudioReceiveStream recv_stream(
helper.rtp_stream_receiver_controller(),
helper.packet_router(),
helper.config(), helper.audio_state(), helper.event_log());
}
@ -246,6 +252,7 @@ TEST(AudioReceiveStreamTest, ReceiveRtpPacket) {
ConfigHelper helper;
helper.config().rtp.transport_cc = true;
internal::AudioReceiveStream recv_stream(
helper.rtp_stream_receiver_controller(),
helper.packet_router(),
helper.config(), helper.audio_state(), helper.event_log());
const int kTransportSequenceNumberValue = 1234;
@ -267,6 +274,7 @@ TEST(AudioReceiveStreamTest, ReceiveRtcpPacket) {
ConfigHelper helper;
helper.config().rtp.transport_cc = true;
internal::AudioReceiveStream recv_stream(
helper.rtp_stream_receiver_controller(),
helper.packet_router(),
helper.config(), helper.audio_state(), helper.event_log());
@ -280,6 +288,7 @@ TEST(AudioReceiveStreamTest, ReceiveRtcpPacket) {
TEST(AudioReceiveStreamTest, GetStats) {
ConfigHelper helper;
internal::AudioReceiveStream recv_stream(
helper.rtp_stream_receiver_controller(),
helper.packet_router(),
helper.config(), helper.audio_state(), helper.event_log());
helper.SetupMockForGetStats();
@ -325,6 +334,7 @@ TEST(AudioReceiveStreamTest, GetStats) {
TEST(AudioReceiveStreamTest, SetGain) {
ConfigHelper helper;
internal::AudioReceiveStream recv_stream(
helper.rtp_stream_receiver_controller(),
helper.packet_router(),
helper.config(), helper.audio_state(), helper.event_log());
EXPECT_CALL(*helper.channel_proxy(),
@ -335,6 +345,7 @@ TEST(AudioReceiveStreamTest, SetGain) {
TEST(AudioReceiveStreamTest, StreamShouldNotBeAddedToMixerWhenVoEReturnsError) {
ConfigHelper helper;
internal::AudioReceiveStream recv_stream(
helper.rtp_stream_receiver_controller(),
helper.packet_router(),
helper.config(), helper.audio_state(), helper.event_log());
@ -347,6 +358,7 @@ TEST(AudioReceiveStreamTest, StreamShouldNotBeAddedToMixerWhenVoEReturnsError) {
TEST(AudioReceiveStreamTest, StreamShouldBeAddedToMixerOnStart) {
ConfigHelper helper;
internal::AudioReceiveStream recv_stream(
helper.rtp_stream_receiver_controller(),
helper.packet_router(),
helper.config(), helper.audio_state(), helper.event_log());

3
webrtc/call/BUILD.gn
View File

@ -38,6 +38,7 @@ rtc_source_set("call_interfaces") {
rtc_source_set("rtp_interfaces") {
sources = [
"rtp_packet_sink_interface.h",
"rtp_stream_receiver_controller_interface.h",
"rtp_transport_controller_send_interface.h",
]
}
@ -46,6 +47,8 @@ rtc_source_set("rtp_receiver") {
sources = [
"rtp_demuxer.cc",
"rtp_demuxer.h",
"rtp_stream_receiver_controller.cc",
"rtp_stream_receiver_controller.h",
"rtx_receive_stream.cc",
"rtx_receive_stream.h",
]

76
webrtc/call/call.cc
View File

@ -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_stream_receiver_controller.h"
#include "webrtc/call/rtp_transport_controller_send.h"
#include "webrtc/config.h"
#include "webrtc/logging/rtc_event_log/rtc_event_log.h"
@ -275,10 +275,10 @@ 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.
RtpStreamReceiverController audio_receiver_controller;
RtpStreamReceiverController video_receiver_controller;
// This extra map is used for receive processing which is
// independent of media type.
@ -486,10 +486,6 @@ rtc::Optional<RtpPacketReceived> Call::ParseRtpPacket(
if (!parsed_packet.Parse(packet, length))
return rtc::Optional<RtpPacketReceived>();
auto it = receive_rtp_config_.find(parsed_packet.Ssrc());
if (it != receive_rtp_config_.end())
parsed_packet.IdentifyExtensions(it->second.extensions);
int64_t arrival_time_ms;
if (packet_time && packet_time->timestamp != -1) {
arrival_time_ms = (packet_time->timestamp + 500) / 1000;
@ -646,12 +642,11 @@ webrtc::AudioReceiveStream* Call::CreateAudioReceiveStream(
TRACE_EVENT0("webrtc", "Call::CreateAudioReceiveStream");
RTC_DCHECK_RUN_ON(&configuration_thread_checker_);
event_log_->LogAudioReceiveStreamConfig(CreateRtcLogStreamConfig(config));
AudioReceiveStream* receive_stream =
new AudioReceiveStream(transport_send_->packet_router(), config,
config_.audio_state, event_log_);
AudioReceiveStream* receive_stream = new AudioReceiveStream(
&audio_receiver_controller, transport_send_->packet_router(), config,
config_.audio_state, event_log_);
{
WriteLockScoped write_lock(*receive_crit_);
audio_rtp_demuxer_.AddSink(config.rtp.remote_ssrc, receive_stream);
receive_rtp_config_[config.rtp.remote_ssrc] =
ReceiveRtpConfig(config.rtp.extensions, UseSendSideBwe(config));
audio_receive_streams_.insert(receive_stream);
@ -683,8 +678,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);
@ -776,19 +769,17 @@ webrtc::VideoReceiveStream* Call::CreateVideoReceiveStream(
TRACE_EVENT0("webrtc", "Call::CreateVideoReceiveStream");
RTC_DCHECK_RUN_ON(&configuration_thread_checker_);
VideoReceiveStream* receive_stream =
new VideoReceiveStream(num_cpu_cores_, transport_send_->packet_router(),
std::move(configuration),
module_process_thread_.get(), call_stats_.get());
VideoReceiveStream* receive_stream = new VideoReceiveStream(
&video_receiver_controller, 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));
{
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);
// 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
@ -817,8 +808,6 @@ 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);
@ -840,16 +829,22 @@ FlexfecReceiveStream* Call::CreateFlexfecReceiveStream(
RTC_DCHECK_RUN_ON(&configuration_thread_checker_);
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);
// Unlike the video and audio receive streams,
// FlexfecReceiveStream implements RtpPacketSinkInterface itself,
// and hence its constructor passes its |this| pointer to
// video_receiver_controller->CreateStream(). Calling the
// constructor while holding |receive_crit_| ensures that we don't
// call OnRtpPacket until the constructor is finished and the
// object is in a valid state.
// TODO(nisse): Fix constructor so that it can be moved outside of
// this locked scope.
receive_stream = new FlexfecReceiveStreamImpl(
&video_receiver_controller, 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());
@ -881,7 +876,6 @@ 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);
}
@ -1302,17 +1296,31 @@ PacketReceiver::DeliveryStatus Call::DeliverRtp(MediaType media_type,
if (!parsed_packet)
return DELIVERY_PACKET_ERROR;
auto it = receive_rtp_config_.find(parsed_packet->Ssrc());
if (it == receive_rtp_config_.end()) {
LOG(LS_ERROR) << "receive_rtp_config_ lookup failed for ssrc "
<< parsed_packet->Ssrc();
// Destruction of the receive stream, including deregistering from the
// RtpDemuxer, is not protected by the |receive_crit_| lock. But
// deregistering in the |receive_rtp_config_| map is protected by that lock.
// So by not passing the packet on to demuxing in this case, we prevent
// incoming packets to be passed on via the demuxer to a receive stream
// which is being torned down.
return DELIVERY_UNKNOWN_SSRC;
}
parsed_packet->IdentifyExtensions(it->second.extensions);
NotifyBweOfReceivedPacket(*parsed_packet, media_type);
if (media_type == MediaType::AUDIO) {
if (audio_rtp_demuxer_.OnRtpPacket(*parsed_packet)) {
if (audio_receiver_controller.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, packet, length);
return DELIVERY_OK;
}
} else if (media_type == MediaType::VIDEO) {
if (video_rtp_demuxer_.OnRtpPacket(*parsed_packet)) {
if (video_receiver_controller.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, packet, length);
@ -1348,7 +1356,7 @@ void Call::OnRecoveredPacket(const uint8_t* packet, size_t length) {
parsed_packet->set_recovered(true);
video_rtp_demuxer_.OnRtpPacket(*parsed_packet);
video_receiver_controller.OnRtpPacket(*parsed_packet);
}
void Call::NotifyBweOfReceivedPacket(const RtpPacketReceived& packet,

18
webrtc/call/flexfec_receive_stream_impl.cc
View File

@ -15,6 +15,7 @@
#include "webrtc/base/checks.h"
#include "webrtc/base/location.h"
#include "webrtc/base/logging.h"
#include "webrtc/call/rtp_stream_receiver_controller_interface.h"
#include "webrtc/modules/rtp_rtcp/include/flexfec_receiver.h"
#include "webrtc/modules/rtp_rtcp/include/receive_statistics.h"
#include "webrtc/modules/rtp_rtcp/include/rtp_rtcp.h"
@ -122,6 +123,7 @@ std::unique_ptr<RtpRtcp> CreateRtpRtcpModule(
} // namespace
FlexfecReceiveStreamImpl::FlexfecReceiveStreamImpl(
RtpStreamReceiverControllerInterface* receiver_controller,
const Config& config,
RecoveredPacketReceiver* recovered_packet_receiver,
RtcpRttStats* rtt_stats,
@ -141,6 +143,22 @@ FlexfecReceiveStreamImpl::FlexfecReceiveStreamImpl(
rtp_rtcp_->SetRTCPStatus(config_.rtcp_mode);
rtp_rtcp_->SetSSRC(config_.local_ssrc);
process_thread_->RegisterModule(rtp_rtcp_.get(), RTC_FROM_HERE);
// Register with transport.
// TODO(nisse): OnRtpPacket in this class delegates all real work to
// |receiver_|. So maybe we don't need to implement RtpPacketSinkInterface
// here at all, we'd then delete the OnRtpPacket method and instead register
// |receiver_| as the RtpPacketSinkInterface for this stream.
// TODO(nisse): Passing |this| from the constructor to the RtpDemuxer, before
// the object is fully initialized, is risky. But it works in this case
// because locking in our caller, Call::CreateFlexfecReceiveStream, ensures
// that the demuxer doesn't call OnRtpPacket before this object is fully
// constructed. Registering |receiver_| instead of |this| would solve this
// problem too.
rtp_stream_receiver_ =
receiver_controller->CreateReceiver(config_.remote_ssrc, this);
for (uint32_t ssrc : config.protected_media_ssrcs)
receiver_controller->AddSink(ssrc, this);
}
FlexfecReceiveStreamImpl::~FlexfecReceiveStreamImpl() {

14
webrtc/call/flexfec_receive_stream_impl.h
View File

@ -26,14 +26,18 @@ class RecoveredPacketReceiver;
class RtcpRttStats;
class RtpPacketReceived;
class RtpRtcp;
class RtpStreamReceiverControllerInterface;
class RtpStreamReceiverInterface;
class FlexfecReceiveStreamImpl : public FlexfecReceiveStream,
public RtpPacketSinkInterface {
public:
FlexfecReceiveStreamImpl(const Config& config,
RecoveredPacketReceiver* recovered_packet_receiver,
RtcpRttStats* rtt_stats,
ProcessThread* process_thread);
FlexfecReceiveStreamImpl(
RtpStreamReceiverControllerInterface* receiver_controller,
const Config& config,
RecoveredPacketReceiver* recovered_packet_receiver,
RtcpRttStats* rtt_stats,
ProcessThread* process_thread);
~FlexfecReceiveStreamImpl() override;
const Config& GetConfig() const { return config_; }
@ -59,6 +63,8 @@ class FlexfecReceiveStreamImpl : public FlexfecReceiveStream,
const std::unique_ptr<ReceiveStatistics> rtp_receive_statistics_;
const std::unique_ptr<RtpRtcp> rtp_rtcp_;
ProcessThread* process_thread_;
std::unique_ptr<RtpStreamReceiverInterface> rtp_stream_receiver_;
};
} // namespace webrtc

9
webrtc/call/flexfec_receive_stream_unittest.cc
View File

@ -12,6 +12,7 @@
#include "webrtc/base/array_view.h"
#include "webrtc/call/flexfec_receive_stream_impl.h"
#include "webrtc/call/rtp_stream_receiver_controller.h"
#include "webrtc/modules/pacing/packet_router.h"
#include "webrtc/modules/rtp_rtcp/include/flexfec_receiver.h"
#include "webrtc/modules/rtp_rtcp/mocks/mock_recovered_packet_receiver.h"
@ -77,7 +78,8 @@ class FlexfecReceiveStreamTest : public ::testing::Test {
protected:
FlexfecReceiveStreamTest()
: config_(CreateDefaultConfig(&rtcp_send_transport_)),
receive_stream_(config_,
receive_stream_(&rtp_stream_receiver_controller_,
config_,
&recovered_packet_receiver_,
&rtt_stats_,
&process_thread_) {}
@ -87,7 +89,7 @@ class FlexfecReceiveStreamTest : public ::testing::Test {
MockRecoveredPacketReceiver recovered_packet_receiver_;
MockRtcpRttStats rtt_stats_;
MockProcessThread process_thread_;
RtpStreamReceiverController rtp_stream_receiver_controller_;
FlexfecReceiveStreamImpl receive_stream_;
};
@ -134,7 +136,8 @@ TEST_F(FlexfecReceiveStreamTest, RecoversPacketWhenStarted) {
// clang-format on
testing::StrictMock<MockRecoveredPacketReceiver> recovered_packet_receiver;
FlexfecReceiveStreamImpl receive_stream(config_, &recovered_packet_receiver,
FlexfecReceiveStreamImpl receive_stream(&rtp_stream_receiver_controller_,
config_, &recovered_packet_receiver,
&rtt_stats_, &process_thread_);
// Do not call back before being started.

58
webrtc/call/rtp_stream_receiver_controller.cc Normal file
View File

@ -0,0 +1,58 @@
/*
* Copyright (c) 2017 The WebRTC project authors. All Rights Reserved.
*
* Use of this source code is governed by a BSD-style license
* that can be found in the LICENSE file in the root of the source
* tree. An additional intellectual property rights grant can be found
* in the file PATENTS. All contributing project authors may
* be found in the AUTHORS file in the root of the source tree.
*/
#include "webrtc/call/rtp_stream_receiver_controller.h"
#include "webrtc/base/ptr_util.h"
namespace webrtc {
RtpStreamReceiverController::Receiver::Receiver(
RtpStreamReceiverController* controller,
uint32_t ssrc,
RtpPacketSinkInterface* sink)
: controller_(controller), sink_(sink) {
controller_->AddSink(ssrc, sink_);
}
RtpStreamReceiverController::Receiver::~Receiver() {
// Don't require return value > 0, since for RTX we currently may
// have multiple Receiver objects with the same sink.
// TODO(nisse): Consider adding a DCHECK when RtxReceiveStream is wired up.
controller_->RemoveSink(sink_);
}
RtpStreamReceiverController::RtpStreamReceiverController() = default;
RtpStreamReceiverController::~RtpStreamReceiverController() = default;
std::unique_ptr<RtpStreamReceiverInterface>
RtpStreamReceiverController::CreateReceiver(
uint32_t ssrc,
RtpPacketSinkInterface* sink) {
return rtc::MakeUnique<Receiver>(this, ssrc, sink);
}
bool RtpStreamReceiverController::OnRtpPacket(const RtpPacketReceived& packet) {
rtc::CritScope cs(&lock_);
return demuxer_.OnRtpPacket(packet);
}
void RtpStreamReceiverController::AddSink(uint32_t ssrc,
RtpPacketSinkInterface* sink) {
rtc::CritScope cs(&lock_);
return demuxer_.AddSink(ssrc, sink);
}
size_t RtpStreamReceiverController::RemoveSink(
const RtpPacketSinkInterface* sink) {
rtc::CritScope cs(&lock_);
return demuxer_.RemoveSink(sink);
}
} // namespace webrtc

72
webrtc/call/rtp_stream_receiver_controller.h Normal file
View File

@ -0,0 +1,72 @@
/*
* Copyright (c) 2017 The WebRTC project authors. All Rights Reserved.
*
* Use of this source code is governed by a BSD-style license
* that can be found in the LICENSE file in the root of the source
* tree. An additional intellectual property rights grant can be found
* 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_STREAM_RECEIVER_CONTROLLER_H_
#define WEBRTC_CALL_RTP_STREAM_RECEIVER_CONTROLLER_H_
#include <memory>
#include "webrtc/base/criticalsection.h"
#include "webrtc/call/rtp_demuxer.h"
#include "webrtc/call/rtp_stream_receiver_controller_interface.h"
namespace webrtc {
class RtpPacketReceived;
// This class represents the RTP receive parsing and demuxing, for a
// single RTP session.
// TODO(nisse): Add RTCP processing, we should aim to terminate RTCP
// and not leave any RTCP processing to individual receive streams.
// TODO(nisse): Extract per-packet processing, including parsing and
// demuxing, into a separate class.
class RtpStreamReceiverController
: public RtpStreamReceiverControllerInterface {
public:
RtpStreamReceiverController();
~RtpStreamReceiverController() override;
// Implements RtpStreamReceiverControllerInterface.
std::unique_ptr<RtpStreamReceiverInterface> CreateReceiver(
uint32_t ssrc,
RtpPacketSinkInterface* sink) override;
// Thread-safe wrappers for the corresponding RtpDemuxer methods.
void AddSink(uint32_t ssrc, RtpPacketSinkInterface* sink) override;
size_t RemoveSink(const RtpPacketSinkInterface* sink) override;
// TODO(nisse): Not yet responsible for parsing.
bool OnRtpPacket(const RtpPacketReceived& packet);
private:
class Receiver : public RtpStreamReceiverInterface {
public:
Receiver(RtpStreamReceiverController* controller,
uint32_t ssrc,
RtpPacketSinkInterface* sink);
~Receiver() override;
private:
RtpStreamReceiverController* const controller_;
RtpPacketSinkInterface* const sink_;
};
// TODO(nisse): Move to a TaskQueue for synchronization. When used
// by Call, we expect construction and all methods but OnRtpPacket
// to be called on the same thread, and OnRtpPacket to be called
// by a single, but possibly distinct, thread. But applications not
// using Call may have use threads differently.
rtc::CriticalSection lock_;
RtpDemuxer demuxer_ GUARDED_BY(&lock_);
};
} // namespace webrtc
#endif // WEBRTC_CALL_RTP_STREAM_RECEIVER_CONTROLLER_H_

47
webrtc/call/rtp_stream_receiver_controller_interface.h Normal file
View File

@ -0,0 +1,47 @@
/*
* Copyright (c) 2017 The WebRTC project authors. All Rights Reserved.
*
* Use of this source code is governed by a BSD-style license
* that can be found in the LICENSE file in the root of the source
* tree. An additional intellectual property rights grant can be found
* 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_STREAM_RECEIVER_CONTROLLER_INTERFACE_H_
#define WEBRTC_CALL_RTP_STREAM_RECEIVER_CONTROLLER_INTERFACE_H_
#include <memory>
#include "webrtc/call/rtp_packet_sink_interface.h"
namespace webrtc {
// An RtpStreamReceiver is responsible for the rtp-specific but
// media-independent state needed for receiving an RTP stream.
// TODO(nisse): Currently, only owns the association between ssrc and
// the stream's RtpPacketSinkInterface. Ownership of corresponding
// objects from modules/rtp_rtcp/ should move to this class (or
// rather, the corresponding implementation class). We should add
// methods for getting rtp receive stats, and for sending RTCP
// messages related to the receive stream.
class RtpStreamReceiverInterface {
public:
virtual ~RtpStreamReceiverInterface() {}
};
// This class acts as a factory for RtpStreamReceiver objects.
class RtpStreamReceiverControllerInterface {
public:
virtual ~RtpStreamReceiverControllerInterface() {}
virtual std::unique_ptr<RtpStreamReceiverInterface> CreateReceiver(
uint32_t ssrc,
RtpPacketSinkInterface* sink) = 0;
// For registering additional sinks, needed for FlexFEC.
virtual void AddSink(uint32_t ssrc, RtpPacketSinkInterface* sink) = 0;
virtual size_t RemoveSink(const RtpPacketSinkInterface* sink) = 0;
};
} // namespace webrtc
#endif // WEBRTC_CALL_RTP_STREAM_RECEIVER_CONTROLLER_INTERFACE_H_

1
webrtc/video/BUILD.gn
View File

@ -260,6 +260,7 @@ if (rtc_include_tests) {
"../base:rtc_base_approved",
"../base:rtc_base_tests_utils",
"../call:call_interfaces",
"../call:rtp_receiver",
"../common_video",
"../logging:rtc_event_log_api",
"../media:rtc_media",

6
webrtc/video/rtp_video_stream_receiver.h
View File

@ -19,6 +19,7 @@
#include "webrtc/base/constructormagic.h"
#include "webrtc/base/criticalsection.h"
#include "webrtc/call/rtp_packet_sink_interface.h"
#include "webrtc/modules/include/module_common_types.h"
#include "webrtc/modules/rtp_rtcp/include/receive_statistics.h"
#include "webrtc/modules/rtp_rtcp/include/remote_ntp_time_estimator.h"
@ -58,6 +59,7 @@ class VideoReceiver;
class RtpVideoStreamReceiver : public RtpData,
public RecoveredPacketReceiver,
public RtpFeedback,
public RtpPacketSinkInterface,
public VCMFrameTypeCallback,
public VCMPacketRequestCallback,
public video_coding::OnReceivedFrameCallback,
@ -96,8 +98,8 @@ class RtpVideoStreamReceiver : public RtpData,
void SignalNetworkState(NetworkState state);
// TODO(nisse): Intended to be part of an RtpPacketReceiver interface.
void OnRtpPacket(const RtpPacketReceived& packet);
// Implements RtpPacketSinkInterface.
void OnRtpPacket(const RtpPacketReceived& packet) override;
// Implements RtpData.
int32_t OnReceivedPayloadData(const uint8_t* payload_data,

25
webrtc/video/video_receive_stream.cc
View File

@ -21,6 +21,7 @@
#include "webrtc/base/logging.h"
#include "webrtc/base/optional.h"
#include "webrtc/base/trace_event.h"
#include "webrtc/call/rtp_stream_receiver_controller_interface.h"
#include "webrtc/common_types.h"
#include "webrtc/common_video/h264/profile_level_id.h"
#include "webrtc/common_video/libyuv/include/webrtc_libyuv.h"
@ -168,11 +169,13 @@ VideoCodec CreateDecoderVideoCodec(const VideoReceiveStream::Decoder& decoder) {
namespace internal {
VideoReceiveStream::VideoReceiveStream(int num_cpu_cores,
PacketRouter* packet_router,
VideoReceiveStream::Config config,
ProcessThread* process_thread,
CallStats* call_stats)
VideoReceiveStream::VideoReceiveStream(
RtpStreamReceiverControllerInterface* receiver_controller,
int num_cpu_cores,
PacketRouter* packet_router,
VideoReceiveStream::Config config,
ProcessThread* process_thread,
CallStats* call_stats)
: transport_adapter_(config.rtcp_send_transport),
config_(std::move(config)),
num_cpu_cores_(num_cpu_cores),
@ -222,6 +225,14 @@ VideoReceiveStream::VideoReceiveStream(int num_cpu_cores,
clock_, jitter_estimator_.get(), timing_.get(), &stats_proxy_));
process_thread_->RegisterModule(&rtp_stream_sync_, RTC_FROM_HERE);
// Register with RtpStreamReceiverController.
media_receiver_ = receiver_controller->CreateReceiver(
config_.rtp.remote_ssrc, &rtp_video_stream_receiver_);
if (config.rtp.rtx_ssrc) {
rtx_receiver_ = receiver_controller->CreateReceiver(
config_.rtp.rtx_ssrc, &rtp_video_stream_receiver_);
}
}
VideoReceiveStream::~VideoReceiveStream() {
@ -241,10 +252,6 @@ bool VideoReceiveStream::DeliverRtcp(const uint8_t* packet, size_t length) {
return rtp_video_stream_receiver_.DeliverRtcp(packet, length);
}
void VideoReceiveStream::OnRtpPacket(const RtpPacketReceived& packet) {
rtp_video_stream_receiver_.OnRtpPacket(packet);
}
void VideoReceiveStream::SetSync(Syncable* audio_syncable) {
RTC_DCHECK_RUN_ON(&worker_thread_checker_);
rtp_stream_sync_.ConfigureSync(audio_syncable);

14
webrtc/video/video_receive_stream.h
View File

@ -36,6 +36,8 @@ class CallStats;
class IvfFileWriter;
class ProcessThread;
class RTPFragmentationHeader;
class RtpStreamReceiverInterface;
class RtpStreamReceiverControllerInterface;
class VCMTiming;
class VCMJitterEstimator;
@ -47,10 +49,10 @@ class VideoReceiveStream : public webrtc::VideoReceiveStream,
public NackSender,
public KeyFrameRequestSender,
public video_coding::OnCompleteFrameCallback,
public Syncable,
public RtpPacketSinkInterface {
public Syncable {
public:
VideoReceiveStream(int num_cpu_cores,
VideoReceiveStream(RtpStreamReceiverControllerInterface* receiver_controller,
int num_cpu_cores,
PacketRouter* packet_router,
VideoReceiveStream::Config config,
ProcessThread* process_thread,
@ -78,9 +80,6 @@ class VideoReceiveStream : public webrtc::VideoReceiveStream,
void EnableEncodedFrameRecording(rtc::PlatformFile file,
size_t byte_limit) override;
// RtpPacketSinkInterface.
void OnRtpPacket(const RtpPacketReceived& packet) override;
// Implements rtc::VideoSinkInterface<VideoFrame>.
void OnFrame(const VideoFrame& video_frame) override;
@ -137,6 +136,9 @@ class VideoReceiveStream : public webrtc::VideoReceiveStream,
// Members for the new jitter buffer experiment.
std::unique_ptr<VCMJitterEstimator> jitter_estimator_;
std::unique_ptr<video_coding::FrameBuffer> frame_buffer_;
std::unique_ptr<RtpStreamReceiverInterface> media_receiver_;
std::unique_ptr<RtpStreamReceiverInterface> rtx_receiver_;
};
} // namespace internal
} // namespace webrtc

14
webrtc/video/video_receive_stream_unittest.cc
View File

@ -16,6 +16,7 @@
#include "webrtc/api/video_codecs/video_decoder.h"
#include "webrtc/base/criticalsection.h"
#include "webrtc/base/event.h"
#include "webrtc/call/rtp_stream_receiver_controller.h"
#include "webrtc/media/base/fakevideorenderer.h"
#include "webrtc/modules/pacing/packet_router.h"
#include "webrtc/modules/rtp_rtcp/source/rtp_packet_to_send.h"
@ -25,15 +26,14 @@
#include "webrtc/system_wrappers/include/clock.h"
#include "webrtc/test/field_trial.h"
namespace webrtc {
namespace {
using testing::_;
using testing::Invoke;
constexpr int kDefaultTimeOutMs = 50;
namespace webrtc {
namespace {
const char kNewJitterBufferFieldTrialEnabled[] =
"WebRTC-NewVideoJitterBuffer/Enabled/";
@ -91,7 +91,7 @@ class VideoReceiveStreamTest : public testing::Test {
config_.decoders.push_back(null_decoder);
video_receive_stream_.reset(new webrtc::internal::VideoReceiveStream(
kDefaultNumCpuCores,
&rtp_stream_receiver_controller_, kDefaultNumCpuCores,
&packet_router_, config_.Copy(), process_thread_.get(), &call_stats_));
}
@ -105,6 +105,7 @@ class VideoReceiveStreamTest : public testing::Test {
MockTransport mock_transport_;
PacketRouter packet_router_;
std::unique_ptr<ProcessThread> process_thread_;
RtpStreamReceiverController rtp_stream_receiver_controller_;
std::unique_ptr<webrtc::internal::VideoReceiveStream> video_receive_stream_;
};
@ -130,9 +131,10 @@ TEST_F(VideoReceiveStreamTest, CreateFrameFromH264FmtpSpropAndIdr) {
EXPECT_CALL(mock_h264_video_decoder_, Decode(_, false, _, _, _));
RtpPacketReceived parsed_packet;
ASSERT_TRUE(parsed_packet.Parse(rtppacket.data(), rtppacket.size()));
video_receive_stream_->OnRtpPacket(parsed_packet);
rtp_stream_receiver_controller_.OnRtpPacket(parsed_packet);
EXPECT_CALL(mock_h264_video_decoder_, Release());
// Make sure the decoder thread had a chance to run.
init_decode_event_.Wait(kDefaultTimeOutMs);
}
} // namespace webrtc

7
webrtc/voice_engine/channel_proxy.h
View File

@ -17,6 +17,7 @@
#include "webrtc/base/constructormagic.h"
#include "webrtc/base/race_checker.h"
#include "webrtc/base/thread_checker.h"
#include "webrtc/call/rtp_packet_sink_interface.h"
#include "webrtc/voice_engine/channel_manager.h"
#include "webrtc/voice_engine/include/voe_rtp_rtcp.h"
@ -50,7 +51,7 @@ class Channel;
// voe::Channel class.
// 2. Provide a refined interface for the stream classes, including assumptions
// on return values and input adaptation.
class ChannelProxy {
class ChannelProxy : public RtpPacketSinkInterface {
public:
ChannelProxy();
explicit ChannelProxy(const ChannelOwner& channel_owner);
@ -94,7 +95,9 @@ class ChannelProxy {
virtual void SetInputMute(bool muted);
virtual void RegisterExternalTransport(Transport* transport);
virtual void DeRegisterExternalTransport();
virtual void OnRtpPacket(const RtpPacketReceived& packet);
// Implements RtpPacketSinkInterface
void OnRtpPacket(const RtpPacketReceived& packet) override;
virtual bool ReceivedRTCPPacket(const uint8_t* packet, size_t length);
virtual const rtc::scoped_refptr<AudioDecoderFactory>&
GetAudioDecoderFactory() const;