end2end_test_renderer.cc 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442
  1. // Copyright 2017 The Chromium Authors. All rights reserved.
  2. // Use of this source code is governed by a BSD-style license that can be
  3. // found in the LICENSE file.
  4. #include "media/remoting/end2end_test_renderer.h"
  5. #include "base/memory/raw_ptr.h"
  6. #include <memory>
  7. #include "base/bind.h"
  8. #include "base/callback.h"
  9. #include "base/callback_helpers.h"
  10. #include "base/check.h"
  11. #include "base/notreached.h"
  12. #include "base/threading/thread_task_runner_handle.h"
  13. #include "components/cast_streaming/public/remoting_proto_utils.h"
  14. #include "media/base/demuxer_stream.h"
  15. #include "media/mojo/common/mojo_data_pipe_read_write.h"
  16. #include "media/mojo/common/mojo_decoder_buffer_converter.h"
  17. #include "media/mojo/mojom/remoting.mojom.h"
  18. #include "media/remoting/courier_renderer.h"
  19. #include "media/remoting/receiver.h"
  20. #include "media/remoting/receiver_controller.h"
  21. #include "media/remoting/renderer_controller.h"
  22. #include "media/remoting/stream_provider.h"
  23. #include "media/remoting/test_utils.h"
  24. #include "mojo/public/cpp/bindings/pending_receiver.h"
  25. #include "mojo/public/cpp/bindings/pending_remote.h"
  26. #include "mojo/public/cpp/bindings/receiver.h"
  27. #include "mojo/public/cpp/bindings/remote.h"
  28. #include "mojo/public/cpp/bindings/self_owned_receiver.h"
  29. #include "mojo/public/cpp/system/data_pipe.h"
  30. using openscreen::cast::RpcMessenger;
  31. namespace media {
  32. namespace remoting {
  33. namespace {
  34. class TestStreamSender final : public mojom::RemotingDataStreamSender {
  35. public:
  36. using SendFrameToSinkCallback =
  37. base::RepeatingCallback<void(uint32_t frame_count,
  38. const std::vector<uint8_t>& data,
  39. DemuxerStream::Type type)>;
  40. TestStreamSender(
  41. mojo::PendingReceiver<mojom::RemotingDataStreamSender> receiver,
  42. mojo::ScopedDataPipeConsumerHandle handle,
  43. DemuxerStream::Type type,
  44. SendFrameToSinkCallback callback)
  45. : receiver_(this, std::move(receiver)),
  46. data_pipe_reader_(std::move(handle)),
  47. type_(type),
  48. send_frame_to_sink_cb_(std::move(callback)) {}
  49. TestStreamSender(const TestStreamSender&) = delete;
  50. TestStreamSender& operator=(const TestStreamSender&) = delete;
  51. ~TestStreamSender() override = default;
  52. // mojom::RemotingDataStreamSender implementation.
  53. void SendFrame(uint32_t frame_size) override {
  54. next_frame_data_.resize(frame_size);
  55. data_pipe_reader_.Read(
  56. next_frame_data_.data(), frame_size,
  57. base::BindOnce(&TestStreamSender::OnFrameRead, base::Unretained(this),
  58. frame_count_++));
  59. }
  60. void CancelInFlightData() override { next_frame_data_.resize(0); }
  61. private:
  62. void OnFrameRead(uint32_t count, bool success) {
  63. DCHECK(success);
  64. if (send_frame_to_sink_cb_)
  65. send_frame_to_sink_cb_.Run(count, next_frame_data_, type_);
  66. next_frame_data_.resize(0);
  67. }
  68. uint32_t frame_count_ = 0;
  69. mojo::Receiver<RemotingDataStreamSender> receiver_;
  70. MojoDataPipeReader data_pipe_reader_;
  71. const DemuxerStream::Type type_;
  72. const SendFrameToSinkCallback send_frame_to_sink_cb_;
  73. std::vector<uint8_t> next_frame_data_;
  74. };
  75. class TestRemoter final : public mojom::Remoter {
  76. public:
  77. using SendMessageToSinkCallback =
  78. base::RepeatingCallback<void(const std::vector<uint8_t>& message)>;
  79. TestRemoter(mojo::PendingRemote<mojom::RemotingSource> source,
  80. SendMessageToSinkCallback send_message_to_sink_cb,
  81. TestStreamSender::SendFrameToSinkCallback send_frame_to_sink_cb)
  82. : source_(std::move(source)),
  83. send_message_to_sink_cb_(std::move(send_message_to_sink_cb)),
  84. send_frame_to_sink_cb_(std::move(send_frame_to_sink_cb)) {}
  85. TestRemoter(const TestRemoter&) = delete;
  86. TestRemoter& operator=(const TestRemoter&) = delete;
  87. ~TestRemoter() override = default;
  88. // mojom::Remoter implementation.
  89. void Start() override { source_->OnStarted(); }
  90. void StartDataStreams(mojo::ScopedDataPipeConsumerHandle audio_pipe,
  91. mojo::ScopedDataPipeConsumerHandle video_pipe,
  92. mojo::PendingReceiver<mojom::RemotingDataStreamSender>
  93. audio_sender_receiver,
  94. mojo::PendingReceiver<mojom::RemotingDataStreamSender>
  95. video_sender_receiver) override {
  96. if (audio_pipe.is_valid()) {
  97. audio_stream_sender_ = std::make_unique<TestStreamSender>(
  98. std::move(audio_sender_receiver), std::move(audio_pipe),
  99. DemuxerStream::AUDIO, send_frame_to_sink_cb_);
  100. }
  101. if (video_pipe.is_valid()) {
  102. video_stream_sender_ = std::make_unique<TestStreamSender>(
  103. std::move(video_sender_receiver), std::move(video_pipe),
  104. DemuxerStream::VIDEO, send_frame_to_sink_cb_);
  105. }
  106. }
  107. void Stop(mojom::RemotingStopReason reason) override {
  108. source_->OnStopped(reason);
  109. }
  110. void SendMessageToSink(const std::vector<uint8_t>& message) override {
  111. if (send_message_to_sink_cb_)
  112. send_message_to_sink_cb_.Run(message);
  113. }
  114. void EstimateTransmissionCapacity(
  115. mojom::Remoter::EstimateTransmissionCapacityCallback callback) override {
  116. std::move(callback).Run(0);
  117. }
  118. // Called when receives RPC messages from receiver.
  119. void OnMessageFromSink(const std::vector<uint8_t>& message) {
  120. source_->OnMessageFromSink(message);
  121. }
  122. private:
  123. mojo::Remote<mojom::RemotingSource> source_;
  124. const SendMessageToSinkCallback send_message_to_sink_cb_;
  125. const TestStreamSender::SendFrameToSinkCallback send_frame_to_sink_cb_;
  126. std::unique_ptr<TestStreamSender> audio_stream_sender_;
  127. std::unique_ptr<TestStreamSender> video_stream_sender_;
  128. };
  129. std::unique_ptr<RendererController> CreateController(
  130. TestRemoter::SendMessageToSinkCallback send_message_to_sink_cb,
  131. TestStreamSender::SendFrameToSinkCallback send_frame_to_sink_cb) {
  132. mojo::PendingRemote<mojom::RemotingSource> remoting_source;
  133. auto remoting_source_receiver =
  134. remoting_source.InitWithNewPipeAndPassReceiver();
  135. mojo::PendingRemote<mojom::Remoter> remoter;
  136. std::unique_ptr<TestRemoter> test_remoter = std::make_unique<TestRemoter>(
  137. std::move(remoting_source), std::move(send_message_to_sink_cb),
  138. std::move(send_frame_to_sink_cb));
  139. mojo::MakeSelfOwnedReceiver(std::move(test_remoter),
  140. remoter.InitWithNewPipeAndPassReceiver());
  141. return std::make_unique<RendererController>(
  142. std::move(remoting_source_receiver), std::move(remoter));
  143. }
  144. } // namespace
  145. class End2EndTestRenderer::TestRemotee : public mojom::Remotee {
  146. public:
  147. explicit TestRemotee(RendererController* controller)
  148. : controller_(controller) {}
  149. ~TestRemotee() override = default;
  150. void OnAudioFrame(uint32_t frame_count,
  151. scoped_refptr<DecoderBuffer> decoder_buffer) {
  152. ::media::mojom::DecoderBufferPtr mojo_buffer =
  153. audio_buffer_writer_->WriteDecoderBuffer(std::move(decoder_buffer));
  154. audio_stream_->ReceiveFrame(frame_count, std::move(mojo_buffer));
  155. }
  156. void OnVideoFrame(uint32_t frame_count,
  157. scoped_refptr<DecoderBuffer> decoder_buffer) {
  158. ::media::mojom::DecoderBufferPtr mojo_buffer =
  159. video_buffer_writer_->WriteDecoderBuffer(std::move(decoder_buffer));
  160. video_stream_->ReceiveFrame(frame_count, std::move(mojo_buffer));
  161. }
  162. void BindMojoReceiver(mojo::PendingReceiver<mojom::Remotee> receiver) {
  163. mojo_receiver_.Bind(std::move(receiver));
  164. }
  165. void OnMessage(const std::vector<uint8_t>& message) {
  166. receiver_controller_->OnMessageFromSource(message);
  167. }
  168. // mojom::Remotee implementation
  169. void OnRemotingSinkReady(
  170. mojo::PendingRemote<::media::mojom::RemotingSink> sink) override {
  171. receiver_controller_.Bind(std::move(sink));
  172. }
  173. void SendMessageToSource(const std::vector<uint8_t>& message) override {
  174. controller_->OnMessageFromSink(message);
  175. }
  176. void StartDataStreams(
  177. mojo::PendingRemote<::media::mojom::RemotingDataStreamReceiver>
  178. audio_stream,
  179. mojo::PendingRemote<::media::mojom::RemotingDataStreamReceiver>
  180. video_stream) override {
  181. if (audio_stream.is_valid()) {
  182. // initialize data pipe for audio data stream receiver
  183. mojo::ScopedDataPipeConsumerHandle audio_data_pipe;
  184. audio_stream_.Bind(std::move(audio_stream));
  185. audio_buffer_writer_ = ::media::MojoDecoderBufferWriter::Create(
  186. GetDefaultDecoderBufferConverterCapacity(
  187. ::media::DemuxerStream::AUDIO),
  188. &audio_data_pipe);
  189. audio_stream_->InitializeDataPipe(std::move(audio_data_pipe));
  190. }
  191. if (video_stream.is_valid()) {
  192. // initialize data pipe for video data stream receiver
  193. mojo::ScopedDataPipeConsumerHandle video_data_pipe;
  194. video_stream_.Bind(std::move(video_stream));
  195. video_buffer_writer_ = ::media::MojoDecoderBufferWriter::Create(
  196. GetDefaultDecoderBufferConverterCapacity(
  197. ::media::DemuxerStream::VIDEO),
  198. &video_data_pipe);
  199. video_stream_->InitializeDataPipe(std::move(video_data_pipe));
  200. }
  201. }
  202. void OnFlushUntil(uint32_t audio_frame_count,
  203. uint32_t video_frame_count) override {}
  204. void OnVideoNaturalSizeChange(const gfx::Size& size) override {}
  205. private:
  206. raw_ptr<RendererController> controller_;
  207. std::unique_ptr<MojoDecoderBufferWriter> audio_buffer_writer_;
  208. std::unique_ptr<MojoDecoderBufferWriter> video_buffer_writer_;
  209. mojo::Remote<mojom::RemotingDataStreamReceiver> audio_stream_;
  210. mojo::Remote<mojom::RemotingDataStreamReceiver> video_stream_;
  211. mojo::Remote<mojom::RemotingSink> receiver_controller_;
  212. mojo::Receiver<mojom::Remotee> mojo_receiver_{this};
  213. };
  214. End2EndTestRenderer::End2EndTestRenderer(std::unique_ptr<Renderer> renderer)
  215. : courier_renderer_initialized_(false), receiver_initialized_(false) {
  216. // create sender components
  217. controller_ = CreateController(
  218. base::BindRepeating(&End2EndTestRenderer::SendMessageToSink,
  219. weak_factory_.GetWeakPtr()),
  220. base::BindRepeating(&End2EndTestRenderer::SendFrameToSink,
  221. weak_factory_.GetWeakPtr()));
  222. courier_renderer_ = std::make_unique<CourierRenderer>(
  223. base::ThreadTaskRunnerHandle::Get(), controller_->GetWeakPtr(), nullptr);
  224. // create receiver components
  225. media_remotee_ = std::make_unique<TestRemotee>(controller_.get());
  226. receiver_controller_ = ReceiverController::GetInstance();
  227. ResetForTesting(receiver_controller_);
  228. receiver_rpc_messenger_ = receiver_controller_->rpc_messenger();
  229. receiver_renderer_handle_ = receiver_rpc_messenger_->GetUniqueHandle();
  230. receiver_rpc_messenger_->RegisterMessageReceiverCallback(
  231. RpcMessenger::kAcquireRendererHandle,
  232. [ptr = weak_factory_.GetWeakPtr()](
  233. std::unique_ptr<openscreen::cast::RpcMessage> message) {
  234. if (ptr) {
  235. ptr->OnReceivedRpc(std::move(message));
  236. }
  237. });
  238. receiver_ = std::make_unique<Receiver>(
  239. receiver_renderer_handle_, sender_renderer_handle_, receiver_controller_,
  240. base::ThreadTaskRunnerHandle::Get(), std::move(renderer),
  241. base::BindOnce(&End2EndTestRenderer::OnAcquireRendererDone,
  242. weak_factory_.GetWeakPtr()));
  243. mojo::PendingRemote<media::mojom::Remotee> remotee;
  244. media_remotee_->BindMojoReceiver(remotee.InitWithNewPipeAndPassReceiver());
  245. receiver_controller_->Initialize(std::move(remotee));
  246. stream_provider_ = std::make_unique<StreamProvider>(
  247. receiver_controller_, base::ThreadTaskRunnerHandle::Get());
  248. }
  249. End2EndTestRenderer::~End2EndTestRenderer() {
  250. receiver_rpc_messenger_->UnregisterMessageReceiverCallback(
  251. RpcMessenger::kAcquireRendererHandle);
  252. }
  253. void End2EndTestRenderer::Initialize(MediaResource* media_resource,
  254. RendererClient* client,
  255. PipelineStatusCallback init_cb) {
  256. init_cb_ = std::move(init_cb);
  257. stream_provider_->Initialize(
  258. nullptr, base::BindOnce(&End2EndTestRenderer::InitializeReceiverRenderer,
  259. weak_factory_.GetWeakPtr()));
  260. courier_renderer_->Initialize(
  261. media_resource, client,
  262. base::BindOnce(&End2EndTestRenderer::OnCourierRendererInitialized,
  263. weak_factory_.GetWeakPtr()));
  264. }
  265. void End2EndTestRenderer::InitializeReceiverRenderer(PipelineStatus status) {
  266. DCHECK(status == PIPELINE_OK);
  267. receiver_->Initialize(
  268. stream_provider_.get(), nullptr,
  269. base::BindOnce(&End2EndTestRenderer::OnReceiverInitialized,
  270. weak_factory_.GetWeakPtr()));
  271. }
  272. void End2EndTestRenderer::OnCourierRendererInitialized(PipelineStatus status) {
  273. DCHECK(status == PIPELINE_OK);
  274. courier_renderer_initialized_ = true;
  275. CompleteInitialize();
  276. }
  277. void End2EndTestRenderer::OnReceiverInitialized(PipelineStatus status) {
  278. DCHECK(status == PIPELINE_OK);
  279. receiver_initialized_ = true;
  280. CompleteInitialize();
  281. }
  282. void End2EndTestRenderer::CompleteInitialize() {
  283. if (!courier_renderer_initialized_ || !receiver_initialized_)
  284. return;
  285. DCHECK(init_cb_);
  286. std::move(init_cb_).Run(PIPELINE_OK);
  287. }
  288. void End2EndTestRenderer::OnReceivedRpc(
  289. std::unique_ptr<openscreen::cast::RpcMessage> message) {
  290. DCHECK(message);
  291. DCHECK_EQ(message->proc(),
  292. openscreen::cast::RpcMessage::RPC_ACQUIRE_RENDERER);
  293. OnAcquireRenderer(std::move(message));
  294. }
  295. void End2EndTestRenderer::OnAcquireRenderer(
  296. std::unique_ptr<openscreen::cast::RpcMessage> message) {
  297. DCHECK(message->has_integer_value());
  298. DCHECK(message->integer_value() != RpcMessenger::kInvalidHandle);
  299. if (sender_renderer_handle_ == RpcMessenger::kInvalidHandle) {
  300. sender_renderer_handle_ = message->integer_value();
  301. receiver_->SetRemoteHandle(sender_renderer_handle_);
  302. }
  303. }
  304. void End2EndTestRenderer::OnAcquireRendererDone(int receiver_renderer_handle) {
  305. openscreen::cast::RpcMessage rpc;
  306. rpc.set_handle(sender_renderer_handle_);
  307. rpc.set_proc(openscreen::cast::RpcMessage::RPC_ACQUIRE_RENDERER_DONE);
  308. rpc.set_integer_value(receiver_renderer_handle);
  309. receiver_rpc_messenger_->SendMessageToRemote(rpc);
  310. }
  311. void End2EndTestRenderer::SetLatencyHint(
  312. absl::optional<base::TimeDelta> latency_hint) {
  313. courier_renderer_->SetLatencyHint(latency_hint);
  314. }
  315. void End2EndTestRenderer::SetPreservesPitch(bool preserves_pitch) {
  316. courier_renderer_->SetPreservesPitch(preserves_pitch);
  317. }
  318. void End2EndTestRenderer::Flush(base::OnceClosure flush_cb) {
  319. courier_renderer_->Flush(std::move(flush_cb));
  320. }
  321. void End2EndTestRenderer::StartPlayingFrom(base::TimeDelta time) {
  322. courier_renderer_->StartPlayingFrom(time);
  323. }
  324. void End2EndTestRenderer::SetPlaybackRate(double playback_rate) {
  325. courier_renderer_->SetPlaybackRate(playback_rate);
  326. }
  327. void End2EndTestRenderer::SetVolume(float volume) {
  328. courier_renderer_->SetVolume(volume);
  329. }
  330. base::TimeDelta End2EndTestRenderer::GetMediaTime() {
  331. return courier_renderer_->GetMediaTime();
  332. }
  333. void End2EndTestRenderer::SendMessageToSink(
  334. const std::vector<uint8_t>& message) {
  335. media_remotee_->OnMessage(message);
  336. }
  337. void End2EndTestRenderer::SendFrameToSink(uint32_t frame_count,
  338. const std::vector<uint8_t>& frame,
  339. DemuxerStream::Type type) {
  340. scoped_refptr<DecoderBuffer> decoder_buffer =
  341. cast_streaming::remoting::ByteArrayToDecoderBuffer(frame.data(),
  342. frame.size());
  343. if (type == DemuxerStream::Type::AUDIO) {
  344. media_remotee_->OnAudioFrame(frame_count, decoder_buffer);
  345. } else if (type == DemuxerStream::Type::VIDEO) {
  346. media_remotee_->OnVideoFrame(frame_count, decoder_buffer);
  347. } else {
  348. NOTREACHED();
  349. }
  350. }
  351. void End2EndTestRenderer::OnMessageFromSink(
  352. std::unique_ptr<std::vector<uint8_t>> message) {
  353. controller_->OnMessageFromSink(*message);
  354. }
  355. void End2EndTestRenderer::OnSelectedVideoTracksChanged(
  356. const std::vector<DemuxerStream*>& enabled_tracks,
  357. base::OnceClosure change_completed_cb) {
  358. courier_renderer_->OnSelectedVideoTracksChanged(
  359. enabled_tracks, std::move(change_completed_cb));
  360. }
  361. void End2EndTestRenderer::OnEnabledAudioTracksChanged(
  362. const std::vector<DemuxerStream*>& enabled_tracks,
  363. base::OnceClosure change_completed_cb) {
  364. courier_renderer_->OnEnabledAudioTracksChanged(
  365. enabled_tracks, std::move(change_completed_cb));
  366. }
  367. } // namespace remoting
  368. } // namespace media