RELAND: Media Remoting end to end integration tests.

This is a re-land of https://codereview.chromium.org/2692593002/.
Moved tests for media remoting pipeline out of general
PipelineIntegrationTest.

-------Description of original change follows-------

Media Remoting: End to end integration tests.

Add end to end integration tests for Media Remoting. Refactors
PipelineIntegrationTest to test both media and media remoting pipeline.
Re-use current tests. No new tests are added in this CL.

BUG=684065

Review-Url: https://codereview.chromium.org/2808583002
Cr-Commit-Position: refs/heads/master@{#466216}

media/remoting/BUILD.gn

@@ -76,13 +76,20 @@ source_set("media_remoting_tests") {
   sources = [
     "courier_renderer_unittest.cc",
     "demuxer_stream_adapter_unittest.cc",
+    "end2end_test_renderer.cc",
+    "end2end_test_renderer.h",
     "fake_media_resource.cc",
     "fake_media_resource.h",
     "fake_remoter.cc",
     "fake_remoter.h",
+    "integration_test.cc",
     "proto_utils_unittest.cc",
+    "receiver.cc",
+    "receiver.h",
     "renderer_controller_unittest.cc",
     "rpc_broker_unittest.cc",
+    "stream_provider.cc",
+    "stream_provider.h",
   ]
 
   deps = [
@@ -93,6 +100,7 @@ source_set("media_remoting_tests") {
     "//media",
     "//media/base:test_support",
     "//media/mojo/interfaces:remoting",
+    "//media/test:pipeline_integration_test_base",
     "//testing/gmock",
     "//testing/gtest",
     "//ui/gfx:test_support",

media/remoting/end2end_test_renderer.cc

@@ -0,0 +1,217 @@
+// 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/end2end_test_renderer.h"
+
+#include "base/bind.h"
+#include "base/bind_helpers.h"
+#include "base/callback.h"
+#include "base/threading/thread_task_runner_handle.h"
+#include "media/mojo/interfaces/remoting.mojom.h"
+#include "media/remoting/courier_renderer.h"
+#include "media/remoting/proto_utils.h"
+#include "media/remoting/receiver.h"
+#include "media/remoting/renderer_controller.h"
+#include "mojo/public/cpp/bindings/strong_binding.h"
+
+namespace media {
+namespace remoting {
+
+namespace {
+
+class TestStreamSender final : public mojom::RemotingDataStreamSender {
+ public:
+  using SendFrameToSinkCallback =
+      base::Callback<void(const std::vector<uint8_t>& data,
+                          DemuxerStream::Type type)>;
+  TestStreamSender(mojom::RemotingDataStreamSenderRequest request,
+                   mojo::ScopedDataPipeConsumerHandle handle,
+                   DemuxerStream::Type type,
+                   const SendFrameToSinkCallback& callback)
+      : binding_(this, std::move(request)),
+        consumer_handle_(std::move(handle)),
+        type_(type),
+        send_frame_to_sink_cb_(callback) {}
+
+  ~TestStreamSender() override {}
+
+  // mojom::RemotingDataStreamSender implementation.
+
+  void ConsumeDataChunk(uint32_t offset,
+                        uint32_t size,
+                        uint32_t total_payload_size) override {
+    next_frame_data_.resize(total_payload_size);
+    MojoResult result = mojo::ReadDataRaw(
+        consumer_handle_.get(), next_frame_data_.data() + offset, &size,
+        MOJO_READ_DATA_FLAG_ALL_OR_NONE);
+    CHECK(result == MOJO_RESULT_OK);
+  }
+
+  void SendFrame() override {
+    if (!send_frame_to_sink_cb_.is_null())
+      send_frame_to_sink_cb_.Run(next_frame_data_, type_);
+    next_frame_data_.resize(0);
+  }
+
+  void CancelInFlightData() override { next_frame_data_.resize(0); }
+
+ private:
+  mojo::Binding<RemotingDataStreamSender> binding_;
+  mojo::ScopedDataPipeConsumerHandle consumer_handle_;
+  const DemuxerStream::Type type_;
+  const SendFrameToSinkCallback send_frame_to_sink_cb_;
+  std::vector<uint8_t> next_frame_data_;
+
+  DISALLOW_COPY_AND_ASSIGN(TestStreamSender);
+};
+
+class TestRemoter final : public mojom::Remoter {
+ public:
+  using SendMessageToSinkCallback =
+      base::Callback<void(const std::vector<uint8_t>& message)>;
+  TestRemoter(
+      mojom::RemotingSourcePtr source,
+      const SendMessageToSinkCallback& send_message_to_sink_cb,
+      const TestStreamSender::SendFrameToSinkCallback& send_frame_to_sink_cb)
+      : source_(std::move(source)),
+        send_message_to_sink_cb_(send_message_to_sink_cb),
+        send_frame_to_sink_cb_(send_frame_to_sink_cb) {}
+
+  ~TestRemoter() override {}
+
+  // mojom::Remoter implementation.
+
+  void Start() override { source_->OnStarted(); }
+
+  void StartDataStreams(
+      mojo::ScopedDataPipeConsumerHandle audio_pipe,
+      mojo::ScopedDataPipeConsumerHandle video_pipe,
+      mojom::RemotingDataStreamSenderRequest audio_sender_request,
+      mojom::RemotingDataStreamSenderRequest video_sender_request) override {
+    if (audio_pipe.is_valid()) {
+      audio_stream_sender_.reset(new TestStreamSender(
+          std::move(audio_sender_request), std::move(audio_pipe),
+          DemuxerStream::AUDIO, send_frame_to_sink_cb_));
+    }
+    if (video_pipe.is_valid()) {
+      video_stream_sender_.reset(new TestStreamSender(
+          std::move(video_sender_request), std::move(video_pipe),
+          DemuxerStream::VIDEO, send_frame_to_sink_cb_));
+    }
+  }
+
+  void Stop(mojom::RemotingStopReason reason) override {
+    source_->OnStopped(reason);
+  }
+
+  void SendMessageToSink(const std::vector<uint8_t>& message) override {
+    if (!send_message_to_sink_cb_.is_null())
+      send_message_to_sink_cb_.Run(message);
+  }
+
+  // Called when receives RPC messages from receiver.
+  void OnMessageFromSink(const std::vector<uint8_t>& message) {
+    source_->OnMessageFromSink(message);
+  }
+
+ private:
+  mojom::RemotingSourcePtr source_;
+  const SendMessageToSinkCallback send_message_to_sink_cb_;
+  const TestStreamSender::SendFrameToSinkCallback send_frame_to_sink_cb_;
+  std::unique_ptr<TestStreamSender> audio_stream_sender_;
+  std::unique_ptr<TestStreamSender> video_stream_sender_;
+
+  DISALLOW_COPY_AND_ASSIGN(TestRemoter);
+};
+
+scoped_refptr<SharedSession> CreateSharedSession(
+    const TestRemoter::SendMessageToSinkCallback& send_message_to_sink_cb,
+    const TestStreamSender::SendFrameToSinkCallback& send_frame_to_sink_cb) {
+  mojom::RemotingSourcePtr remoting_source;
+  mojom::RemotingSourceRequest remoting_source_request(&remoting_source);
+  mojom::RemoterPtr remoter;
+  std::unique_ptr<TestRemoter> test_remoter = base::MakeUnique<TestRemoter>(
+      std::move(remoting_source), send_message_to_sink_cb,
+      send_frame_to_sink_cb);
+  mojo::MakeStrongBinding(std::move(test_remoter), mojo::MakeRequest(&remoter));
+  return new SharedSession(std::move(remoting_source_request),
+                           std::move(remoter));
+}
+
+}  // namespace
+
+End2EndTestRenderer::End2EndTestRenderer(std::unique_ptr<Renderer> renderer)
+    : receiver_rpc_broker_(base::Bind(&End2EndTestRenderer::OnMessageFromSink,
+                                      base::Unretained(this))),
+      receiver_(new Receiver(std::move(renderer), &receiver_rpc_broker_)),
+      weak_factory_(this) {
+  shared_session_ =
+      CreateSharedSession(base::Bind(&End2EndTestRenderer::SendMessageToSink,
+                                     weak_factory_.GetWeakPtr()),
+                          base::Bind(&End2EndTestRenderer::SendFrameToSink,
+                                     weak_factory_.GetWeakPtr()));
+  controller_.reset(new RendererController(shared_session_));
+  courier_renderer_.reset(new CourierRenderer(
+      base::ThreadTaskRunnerHandle::Get(), controller_->GetWeakPtr(), nullptr));
+}
+
+End2EndTestRenderer::~End2EndTestRenderer() {}
+
+void End2EndTestRenderer::Initialize(MediaResource* media_resource,
+                                     RendererClient* client,
+                                     const PipelineStatusCB& init_cb) {
+  courier_renderer_->Initialize(media_resource, client, init_cb);
+}
+
+void End2EndTestRenderer::SetCdm(CdmContext* cdm_context,
+                                 const CdmAttachedCB& cdc_attached_cb) {
+  // TODO(xjz): Add the implementation when media remoting starts supporting
+  // encrypted contents.
+  NOTIMPLEMENTED() << "Media Remoting doesn't support EME for now.";
+}
+
+void End2EndTestRenderer::Flush(const base::Closure& flush_cb) {
+  courier_renderer_->Flush(flush_cb);
+}
+
+void End2EndTestRenderer::StartPlayingFrom(base::TimeDelta time) {
+  courier_renderer_->StartPlayingFrom(time);
+}
+
+void End2EndTestRenderer::SetPlaybackRate(double playback_rate) {
+  courier_renderer_->SetPlaybackRate(playback_rate);
+}
+
+void End2EndTestRenderer::SetVolume(float volume) {
+  courier_renderer_->SetVolume(volume);
+}
+
+base::TimeDelta End2EndTestRenderer::GetMediaTime() {
+  return courier_renderer_->GetMediaTime();
+}
+
+void End2EndTestRenderer::SendMessageToSink(
+    const std::vector<uint8_t>& message) {
+  std::unique_ptr<pb::RpcMessage> rpc(new pb::RpcMessage());
+  if (!rpc->ParseFromArray(message.data(), message.size())) {
+    VLOG(1) << __func__ << ": Received corrupted Rpc message.";
+    return;
+  }
+  receiver_rpc_broker_.ProcessMessageFromRemote(std::move(rpc));
+}
+
+void End2EndTestRenderer::SendFrameToSink(const std::vector<uint8_t>& frame,
+                                          DemuxerStream::Type type) {
+  scoped_refptr<DecoderBuffer> decoder_buffer =
+      ByteArrayToDecoderBuffer(frame.data(), frame.size());
+  receiver_->OnReceivedBuffer(type, decoder_buffer);
+}
+
+void End2EndTestRenderer::OnMessageFromSink(
+    std::unique_ptr<std::vector<uint8_t>> message) {
+  shared_session_->OnMessageFromSink(*message);
+}
+
+}  // namespace remoting
+}  // namespace media

media/remoting/end2end_test_renderer.h

@@ -0,0 +1,71 @@
+// 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.
+
+#ifndef MEDIA_REMOTING_END2END_RENDERER_H_
+#define MEDIA_REMOTING_END2END_RENDERER_H_
+
+#include <vector>
+
+#include "base/memory/weak_ptr.h"
+#include "media/base/demuxer_stream.h"
+#include "media/base/renderer.h"
+#include "media/remoting/rpc_broker.h"
+
+namespace media {
+namespace remoting {
+
+class SharedSession;
+class RendererController;
+class CourierRenderer;
+class Receiver;
+
+// Simulates the media remoting pipeline.
+class End2EndTestRenderer final : public Renderer {
+ public:
+  explicit End2EndTestRenderer(std::unique_ptr<Renderer> renderer);
+  ~End2EndTestRenderer() override;
+
+  // Renderer implementation.
+  void Initialize(MediaResource* media_resource,
+                  RendererClient* client,
+                  const PipelineStatusCB& init_cb) override;
+  void SetCdm(CdmContext* cdm_context,
+              const CdmAttachedCB& cdm_attached_cb) override;
+  void Flush(const base::Closure& flush_cb) override;
+  void StartPlayingFrom(base::TimeDelta time) override;
+  void SetPlaybackRate(double playback_rate) override;
+  void SetVolume(float volume) override;
+  base::TimeDelta GetMediaTime() override;
+
+ private:
+  // Called to send RPC messages to |receiver_|.
+  void SendMessageToSink(const std::vector<uint8_t>& message);
+
+  // Called to send frame data to |receiver_|.
+  void SendFrameToSink(const std::vector<uint8_t>& data,
+                       DemuxerStream::Type type);
+
+  // Called when receives RPC messages from |receiver_|.
+  void OnMessageFromSink(std::unique_ptr<std::vector<uint8_t>> message);
+
+  // The session that is used by |controller_| to create the data pipes.
+  scoped_refptr<SharedSession> shared_session_;
+  std::unique_ptr<RendererController> controller_;
+  std::unique_ptr<CourierRenderer> courier_renderer_;
+
+  // The RpcBroker to handle the RPC messages to/from |receiver_|.
+  RpcBroker receiver_rpc_broker_;
+
+  // A receiver that renders media streams.
+  std::unique_ptr<Receiver> receiver_;
+
+  base::WeakPtrFactory<End2EndTestRenderer> weak_factory_;
+
+  DISALLOW_COPY_AND_ASSIGN(End2EndTestRenderer);
+};
+
+}  // namespace remoting
+}  // namespace media
+
+#endif  // MEDIA_REMOTING_END2END_RENDERER_H_