media/remoting/receiver.cc

   1 // Copyright 2017 The Chromium Authors
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "media/remoting/receiver.h"
   6
   7 #include <utility>
   8
   9 #include "base/functional/bind.h"
  10 #include "base/functional/callback.h"
  11 #include "base/notreached.h"
  12 #include "base/task/bind_post_task.h"
  13 #include "base/task/sequenced_task_runner.h"
  14 #include "base/task/single_thread_task_runner.h"
  15 #include "media/base/decoder_buffer.h"
  16 #include "media/base/renderer.h"
  17 #include "media/cast/openscreen/remoting_message_factories.h"
  18 #include "media/cast/openscreen/remoting_proto_enum_utils.h"
  19 #include "media/cast/openscreen/remoting_proto_utils.h"
  20 #include "media/remoting/receiver_controller.h"
  21 #include "media/remoting/stream_provider.h"
  22
  23 using openscreen::cast::RpcMessenger;
  24
  25 namespace media {
  26 namespace remoting {
  27 namespace {
  28
  29 // The period to send the TimeUpdate RPC message to update the media time on
  30 // sender side.
  31 constexpr base::TimeDelta kTimeUpdateInterval = base::Milliseconds(250);
  32
  33 }  // namespace
  34
  35 Receiver::Receiver(
  36     int rpc_handle,
  37     int remote_handle,
  38     ReceiverController* receiver_controller,
  39     const scoped_refptr<base::SequencedTaskRunner>& media_task_runner,
  40     std::unique_ptr<Renderer> renderer,
  41     base::OnceCallback<void(int)> acquire_renderer_done_cb)
  42     : rpc_handle_(rpc_handle),
  43       remote_handle_(remote_handle),
  44       receiver_controller_(receiver_controller),
  45       rpc_messenger_(receiver_controller_->rpc_messenger()),
  46       main_task_runner_(base::SingleThreadTaskRunner::GetCurrentDefault()),
  47       media_task_runner_(media_task_runner),
  48       renderer_(std::move(renderer)),
  49       acquire_renderer_done_cb_(std::move(acquire_renderer_done_cb)) {
  50   DCHECK(rpc_handle_ != RpcMessenger::kInvalidHandle);
  51   DCHECK(receiver_controller_);
  52   DCHECK(rpc_messenger_);
  53   DCHECK(renderer_);
  54
  55   // Note: The constructor is running on the main thread, but will be destroyed
  56   // on the media thread. Therefore, all weak pointers must be dereferenced on
  57   // the media thread.
  58   auto receive_callback = base::BindPostTask(
  59       media_task_runner_,
  60       BindRepeating(&Receiver::OnReceivedRpc, weak_factory_.GetWeakPtr()));
  61
  62   // Listening all renderer rpc messages.
  63   rpc_messenger_->RegisterMessageReceiverCallback(
  64       rpc_handle_, [cb = std::move(receive_callback)](
  65                        std::unique_ptr<openscreen::cast::RpcMessage> message) {
  66         cb.Run(std::move(message));
  67       });
  68
  69   VerifyAcquireRendererDone();
  70 }
  71
  72 Receiver::~Receiver() {
  73   rpc_messenger_->UnregisterMessageReceiverCallback(rpc_handle_);
  74 }
  75
  76 // Receiver::Initialize() will be called by the local pipeline, it would only
  77 // keep the |init_cb| in order to continue the initialization once it receives
  78 // RPC_R_INITIALIZE, which means Receiver::OnRpcInitialize() is called.
  79 void Receiver::Initialize(MediaResource* media_resource,
  80                           RendererClient* client,
  81                           PipelineStatusCallback init_cb) {
  82   demuxer_ = media_resource;
  83   init_cb_ = std::move(init_cb);
  84   ShouldInitializeRenderer();
  85 }
  86
  87 /* CDM is not supported for remoting media */
  88 void Receiver::SetCdm(CdmContext* cdm_context, CdmAttachedCB cdm_attached_cb) {
  89   NOTREACHED();
  90 }
  91
  92 // No-op. Controlled by sender via RPC calls instead.
  93 void Receiver::SetLatencyHint(absl::optional<base::TimeDelta> latency_hint) {}
  94
  95 // No-op. Controlled by sender via RPC calls instead.
  96 void Receiver::Flush(base::OnceClosure flush_cb) {}
  97
  98 // No-op. Controlled by sender via RPC calls instead.
  99 void Receiver::StartPlayingFrom(base::TimeDelta time) {}
 100
 101 // No-op. Controlled by sender via RPC calls instead.
 102 void Receiver::SetPlaybackRate(double playback_rate) {}
 103
 104 // No-op. Controlled by sender via RPC calls instead.
 105 void Receiver::SetVolume(float volume) {}
 106
 107 // No-op. Controlled by sender via RPC calls instead.
 108 base::TimeDelta Receiver::GetMediaTime() {
 109   return base::TimeDelta();
 110 }
 111
 112 RendererType Receiver::GetRendererType() {
 113   return RendererType::kRemoting;
 114 }
 115
 116 void Receiver::SendRpcMessageOnMainThread(
 117     std::unique_ptr<openscreen::cast::RpcMessage> message) {
 118   // |rpc_messenger_| is owned by |receiver_controller_| which is a singleton
 119   // per process, so it's safe to use Unretained() here.
 120   main_task_runner_->PostTask(
 121       FROM_HERE, base::BindOnce(&RpcMessenger::SendMessageToRemote,
 122                                 base::Unretained(rpc_messenger_), *message));
 123 }
 124
 125 void Receiver::OnReceivedRpc(
 126     std::unique_ptr<openscreen::cast::RpcMessage> message) {
 127   DCHECK(main_task_runner_->RunsTasksInCurrentSequence());
 128   DCHECK(message);
 129
 130   media::cast::DispatchRendererRpcCall(message.get(), this);
 131 }
 132
 133 void Receiver::SetRemoteHandle(int remote_handle) {
 134   DCHECK_NE(remote_handle, RpcMessenger::kInvalidHandle);
 135   DCHECK_EQ(remote_handle_, RpcMessenger::kInvalidHandle);
 136   remote_handle_ = remote_handle;
 137   VerifyAcquireRendererDone();
 138 }
 139
 140 void Receiver::VerifyAcquireRendererDone() {
 141   if (remote_handle_ == RpcMessenger::kInvalidHandle)
 142     return;
 143
 144   DCHECK(acquire_renderer_done_cb_);
 145   std::move(acquire_renderer_done_cb_).Run(rpc_handle_);
 146 }
 147
 148 void Receiver::OnRpcInitialize() {
 149   DCHECK(renderer_);
 150   rpc_initialize_received_ = true;
 151   ShouldInitializeRenderer();
 152 }
 153
 154 void Receiver::ShouldInitializeRenderer() {
 155   // ShouldInitializeRenderer() will be called from Initialize() and
 156   // OnRpcInitialize() in different orders.
 157   //
 158   // |renderer_| must be initialized when both Initialize() and
 159   // OnRpcInitialize() are called.
 160   if (!rpc_initialize_received_ || !init_cb_)
 161     return;
 162
 163   DCHECK(main_task_runner_->RunsTasksInCurrentSequence());
 164   DCHECK(renderer_);
 165   DCHECK(demuxer_);
 166   renderer_->Initialize(demuxer_, this,
 167                         base::BindOnce(&Receiver::OnRendererInitialized,
 168                                        weak_factory_.GetWeakPtr()));
 169 }
 170
 171 void Receiver::OnRendererInitialized(PipelineStatus status) {
 172   DCHECK(main_task_runner_->RunsTasksInCurrentSequence());
 173   DCHECK(init_cb_);
 174   std::move(init_cb_).Run(status);
 175
 176   auto rpc = media::cast::CreateMessageForInitializationComplete(status ==
 177                                                                  PIPELINE_OK);
 178   rpc->set_handle(remote_handle_);
 179   SendRpcMessageOnMainThread(std::move(rpc));
 180 }
 181
 182 void Receiver::OnRpcSetPlaybackRate(double playback_rate) {
 183   DCHECK(main_task_runner_->RunsTasksInCurrentSequence());
 184
 185   renderer_->SetPlaybackRate(playback_rate);
 186
 187   if (playback_rate == 0.0) {
 188     if (time_update_timer_.IsRunning()) {
 189       time_update_timer_.Stop();
 190       // Send one final media time update since the sender will not get any
 191       // until playback resumes.
 192       SendMediaTimeUpdate();
 193     }
 194   } else {
 195     ScheduleMediaTimeUpdates();
 196   }
 197 }
 198
 199 void Receiver::OnRpcFlush(uint32_t audio_count, uint32_t video_count) {
 200   DCHECK(main_task_runner_->RunsTasksInCurrentSequence());
 201
 202   receiver_controller_->OnRendererFlush(audio_count, video_count);
 203
 204   time_update_timer_.Stop();
 205   renderer_->Flush(
 206       base::BindOnce(&Receiver::OnFlushDone, weak_factory_.GetWeakPtr()));
 207 }
 208
 209 void Receiver::OnFlushDone() {
 210   auto rpc = media::cast::CreateMessageForFlushComplete();
 211   rpc->set_handle(remote_handle_);
 212   SendRpcMessageOnMainThread(std::move(rpc));
 213 }
 214
 215 void Receiver::OnRpcStartPlayingFrom(base::TimeDelta time) {
 216   DCHECK(main_task_runner_->RunsTasksInCurrentSequence());
 217
 218   renderer_->StartPlayingFrom(time);
 219   ScheduleMediaTimeUpdates();
 220 }
 221
 222 void Receiver::ScheduleMediaTimeUpdates() {
 223   if (time_update_timer_.IsRunning())
 224     return;
 225   SendMediaTimeUpdate();
 226   time_update_timer_.Start(FROM_HERE, kTimeUpdateInterval,
 227                            base::BindRepeating(&Receiver::SendMediaTimeUpdate,
 228                                                weak_factory_.GetWeakPtr()));
 229 }
 230
 231 void Receiver::OnRpcSetVolume(double volume) {
 232   DCHECK(main_task_runner_->RunsTasksInCurrentSequence());
 233   renderer_->SetVolume(volume);
 234 }
 235
 236 void Receiver::SendMediaTimeUpdate() {
 237   // Issues RPC_RC_ONTIMEUPDATE RPC message.
 238   auto rpc =
 239       media::cast::CreateMessageForMediaTimeUpdate(renderer_->GetMediaTime());
 240   rpc->set_handle(remote_handle_);
 241   SendRpcMessageOnMainThread(std::move(rpc));
 242 }
 243
 244 void Receiver::OnError(PipelineStatus status) {
 245   auto rpc = media::cast::CreateMessageForError();
 246   rpc->set_handle(remote_handle_);
 247   SendRpcMessageOnMainThread(std::move(rpc));
 248 }
 249
 250 void Receiver::OnFallback(PipelineStatus status) {
 251   NOTREACHED();
 252 }
 253
 254 void Receiver::OnEnded() {
 255   auto rpc = media::cast::CreateMessageForMediaEnded();
 256   rpc->set_handle(remote_handle_);
 257   SendRpcMessageOnMainThread(std::move(rpc));
 258   time_update_timer_.Stop();
 259 }
 260
 261 void Receiver::OnStatisticsUpdate(const PipelineStatistics& stats) {
 262   auto rpc = media::cast::CreateMessageForStatisticsUpdate(stats);
 263   rpc->set_handle(remote_handle_);
 264   SendRpcMessageOnMainThread(std::move(rpc));
 265 }
 266
 267 void Receiver::OnBufferingStateChange(BufferingState state,
 268                                       BufferingStateChangeReason reason) {
 269   auto rpc = media::cast::CreateMessageForBufferingStateChange(state);
 270   rpc->set_handle(remote_handle_);
 271   SendRpcMessageOnMainThread(std::move(rpc));
 272 }
 273
 274 void Receiver::OnWaiting(WaitingReason reason) {
 275   // Media Remoting has not implemented this concept.
 276   NOTIMPLEMENTED();
 277 }
 278
 279 void Receiver::OnAudioConfigChange(const AudioDecoderConfig& config) {
 280   auto rpc = media::cast::CreateMessageForAudioConfigChange(config);
 281   rpc->set_handle(remote_handle_);
 282   SendRpcMessageOnMainThread(std::move(rpc));
 283 }
 284
 285 void Receiver::OnVideoConfigChange(const VideoDecoderConfig& config) {
 286   auto rpc = media::cast::CreateMessageForVideoConfigChange(config);
 287   rpc->set_handle(remote_handle_);
 288   SendRpcMessageOnMainThread(std::move(rpc));
 289 }
 290
 291 void Receiver::OnVideoNaturalSizeChange(const gfx::Size& size) {
 292   auto rpc = media::cast::CreateMessageForVideoNaturalSizeChange(size);
 293   rpc->set_handle(remote_handle_);
 294   SendRpcMessageOnMainThread(std::move(rpc));
 295
 296   // Notify the host.
 297   receiver_controller_->OnVideoNaturalSizeChange(size);
 298 }
 299
 300 void Receiver::OnVideoOpacityChange(bool opaque) {
 301   auto rpc = media::cast::CreateMessageForVideoOpacityChange(opaque);
 302   rpc->set_handle(remote_handle_);
 303   SendRpcMessageOnMainThread(std::move(rpc));
 304 }
 305
 306 void Receiver::OnVideoFrameRateChange(absl::optional<int>) {}
 307
 308 }  // namespace remoting
 309 }  // namespace media