[Cast] Add Translation Layer between CmaBackend and gRPC Channel

This CL defines a layer in between the gRPC Channel and the CmaBackend,
which is responsible for converting between internal and external
entity types, and ensuring that calls to the CmaBackend occur on the
proper thread.

Bug: b/167426091
Change-Id: I7bf46475f147a62500f155eb670b4983de4ba824
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2519156
Commit-Queue: Ryan Keane <rwkeane@google.com>
Reviewed-by: mark a. foltz <mfoltz@chromium.org>
Reviewed-by: Yuchen Liu <yucliu@chromium.org>
Cr-Commit-Position: refs/heads/master@{#834565}

chromecast/media/cma/BUILD.gn

@@ -81,6 +81,7 @@ cast_source_set("unittests") {
   if (!is_android) {
     sources += [
       "backend/proxy/cma_backend_proxy_unittest.cc",
+      "backend/proxy/proxy_call_translator_unittest.cc",
       "backend/proxy/push_buffer_queue_unittest.cc",
     ]
     deps += [

chromecast/media/cma/backend/proxy/BUILD.gn

@@ -38,9 +38,12 @@ cast_source_set("proxy") {
     "cast_runtime_audio_channel_broker.h",
     "cma_backend_proxy.cc",
     "cma_backend_proxy.h",
+    "cma_proxy_handler.h",
     "multizone_audio_decoder_proxy.h",
     "multizone_audio_decoder_proxy_impl.cc",
     "multizone_audio_decoder_proxy_impl.h",
+    "proxy_call_translator.cc",
+    "proxy_call_translator.h",
     "push_buffer_queue.cc",
     "push_buffer_queue.h",
   ]
@@ -51,6 +54,7 @@ cast_source_set("proxy") {
     "//chromecast/base",
     "//chromecast/media/api",
     "//chromecast/public/media",
+    "//third_party/grpc:grpc++",
     "//third_party/protobuf:protobuf_full",
   ]
 }

chromecast/media/cma/backend/proxy/cma_proxy_handler.h

+// Copyright 2020 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 CHROMECAST_MEDIA_CMA_BACKEND_PROXY_CMA_PROXY_HANDLER_H_
+#define CHROMECAST_MEDIA_CMA_BACKEND_PROXY_CMA_PROXY_HANDLER_H_
+
+#include "base/memory/ref_counted.h"
+
+namespace chromecast {
+
+class TaskRunner;
+
+namespace media {
+
+struct AudioConfig;
+
+class CmaProxyHandler {
+ public:
+  // The mode in which Cast Core should operate.
+  enum class AudioDecoderOperationMode {
+    // Both multiroom and audio rendering is enabled.
+    kAll = 0,
+
+    // Only multiroom is enabled and audio rendering is disabled.  This should
+    // be used if the runtime is taking over responsibility for rendering audio.
+    kMultiroomOnly = 1,
+
+    // Only audio rendering is enabled and multiroom is disabled.
+    kAudioOnly = 2
+  };
+
+  // The current state of the remote CMA backend.
+  enum class PipelineState {
+    kUninitialized = 0,
+    kStopped = 1,
+    kPlaying = 2,
+    kPaused = 3,
+  };
+
+  // Observer for changes on the remote client.
+  class Client {
+   public:
+    virtual ~Client() = default;
+
+    // Called when an error occurs upon calling any gRPC Method.
+    virtual void OnError() = 0;
+
+    // Called when the Start(), Stop(), Pause(), or Resume() methods
+    // successfully change the current pipeline state.
+    virtual void OnPipelineStateChange(PipelineState state) = 0;
+
+    // Called following the successful processing of a batch of PushBuffer
+    // calls.
+    virtual void OnBytesDecoded(int64_t decoded_byte_count) = 0;
+  };
+
+  virtual ~CmaProxyHandler() = default;
+
+  // Create a new implementation-specific CmaProxyHandler. Each provided
+  // object must exist for the duration of the created instance's lifetime, and
+  // all callbacks for |client| will be called on |task_runner|.
+  static std::unique_ptr<CmaProxyHandler> Create(TaskRunner* task_runner,
+                                                 Client* client);
+
+  // Calls to the corresponding gRPC Methods. These functions may be called from
+  // any thread.
+  virtual void Initialize(const std::string& cast_session_id,
+                          AudioDecoderOperationMode decoder_mode) = 0;
+  virtual void Start(int64_t start_pts) = 0;
+  virtual void Stop() = 0;
+  virtual void Pause() = 0;
+  virtual void Resume() = 0;
+  virtual void SetPlaybackRate(float rate) = 0;
+  virtual void SetVolume(float multiplier) = 0;
+
+  // Push the provided data or config to a queue, for processing at a later
+  // point when resources are available. Returns true if the data was
+  // successfully pushed to the queue and false otherwise. These functions may
+  // be called from any thread.
+  //
+  // NOTES:
+  // - SetConfig is expected to be called prior to any PushBuffer calls.
+  // - SetConfig may be called later on as-well, after which time the new config
+  //   will be used for all following PushBuffer calls.
+  virtual bool SetConfig(const AudioConfig& config) = 0;
+  virtual bool PushBuffer(scoped_refptr<DecoderBufferBase> buffer) = 0;
+};
+
+}  // namespace media
+}  // namespace chromecast
+
+#endif  // CHROMECAST_MEDIA_CMA_BACKEND_PROXY_CMA_PROXY_HANDLER_H_

chromecast/media/cma/backend/proxy/proxy_call_translator.h

+// Copyright 2020 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 CHROMECAST_MEDIA_CMA_BACKEND_PROXY_PROXY_CALL_TRANSLATOR_H_
+#define CHROMECAST_MEDIA_CMA_BACKEND_PROXY_PROXY_CALL_TRANSLATOR_H_
+
+#include <memory>
+
+#include "base/memory/ref_counted.h"
+#include "base/optional.h"
+#include "chromecast/media/api/decoder_buffer_base.h"
+#include "chromecast/media/cma/backend/proxy/cast_runtime_audio_channel_broker.h"
+#include "chromecast/media/cma/backend/proxy/cma_proxy_handler.h"
+
+namespace chromecast {
+
+class TaskRunner;
+
+namespace media {
+
+struct AudioConfig;
+
+// This class is responsible for translating between entities used by the
+// client CmaBackend and entities used by the internal gRPC Channel.
+// Calls made to all methods of this class may be made from any thread.
+class ProxyCallTranslator : public CmaProxyHandler,
+                            public CastRuntimeAudioChannelBroker::Handler {
+ public:
+  // Creates a new ProxyCallTranslator. All provided entities must exist for the
+  // duration of this instance's lifetime. All calls to |client| will be made
+  // on |client_task_runner|.
+  ProxyCallTranslator(TaskRunner* client_task_runner,
+                      CmaProxyHandler::Client* client);
+  ~ProxyCallTranslator() override;
+
+  // CmaProxyHandler overrides:
+  void Initialize(
+      const std::string& cast_session_id,
+      CmaProxyHandler::AudioDecoderOperationMode decoder_mode) override;
+  void Start(int64_t start_pts) override;
+  void Stop() override;
+  void Pause() override;
+  void Resume() override;
+  void SetPlaybackRate(float rate) override;
+  void SetVolume(float multiplier) override;
+  bool SetConfig(const AudioConfig& config) override;
+  bool PushBuffer(scoped_refptr<DecoderBufferBase> buffer) override;
+
+ private:
+  friend class ProxyCallTranslatorTest;
+
+  using MediaTime = CastRuntimeAudioChannelBroker::Handler::MediaTime;
+  using PipelineState = CastRuntimeAudioChannelBroker::Handler::PipelineState;
+  using PushBufferRequest =
+      CastRuntimeAudioChannelBroker::Handler::PushBufferRequest;
+
+  ProxyCallTranslator(
+      TaskRunner* client_task_runner,
+      CmaProxyHandler::Client* client,
+      std::unique_ptr<CastRuntimeAudioChannelBroker> decoder_channel);
+
+  // CastRuntimeAudioChannelBroker::Handler overrides:
+  PushBufferRequest GetBufferedData() override;
+  bool HasBufferedData() override;
+  void HandleInitializeResponse(
+      CastRuntimeAudioChannelBroker::StatusCode status) override;
+  void HandleStateChangeResponse(
+      PipelineState state,
+      CastRuntimeAudioChannelBroker::StatusCode status) override;
+  void HandleSetVolumeResponse(
+      CastRuntimeAudioChannelBroker::StatusCode status) override;
+  void HandleSetPlaybackResponse(
+      CastRuntimeAudioChannelBroker::StatusCode status) override;
+  void HandlePushBufferResponse(
+      int64_t decoded_bytes,
+      CastRuntimeAudioChannelBroker::StatusCode status) override;
+  void HandleGetMediaTimeResponse(
+      base::Optional<MediaTime> time,
+      CastRuntimeAudioChannelBroker::StatusCode status) override;
+
+  bool ProcessPushBufferRequest(PushBufferRequest request);
+
+  // Helper to share error handling code.
+  bool HandleError(CastRuntimeAudioChannelBroker::StatusCode status);
+
+  // Helpers to simplify use of callbacks for tasks posted to
+  // |client_task_runner_|.
+  void OnErrorTask();
+  void OnPipelineStateChangeTask(CmaProxyHandler::PipelineState state);
+  void OnBytesDecodedTask(int64_t decoded_byte_count);
+
+  std::unique_ptr<CastRuntimeAudioChannelBroker> decoder_channel_;
+  TaskRunner* const client_task_runner_;
+  CmaProxyHandler::Client* const client_;
+
+  // NOTE: All weak_ptrs created from this factory must be dereferenced on
+  // |client_task_runner_|. Unfortunately, due to the structure of the
+  // chromecast::TaskRunner class, weak_ptr validation is not guaranteed so this
+  // assumption cannot be validated outside of the WeakPtr class.
+  base::WeakPtrFactory<ProxyCallTranslator> weak_factory_;
+};
+
+}  // namespace media
+}  // namespace chromecast
+
+#endif  // CHROMECAST_MEDIA_CMA_BACKEND_PROXY_PROXY_CALL_TRANSLATOR_H_

chromecast/media/cma/backend/proxy/proxy_call_translator_unittest.cc

+// Copyright 2020 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 "chromecast/media/cma/backend/proxy/proxy_call_translator.h"
+
+#include <memory>
+
+#include "base/memory/scoped_refptr.h"
+#include "base/test/test_simple_task_runner.h"
+#include "chromecast/base/task_runner_impl.h"
+#include "chromecast/media/cma/backend/proxy/cast_runtime_audio_channel_broker.h"
+#include "chromecast/media/cma/backend/proxy/cma_proxy_handler.h"
+#include "testing/gmock/include/gmock/gmock.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace chromecast {
+namespace media {
+namespace {
+
+class MockTranslatorClient : public CmaProxyHandler::Client {
+ public:
+  ~MockTranslatorClient() override = default;
+
+  MOCK_METHOD0(OnError, void());
+  MOCK_METHOD1(OnPipelineStateChange, void(CmaProxyHandler::PipelineState));
+  MOCK_METHOD1(OnBytesDecoded, void(int64_t));
+  MOCK_METHOD2(OnMediaTimeUpdate, void(int64_t, base::TimeTicks));
+  MOCK_METHOD0(OnEndOfStream, void());
+};
+
+class MockDecoderChannel : public CastRuntimeAudioChannelBroker {
+ public:
+  ~MockDecoderChannel() override = default;
+
+  MOCK_METHOD2(
+      InitializeAsync,
+      void(const std::string&,
+           CastRuntimeAudioChannelBroker::CastAudioDecoderMode decoder_mode));
+  MOCK_METHOD1(SetVolumeAsync, void(float));
+  MOCK_METHOD1(SetPlaybackAsync, void(double));
+  MOCK_METHOD0(GetMediaTimeAsync, void());
+  MOCK_METHOD1(StartAsync, void(int64_t));
+  MOCK_METHOD0(StopAsync, void());
+  MOCK_METHOD0(PauseAsync, void());
+  MOCK_METHOD0(ResumeAsync, void());
+};
+
+}  // namespace
+
+class ProxyCallTranslatorTest : public testing::Test {
+ public:
+  ProxyCallTranslatorTest()
+      : task_runner_(new base::TestSimpleTaskRunner()),
+        chromecast_task_runner_(task_runner_),
+        decoder_channel_not_owned_(
+            std::make_unique<testing::StrictMock<MockDecoderChannel>>()),
+        decoder_channel_(decoder_channel_not_owned_.get()),
+        translator_(&chromecast_task_runner_,
+                    &translator_client_,
+                    std::move(decoder_channel_not_owned_)),
+        translator_as_handler_(
+            static_cast<CastRuntimeAudioChannelBroker::Handler*>(
+                &translator_)) {}
+
+  ~ProxyCallTranslatorTest() override = default;
+
+ protected:
+  scoped_refptr<base::TestSimpleTaskRunner> task_runner_;
+  TaskRunnerImpl chromecast_task_runner_;
+  testing::StrictMock<MockTranslatorClient> translator_client_;
+  std::unique_ptr<testing::StrictMock<MockDecoderChannel>>
+      decoder_channel_not_owned_;
+  testing::StrictMock<MockDecoderChannel>* decoder_channel_;
+
+  ProxyCallTranslator translator_;
+  CastRuntimeAudioChannelBroker::Handler* translator_as_handler_;
+
+  const CastRuntimeAudioChannelBroker::StatusCode failure_status_ =
+      CastRuntimeAudioChannelBroker::StatusCode::kUnknown;
+  const CastRuntimeAudioChannelBroker::StatusCode success_status_ =
+      CastRuntimeAudioChannelBroker::StatusCode::kOk;
+};
+
+TEST_F(ProxyCallTranslatorTest, TestExternalInitialize) {
+  std::string session_id = "foo";
+  EXPECT_CALL(
+      *decoder_channel_,
+      InitializeAsync(session_id, cast::media::CAST_AUDIO_DECODER_MODE_ALL));
+  translator_.Initialize(session_id,
+                         CmaProxyHandler::AudioDecoderOperationMode::kAll);
+
+  session_id = "bar";
+  EXPECT_CALL(
+      *decoder_channel_,
+      InitializeAsync(session_id,
+                      cast::media::CAST_AUDIO_DECODER_MODE_MULTIROOM_ONLY));
+  translator_.Initialize(
+      session_id, CmaProxyHandler::AudioDecoderOperationMode::kMultiroomOnly);
+
+  session_id = "foobar";
+  EXPECT_CALL(*decoder_channel_,
+              InitializeAsync(session_id,
+                              cast::media::CAST_AUDIO_DECODER_MODE_AUDIO_ONLY));
+  translator_.Initialize(
+      session_id, CmaProxyHandler::AudioDecoderOperationMode::kAudioOnly);
+}
+
+TEST_F(ProxyCallTranslatorTest, TestExternalStart) {
+  constexpr int64_t start_pts = 42;
+  EXPECT_CALL(*decoder_channel_, StartAsync(start_pts));
+  translator_.Start(start_pts);
+}
+
+TEST_F(ProxyCallTranslatorTest, TestExternalStop) {
+  EXPECT_CALL(*decoder_channel_, StopAsync());
+  translator_.Stop();
+}
+
+TEST_F(ProxyCallTranslatorTest, TestExternalPause) {
+  EXPECT_CALL(*decoder_channel_, PauseAsync());
+  translator_.Pause();
+}
+
+TEST_F(ProxyCallTranslatorTest, TestExternalResume) {
+  EXPECT_CALL(*decoder_channel_, ResumeAsync());
+  translator_.Resume();
+}
+
+TEST_F(ProxyCallTranslatorTest, TestExternalSetPlaybackRate) {
+  constexpr float rate = 42;
+  EXPECT_CALL(*decoder_channel_, SetPlaybackAsync(rate));
+  translator_.SetPlaybackRate(rate);
+}
+
+TEST_F(ProxyCallTranslatorTest, TestExternalSetVolume) {
+  constexpr float multiplier = 42;
+  EXPECT_CALL(*decoder_channel_, SetVolumeAsync(multiplier));
+  translator_.SetVolume(multiplier);
+}
+
+TEST_F(ProxyCallTranslatorTest, TestInternalHandleInitializeFailure) {
+  EXPECT_CALL(translator_client_, OnError());
+  translator_as_handler_->HandleInitializeResponse(failure_status_);
+  task_runner_->RunPendingTasks();
+}
+
+TEST_F(ProxyCallTranslatorTest, TestHandleStateChangeFailure) {
+  EXPECT_CALL(translator_client_, OnError());
+  translator_as_handler_->HandleStateChangeResponse(
+      CastRuntimeAudioChannelBroker::Handler::PipelineState::
+          PIPELINE_STATE_UNINITIALIZED,
+      failure_status_);
+  task_runner_->RunPendingTasks();
+}
+
+TEST_F(ProxyCallTranslatorTest, TestHandleSetVolumeFailure) {
+  EXPECT_CALL(translator_client_, OnError());
+  translator_as_handler_->HandleSetVolumeResponse(failure_status_);
+  task_runner_->RunPendingTasks();
+}
+
+TEST_F(ProxyCallTranslatorTest, TesSetPlaybackFailure) {
+  EXPECT_CALL(translator_client_, OnError());
+  translator_as_handler_->HandleSetPlaybackResponse(failure_status_);
+  task_runner_->RunPendingTasks();
+}
+
+TEST_F(ProxyCallTranslatorTest, TestPushBufferFailure) {
+  EXPECT_CALL(translator_client_, OnError());
+  translator_as_handler_->HandlePushBufferResponse(42, failure_status_);
+  task_runner_->RunPendingTasks();
+}
+
+TEST_F(ProxyCallTranslatorTest, TestStateChangeSuccess) {
+  EXPECT_CALL(
+      translator_client_,
+      OnPipelineStateChange(CmaProxyHandler::PipelineState::kUninitialized));
+  translator_as_handler_->HandleStateChangeResponse(
+      CastRuntimeAudioChannelBroker::Handler::PipelineState::
+          PIPELINE_STATE_UNINITIALIZED,
+      success_status_);
+  task_runner_->RunPendingTasks();
+
+  EXPECT_CALL(translator_client_,
+              OnPipelineStateChange(CmaProxyHandler::PipelineState::kStopped));
+  translator_as_handler_->HandleStateChangeResponse(
+      CastRuntimeAudioChannelBroker::Handler::PipelineState::
+          PIPELINE_STATE_STOPPED,
+      success_status_);
+  task_runner_->RunPendingTasks();
+
+  EXPECT_CALL(translator_client_,
+              OnPipelineStateChange(CmaProxyHandler::PipelineState::kPlaying));
+  translator_as_handler_->HandleStateChangeResponse(
+      CastRuntimeAudioChannelBroker::Handler::PipelineState::
+          PIPELINE_STATE_PLAYING,
+      success_status_);
+  task_runner_->RunPendingTasks();
+
+  EXPECT_CALL(translator_client_,
+              OnPipelineStateChange(CmaProxyHandler::PipelineState::kPaused));
+  translator_as_handler_->HandleStateChangeResponse(
+      CastRuntimeAudioChannelBroker::Handler::PipelineState::
+          PIPELINE_STATE_PAUSED,
+      success_status_);
+  task_runner_->RunPendingTasks();
+}
+
+TEST_F(ProxyCallTranslatorTest, TestPushBufferSuccess) {
+  EXPECT_CALL(translator_client_, OnBytesDecoded(42));
+  translator_as_handler_->HandlePushBufferResponse(42, success_status_);
+  task_runner_->RunPendingTasks();
+
+  EXPECT_CALL(translator_client_, OnBytesDecoded(112358));
+  translator_as_handler_->HandlePushBufferResponse(112358, success_status_);
+  task_runner_->RunPendingTasks();
+}
+
+}  // namespace media
+}  // namespace chromecast