// Copyright 2020 The Chromium Authors
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
#include "components/cast_streaming/test/cast_streaming_test_sender.h"
#include "base/containers/flat_map.h"
#include "base/logging.h"
#include "base/memory/raw_ptr.h"
#include "base/run_loop.h"
#include "base/task/sequenced_task_runner.h"
#include "components/cast_streaming/test/cast_message_port_sender_impl.h"
#include "media/cast/openscreen/config_conversions.h"
#include "third_party/openscreen/src/platform/base/span.h"
namespace cast_streaming {
namespace {
const char kSenderId[] = "testSenderId";
const char kReceiverId[] = "testReceiverId";
// Converts |decoder_buffer| into an Open Screen EncodedFrame.
openscreen::cast::EncodedFrame DecoderBufferToEncodedFrame(
const media::DecoderBuffer* decoder_buffer,
openscreen::cast::FrameId frame_id,
openscreen::cast::FrameId* last_referenced_frame_id,
int rtp_timebase) {
CHECK(decoder_buffer);
CHECK(!decoder_buffer->end_of_stream());
openscreen::cast::EncodedFrame encoded_frame;
encoded_frame.frame_id = frame_id;
if (decoder_buffer->is_key_frame()) {
encoded_frame.dependency =
openscreen::cast::EncodedFrame::Dependency::kKeyFrame;
*last_referenced_frame_id = encoded_frame.frame_id;
} else {
encoded_frame.dependency =
openscreen::cast::EncodedFrame::Dependency::kDependent;
}
encoded_frame.referenced_frame_id = *last_referenced_frame_id;
std::chrono::milliseconds timestamp(
decoder_buffer->timestamp().InMilliseconds());
encoded_frame.rtp_timestamp =
openscreen::cast::RtpTimeTicks::FromTimeSinceOrigin<
std::chrono::milliseconds>(timestamp, rtp_timebase);
encoded_frame.reference_time = openscreen::Clock::time_point(timestamp);
encoded_frame.data = openscreen::ByteView(decoder_buffer->writable_data(),
decoder_buffer->size());
return encoded_frame;
}
} // namespace
class CastStreamingTestSender::SenderObserver final
: public openscreen::cast::Sender::Observer {
public:
explicit SenderObserver(std::unique_ptr<openscreen::cast::Sender> sender)
: sender_(std::move(sender)) {
CHECK(sender_);
sender_->SetObserver(this);
}
~SenderObserver() override { sender_->SetObserver(nullptr); }
SenderObserver(const SenderObserver&) = delete;
SenderObserver& operator=(const SenderObserver&) = delete;
bool EnqueueBuffer(scoped_refptr<media::DecoderBuffer> buffer) {
VLOG(3) << __func__;
openscreen::cast::FrameId frame_id = sender_->GetNextFrameId();
openscreen::cast::Sender::EnqueueFrameResult result =
sender_->EnqueueFrame(DecoderBufferToEncodedFrame(
buffer.get(), frame_id, &last_reference_frame_id_,
sender_->rtp_timebase()));
if (result != openscreen::cast::Sender::EnqueueFrameResult::OK) {
LOG(ERROR) << "Failed to enqueue buffer " << result;
return false;
}
buffer_map_.emplace(frame_id, std::move(buffer));
return true;
}
private:
// openscreen::cast::Sender::Observer implementation.
void OnFrameCanceled(openscreen::cast::FrameId frame_id) final {
VLOG(3) << __func__ << ". frame_id: " << frame_id;
buffer_map_.erase(frame_id);
}
void OnPictureLost() final {}
openscreen::cast::FrameId last_reference_frame_id_;
std::unique_ptr<openscreen::cast::Sender> sender_;
base::flat_map<openscreen::cast::FrameId, scoped_refptr<media::DecoderBuffer>>
buffer_map_;
};
CastStreamingTestSender::CastStreamingTestSender()
: task_runner_(base::SequencedTaskRunner::GetCurrentDefault()),
environment_(&openscreen::Clock::now,
task_runner_,
openscreen::IPEndpoint::kAnyV4()) {}
CastStreamingTestSender::~CastStreamingTestSender() = default;
void CastStreamingTestSender::Start(
std::unique_ptr<cast_api_bindings::MessagePort> message_port,
net::IPAddress receiver_address,
std::optional<media::AudioDecoderConfig> audio_config,
std::optional<media::VideoDecoderConfig> video_config) {
VLOG(1) << __func__;
CHECK(!has_startup_completed_);
CHECK(!sender_session_);
CHECK(audio_config || video_config);
// Instantiate the |sender_session_|.
message_port_ = std::make_unique<CastMessagePortSenderImpl>(
std::move(message_port),
base::BindOnce(&CastStreamingTestSender::OnCastChannelClosed,
base::Unretained(this)),
base::BindOnce(&CastStreamingTestSender::OnSystemSenderMessageReceived,
base::Unretained(this)));
sender_session_ = std::make_unique<openscreen::cast::SenderSession>(
openscreen::cast::SenderSession::Configuration{
openscreen::IPAddress::kV4LoopbackAddress(), *this, &environment_,
message_port_.get(), kSenderId, kReceiverId,
true /* use_android_rtp_hack */});
if (audio_config) {
audio_configs_.push_back(
media::cast::ToAudioCaptureConfig(audio_config.value()));
}
if (video_config) {
video_configs_.push_back(
media::cast::ToVideoCaptureConfig(video_config.value()));
}
}
void CastStreamingTestSender::Stop() {
VLOG(1) << __func__;
// Senders must be deconstructed before the session that hosts them.
audio_sender_observer_.reset();
video_sender_observer_.reset();
// Disconnect the message port before destructing its client.
if (message_port_) {
// TODO(crbug.com/42050578): CastMessagePortSender should be RAII and clean
// itself during the destruction instead of relying the client to call its
// ResetClient function.
message_port_->ResetClient();
}
sender_session_.reset();
message_port_.reset();
audio_decoder_config_.reset();
video_decoder_config_.reset();
has_startup_completed_ = false;
if (sender_stopped_closure_) {
std::move(sender_stopped_closure_).Run();
}
}
void CastStreamingTestSender::SendAudioBuffer(
scoped_refptr<media::DecoderBuffer> audio_buffer) {
VLOG(3) << __func__;
CHECK(audio_sender_observer_);
if (!audio_sender_observer_->EnqueueBuffer(audio_buffer)) {
// The error has already been logged in EnqueueBuffer().
Stop();
}
}
void CastStreamingTestSender::SendVideoBuffer(
scoped_refptr<media::DecoderBuffer> video_buffer) {
VLOG(3) << __func__;
CHECK(video_sender_observer_);
if (!video_sender_observer_->EnqueueBuffer(video_buffer)) {
// The error has already been logged in EnqueueBuffer().
Stop();
}
}
bool CastStreamingTestSender::RunUntilActive() {
VLOG(1) << __func__;
while (!has_startup_completed_) {
base::RunLoop run_loop;
CHECK(!sender_started_closure_);
sender_started_closure_ = run_loop.QuitClosure();
run_loop.Run();
}
return !!sender_session_;
}
void CastStreamingTestSender::RunUntilStopped() {
VLOG(1) << __func__;
while (has_startup_completed_) {
base::RunLoop run_loop;
CHECK(!sender_stopped_closure_);
sender_stopped_closure_ = run_loop.QuitClosure();
run_loop.Run();
}
}
void CastStreamingTestSender::OnCastChannelClosed() {
VLOG(1) << __func__;
Stop();
}
void CastStreamingTestSender::OnSystemSenderMessageReceived() {
openscreen::Error error = sender_session_->Negotiate(
std::move(audio_configs_), std::move(video_configs_));
if (error != openscreen::Error::None()) {
LOG(ERROR) << "Failed to start sender session. " << error.ToString();
sender_session_.reset();
message_port_.reset();
has_startup_completed_ = true;
if (sender_started_closure_) {
std::move(sender_started_closure_).Run();
}
}
}
void CastStreamingTestSender::OnNegotiated(
const openscreen::cast::SenderSession* session,
openscreen::cast::SenderSession::ConfiguredSenders senders,
openscreen::cast::capture_recommendations::Recommendations
capture_recommendations) {
VLOG(1) << __func__;
CHECK_EQ(session, sender_session_.get());
CHECK(senders.audio_sender || senders.video_sender);
if (senders.audio_sender) {
audio_sender_observer_ =
std::make_unique<SenderObserver>(std::move(senders.audio_sender));
audio_decoder_config_ =
media::cast::ToAudioDecoderConfig(senders.audio_config);
}
if (senders.video_sender) {
video_sender_observer_ =
std::make_unique<SenderObserver>(std::move(senders.video_sender));
video_decoder_config_ =
media::cast::ToVideoDecoderConfig(senders.video_config);
}
has_startup_completed_ = true;
if (sender_started_closure_) {
std::move(sender_started_closure_).Run();
}
}
void CastStreamingTestSender::OnError(
const openscreen::cast::SenderSession* session,
const openscreen::Error& error) {
LOG(ERROR) << "Sender Session error: " << error.ToString();
CHECK_EQ(session, sender_session_.get());
Stop();
}
} // namespace cast_streaming
Codebase Browser
chromium