Media Remoting: End to end integration tests.

media/remoting/remote_stream_provider.cc

Index: media/remoting/remote_stream_provider.cc
diff --git a/media/remoting/remote_stream_provider.cc b/media/remoting/remote_stream_provider.cc
new file mode 100644
index 0000000000000000000000000000000000000000..b81114bbab0603e556e4b2392dfb83e0dc5f2332
--- /dev/null
+++ b/media/remoting/remote_stream_provider.cc
@@ -0,0 +1,331 @@
+// 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_stream_provider.h"
+
+#include "base/bind.h"
+#include "base/callback.h"
+#include "base/callback_helpers.h"
+#include "media/base/decoder_buffer.h"
+#include "media/base/video_rotation.h"
+#include "media/remoting/proto_enum_utils.h"
+#include "media/remoting/proto_utils.h"
+
+namespace media {
+namespace remoting {
+
+namespace {
+// The number of frames requested in each ReadUntil RPC message.
+constexpr int kNumFramesInEachReadUntil = 10;
+}
+
+RemoteMediaStream::RemoteMediaStream(RpcBroker* rpc_broker,
+                                     Type type,
+                                     int remote_handle)
+    : rpc_broker_(rpc_broker),
+      type_(type),
+      remote_handle_(remote_handle),
+      rpc_handle_(rpc_broker_->GetUniqueHandle()),
+      weak_factory_(this) {
+  DCHECK(remote_handle_ != RpcBroker::kInvalidHandle);
+  rpc_broker_->RegisterMessageReceiverCallback(
+      rpc_handle_, base::Bind(&RemoteMediaStream::OnReceivedRpc,
+                              weak_factory_.GetWeakPtr()));
+}
+
+RemoteMediaStream::~RemoteMediaStream() {
+  rpc_broker_->UnregisterMessageReceiverCallback(rpc_handle_);
+}
+
+void RemoteMediaStream::Initialize(const base::Closure& init_done_cb) {
+  DCHECK(init_done_callback_.is_null());
+  DCHECK(!init_done_cb.is_null());
+  init_done_callback_ = init_done_cb;
+
+  DVLOG(3) << __func__ << "Issues RpcMessage::RPC_DS_INITIALIZE with "
+           << "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_DS_INITIALIZE);
+  rpc->set_integer_value(rpc_handle_);
+  rpc_broker_->SendMessageToRemote(std::move(rpc));
+}
+
+void RemoteMediaStream::OnReceivedRpc(std::unique_ptr<pb::RpcMessage> message) {
+  DCHECK(message->handle() == rpc_handle_);
+
+  switch (message->proc()) {
+    case pb::RpcMessage::RPC_DS_INITIALIZE_CALLBACK:
+      OnInitializeCallback(std::move(message));
+      break;
+    case pb::RpcMessage::RPC_DS_READUNTIL_CALLBACK:
+      OnReadUntilCallback(std::move(message));
+      break;
+    default:
+      VLOG(3) << __func__ << "Unknow RPC message.";
+  }
+}
+
+void RemoteMediaStream::OnInitializeCallback(
+    std::unique_ptr<pb::RpcMessage> message) {
+  DVLOG(3) << __func__ << "Receives RPC_DS_INITIALIZE_CALLBACK message.";
+  const pb::DemuxerStreamInitializeCallback callback_message =
+      message->demuxerstream_initializecb_rpc();
+  DCHECK(callback_message.type() == type_);

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

These four DCHECKs() should be if-statements instead, since they are operating
on possibly-invalid data from the remote peer. DCHECKs() are meant to test
internal, local assumptions about the state of code; and not assumptions about a
remote actor's internal state. Generally, in cases where the sender or receiver
discover an inconsistency, they should simply fatal-error and shutdown
immediately.

Something like:

  if (callback_message.type() != type_) {
    SendInitializeRpcFailed("wrong type");
    return;
  }
  if (!audio_decoder_config_.empty() ||
      !video_decoder_config_.empty() ||
      init_done_callback.is_null()) {
    SendInitializeRpcFailed("already initialized");
    return
  }

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

Done.

+  DCHECK(audio_decoder_config_.empty());
+  DCHECK(video_decoder_config_.empty());
+  DCHECK(!init_done_callback_.is_null());
+
+  if (type_ == DemuxerStream::AUDIO &&
+      callback_message.has_audio_decoder_config()) {
+    const pb::AudioDecoderConfig audio_message =
+        callback_message.audio_decoder_config();
+    UpdateConfig(&audio_message, nullptr);
+  } else if (type_ == DemuxerStream::VIDEO &&
+             callback_message.has_video_decoder_config()) {
+    const pb::VideoDecoderConfig video_message =
+        callback_message.video_decoder_config();
+    UpdateConfig(nullptr, &video_message);
+  } else {
+    NOTREACHED();

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

Replace with something like:

  SendInitializeRpcFailed("config missing in callback message");

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

Done.

+  }
+  base::ResetAndReturn(&init_done_callback_).Run();
+}
+
+void RemoteMediaStream::OnReadUntilCallback(
+    std::unique_ptr<pb::RpcMessage> message) {
+  DVLOG(3) << __func__ << ": Receives RPC_DS_READUNTIL_CALLBACK message.";
+  DCHECK(read_until_sent_);

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

DCHECK() --> if ()

...and rather than me point all these out throughout the file, please go through
and fix them all. You may want to employ a common fatal-error-handling routine
(like what we did for DemuxerStreamAdapter).

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

Done.

+  read_until_sent_ = false;
+  const pb::DemuxerStreamReadUntilCallback callback_message =
+      message->demuxerstream_readuntilcb_rpc();
+  last_read_until_count_ = callback_message.count();
+  if (ToDemuxerStreamStatus(callback_message.status()) == kConfigChanged) {
+    if (callback_message.has_audio_decoder_config()) {
+      const pb::AudioDecoderConfig audio_message =
+          callback_message.audio_decoder_config();
+      UpdateConfig(&audio_message, nullptr);
+    }
+    if (callback_message.has_video_decoder_config()) {
+      const pb::VideoDecoderConfig video_message =
+          callback_message.video_decoder_config();
+      UpdateConfig(nullptr, &video_message);
+    }
+    config_changed_ = true;
+    if (buffers_.empty() && !read_complete_callback_.is_null())
+      CompleteRead(DemuxerStream::kConfigChanged);
+    return;
+  }
+  if (buffers_.empty() && !read_complete_callback_.is_null())
+    SendReadUntil();
+}
+
+void RemoteMediaStream::UpdateConfig(
+    const pb::AudioDecoderConfig* audio_message,
+    const pb::VideoDecoderConfig* video_message) {
+  if (type_ == AUDIO) {
+    DCHECK(audio_message && !video_message);
+    AudioDecoderConfig audio_config;
+    ConvertProtoToAudioDecoderConfig(*audio_message, &audio_config);
+    DCHECK(audio_config.IsValidConfig());
+    audio_decoder_config_.push_back(audio_config);
+  } else if (type_ == VIDEO) {
+    DCHECK(video_message && !audio_message);
+    VideoDecoderConfig video_config;
+    ConvertProtoToVideoDecoderConfig(*video_message, &video_config);
+    DCHECK(video_config.IsValidConfig());
+    video_decoder_config_.push_back(video_config);
+  } else {
+    VLOG(1) << __func__ << ": Only supports video or audio stream.";
+  }
+}
+
+void RemoteMediaStream::SendReadUntil() {
+  if (read_until_sent_)
+    return;
+  DVLOG(3) << "Issues RPC_DS_READUNTIL RPC message to remote_handle_="
+           << remote_handle_ << " with callback handle=" << rpc_handle_
+           << " count=" << last_read_until_count_;
+
+  std::unique_ptr<pb::RpcMessage> rpc(new pb::RpcMessage());
+  rpc->set_handle(remote_handle_);
+  rpc->set_proc(pb::RpcMessage::RPC_DS_READUNTIL);
+  auto* message = rpc->mutable_demuxerstream_readuntil_rpc();
+  last_read_until_count_ += kNumFramesInEachReadUntil;
+  message->set_count(last_read_until_count_);
+  message->set_callback_handle(rpc_handle_);
+  rpc_broker_->SendMessageToRemote(std::move(rpc));
+  read_until_sent_ = true;
+}
+
+void RemoteMediaStream::FlushUntil(int count) {
+  while (!buffers_.empty()) {
+    buffers_.pop_front();
+  }
+
+  last_read_until_count_ = count;
+  if (!read_complete_callback_.is_null())
+    CompleteRead(DemuxerStream::kAborted);
+  read_until_sent_ = false;
+}
+
+void RemoteMediaStream::Read(const ReadCB& read_cb) {
+  DVLOG(3) << __func__;
+  DCHECK(read_complete_callback_.is_null());
+  read_complete_callback_ = read_cb;
+  if (buffers_.empty() && config_changed_) {
+    CompleteRead(DemuxerStream::kConfigChanged);
+    return;
+  }
+
+  // Wait for more data.
+  if (buffers_.empty()) {
+    SendReadUntil();
+    return;
+  }
+
+  CompleteRead(DemuxerStream::kOk);
+}
+
+void RemoteMediaStream::CompleteRead(DemuxerStream::Status status) {
+  DVLOG(3) << __func__ << ": " << status;
+  switch (status) {
+    case DemuxerStream::kConfigChanged:
+      if (type_ == AUDIO) {
+        DCHECK(audio_decoder_config_.size() == 2);

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

This is weird. Looks like we create a whole deque to just store 0, 1, or 2
elements. Feels like we should have a simpler scheme for updating config. Maybe
just two class members, like:

  AudioDecoderConfig audio_decoder_config_;
  AudioDecoderConfig next_audio_decoder_config_;

And, here, then we can just:

  audio_decoder_config_ = next_audio_decoder_config_;

...and same for video, of course.

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

Done.

+        audio_decoder_config_.pop_front();
+      } else {
+        DCHECK(video_decoder_config_.size() == 2);
+        video_decoder_config_.pop_front();
+      }
+      config_changed_ = false;
+      base::ResetAndReturn(&read_complete_callback_).Run(status, nullptr);
+      return;
+    case DemuxerStream::kAborted:
+      base::ResetAndReturn(&read_complete_callback_).Run(status, nullptr);
+      return;
+    case DemuxerStream::kOk:
+      DCHECK(!buffers_.empty());
+      scoped_refptr<DecoderBuffer> frame_data = buffers_.front();
+      buffers_.pop_front();
+      base::ResetAndReturn(&read_complete_callback_).Run(status, frame_data);
+      return;
+  }
+}
+
+AudioDecoderConfig RemoteMediaStream::audio_decoder_config() {
+  DVLOG(3) << __func__;
+  DCHECK(type_ == DemuxerStream::AUDIO);
+  DCHECK(!audio_decoder_config_.empty());
+  return audio_decoder_config_.front();
+}
+
+VideoDecoderConfig RemoteMediaStream::video_decoder_config() {
+  DVLOG(3) << __func__;
+  DCHECK(type_ == DemuxerStream::VIDEO);
+  DCHECK(!video_decoder_config_.empty());
+  return video_decoder_config_.front();
+}
+
+DemuxerStream::Type RemoteMediaStream::type() const {
+  return type_;
+}
+
+bool RemoteMediaStream::SupportsConfigChanges() {
+  return true;
+}
+
+VideoRotation RemoteMediaStream::video_rotation() {
+  return VideoRotation::VIDEO_ROTATION_0;
+}
+
+bool RemoteMediaStream::enabled() const {
+  return true;
+}
+
+void RemoteMediaStream::AppendBuffer(scoped_refptr<DecoderBuffer> buffer) {
+  DVLOG(3) << __func__;
+  buffers_.push_back(buffer);
+  if (!read_complete_callback_.is_null())
+    CompleteRead(DemuxerStream::kOk);
+}
+
+RemoteStreamProvider::RemoteStreamProvider(RpcBroker* rpc_broker)
+    : rpc_broker_(rpc_broker), weak_factory_(this) {}
+
+RemoteStreamProvider::~RemoteStreamProvider() {}
+
+void RemoteStreamProvider::Initialize(int remote_audio_handle,
+                                      int remote_video_handle,
+                                      const base::Closure& callback) {
+  DVLOG(3) << __func__ << ": remote_audio_handle=" << remote_audio_handle
+           << " remote_video_handle=" << remote_video_handle;
+  DCHECK(init_done_callback_.is_null());
+  DCHECK(remote_audio_handle != RpcBroker::kInvalidHandle ||
+         remote_video_handle != RpcBroker::kInvalidHandle);
+
+  init_done_callback_ = callback;
+  if (remote_audio_handle != RpcBroker::kInvalidHandle) {
+    audio_stream_.reset(new RemoteMediaStream(rpc_broker_, DemuxerStream::AUDIO,
+                                              remote_audio_handle));
+    audio_stream_->Initialize(
+        base::Bind(&RemoteStreamProvider::AudioStreamInitialized,
+                   weak_factory_.GetWeakPtr()));
+  }
+  if (remote_video_handle != RpcBroker::kInvalidHandle) {
+    video_stream_.reset(new RemoteMediaStream(rpc_broker_, DemuxerStream::VIDEO,
+                                              remote_video_handle));
+    video_stream_->Initialize(
+        base::Bind(&RemoteStreamProvider::VideoStreamInitialized,
+                   weak_factory_.GetWeakPtr()));
+  }
+}
+
+void RemoteStreamProvider::AudioStreamInitialized() {
+  DCHECK(!init_done_callback_.is_null());
+  audio_stream_initialized_ = true;
+  if (video_stream_initialized_ || !video_stream_)
+    base::ResetAndReturn(&init_done_callback_).Run();
+}
+
+void RemoteStreamProvider::VideoStreamInitialized() {
+  DCHECK(!init_done_callback_.is_null());
+  video_stream_initialized_ = true;
+  if (audio_stream_initialized_ || !audio_stream_)
+    base::ResetAndReturn(&init_done_callback_).Run();
+}
+
+DemuxerStream* RemoteStreamProvider::GetStream(DemuxerStream::Type type) {
+  if (type == DemuxerStream::AUDIO)
+    return audio_stream_.get();
+  else if (type == DemuxerStream::VIDEO)
+    return video_stream_.get();
+  else
+    VLOG(1) << __func__ << ": Only supports video or audio stream.";
+  return nullptr;
+}
+
+void RemoteStreamProvider::AppendBuffer(DemuxerStream::Type type,
+                                        scoped_refptr<DecoderBuffer> buffer) {
+  if (type == DemuxerStream::AUDIO)
+    audio_stream_->AppendBuffer(buffer);
+  else if (type == DemuxerStream::VIDEO)
+    video_stream_->AppendBuffer(buffer);
+  else
+    VLOG(1) << __func__ << ": Only supports video or audio stream.";
+}
+
+void RemoteStreamProvider::FlushUntil(DemuxerStream::Type type, int count) {
+  DVLOG(3) << __func__ << ": type=" << type << " count=" << count;
+  if (type == DemuxerStream::AUDIO)
+    audio_stream_->FlushUntil(count);
+  else if (type == DemuxerStream::VIDEO)
+    video_stream_->FlushUntil(count);
+  else
+    VLOG(1) << __func__ << ": Only supports video or audio stream.";
+}
+
+}  // namespace remoting
+}  // namespace media