stream_provider.cc 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653
  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/stream_provider.h"
  5. #include <vector>
  6. #include "base/bind.h"
  7. #include "base/callback.h"
  8. #include "base/callback_helpers.h"
  9. #include "base/containers/circular_deque.h"
  10. #include "base/logging.h"
  11. #include "base/task/bind_post_task.h"
  12. #include "base/task/single_thread_task_runner.h"
  13. #include "base/threading/thread_task_runner_handle.h"
  14. #include "components/cast_streaming/public/remoting_proto_enum_utils.h"
  15. #include "components/cast_streaming/public/remoting_proto_utils.h"
  16. #include "media/base/bind_to_current_loop.h"
  17. #include "media/base/decoder_buffer.h"
  18. #include "media/base/video_transformation.h"
  19. #include "media/mojo/common/mojo_decoder_buffer_converter.h"
  20. #include "media/remoting/receiver_controller.h"
  21. #include "third_party/openscreen/src/cast/streaming/rpc_messenger.h"
  22. using openscreen::cast::RpcMessenger;
  23. namespace media {
  24. namespace remoting {
  25. namespace {
  26. // The number of frames requested in each ReadUntil RPC message.
  27. constexpr int kNumFramesInEachReadUntil = 10;
  28. }
  29. // static
  30. void StreamProvider::MediaStream::CreateOnMainThread(
  31. RpcMessenger* rpc_messenger,
  32. Type type,
  33. int32_t handle,
  34. const scoped_refptr<base::SingleThreadTaskRunner>& media_task_runner,
  35. base::OnceCallback<void(MediaStream::UniquePtr)> callback) {
  36. MediaStream::UniquePtr stream(
  37. new MediaStream(rpc_messenger, type, handle, media_task_runner),
  38. &DestructionHelper);
  39. std::move(callback).Run(std::move(stream));
  40. }
  41. // static
  42. void StreamProvider::MediaStream::DestructionHelper(MediaStream* stream) {
  43. stream->Destroy();
  44. }
  45. StreamProvider::MediaStream::MediaStream(
  46. RpcMessenger* rpc_messenger,
  47. Type type,
  48. int remote_handle,
  49. const scoped_refptr<base::SingleThreadTaskRunner>& media_task_runner)
  50. : main_task_runner_(base::ThreadTaskRunnerHandle::Get()),
  51. media_task_runner_(media_task_runner),
  52. rpc_messenger_(rpc_messenger),
  53. type_(type),
  54. remote_handle_(remote_handle),
  55. rpc_handle_(rpc_messenger_->GetUniqueHandle()) {
  56. DCHECK(remote_handle_ != RpcMessenger::kInvalidHandle);
  57. media_weak_this_ = media_weak_factory_.GetWeakPtr();
  58. auto receive_callback = base::BindPostTask(
  59. media_task_runner_,
  60. BindRepeating(&MediaStream::OnReceivedRpc, media_weak_this_));
  61. rpc_messenger_->RegisterMessageReceiverCallback(
  62. rpc_handle_, [cb = std::move(receive_callback)](
  63. std::unique_ptr<openscreen::cast::RpcMessage> message) {
  64. cb.Run(std::move(message));
  65. });
  66. }
  67. StreamProvider::MediaStream::~MediaStream() {
  68. DCHECK(main_task_runner_->BelongsToCurrentThread());
  69. rpc_messenger_->UnregisterMessageReceiverCallback(rpc_handle_);
  70. }
  71. void StreamProvider::MediaStream::Destroy() {
  72. DCHECK(media_task_runner_->BelongsToCurrentThread());
  73. // Invalid weak pointers to prevent |this| from receiving RPC calls on the
  74. // media thread.
  75. media_weak_factory_.InvalidateWeakPtrs();
  76. // Unbind all mojo pipes and bindings.
  77. receiver_.reset();
  78. decoder_buffer_reader_.reset();
  79. // After invalidating all weak ptrs of |media_weak_factory_|, MediaStream
  80. // won't be access anymore, so using |this| here is safe.
  81. main_task_runner_->DeleteSoon(FROM_HERE, this);
  82. }
  83. void StreamProvider::MediaStream::SendRpcMessageOnMainThread(
  84. std::unique_ptr<openscreen::cast::RpcMessage> message) {
  85. // |rpc_messenger_| is owned by |receiver_controller_| which is a singleton
  86. // per process, so it's safe to use Unretained() here.
  87. main_task_runner_->PostTask(
  88. FROM_HERE, base::BindOnce(&RpcMessenger::SendMessageToRemote,
  89. base::Unretained(rpc_messenger_), *message));
  90. }
  91. void StreamProvider::MediaStream::Initialize(
  92. base::OnceClosure init_done_callback) {
  93. DCHECK(media_task_runner_->BelongsToCurrentThread());
  94. DCHECK(init_done_callback);
  95. if (init_done_callback_) {
  96. OnError("Duplicate initialization");
  97. return;
  98. }
  99. init_done_callback_ = std::move(init_done_callback);
  100. auto rpc = std::make_unique<openscreen::cast::RpcMessage>();
  101. rpc->set_handle(remote_handle_);
  102. rpc->set_proc(openscreen::cast::RpcMessage::RPC_DS_INITIALIZE);
  103. rpc->set_integer_value(rpc_handle_);
  104. SendRpcMessageOnMainThread(std::move(rpc));
  105. }
  106. void StreamProvider::MediaStream::InitializeDataPipe(
  107. mojo::ScopedDataPipeConsumerHandle data_pipe) {
  108. DCHECK(media_task_runner_->BelongsToCurrentThread());
  109. decoder_buffer_reader_ =
  110. std::make_unique<MojoDecoderBufferReader>(std::move(data_pipe));
  111. CompleteInitialize();
  112. }
  113. void StreamProvider::MediaStream::ReceiveFrame(uint32_t count,
  114. mojom::DecoderBufferPtr buffer) {
  115. DCHECK(media_task_runner_->BelongsToCurrentThread());
  116. DCHECK(decoder_buffer_reader_);
  117. auto callback = BindToCurrentLoop(
  118. base::BindOnce(&MediaStream::AppendBuffer, media_weak_this_, count));
  119. decoder_buffer_reader_->ReadDecoderBuffer(std::move(buffer),
  120. std::move(callback));
  121. }
  122. void StreamProvider::MediaStream::FlushUntil(uint32_t count) {
  123. DCHECK(media_task_runner_->BelongsToCurrentThread());
  124. if (count < current_frame_count_)
  125. return;
  126. uint32_t buffers_to_erase = count - current_frame_count_;
  127. if (buffers_to_erase > buffers_.size()) {
  128. buffers_.clear();
  129. } else {
  130. buffers_.erase(buffers_.begin(), buffers_.begin() + buffers_to_erase);
  131. }
  132. current_frame_count_ = count;
  133. if (!read_complete_callback_.is_null())
  134. CompleteRead(DemuxerStream::kAborted);
  135. read_until_sent_ = false;
  136. }
  137. void StreamProvider::MediaStream::OnReceivedRpc(
  138. std::unique_ptr<openscreen::cast::RpcMessage> message) {
  139. DCHECK(media_task_runner_->BelongsToCurrentThread());
  140. DCHECK(message->handle() == rpc_handle_);
  141. switch (message->proc()) {
  142. case openscreen::cast::RpcMessage::RPC_DS_INITIALIZE_CALLBACK:
  143. OnInitializeCallback(std::move(message));
  144. break;
  145. case openscreen::cast::RpcMessage::RPC_DS_READUNTIL_CALLBACK:
  146. OnReadUntilCallback(std::move(message));
  147. break;
  148. default:
  149. VLOG(3) << __func__ << "Unknow RPC message.";
  150. }
  151. }
  152. void StreamProvider::MediaStream::OnInitializeCallback(
  153. std::unique_ptr<openscreen::cast::RpcMessage> message) {
  154. DCHECK(media_task_runner_->BelongsToCurrentThread());
  155. const openscreen::cast::DemuxerStreamInitializeCallback callback_message =
  156. message->demuxerstream_initializecb_rpc();
  157. if (callback_message.type() != type_) {
  158. OnError("Wrong type");
  159. return;
  160. }
  161. if ((type_ == DemuxerStream::AUDIO &&
  162. audio_decoder_config_.IsValidConfig()) ||
  163. (type_ == DemuxerStream::VIDEO &&
  164. video_decoder_config_.IsValidConfig())) {
  165. OnError("Duplicate initialization");
  166. return;
  167. }
  168. if (type_ == DemuxerStream::AUDIO &&
  169. callback_message.has_audio_decoder_config()) {
  170. const openscreen::cast::AudioDecoderConfig audio_message =
  171. callback_message.audio_decoder_config();
  172. cast_streaming::remoting::ConvertProtoToAudioDecoderConfig(
  173. audio_message, &audio_decoder_config_);
  174. if (!audio_decoder_config_.IsValidConfig()) {
  175. OnError("Invalid audio config");
  176. return;
  177. }
  178. } else if (type_ == DemuxerStream::VIDEO &&
  179. callback_message.has_video_decoder_config()) {
  180. const openscreen::cast::VideoDecoderConfig video_message =
  181. callback_message.video_decoder_config();
  182. cast_streaming::remoting::ConvertProtoToVideoDecoderConfig(
  183. video_message, &video_decoder_config_);
  184. if (!video_decoder_config_.IsValidConfig()) {
  185. OnError("Invalid video config");
  186. return;
  187. }
  188. } else {
  189. OnError("Config missing");
  190. return;
  191. }
  192. rpc_initialized_ = true;
  193. CompleteInitialize();
  194. }
  195. void StreamProvider::MediaStream::CompleteInitialize() {
  196. DCHECK(media_task_runner_->BelongsToCurrentThread());
  197. // Initialization finished when received RPC_DS_INITIALIZE_CALLBACK and
  198. // |decoder_buffer_reader_| is created.
  199. if (!rpc_initialized_ || !decoder_buffer_reader_)
  200. return;
  201. if (!init_done_callback_) {
  202. OnError("Initialize callback missing");
  203. return;
  204. }
  205. std::move(init_done_callback_).Run();
  206. }
  207. void StreamProvider::MediaStream::OnReadUntilCallback(
  208. std::unique_ptr<openscreen::cast::RpcMessage> message) {
  209. DCHECK(media_task_runner_->BelongsToCurrentThread());
  210. if (!read_until_sent_) {
  211. OnError("Unexpected ReadUntilCallback");
  212. return;
  213. }
  214. read_until_sent_ = false;
  215. const openscreen::cast::DemuxerStreamReadUntilCallback callback_message =
  216. message->demuxerstream_readuntilcb_rpc();
  217. total_received_frame_count_ = callback_message.count();
  218. if (cast_streaming::remoting::ToDemuxerStreamStatus(
  219. callback_message.status()) == kConfigChanged) {
  220. if (callback_message.has_audio_decoder_config()) {
  221. const openscreen::cast::AudioDecoderConfig audio_message =
  222. callback_message.audio_decoder_config();
  223. UpdateAudioConfig(audio_message);
  224. }
  225. if (callback_message.has_video_decoder_config()) {
  226. const openscreen::cast::VideoDecoderConfig video_message =
  227. callback_message.video_decoder_config();
  228. UpdateVideoConfig(video_message);
  229. }
  230. if (buffers_.empty() && !read_complete_callback_.is_null())
  231. CompleteRead(DemuxerStream::kConfigChanged);
  232. return;
  233. }
  234. if (buffers_.empty() && !read_complete_callback_.is_null())
  235. SendReadUntil();
  236. }
  237. void StreamProvider::MediaStream::UpdateAudioConfig(
  238. const openscreen::cast::AudioDecoderConfig& audio_message) {
  239. DCHECK(type_ == AUDIO);
  240. AudioDecoderConfig audio_config;
  241. cast_streaming::remoting::ConvertProtoToAudioDecoderConfig(audio_message,
  242. &audio_config);
  243. if (!audio_config.IsValidConfig()) {
  244. OnError("Invalid audio config");
  245. return;
  246. }
  247. next_audio_decoder_config_ = audio_config;
  248. }
  249. void StreamProvider::MediaStream::UpdateVideoConfig(
  250. const openscreen::cast::VideoDecoderConfig& video_message) {
  251. DCHECK(type_ == VIDEO);
  252. VideoDecoderConfig video_config;
  253. cast_streaming::remoting::ConvertProtoToVideoDecoderConfig(video_message,
  254. &video_config);
  255. if (!video_config.IsValidConfig()) {
  256. OnError("Invalid video config");
  257. return;
  258. }
  259. next_video_decoder_config_ = video_config;
  260. }
  261. void StreamProvider::MediaStream::SendReadUntil() {
  262. if (read_until_sent_)
  263. return;
  264. std::unique_ptr<openscreen::cast::RpcMessage> rpc(
  265. new openscreen::cast::RpcMessage());
  266. rpc->set_handle(remote_handle_);
  267. rpc->set_proc(openscreen::cast::RpcMessage::RPC_DS_READUNTIL);
  268. auto* message = rpc->mutable_demuxerstream_readuntil_rpc();
  269. message->set_count(total_received_frame_count_ + kNumFramesInEachReadUntil);
  270. message->set_callback_handle(rpc_handle_);
  271. SendRpcMessageOnMainThread(std::move(rpc));
  272. read_until_sent_ = true;
  273. }
  274. void StreamProvider::MediaStream::Read(ReadCB read_cb) {
  275. DCHECK(media_task_runner_->BelongsToCurrentThread());
  276. DCHECK(read_complete_callback_.is_null());
  277. DCHECK(read_cb);
  278. read_complete_callback_ = std::move(read_cb);
  279. if (buffers_.empty() && (next_audio_decoder_config_.IsValidConfig() ||
  280. next_video_decoder_config_.IsValidConfig())) {
  281. CompleteRead(DemuxerStream::kConfigChanged);
  282. return;
  283. }
  284. // Wait for more data.
  285. if (buffers_.empty()) {
  286. SendReadUntil();
  287. return;
  288. }
  289. CompleteRead(DemuxerStream::kOk);
  290. }
  291. void StreamProvider::MediaStream::CompleteRead(DemuxerStream::Status status) {
  292. DCHECK(media_task_runner_->BelongsToCurrentThread());
  293. switch (status) {
  294. case DemuxerStream::kConfigChanged:
  295. if (next_audio_decoder_config_.IsValidConfig()) {
  296. audio_decoder_config_ = next_audio_decoder_config_;
  297. next_audio_decoder_config_ = media::AudioDecoderConfig();
  298. }
  299. if (next_video_decoder_config_.IsValidConfig()) {
  300. video_decoder_config_ = next_video_decoder_config_;
  301. next_video_decoder_config_ = media::VideoDecoderConfig();
  302. }
  303. std::move(read_complete_callback_).Run(status, nullptr);
  304. return;
  305. case DemuxerStream::kAborted:
  306. case DemuxerStream::kError:
  307. std::move(read_complete_callback_).Run(status, nullptr);
  308. return;
  309. case DemuxerStream::kOk:
  310. DCHECK(read_complete_callback_);
  311. DCHECK(!buffers_.empty());
  312. DCHECK_LT(current_frame_count_, buffered_frame_count_);
  313. scoped_refptr<DecoderBuffer> frame_data = buffers_.front();
  314. buffers_.pop_front();
  315. ++current_frame_count_;
  316. std::move(read_complete_callback_).Run(status, frame_data);
  317. return;
  318. }
  319. }
  320. AudioDecoderConfig StreamProvider::MediaStream::audio_decoder_config() {
  321. DCHECK(type_ == DemuxerStream::AUDIO);
  322. return audio_decoder_config_;
  323. }
  324. VideoDecoderConfig StreamProvider::MediaStream::video_decoder_config() {
  325. DCHECK(type_ == DemuxerStream::VIDEO);
  326. return video_decoder_config_;
  327. }
  328. DemuxerStream::Type StreamProvider::MediaStream::type() const {
  329. return type_;
  330. }
  331. StreamLiveness StreamProvider::MediaStream::liveness() const {
  332. return StreamLiveness::kLive;
  333. }
  334. bool StreamProvider::MediaStream::SupportsConfigChanges() {
  335. return true;
  336. }
  337. void StreamProvider::MediaStream::AppendBuffer(
  338. uint32_t count,
  339. scoped_refptr<DecoderBuffer> buffer) {
  340. DCHECK(media_task_runner_->BelongsToCurrentThread());
  341. // Drop flushed frame.
  342. if (count < current_frame_count_)
  343. return;
  344. // Continuity check.
  345. DCHECK(buffers_.empty() || buffered_frame_count_ == count);
  346. buffers_.push_back(buffer);
  347. buffered_frame_count_ = count + 1;
  348. if (!read_complete_callback_.is_null())
  349. CompleteRead(DemuxerStream::kOk);
  350. }
  351. void StreamProvider::MediaStream::OnError(const std::string& error) {
  352. auto rpc = std::make_unique<openscreen::cast::RpcMessage>();
  353. rpc->set_handle(remote_handle_);
  354. rpc->set_proc(openscreen::cast::RpcMessage::RPC_DS_ONERROR);
  355. SendRpcMessageOnMainThread(std::move(rpc));
  356. }
  357. StreamProvider::StreamProvider(
  358. ReceiverController* receiver_controller,
  359. const scoped_refptr<base::SingleThreadTaskRunner>& media_task_runner)
  360. : main_task_runner_(base::ThreadTaskRunnerHandle::Get()),
  361. media_task_runner_(media_task_runner),
  362. receiver_controller_(receiver_controller),
  363. rpc_messenger_(receiver_controller_->rpc_messenger()) {
  364. DCHECK(receiver_controller_);
  365. DCHECK(rpc_messenger_);
  366. media_weak_this_ = media_weak_factory_.GetWeakPtr();
  367. auto callback = base::BindPostTask(
  368. media_task_runner_,
  369. base::BindRepeating(&StreamProvider::OnReceivedRpc, media_weak_this_));
  370. rpc_messenger_->RegisterMessageReceiverCallback(
  371. RpcMessenger::kAcquireDemuxerHandle,
  372. [cb = std::move(callback)](
  373. std::unique_ptr<openscreen::cast::RpcMessage> message) {
  374. cb.Run(std::move(message));
  375. });
  376. }
  377. StreamProvider::~StreamProvider() {
  378. DCHECK(main_task_runner_->BelongsToCurrentThread());
  379. rpc_messenger_->UnregisterMessageReceiverCallback(
  380. RpcMessenger::kAcquireDemuxerHandle);
  381. }
  382. std::string StreamProvider::GetDisplayName() const {
  383. return "media::remoting::StreamProvider";
  384. }
  385. void StreamProvider::Initialize(DemuxerHost* host,
  386. PipelineStatusCallback status_cb) {
  387. DCHECK(media_task_runner_->BelongsToCurrentThread());
  388. init_done_callback_ = std::move(status_cb);
  389. CompleteInitialize();
  390. }
  391. void StreamProvider::AbortPendingReads() {}
  392. void StreamProvider::StartWaitingForSeek(base::TimeDelta seek_time) {}
  393. void StreamProvider::CancelPendingSeek(base::TimeDelta seek_time) {}
  394. void StreamProvider::Seek(base::TimeDelta time,
  395. PipelineStatusCallback seek_cb) {
  396. media_task_runner_->PostTask(FROM_HERE,
  397. base::BindOnce(std::move(seek_cb), PIPELINE_OK));
  398. }
  399. void StreamProvider::Stop() {}
  400. base::TimeDelta StreamProvider::GetStartTime() const {
  401. return base::TimeDelta();
  402. }
  403. base::Time StreamProvider::GetTimelineOffset() const {
  404. return base::Time();
  405. }
  406. int64_t StreamProvider::GetMemoryUsage() const {
  407. return 0;
  408. }
  409. absl::optional<container_names::MediaContainerName>
  410. StreamProvider::GetContainerForMetrics() const {
  411. return absl::optional<container_names::MediaContainerName>();
  412. }
  413. void StreamProvider::OnEnabledAudioTracksChanged(
  414. const std::vector<MediaTrack::Id>& track_ids,
  415. base::TimeDelta curr_time,
  416. TrackChangeCB change_completed_cb) {
  417. std::vector<DemuxerStream*> streams;
  418. std::move(change_completed_cb).Run(DemuxerStream::AUDIO, streams);
  419. DVLOG(1) << "Track changes are not supported.";
  420. }
  421. void StreamProvider::OnSelectedVideoTrackChanged(
  422. const std::vector<MediaTrack::Id>& track_ids,
  423. base::TimeDelta curr_time,
  424. TrackChangeCB change_completed_cb) {
  425. std::vector<DemuxerStream*> streams;
  426. std::move(change_completed_cb).Run(DemuxerStream::VIDEO, streams);
  427. DVLOG(1) << "Track changes are not supported.";
  428. }
  429. void StreamProvider::Destroy() {
  430. DCHECK(media_task_runner_->BelongsToCurrentThread());
  431. if (init_done_callback_)
  432. std::move(init_done_callback_).Run(PIPELINE_ERROR_ABORT);
  433. // Invalid weak pointers to prevent |this| from receiving RPC calls on the
  434. // media thread.
  435. media_weak_factory_.InvalidateWeakPtrs();
  436. audio_stream_.reset();
  437. video_stream_.reset();
  438. // After invalidating all weak ptrs of |media_weak_factory_|, StreamProvider
  439. // won't be access anymore, so using |this| here is safe.
  440. main_task_runner_->DeleteSoon(FROM_HERE, this);
  441. }
  442. void StreamProvider::OnReceivedRpc(
  443. std::unique_ptr<openscreen::cast::RpcMessage> message) {
  444. switch (message->proc()) {
  445. case openscreen::cast::RpcMessage::RPC_ACQUIRE_DEMUXER:
  446. OnAcquireDemuxer(std::move(message));
  447. break;
  448. default:
  449. VLOG(3) << __func__ << "Unknown RPC message.";
  450. }
  451. }
  452. void StreamProvider::OnAcquireDemuxer(
  453. std::unique_ptr<openscreen::cast::RpcMessage> message) {
  454. DCHECK(media_task_runner_->BelongsToCurrentThread());
  455. DCHECK(message->has_acquire_demuxer_rpc());
  456. int32_t audio_demuxer_handle =
  457. message->acquire_demuxer_rpc().audio_demuxer_handle();
  458. int32_t video_demuxer_handle =
  459. message->acquire_demuxer_rpc().video_demuxer_handle();
  460. has_audio_ = audio_demuxer_handle != RpcMessenger::kInvalidHandle;
  461. has_video_ = video_demuxer_handle != RpcMessenger::kInvalidHandle;
  462. DCHECK(has_audio_ || has_video_);
  463. if (has_audio_) {
  464. auto callback = BindToCurrentLoop(base::BindOnce(
  465. &StreamProvider::OnAudioStreamCreated, media_weak_this_));
  466. main_task_runner_->PostTask(
  467. FROM_HERE,
  468. base::BindOnce(&MediaStream::CreateOnMainThread, rpc_messenger_,
  469. DemuxerStream::AUDIO, audio_demuxer_handle,
  470. media_task_runner_, std::move(callback)));
  471. }
  472. if (has_video_) {
  473. auto callback = BindToCurrentLoop(base::BindOnce(
  474. &StreamProvider::OnVideoStreamCreated, media_weak_this_));
  475. main_task_runner_->PostTask(
  476. FROM_HERE,
  477. base::BindOnce(&MediaStream::CreateOnMainThread, rpc_messenger_,
  478. DemuxerStream::VIDEO, video_demuxer_handle,
  479. media_task_runner_, std::move(callback)));
  480. }
  481. }
  482. void StreamProvider::OnAudioStreamCreated(MediaStream::UniquePtr stream) {
  483. DCHECK(media_task_runner_->BelongsToCurrentThread());
  484. audio_stream_ = std::move(stream);
  485. audio_stream_->Initialize(base::BindOnce(
  486. &StreamProvider::OnAudioStreamInitialized, media_weak_this_));
  487. InitializeDataPipe();
  488. }
  489. void StreamProvider::OnVideoStreamCreated(MediaStream::UniquePtr stream) {
  490. DCHECK(media_task_runner_->BelongsToCurrentThread());
  491. video_stream_ = std::move(stream);
  492. video_stream_->Initialize(base::BindOnce(
  493. &StreamProvider::OnVideoStreamInitialized, media_weak_this_));
  494. InitializeDataPipe();
  495. }
  496. void StreamProvider::InitializeDataPipe() {
  497. DCHECK(media_task_runner_->BelongsToCurrentThread());
  498. if ((has_audio_ && !audio_stream_) || (has_video_ && !video_stream_))
  499. return;
  500. receiver_controller_->StartDataStreams(
  501. has_audio_ ? audio_stream_->BindNewPipeAndPassRemote()
  502. : mojo::NullRemote(),
  503. has_video_ ? video_stream_->BindNewPipeAndPassRemote()
  504. : mojo::NullRemote());
  505. }
  506. void StreamProvider::OnAudioStreamInitialized() {
  507. audio_stream_initialized_ = true;
  508. CompleteInitialize();
  509. }
  510. void StreamProvider::OnVideoStreamInitialized() {
  511. video_stream_initialized_ = true;
  512. CompleteInitialize();
  513. }
  514. void StreamProvider::CompleteInitialize() {
  515. DCHECK(media_task_runner_->BelongsToCurrentThread());
  516. // Haven't receive RpcAcquireRenderer message
  517. if (!has_audio_ && !has_video_)
  518. return;
  519. if ((has_audio_ && !audio_stream_initialized_) ||
  520. (has_video_ && !video_stream_initialized_) || !init_done_callback_)
  521. return;
  522. // |init_done_callback_| should be called on |media_task_runner_|.
  523. std::move(init_done_callback_).Run(PIPELINE_OK);
  524. }
  525. std::vector<DemuxerStream*> StreamProvider::GetAllStreams() {
  526. std::vector<DemuxerStream*> streams;
  527. if (audio_stream_)
  528. streams.push_back(audio_stream_.get());
  529. if (video_stream_)
  530. streams.push_back(video_stream_.get());
  531. return streams;
  532. }
  533. } // namespace remoting
  534. } // namespace media
  535. namespace std {
  536. void default_delete<media::remoting::StreamProvider>::operator()(
  537. media::remoting::StreamProvider* ptr) const {
  538. ptr->Destroy();
  539. }
  540. } // namespace std