RELAND: Media Remoting end to end integration tests.

media/remoting/receiver.cc

+// 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/receiver.h"
+
+#include "base/bind.h"
+#include "base/callback.h"
+#include "media/base/decoder_buffer.h"
+#include "media/base/renderer.h"
+#include "media/remoting/proto_enum_utils.h"
+#include "media/remoting/stream_provider.h"
+
+namespace media {
+namespace remoting {
+namespace {
+
+// The period to send the TimeUpdate RPC message to update the media time on
+// sender side.
+constexpr base::TimeDelta kTimeUpdateInterval =
+    base::TimeDelta::FromMilliseconds(250);
+
+}  // namespace
+
+Receiver::Receiver(std::unique_ptr<Renderer> renderer, RpcBroker* rpc_broker)
+    : renderer_(std::move(renderer)),
+      rpc_broker_(rpc_broker),
+      rpc_handle_(rpc_broker_->GetUniqueHandle()),
+      weak_factory_(this) {
+  DCHECK(renderer_);
+  DCHECK(rpc_broker_);
+  rpc_broker_->RegisterMessageReceiverCallback(
+      rpc_handle_,
+      base::Bind(&Receiver::OnReceivedRpc, weak_factory_.GetWeakPtr()));
+  rpc_broker_->RegisterMessageReceiverCallback(
+      RpcBroker::kAcquireHandle,
+      base::Bind(&Receiver::OnReceivedRpc, weak_factory_.GetWeakPtr()));
+}
+
+Receiver::~Receiver() {
+  rpc_broker_->UnregisterMessageReceiverCallback(rpc_handle_);
+  rpc_broker_->UnregisterMessageReceiverCallback(RpcBroker::kAcquireHandle);
+}
+
+void Receiver::OnReceivedRpc(std::unique_ptr<pb::RpcMessage> message) {
+  DCHECK(message);
+  switch (message->proc()) {
+    case pb::RpcMessage::RPC_ACQUIRE_RENDERER:
+      AcquireRenderer(std::move(message));
+      break;
+    case pb::RpcMessage::RPC_R_FLUSHUNTIL:
+      FlushUntil(std::move(message));
+      break;
+    case pb::RpcMessage::RPC_R_STARTPLAYINGFROM:
+      StartPlayingFrom(std::move(message));
+      break;
+    case pb::RpcMessage::RPC_R_SETPLAYBACKRATE:
+      SetPlaybackRate(std::move(message));
+      break;
+    case pb::RpcMessage::RPC_R_SETVOLUME:
+      SetVolume(std::move(message));
+      break;
+    case pb::RpcMessage::RPC_R_INITIALIZE:
+      Initialize(std::move(message));
+      break;
+    default:
+      VLOG(1) << __func__ << ": Unknow RPC message. proc=" << message->proc();
+  }
+}
+
+void Receiver::AcquireRenderer(std::unique_ptr<pb::RpcMessage> message) {
+  DVLOG(3) << __func__ << ": Receives RPC_ACQUIRE_RENDERER with remote_handle= "
+           << message->integer_value();
+
+  remote_handle_ = message->integer_value();
+  if (stream_provider_) {
+    VLOG(1) << "Acquire renderer error: Already aquired.";
+    OnError(PipelineStatus::PIPELINE_ERROR_DECODE);
+    return;
+  }
+
+  stream_provider_.reset(new StreamProvider(
+      rpc_broker_, base::Bind(&Receiver::OnError, weak_factory_.GetWeakPtr(),
+                              PipelineStatus::PIPELINE_ERROR_DECODE)));
+
+  DVLOG(3) << __func__
+           << ": Issues RPC_ACQUIRE_RENDERER_DONE RPC message. remote_handle="
+           << remote_handle_ << " rpc_handle=" << rpc_handle_;
+  std::unique_ptr<pb::RpcMessage> rpc(new pb::RpcMessage());
+  rpc->set_handle(remote_handle_);
+  rpc->set_proc(pb::RpcMessage::RPC_ACQUIRE_RENDERER_DONE);
+  rpc->set_integer_value(rpc_handle_);
+  rpc_broker_->SendMessageToRemote(std::move(rpc));
+}
+
+void Receiver::Initialize(std::unique_ptr<pb::RpcMessage> message) {
+  DCHECK(stream_provider_);
+  DVLOG(3) << __func__ << ": Receives RPC_R_INITIALIZE with callback handle= "
+           << message->renderer_initialize_rpc().callback_handle();
+  DCHECK(message->renderer_initialize_rpc().callback_handle() ==
+         remote_handle_);
+  if (!stream_provider_)
+    OnRendererInitialized(PipelineStatus::PIPELINE_ERROR_INITIALIZATION_FAILED);
+
+  stream_provider_->Initialize(
+      message->renderer_initialize_rpc().audio_demuxer_handle(),
+      message->renderer_initialize_rpc().video_demuxer_handle(),
+      base::Bind(&Receiver::OnStreamInitialized, weak_factory_.GetWeakPtr()));
+}
+
+void Receiver::OnStreamInitialized() {
+  DCHECK(stream_provider_);
+  renderer_->Initialize(
+      stream_provider_.get(), this,
+      base::Bind(&Receiver::OnRendererInitialized, weak_factory_.GetWeakPtr()));
+}
+
+void Receiver::OnRendererInitialized(PipelineStatus status) {
+  DVLOG(3) << __func__ << ": Issues RPC_R_INITIALIZE_CALLBACK RPC message."
+           << "remote_handle=" << remote_handle_;
+
+  std::unique_ptr<pb::RpcMessage> rpc(new pb::RpcMessage());
+  rpc->set_handle(remote_handle_);
+  rpc->set_proc(pb::RpcMessage::RPC_R_INITIALIZE_CALLBACK);
+  rpc->set_boolean_value(status == PIPELINE_OK);
+  rpc_broker_->SendMessageToRemote(std::move(rpc));
+}
+
+void Receiver::SetPlaybackRate(std::unique_ptr<pb::RpcMessage> message) {
+  const double playback_rate = message->double_value();
+  DVLOG(3) << __func__
+           << ": Receives RPC_R_SETPLAYBACKRATE with rate=" << playback_rate;
+  renderer_->SetPlaybackRate(playback_rate);
+
+  if (playback_rate == 0.0) {
+    if (time_update_timer_.IsRunning()) {
+      time_update_timer_.Stop();
+      // Send one final media time update since the sender will not get any
+      // until playback resumes.
+      SendMediaTimeUpdate();
+    }
+  } else {
+    ScheduleMediaTimeUpdates();
+  }
+}
+
+void Receiver::FlushUntil(std::unique_ptr<pb::RpcMessage> message) {
+  DVLOG(3) << __func__ << ": Receives RPC_R_FLUSHUNTIL RPC message.";
+
+  const pb::RendererFlushUntil flush_message =
+      message->renderer_flushuntil_rpc();
+  DCHECK_EQ(flush_message.callback_handle(), remote_handle_);
+  if (stream_provider_) {
+    if (flush_message.has_audio_count()) {
+      stream_provider_->FlushUntil(DemuxerStream::AUDIO,
+                                   flush_message.audio_count());
+    }
+    if (flush_message.has_video_count()) {
+      stream_provider_->FlushUntil(DemuxerStream::VIDEO,
+                                   flush_message.video_count());
+    }
+  }
+  time_update_timer_.Stop();
+  renderer_->Flush(
+      base::Bind(&Receiver::OnFlushDone, weak_factory_.GetWeakPtr()));
+}
+
+void Receiver::OnFlushDone() {
+  DVLOG(3) << __func__ << ": Issues RPC_R_FLUSHUNTIL_CALLBACK RPC message.";
+
+  std::unique_ptr<pb::RpcMessage> rpc(new pb::RpcMessage());
+  rpc->set_handle(remote_handle_);
+  rpc->set_proc(pb::RpcMessage::RPC_R_FLUSHUNTIL_CALLBACK);
+  rpc_broker_->SendMessageToRemote(std::move(rpc));
+}
+
+void Receiver::StartPlayingFrom(std::unique_ptr<pb::RpcMessage> message) {
+  DVLOG(3) << __func__ << ": Receives RPC_R_STARTPLAYINGFROM message.";
+  base::TimeDelta time =
+      base::TimeDelta::FromMicroseconds(message->integer64_value());
+  renderer_->StartPlayingFrom(time);
+  ScheduleMediaTimeUpdates();
+}
+
+void Receiver::ScheduleMediaTimeUpdates() {
+  if (time_update_timer_.IsRunning())
+    return;
+  SendMediaTimeUpdate();
+  time_update_timer_.Start(
+      FROM_HERE, kTimeUpdateInterval,
+      base::Bind(&Receiver::SendMediaTimeUpdate, weak_factory_.GetWeakPtr()));
+}
+
+void Receiver::SetVolume(std::unique_ptr<pb::RpcMessage> message) {
+  DVLOG(3) << __func__ << ": Receives RPC_R_SETVOLUME message.";
+  renderer_->SetVolume(message->double_value());
+}
+
+void Receiver::SendMediaTimeUpdate() {
+  // Issues RPC_RC_ONTIMEUPDATE RPC message.
+  std::unique_ptr<pb::RpcMessage> rpc(new pb::RpcMessage());
+  rpc->set_handle(remote_handle_);
+  rpc->set_proc(pb::RpcMessage::RPC_RC_ONTIMEUPDATE);
+  auto* message = rpc->mutable_rendererclient_ontimeupdate_rpc();
+  base::TimeDelta media_time = renderer_->GetMediaTime();
+  message->set_time_usec(media_time.InMicroseconds());
+  base::TimeDelta max_time = media_time;
+  message->set_max_time_usec(max_time.InMicroseconds());
+  DVLOG(3) << __func__ << ": Issues RPC_RC_ONTIMEUPDATE message."
+           << " media_time = " << media_time.InMicroseconds()
+           << " max_time= " << max_time.InMicroseconds();
+  rpc_broker_->SendMessageToRemote(std::move(rpc));
+}
+
+void Receiver::OnReceivedBuffer(DemuxerStream::Type type,
+                                scoped_refptr<DecoderBuffer> buffer) {
+  DVLOG(3) << __func__
+           << ": type=" << (type == DemuxerStream::AUDIO ? "Audio" : "Video");
+  DCHECK(stream_provider_);
+  stream_provider_->AppendBuffer(type, buffer);
+}
+
+void Receiver::OnError(PipelineStatus status) {
+  VLOG(1) << __func__ << ": Issues RPC_RC_ONERROR message.";
+  std::unique_ptr<pb::RpcMessage> rpc(new pb::RpcMessage());
+  rpc->set_handle(remote_handle_);
+  rpc->set_proc(pb::RpcMessage::RPC_RC_ONERROR);
+  rpc_broker_->SendMessageToRemote(std::move(rpc));
+}
+
+void Receiver::OnEnded() {
+  DVLOG(3) << __func__ << ": Issues RPC_RC_ONENDED message.";
+  std::unique_ptr<pb::RpcMessage> rpc(new pb::RpcMessage());
+  rpc->set_handle(remote_handle_);
+  rpc->set_proc(pb::RpcMessage::RPC_RC_ONENDED);
+  rpc_broker_->SendMessageToRemote(std::move(rpc));
+  time_update_timer_.Stop();
+}
+
+void Receiver::OnStatisticsUpdate(const PipelineStatistics& stats) {
+  DVLOG(3) << __func__ << ": Issues RPC_RC_ONSTATISTICSUPDATE message.";
+  std::unique_ptr<pb::RpcMessage> rpc(new pb::RpcMessage());
+  rpc->set_handle(remote_handle_);
+  rpc->set_proc(pb::RpcMessage::RPC_RC_ONSTATISTICSUPDATE);
+  auto* message = rpc->mutable_rendererclient_onstatisticsupdate_rpc();
+  message->set_audio_bytes_decoded(stats.audio_bytes_decoded);
+  message->set_video_bytes_decoded(stats.video_bytes_decoded);
+  message->set_video_frames_decoded(stats.video_frames_decoded);
+  message->set_video_frames_dropped(stats.video_frames_dropped);
+  message->set_audio_memory_usage(stats.audio_memory_usage);
+  message->set_video_memory_usage(stats.video_memory_usage);
+  rpc_broker_->SendMessageToRemote(std::move(rpc));
+}
+
+void Receiver::OnBufferingStateChange(BufferingState state) {
+  DVLOG(3) << __func__
+           << ": Issues RPC_RC_ONBUFFERINGSTATECHANGE message: state=" << state;
+  std::unique_ptr<pb::RpcMessage> rpc(new pb::RpcMessage());
+  rpc->set_handle(remote_handle_);
+  rpc->set_proc(pb::RpcMessage::RPC_RC_ONBUFFERINGSTATECHANGE);
+  auto* message = rpc->mutable_rendererclient_onbufferingstatechange_rpc();
+  message->set_state(ToProtoMediaBufferingState(state).value());
+  rpc_broker_->SendMessageToRemote(std::move(rpc));
+}
+
+void Receiver::OnWaitingForDecryptionKey() {
+  DVLOG(3) << __func__ << ": Issues RPC_RC_ONWAITINGFORDECRYPTIONKEY message.";
+  std::unique_ptr<pb::RpcMessage> rpc(new pb::RpcMessage());
+  rpc->set_handle(remote_handle_);
+  rpc->set_proc(pb::RpcMessage::RPC_RC_ONWAITINGFORDECRYPTIONKEY);
+  rpc_broker_->SendMessageToRemote(std::move(rpc));
+}
+
+void Receiver::OnVideoNaturalSizeChange(const gfx::Size& size) {
+  DVLOG(3) << __func__ << ": Issues RPC_RC_ONVIDEONATURALSIZECHANGE message.";
+  std::unique_ptr<pb::RpcMessage> rpc(new pb::RpcMessage());
+  rpc->set_handle(remote_handle_);
+  rpc->set_proc(pb::RpcMessage::RPC_RC_ONVIDEONATURALSIZECHANGE);
+  auto* message = rpc->mutable_rendererclient_onvideonatualsizechange_rpc();
+  message->set_width(size.width());
+  message->set_height(size.height());
+  rpc_broker_->SendMessageToRemote(std::move(rpc));
+}
+
+void Receiver::OnVideoOpacityChange(bool opaque) {
+  DVLOG(3) << __func__ << ": Issues RPC_RC_ONVIDEOOPACITYCHANGE message.";
+  std::unique_ptr<pb::RpcMessage> rpc(new pb::RpcMessage());
+  rpc->set_handle(remote_handle_);
+  rpc->set_proc(pb::RpcMessage::RPC_RC_ONVIDEOOPACITYCHANGE);
+  rpc->set_boolean_value(opaque);
+  rpc_broker_->SendMessageToRemote(std::move(rpc));
+}
+
+void Receiver::OnDurationChange(base::TimeDelta duration) {
+  DVLOG(3) << __func__ << ": Issues RPC_RC_ONDURATIONCHANGE message.";
+  std::unique_ptr<pb::RpcMessage> rpc(new pb::RpcMessage());
+  rpc->set_handle(remote_handle_);
+  rpc->set_proc(pb::RpcMessage::RPC_RC_ONDURATIONCHANGE);
+  rpc->set_integer_value(duration.InMicroseconds());
+  rpc_broker_->SendMessageToRemote(std::move(rpc));
+}
+
+}  // namespace remoting
+}  // namespace media