Media Remoting: End to end integration tests.

media/remoting/remote_receiver.cc

Index: media/remoting/remote_receiver.cc
diff --git a/media/remoting/remote_receiver.cc b/media/remoting/remote_receiver.cc
new file mode 100644
index 0000000000000000000000000000000000000000..32f198c8afc41fdea686e38f0ae9d41efab5007d
--- /dev/null
+++ b/media/remoting/remote_receiver.cc
@@ -0,0 +1,279 @@
+// 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/remote_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/remote_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
+
+RemoteReceiver::RemoteReceiver(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(&RemoteReceiver::OnReceivedRpc, weak_factory_.GetWeakPtr()));
+  rpc_broker_->RegisterMessageReceiverCallback(
+      RpcBroker::kAcquireHandle,
+      base::Bind(&RemoteReceiver::OnReceivedRpc, weak_factory_.GetWeakPtr()));
+}
+
+RemoteReceiver::~RemoteReceiver() {
+  rpc_broker_->UnregisterMessageReceiverCallback(rpc_handle_);
+  rpc_broker_->UnregisterMessageReceiverCallback(RpcBroker::kAcquireHandle);
+}
+
+void RemoteReceiver::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 RemoteReceiver::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();
+  stream_provider_.reset(new RemoteStreamProvider(rpc_broker_));

[miu, 2017/03/29 01:39:13]

if (stream_provider_) {
  return acquire renderer failed RPC;
}
stream_provider_.reset(new ...);

[xjz, 2017/03/30 23:21:30]

We don't have a RPC for acquire failure. Called OnError() instead, which will
send the error RPC message to sender and ends the remoting session.

+
+  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 RemoteReceiver::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_);
+  stream_provider_->Initialize(

[miu, 2017/03/29 01:39:14]

if (!stream_provider_) {
  return initialize failed RPC;
}

[xjz, 2017/03/30 23:21:30]

Done.

+      message->renderer_initialize_rpc().audio_demuxer_handle(),
+      message->renderer_initialize_rpc().video_demuxer_handle(),
+      base::Bind(&RemoteReceiver::OnStreamInitialized,
+                 weak_factory_.GetWeakPtr()));
+}
+
+void RemoteReceiver::OnStreamInitialized() {
+  renderer_->Initialize(stream_provider_.get(), this,

[miu, 2017/03/29 01:39:13]

DCHECK(stream_provider_);

[xjz, 2017/03/30 23:21:30]

Done.

+                        base::Bind(&RemoteReceiver::OnRendererInitialized,
+                                   weak_factory_.GetWeakPtr()));
+}
+
+void RemoteReceiver::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 RemoteReceiver::SetPlaybackRate(std::unique_ptr<pb::RpcMessage> message) {
+  playback_rate_ = message->double_value();

[miu, 2017/03/29 01:39:14]

Per comment in header file, instead of this, how about:

  void RemoteReceiver::SetPlaybackRate(std::unique_ptr<pb::RpcMessage> message)
{
    const double playback_rate = message->double_value();
    DVLOG...
    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.
        UpdateMediaTime();
      }
    } else {
      ScheduleMediaTimeUpdates();
    }
  }

...and the helper method (called from here and from StartPlayingFrom()):

  void RemoteReceiver::ScheduleMediaTimeUpdates() {
    if (time_update_timer_.IsRunning())
      return;
    SendMediaTimeUpdate();  // Send one right away too!
    time_update_timer_.Start(...);
  }

[xjz, 2017/03/30 23:21:30]

Done.

+  DVLOG(3) << __func__
+           << ": Receives RPC_R_SETPLAYBACKRATE with rate=" << playback_rate_;
+  renderer_->SetPlaybackRate(playback_rate_);
+}
+
+void RemoteReceiver::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 (flush_message.has_audio_count()) {

[miu, 2017/03/29 01:39:13]

if (stream_provider_) {
  if (flush_message.has_audio_count()) {
    stream_provider_->FlushUntil(...);
  }
  if (flush_message.has_video_count()) {
    stream_provider_->FlushUntil(...);
  }
}

[xjz, 2017/03/30 23:21:30]

Done.

+    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(&RemoteReceiver::OnFlushDone, weak_factory_.GetWeakPtr()));
+}
+
+void RemoteReceiver::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 RemoteReceiver::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);
+  // Schedule period report timer.

[miu, 2017/03/29 01:39:14]

Per above suggestions, the rest of the method is just a call to
ScheduleMediaTimeUpdates().

[xjz, 2017/03/30 23:21:30]

Done.

+  UpdateMediaTime();
+  time_update_timer_.Start(
+      FROM_HERE, kTimeUpdateInterval,
+      base::Bind(&RemoteReceiver::UpdateMediaTime, weak_factory_.GetWeakPtr()));
+}
+
+void RemoteReceiver::SetVolume(std::unique_ptr<pb::RpcMessage> message) {
+  DVLOG(3) << __func__ << ": Receives RPC_R_SETVOLUME message.";
+  renderer_->SetVolume(message->double_value());
+}
+
+void RemoteReceiver::UpdateMediaTime() {
+  // 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;
+  // Allow some slop to account for delays in scheduling time update tasks.
+  if (time_update_timer_.IsRunning() && (playback_rate_ > 0)) {
+    max_time += kTimeUpdateInterval;
+    max_time += kTimeUpdateInterval;
+  }
+  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 RemoteReceiver::OnReceivedBuffer(DemuxerStream::Type type,
+                                      scoped_refptr<DecoderBuffer> buffer) {
+  DVLOG(3) << __func__;
+  DCHECK(stream_provider_);
+  stream_provider_->AppendBuffer(type, buffer);
+}
+
+void RemoteReceiver::OnError(PipelineStatus status) {
+  DVLOG(3) << __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 RemoteReceiver::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 RemoteReceiver::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 RemoteReceiver::OnBufferingStateChange(BufferingState state) {
+  DVLOG(3) << __func__ << ": Issues RPC_RC_ONBUFFERINGSTATECHANGE message.";
+  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 RemoteReceiver::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 RemoteReceiver::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 RemoteReceiver::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 RemoteReceiver::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