123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442 |
- // Copyright 2017 The Chromium Authors. All rights reserved.
- // Use of this source code is governed by a BSD-style license that can be
- // found in the LICENSE file.
- #include "media/remoting/end2end_test_renderer.h"
- #include "base/memory/raw_ptr.h"
- #include <memory>
- #include "base/bind.h"
- #include "base/callback.h"
- #include "base/callback_helpers.h"
- #include "base/check.h"
- #include "base/notreached.h"
- #include "base/threading/thread_task_runner_handle.h"
- #include "components/cast_streaming/public/remoting_proto_utils.h"
- #include "media/base/demuxer_stream.h"
- #include "media/mojo/common/mojo_data_pipe_read_write.h"
- #include "media/mojo/common/mojo_decoder_buffer_converter.h"
- #include "media/mojo/mojom/remoting.mojom.h"
- #include "media/remoting/courier_renderer.h"
- #include "media/remoting/receiver.h"
- #include "media/remoting/receiver_controller.h"
- #include "media/remoting/renderer_controller.h"
- #include "media/remoting/stream_provider.h"
- #include "media/remoting/test_utils.h"
- #include "mojo/public/cpp/bindings/pending_receiver.h"
- #include "mojo/public/cpp/bindings/pending_remote.h"
- #include "mojo/public/cpp/bindings/receiver.h"
- #include "mojo/public/cpp/bindings/remote.h"
- #include "mojo/public/cpp/bindings/self_owned_receiver.h"
- #include "mojo/public/cpp/system/data_pipe.h"
- using openscreen::cast::RpcMessenger;
- namespace media {
- namespace remoting {
- namespace {
- class TestStreamSender final : public mojom::RemotingDataStreamSender {
- public:
- using SendFrameToSinkCallback =
- base::RepeatingCallback<void(uint32_t frame_count,
- const std::vector<uint8_t>& data,
- DemuxerStream::Type type)>;
- TestStreamSender(
- mojo::PendingReceiver<mojom::RemotingDataStreamSender> receiver,
- mojo::ScopedDataPipeConsumerHandle handle,
- DemuxerStream::Type type,
- SendFrameToSinkCallback callback)
- : receiver_(this, std::move(receiver)),
- data_pipe_reader_(std::move(handle)),
- type_(type),
- send_frame_to_sink_cb_(std::move(callback)) {}
- TestStreamSender(const TestStreamSender&) = delete;
- TestStreamSender& operator=(const TestStreamSender&) = delete;
- ~TestStreamSender() override = default;
- // mojom::RemotingDataStreamSender implementation.
- void SendFrame(uint32_t frame_size) override {
- next_frame_data_.resize(frame_size);
- data_pipe_reader_.Read(
- next_frame_data_.data(), frame_size,
- base::BindOnce(&TestStreamSender::OnFrameRead, base::Unretained(this),
- frame_count_++));
- }
- void CancelInFlightData() override { next_frame_data_.resize(0); }
- private:
- void OnFrameRead(uint32_t count, bool success) {
- DCHECK(success);
- if (send_frame_to_sink_cb_)
- send_frame_to_sink_cb_.Run(count, next_frame_data_, type_);
- next_frame_data_.resize(0);
- }
- uint32_t frame_count_ = 0;
- mojo::Receiver<RemotingDataStreamSender> receiver_;
- MojoDataPipeReader data_pipe_reader_;
- const DemuxerStream::Type type_;
- const SendFrameToSinkCallback send_frame_to_sink_cb_;
- std::vector<uint8_t> next_frame_data_;
- };
- class TestRemoter final : public mojom::Remoter {
- public:
- using SendMessageToSinkCallback =
- base::RepeatingCallback<void(const std::vector<uint8_t>& message)>;
- TestRemoter(mojo::PendingRemote<mojom::RemotingSource> source,
- SendMessageToSinkCallback send_message_to_sink_cb,
- TestStreamSender::SendFrameToSinkCallback send_frame_to_sink_cb)
- : source_(std::move(source)),
- send_message_to_sink_cb_(std::move(send_message_to_sink_cb)),
- send_frame_to_sink_cb_(std::move(send_frame_to_sink_cb)) {}
- TestRemoter(const TestRemoter&) = delete;
- TestRemoter& operator=(const TestRemoter&) = delete;
- ~TestRemoter() override = default;
- // mojom::Remoter implementation.
- void Start() override { source_->OnStarted(); }
- void StartDataStreams(mojo::ScopedDataPipeConsumerHandle audio_pipe,
- mojo::ScopedDataPipeConsumerHandle video_pipe,
- mojo::PendingReceiver<mojom::RemotingDataStreamSender>
- audio_sender_receiver,
- mojo::PendingReceiver<mojom::RemotingDataStreamSender>
- video_sender_receiver) override {
- if (audio_pipe.is_valid()) {
- audio_stream_sender_ = std::make_unique<TestStreamSender>(
- std::move(audio_sender_receiver), std::move(audio_pipe),
- DemuxerStream::AUDIO, send_frame_to_sink_cb_);
- }
- if (video_pipe.is_valid()) {
- video_stream_sender_ = std::make_unique<TestStreamSender>(
- std::move(video_sender_receiver), std::move(video_pipe),
- DemuxerStream::VIDEO, send_frame_to_sink_cb_);
- }
- }
- void Stop(mojom::RemotingStopReason reason) override {
- source_->OnStopped(reason);
- }
- void SendMessageToSink(const std::vector<uint8_t>& message) override {
- if (send_message_to_sink_cb_)
- send_message_to_sink_cb_.Run(message);
- }
- void EstimateTransmissionCapacity(
- mojom::Remoter::EstimateTransmissionCapacityCallback callback) override {
- std::move(callback).Run(0);
- }
- // Called when receives RPC messages from receiver.
- void OnMessageFromSink(const std::vector<uint8_t>& message) {
- source_->OnMessageFromSink(message);
- }
- private:
- mojo::Remote<mojom::RemotingSource> source_;
- const SendMessageToSinkCallback send_message_to_sink_cb_;
- const TestStreamSender::SendFrameToSinkCallback send_frame_to_sink_cb_;
- std::unique_ptr<TestStreamSender> audio_stream_sender_;
- std::unique_ptr<TestStreamSender> video_stream_sender_;
- };
- std::unique_ptr<RendererController> CreateController(
- TestRemoter::SendMessageToSinkCallback send_message_to_sink_cb,
- TestStreamSender::SendFrameToSinkCallback send_frame_to_sink_cb) {
- mojo::PendingRemote<mojom::RemotingSource> remoting_source;
- auto remoting_source_receiver =
- remoting_source.InitWithNewPipeAndPassReceiver();
- mojo::PendingRemote<mojom::Remoter> remoter;
- std::unique_ptr<TestRemoter> test_remoter = std::make_unique<TestRemoter>(
- std::move(remoting_source), std::move(send_message_to_sink_cb),
- std::move(send_frame_to_sink_cb));
- mojo::MakeSelfOwnedReceiver(std::move(test_remoter),
- remoter.InitWithNewPipeAndPassReceiver());
- return std::make_unique<RendererController>(
- std::move(remoting_source_receiver), std::move(remoter));
- }
- } // namespace
- class End2EndTestRenderer::TestRemotee : public mojom::Remotee {
- public:
- explicit TestRemotee(RendererController* controller)
- : controller_(controller) {}
- ~TestRemotee() override = default;
- void OnAudioFrame(uint32_t frame_count,
- scoped_refptr<DecoderBuffer> decoder_buffer) {
- ::media::mojom::DecoderBufferPtr mojo_buffer =
- audio_buffer_writer_->WriteDecoderBuffer(std::move(decoder_buffer));
- audio_stream_->ReceiveFrame(frame_count, std::move(mojo_buffer));
- }
- void OnVideoFrame(uint32_t frame_count,
- scoped_refptr<DecoderBuffer> decoder_buffer) {
- ::media::mojom::DecoderBufferPtr mojo_buffer =
- video_buffer_writer_->WriteDecoderBuffer(std::move(decoder_buffer));
- video_stream_->ReceiveFrame(frame_count, std::move(mojo_buffer));
- }
- void BindMojoReceiver(mojo::PendingReceiver<mojom::Remotee> receiver) {
- mojo_receiver_.Bind(std::move(receiver));
- }
- void OnMessage(const std::vector<uint8_t>& message) {
- receiver_controller_->OnMessageFromSource(message);
- }
- // mojom::Remotee implementation
- void OnRemotingSinkReady(
- mojo::PendingRemote<::media::mojom::RemotingSink> sink) override {
- receiver_controller_.Bind(std::move(sink));
- }
- void SendMessageToSource(const std::vector<uint8_t>& message) override {
- controller_->OnMessageFromSink(message);
- }
- void StartDataStreams(
- mojo::PendingRemote<::media::mojom::RemotingDataStreamReceiver>
- audio_stream,
- mojo::PendingRemote<::media::mojom::RemotingDataStreamReceiver>
- video_stream) override {
- if (audio_stream.is_valid()) {
- // initialize data pipe for audio data stream receiver
- mojo::ScopedDataPipeConsumerHandle audio_data_pipe;
- audio_stream_.Bind(std::move(audio_stream));
- audio_buffer_writer_ = ::media::MojoDecoderBufferWriter::Create(
- GetDefaultDecoderBufferConverterCapacity(
- ::media::DemuxerStream::AUDIO),
- &audio_data_pipe);
- audio_stream_->InitializeDataPipe(std::move(audio_data_pipe));
- }
- if (video_stream.is_valid()) {
- // initialize data pipe for video data stream receiver
- mojo::ScopedDataPipeConsumerHandle video_data_pipe;
- video_stream_.Bind(std::move(video_stream));
- video_buffer_writer_ = ::media::MojoDecoderBufferWriter::Create(
- GetDefaultDecoderBufferConverterCapacity(
- ::media::DemuxerStream::VIDEO),
- &video_data_pipe);
- video_stream_->InitializeDataPipe(std::move(video_data_pipe));
- }
- }
- void OnFlushUntil(uint32_t audio_frame_count,
- uint32_t video_frame_count) override {}
- void OnVideoNaturalSizeChange(const gfx::Size& size) override {}
- private:
- raw_ptr<RendererController> controller_;
- std::unique_ptr<MojoDecoderBufferWriter> audio_buffer_writer_;
- std::unique_ptr<MojoDecoderBufferWriter> video_buffer_writer_;
- mojo::Remote<mojom::RemotingDataStreamReceiver> audio_stream_;
- mojo::Remote<mojom::RemotingDataStreamReceiver> video_stream_;
- mojo::Remote<mojom::RemotingSink> receiver_controller_;
- mojo::Receiver<mojom::Remotee> mojo_receiver_{this};
- };
- End2EndTestRenderer::End2EndTestRenderer(std::unique_ptr<Renderer> renderer)
- : courier_renderer_initialized_(false), receiver_initialized_(false) {
- // create sender components
- controller_ = CreateController(
- base::BindRepeating(&End2EndTestRenderer::SendMessageToSink,
- weak_factory_.GetWeakPtr()),
- base::BindRepeating(&End2EndTestRenderer::SendFrameToSink,
- weak_factory_.GetWeakPtr()));
- courier_renderer_ = std::make_unique<CourierRenderer>(
- base::ThreadTaskRunnerHandle::Get(), controller_->GetWeakPtr(), nullptr);
- // create receiver components
- media_remotee_ = std::make_unique<TestRemotee>(controller_.get());
- receiver_controller_ = ReceiverController::GetInstance();
- ResetForTesting(receiver_controller_);
- receiver_rpc_messenger_ = receiver_controller_->rpc_messenger();
- receiver_renderer_handle_ = receiver_rpc_messenger_->GetUniqueHandle();
- receiver_rpc_messenger_->RegisterMessageReceiverCallback(
- RpcMessenger::kAcquireRendererHandle,
- [ptr = weak_factory_.GetWeakPtr()](
- std::unique_ptr<openscreen::cast::RpcMessage> message) {
- if (ptr) {
- ptr->OnReceivedRpc(std::move(message));
- }
- });
- receiver_ = std::make_unique<Receiver>(
- receiver_renderer_handle_, sender_renderer_handle_, receiver_controller_,
- base::ThreadTaskRunnerHandle::Get(), std::move(renderer),
- base::BindOnce(&End2EndTestRenderer::OnAcquireRendererDone,
- weak_factory_.GetWeakPtr()));
- mojo::PendingRemote<media::mojom::Remotee> remotee;
- media_remotee_->BindMojoReceiver(remotee.InitWithNewPipeAndPassReceiver());
- receiver_controller_->Initialize(std::move(remotee));
- stream_provider_ = std::make_unique<StreamProvider>(
- receiver_controller_, base::ThreadTaskRunnerHandle::Get());
- }
- End2EndTestRenderer::~End2EndTestRenderer() {
- receiver_rpc_messenger_->UnregisterMessageReceiverCallback(
- RpcMessenger::kAcquireRendererHandle);
- }
- void End2EndTestRenderer::Initialize(MediaResource* media_resource,
- RendererClient* client,
- PipelineStatusCallback init_cb) {
- init_cb_ = std::move(init_cb);
- stream_provider_->Initialize(
- nullptr, base::BindOnce(&End2EndTestRenderer::InitializeReceiverRenderer,
- weak_factory_.GetWeakPtr()));
- courier_renderer_->Initialize(
- media_resource, client,
- base::BindOnce(&End2EndTestRenderer::OnCourierRendererInitialized,
- weak_factory_.GetWeakPtr()));
- }
- void End2EndTestRenderer::InitializeReceiverRenderer(PipelineStatus status) {
- DCHECK(status == PIPELINE_OK);
- receiver_->Initialize(
- stream_provider_.get(), nullptr,
- base::BindOnce(&End2EndTestRenderer::OnReceiverInitialized,
- weak_factory_.GetWeakPtr()));
- }
- void End2EndTestRenderer::OnCourierRendererInitialized(PipelineStatus status) {
- DCHECK(status == PIPELINE_OK);
- courier_renderer_initialized_ = true;
- CompleteInitialize();
- }
- void End2EndTestRenderer::OnReceiverInitialized(PipelineStatus status) {
- DCHECK(status == PIPELINE_OK);
- receiver_initialized_ = true;
- CompleteInitialize();
- }
- void End2EndTestRenderer::CompleteInitialize() {
- if (!courier_renderer_initialized_ || !receiver_initialized_)
- return;
- DCHECK(init_cb_);
- std::move(init_cb_).Run(PIPELINE_OK);
- }
- void End2EndTestRenderer::OnReceivedRpc(
- std::unique_ptr<openscreen::cast::RpcMessage> message) {
- DCHECK(message);
- DCHECK_EQ(message->proc(),
- openscreen::cast::RpcMessage::RPC_ACQUIRE_RENDERER);
- OnAcquireRenderer(std::move(message));
- }
- void End2EndTestRenderer::OnAcquireRenderer(
- std::unique_ptr<openscreen::cast::RpcMessage> message) {
- DCHECK(message->has_integer_value());
- DCHECK(message->integer_value() != RpcMessenger::kInvalidHandle);
- if (sender_renderer_handle_ == RpcMessenger::kInvalidHandle) {
- sender_renderer_handle_ = message->integer_value();
- receiver_->SetRemoteHandle(sender_renderer_handle_);
- }
- }
- void End2EndTestRenderer::OnAcquireRendererDone(int receiver_renderer_handle) {
- openscreen::cast::RpcMessage rpc;
- rpc.set_handle(sender_renderer_handle_);
- rpc.set_proc(openscreen::cast::RpcMessage::RPC_ACQUIRE_RENDERER_DONE);
- rpc.set_integer_value(receiver_renderer_handle);
- receiver_rpc_messenger_->SendMessageToRemote(rpc);
- }
- void End2EndTestRenderer::SetLatencyHint(
- absl::optional<base::TimeDelta> latency_hint) {
- courier_renderer_->SetLatencyHint(latency_hint);
- }
- void End2EndTestRenderer::SetPreservesPitch(bool preserves_pitch) {
- courier_renderer_->SetPreservesPitch(preserves_pitch);
- }
- void End2EndTestRenderer::Flush(base::OnceClosure flush_cb) {
- courier_renderer_->Flush(std::move(flush_cb));
- }
- void End2EndTestRenderer::StartPlayingFrom(base::TimeDelta time) {
- courier_renderer_->StartPlayingFrom(time);
- }
- void End2EndTestRenderer::SetPlaybackRate(double playback_rate) {
- courier_renderer_->SetPlaybackRate(playback_rate);
- }
- void End2EndTestRenderer::SetVolume(float volume) {
- courier_renderer_->SetVolume(volume);
- }
- base::TimeDelta End2EndTestRenderer::GetMediaTime() {
- return courier_renderer_->GetMediaTime();
- }
- void End2EndTestRenderer::SendMessageToSink(
- const std::vector<uint8_t>& message) {
- media_remotee_->OnMessage(message);
- }
- void End2EndTestRenderer::SendFrameToSink(uint32_t frame_count,
- const std::vector<uint8_t>& frame,
- DemuxerStream::Type type) {
- scoped_refptr<DecoderBuffer> decoder_buffer =
- cast_streaming::remoting::ByteArrayToDecoderBuffer(frame.data(),
- frame.size());
- if (type == DemuxerStream::Type::AUDIO) {
- media_remotee_->OnAudioFrame(frame_count, decoder_buffer);
- } else if (type == DemuxerStream::Type::VIDEO) {
- media_remotee_->OnVideoFrame(frame_count, decoder_buffer);
- } else {
- NOTREACHED();
- }
- }
- void End2EndTestRenderer::OnMessageFromSink(
- std::unique_ptr<std::vector<uint8_t>> message) {
- controller_->OnMessageFromSink(*message);
- }
- void End2EndTestRenderer::OnSelectedVideoTracksChanged(
- const std::vector<DemuxerStream*>& enabled_tracks,
- base::OnceClosure change_completed_cb) {
- courier_renderer_->OnSelectedVideoTracksChanged(
- enabled_tracks, std::move(change_completed_cb));
- }
- void End2EndTestRenderer::OnEnabledAudioTracksChanged(
- const std::vector<DemuxerStream*>& enabled_tracks,
- base::OnceClosure change_completed_cb) {
- courier_renderer_->OnEnabledAudioTracksChanged(
- enabled_tracks, std::move(change_completed_cb));
- }
- } // namespace remoting
- } // namespace media
|