[Cast] Use gRPC Pipeline in CmaBackendProxy

This CL Updates CmaBackendProxy to make the following changes:
- Calls to CmaBackendProxy now call into the CmaProxyHandler.
- Calls to MultizoneAudioDecoderProxyImpl now also are duplicated to the
  underlying CmaBackend::AudioDecoder (when appropriate).

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

chromecast/media/cma/backend/cma_backend_factory_impl.cc

@@ -29,7 +29,7 @@ std::unique_ptr<CmaBackend> CmaBackendFactoryImpl::CreateBackend(
       media_pipeline_backend_manager_->CreateBackend(params);
 
 #if BUILDFLAG(ENABLE_CHROMIUM_RUNTIME_CAST_RENDERER)
-  backend = std::make_unique<CmaBackendProxy>(std::move(backend));
+  backend = std::make_unique<CmaBackendProxy>(params, std::move(backend));
 #endif  // BUILDFLAG(ENABLE_CHROMIUM_RUNTIME_CAST_RENDERER)
 
   return backend;

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

@@ -12,6 +12,7 @@
 #include "base/notreached.h"
 #include "chromecast/media/cma/backend/proxy/multizone_audio_decoder_proxy_impl.h"
 #include "chromecast/public/media/decoder_config.h"
+#include "chromecast/public/media/media_pipeline_device_params.h"
 
 namespace chromecast {
 namespace media {
@@ -25,20 +26,19 @@ int64_t kMaxAllowedPtsDrift = 500;
 
 }  // namespace
 
-CmaBackendProxy::CmaBackendProxy(
-    std::unique_ptr<CmaBackend> delegated_video_pipeline)
-    : CmaBackendProxy(
-          std::move(delegated_video_pipeline),
-          base::BindOnce([]() -> std::unique_ptr<MultizoneAudioDecoderProxy> {
-            return std::make_unique<MultizoneAudioDecoderProxyImpl>();
-          })) {}
+CmaBackendProxy::CmaBackendProxy(const MediaPipelineDeviceParams& params,
+                                 std::unique_ptr<CmaBackend> delegated_pipeline)
+    : CmaBackendProxy(base::BindOnce(&CmaBackendProxy::CreateAudioDecoderProxy,
+                                     base::Unretained(this),
+                                     params),
+                      std::move(delegated_pipeline)) {}
 
 CmaBackendProxy::CmaBackendProxy(
-    std::unique_ptr<CmaBackend> delegated_video_pipeline,
-    CmaBackendProxy::AudioDecoderFactoryCB audio_decoder_factory)
-    : delegated_video_pipeline_(std::move(delegated_video_pipeline)),
+    CmaBackendProxy::AudioDecoderFactoryCB audio_decoder_factory,
+    std::unique_ptr<CmaBackend> delegated_pipeline)
+    : delegated_pipeline_(std::move(delegated_pipeline)),
       audio_decoder_factory_(std::move(audio_decoder_factory)) {
-  DCHECK(delegated_video_pipeline_);
+  DCHECK(delegated_pipeline_);
   DCHECK(audio_decoder_factory_);
 }
 
@@ -53,28 +53,36 @@ CmaBackend::AudioDecoder* CmaBackendProxy::CreateAudioDecoder() {
 
 CmaBackend::VideoDecoder* CmaBackendProxy::CreateVideoDecoder() {
   has_video_decoder_ = true;
-  return delegated_video_pipeline_->CreateVideoDecoder();
+  return delegated_pipeline_->CreateVideoDecoder();
 }
 
 bool CmaBackendProxy::Initialize() {
-  if (has_video_decoder_ && !delegated_video_pipeline_->Initialize()) {
-    return false;
+  if (audio_decoder_) {
+    audio_decoder_->Initialize();
   }
 
-  return !audio_decoder_ || audio_decoder_->Initialize();
+  if (audio_decoder_ || has_video_decoder_) {
+    return delegated_pipeline_->Initialize();
+  } else {
+    return true;
+  }
 }
 
 bool CmaBackendProxy::Start(int64_t start_pts) {
-  if (has_video_decoder_ && !delegated_video_pipeline_->Start(start_pts)) {
-    return false;
+  if (audio_decoder_) {
+    audio_decoder_->Start(start_pts);
   }
 
-  return !audio_decoder_ || audio_decoder_->Start(start_pts);
+  if (audio_decoder_ || has_video_decoder_) {
+    return delegated_pipeline_->Start(start_pts);
+  } else {
+    return true;
+  }
 }
 
 void CmaBackendProxy::Stop() {
-  if (has_video_decoder_) {
-    delegated_video_pipeline_->Stop();
+  if (has_video_decoder_ || audio_decoder_) {
+    delegated_pipeline_->Stop();
   }
 
   if (audio_decoder_) {
@@ -83,60 +91,58 @@ void CmaBackendProxy::Stop() {
 }
 
 bool CmaBackendProxy::Pause() {
-  bool result = true;
-
-  if (has_video_decoder_) {
-    result &= delegated_video_pipeline_->Pause();
-  }
-
   if (audio_decoder_) {
-    result &= audio_decoder_->Pause();
+    audio_decoder_->Pause();
   }
 
-  return result;
+  if (audio_decoder_ || has_video_decoder_) {
+    return delegated_pipeline_->Pause();
+  } else {
+    return true;
+  }
 }
 
 bool CmaBackendProxy::Resume() {
-  if (has_video_decoder_ && !delegated_video_pipeline_->Resume()) {
-    return false;
+  if (audio_decoder_) {
+    audio_decoder_->Resume();
   }
 
-  return !audio_decoder_ || audio_decoder_->Resume();
+  if (audio_decoder_ || has_video_decoder_) {
+    return delegated_pipeline_->Resume();
+  } else {
+    return true;
+  }
 }
 
 int64_t CmaBackendProxy::GetCurrentPts() {
-  if (audio_decoder_ && has_video_decoder_) {
+  if (audio_decoder_) {
     const int64_t audio_pts = audio_decoder_->GetCurrentPts();
-    const int64_t video_pts = delegated_video_pipeline_->GetCurrentPts();
+    const int64_t video_pts = delegated_pipeline_->GetCurrentPts();
     const int64_t min = std::min(audio_pts, video_pts);
     LOG_IF(WARNING, std::max(audio_pts, video_pts) - min > kMaxAllowedPtsDrift);
     return min;
-  } else if (audio_decoder_) {
-    return audio_decoder_->GetCurrentPts();
   } else if (has_video_decoder_) {
-    return delegated_video_pipeline_->GetCurrentPts();
+    return delegated_pipeline_->GetCurrentPts();
   } else {
     return std::numeric_limits<int64_t>::min();
   }
 }
 
 bool CmaBackendProxy::SetPlaybackRate(float rate) {
-  bool result = true;
-
-  if (has_video_decoder_) {
-    result &= delegated_video_pipeline_->SetPlaybackRate(rate);
-  }
-
   if (audio_decoder_) {
-    result &= audio_decoder_->SetPlaybackRate(rate);
+    audio_decoder_->SetPlaybackRate(rate);
   }
 
-  return result;
+  if (audio_decoder_ || has_video_decoder_) {
+    return delegated_pipeline_->SetPlaybackRate(rate);
+  } else {
+    return true;
+  }
 }
 
 void CmaBackendProxy::LogicalPause() {
-  if (has_video_decoder_) {
-    delegated_video_pipeline_->LogicalPause();
+  if (has_video_decoder_ || audio_decoder_) {
+    delegated_pipeline_->LogicalPause();
   }
 
   if (audio_decoder_) {
@@ -145,8 +151,8 @@ void CmaBackendProxy::LogicalPause() {
 }
 
 void CmaBackendProxy::LogicalResume() {
-  if (has_video_decoder_) {
-    delegated_video_pipeline_->LogicalResume();
+  if (has_video_decoder_ || audio_decoder_) {
+    delegated_pipeline_->LogicalResume();
   }
 
   if (audio_decoder_) {
@@ -154,5 +160,14 @@ void CmaBackendProxy::LogicalResume() {
   }
 }
 
+std::unique_ptr<MultizoneAudioDecoderProxy>
+CmaBackendProxy::CreateAudioDecoderProxy(
+    const MediaPipelineDeviceParams& params) {
+  CmaBackend::AudioDecoder* downstream_decoder =
+      delegated_pipeline_->CreateAudioDecoder();
+  return std::make_unique<MultizoneAudioDecoderProxyImpl>(params,
+                                                          downstream_decoder);
+}
+
 }  // namespace media
 }  // namespace chromecast

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

@@ -17,11 +17,14 @@
 namespace chromecast {
 namespace media {
 
-// This class is used to proxy audio data to an external
-// CmaBackend::AudioDecoder over gRPC, while delegating video decoding to an
-// alternate CMA Backend.
+struct MediaPipelineDeviceParams;
+
+// This class is used to intercept audio data and control messages being sent to
+// the local CmaBackend and forward it to a remote instance prior to local
+// processing in the underlying CmaBackend.
+//
 // NOTE: By design, this class does NOT handle a/v sync drift between
-// |audio_decoder_| and |delegated_video_pipeline_|.
+// |audio_decoder_| and |delegated_pipeline_|.
 class CmaBackendProxy : public CmaBackend {
  public:
   using AudioDecoderFactoryCB =
@@ -29,11 +32,15 @@ class CmaBackendProxy : public CmaBackend {
 
   // Creates a new CmaBackendProxy such that all video processing is delegated
   // to |delegated_video_pipeline|.
-  explicit CmaBackendProxy(
-      std::unique_ptr<CmaBackend> delegated_video_pipeline);
+  CmaBackendProxy(const MediaPipelineDeviceParams& params,
+                  std::unique_ptr<CmaBackend> delegated_video_pipeline);
   ~CmaBackendProxy() override;
 
   // MediaPipelineBackend implementation:
+  //
+  // NOTE: Each of these calls must forward the call both to the gRPC Pipe (if
+  // the audio decoder exists) and to the local audio decoder (if either audio
+  // or video decoding is occuring).
   CmaBackend::AudioDecoder* CreateAudioDecoder() override;
   CmaBackend::VideoDecoder* CreateVideoDecoder() override;
   bool Initialize() override;
@@ -50,19 +57,23 @@ class CmaBackendProxy : public CmaBackend {
   friend class CmaBackendProxyTest;
 
   // Creates a new CmaBackendProxy such that all video processing is delegated
-  // to |delegated_video_pipeline| and all audio processing is delegated to a
+  // to |delegated_pipeline| and all audio processing is delegated to a
   // new MultizoneAudioDecoderProxy created by |audio_decoder_factory|.
-  CmaBackendProxy(std::unique_ptr<CmaBackend> delegated_video_pipeline,
-                  AudioDecoderFactoryCB audio_decoder_factory);
+  CmaBackendProxy(AudioDecoderFactoryCB audio_decoder_factory,
+                  std::unique_ptr<CmaBackend> delegated_pipeline);
+
+  // Helper to create the AudioDecoder instance, used with base::bind.
+  std::unique_ptr<MultizoneAudioDecoderProxy> CreateAudioDecoderProxy(
+      const MediaPipelineDeviceParams& params);
 
   // The audio decoder to which audio operations should be delegated.
   std::unique_ptr<MultizoneAudioDecoderProxy> audio_decoder_;
 
   // The CMA Backend to which all video decoding is delegated.
-  std::unique_ptr<CmaBackend> delegated_video_pipeline_;
+  std::unique_ptr<CmaBackend> delegated_pipeline_;
 
   // Determines whether a video decoder is being used. If not, calls should not
-  // be delegated to the |delegated_video_pipeline_|, as it may not behave as
+  // be delegated to the |delegated_pipeline_|, as it may not behave as
   // expected when neither the audio or video decoders are initialized.
   bool has_video_decoder_ = false;
 

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

@@ -27,12 +27,12 @@ class MultizoneAudioDecoderProxy : public CmaBackend::AudioDecoder {
 
   ~MultizoneAudioDecoderProxy() override = default;
 
-  virtual bool Initialize() = 0;
-  virtual bool Start(int64_t start_pts) = 0;
+  virtual void Initialize() = 0;
+  virtual void Start(int64_t start_pts) = 0;
   virtual void Stop() = 0;
-  virtual bool Pause() = 0;
-  virtual bool Resume() = 0;
-  virtual bool SetPlaybackRate(float rate) = 0;
+  virtual void Pause() = 0;
+  virtual void Resume() = 0;
+  virtual void SetPlaybackRate(float rate) = 0;
   virtual void LogicalPause() = 0;
   virtual void LogicalResume() = 0;
   virtual int64_t GetCurrentPts() const = 0;

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

@@ -6,93 +6,143 @@
 
 #include "base/notreached.h"
 #include "chromecast/public/media/decoder_config.h"
+#include "chromecast/public/media/media_pipeline_device_params.h"
 
 namespace chromecast {
 namespace media {
 
-MultizoneAudioDecoderProxyImpl::MultizoneAudioDecoderProxyImpl() = default;
+MultizoneAudioDecoderProxyImpl::MultizoneAudioDecoderProxyImpl(
+    const MediaPipelineDeviceParams& params,
+    CmaBackend::AudioDecoder* downstream_decoder)
+    : cast_session_id_(params.session_id),
+      decoder_mode_(CmaProxyHandler::AudioDecoderOperationMode::kMultiroomOnly),
+      downstream_decoder_(downstream_decoder),
+      proxy_handler_(CmaProxyHandler::Create(params.task_runner, this)) {
+  DCHECK(downstream_decoder_);
+  DCHECK(proxy_handler_);
+
+  DETACH_FROM_SEQUENCE(sequence_checker_);
+}
 
 MultizoneAudioDecoderProxyImpl::~MultizoneAudioDecoderProxyImpl() = default;
 
-bool MultizoneAudioDecoderProxyImpl::Initialize() {
-  NOTIMPLEMENTED();
-  return true;
+void MultizoneAudioDecoderProxyImpl::Initialize() {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+  proxy_handler_->Initialize(cast_session_id_, decoder_mode_);
 }
 
-bool MultizoneAudioDecoderProxyImpl::Start(int64_t start_pts) {
-  NOTIMPLEMENTED();
-  return true;
+void MultizoneAudioDecoderProxyImpl::Start(int64_t start_pts) {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+  proxy_handler_->Start(start_pts);
 }
 
 void MultizoneAudioDecoderProxyImpl::Stop() {
-  NOTIMPLEMENTED();
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+  proxy_handler_->Stop();
 }
 
-bool MultizoneAudioDecoderProxyImpl::Pause() {
-  NOTIMPLEMENTED();
-  return true;
+void MultizoneAudioDecoderProxyImpl::Pause() {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+  proxy_handler_->Pause();
 }
 
-bool MultizoneAudioDecoderProxyImpl::Resume() {
-  NOTIMPLEMENTED();
-  return true;
+void MultizoneAudioDecoderProxyImpl::Resume() {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+  proxy_handler_->Resume();
 }
 
-bool MultizoneAudioDecoderProxyImpl::SetPlaybackRate(float rate) {
-  NOTIMPLEMENTED();
-  return true;
+void MultizoneAudioDecoderProxyImpl::SetPlaybackRate(float rate) {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+  proxy_handler_->SetPlaybackRate(rate);
 }
 
 void MultizoneAudioDecoderProxyImpl::LogicalPause() {
-  NOTIMPLEMENTED();
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+  // There is intentionally no proxy implementation of this method.
 }
 
 void MultizoneAudioDecoderProxyImpl::LogicalResume() {
-  NOTIMPLEMENTED();
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+  // There is intentionally no proxy implementation of this method.
 }
 
 int64_t MultizoneAudioDecoderProxyImpl::GetCurrentPts() const {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+
+  // This will be implemented as part of audio-audio sync.
+  NOTIMPLEMENTED();
   return pts_offset_;
 }
 
 void MultizoneAudioDecoderProxyImpl::SetDelegate(Delegate* delegate) {
-  NOTIMPLEMENTED();
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+  downstream_decoder_->SetDelegate(delegate);
 }
 
 MultizoneAudioDecoderProxy::BufferStatus
 MultizoneAudioDecoderProxyImpl::PushBuffer(
     scoped_refptr<DecoderBufferBase> buffer) {
-  NOTIMPLEMENTED();
-  return BufferStatus::kBufferSuccess;
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+
+  if (!proxy_handler_->PushBuffer(buffer)) {
+    return BufferStatus::kBufferFailed;
+  }
+
+  return downstream_decoder_->PushBuffer(std::move(buffer));
 }
 
 bool MultizoneAudioDecoderProxyImpl::SetConfig(const AudioConfig& config) {
-  NOTIMPLEMENTED();
-  return true;
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+  return proxy_handler_->SetConfig(config) &&
+         downstream_decoder_->SetConfig(config);
 }
 
 bool MultizoneAudioDecoderProxyImpl::SetVolume(float multiplier) {
-  NOTIMPLEMENTED();
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+
+  // The proxy implementation of this method is INTENTIONALLY not called here.
   return true;
 }
 
 MultizoneAudioDecoderProxyImpl::RenderingDelay
 MultizoneAudioDecoderProxyImpl::GetRenderingDelay() {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+
+  // This will be implemented as part of audio-audio sync.
   NOTIMPLEMENTED();
   return RenderingDelay{};
 }
 
 void MultizoneAudioDecoderProxyImpl::GetStatistics(Statistics* statistics) {
-  NOTIMPLEMENTED();
+  DCHECK(statistics);
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+  statistics->decoded_bytes = bytes_decoded_;
 }
 
 bool MultizoneAudioDecoderProxyImpl::RequiresDecryption() {
-  NOTIMPLEMENTED();
-  return true;
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+  return downstream_decoder_->RequiresDecryption();
 }
 
 void MultizoneAudioDecoderProxyImpl::SetObserver(Observer* observer) {
-  NOTIMPLEMENTED();
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+  downstream_decoder_->SetObserver(observer);
+}
+
+void MultizoneAudioDecoderProxyImpl::OnError() {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+  NOTREACHED();
+}
+
+void MultizoneAudioDecoderProxyImpl::OnPipelineStateChange(
+    CmaProxyHandler::PipelineState state) {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+}
+
+void MultizoneAudioDecoderProxyImpl::OnBytesDecoded(
+    int64_t decoded_byte_count) {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+  bytes_decoded_ = decoded_byte_count;
 }
 
 }  // namespace media

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

@@ -8,33 +8,44 @@
 #include <limits>
 
 #include "base/memory/ref_counted.h"
+#include "base/sequence_checker.h"
 #include "chromecast/media/api/cma_backend.h"
 #include "chromecast/media/api/decoder_buffer_base.h"
+#include "chromecast/media/cma/backend/proxy/cma_proxy_handler.h"
 #include "chromecast/media/cma/backend/proxy/multizone_audio_decoder_proxy.h"
 
 namespace chromecast {
 namespace media {
 
 struct AudioConfig;
+struct MediaPipelineDeviceParams;
 
 // This class is used to proxy audio data to an external
 // CmaBackend::AudioDecoder over gRPC.
-class MultizoneAudioDecoderProxyImpl : public MultizoneAudioDecoderProxy {
+class MultizoneAudioDecoderProxyImpl : public MultizoneAudioDecoderProxy,
+                                       public CmaProxyHandler::Client {
  public:
   // Creates a new MultizoneAudioDecoderProxy, such that in the event of an
   // unrecoverable error, |fatal_error_callback| will be called. Fallowing this
   // call, this instance will be in an undefined state.
-  MultizoneAudioDecoderProxyImpl();
+  MultizoneAudioDecoderProxyImpl(const MediaPipelineDeviceParams& params,
+                                 CmaBackend::AudioDecoder* downstream_decoder);
   ~MultizoneAudioDecoderProxyImpl() override;
 
   // MultizoneAudioDecoderProxy implementation:
-  bool Initialize() override;
-  bool Start(int64_t start_pts) override;
+  //
+  // Note that the methods implementing of CmaBackend::AudioDecoder (which
+  // MultizoneAudioDecoderProxy extends) must call both into the downstream
+  // decoder and into the |proxy_handler_|, so that audio can be processed both
+  // locally and remotely. The remaining methods should NOT call into the
+  // downstream CmaBackend, as this is the responsibility of the caller.
+  void Initialize() override;
+  void Start(int64_t start_pts) override;
   void Stop() override;
-  bool Pause() override;
-  bool Resume() override;
+  void Pause() override;
+  void Resume() override;
   int64_t GetCurrentPts() const override;
-  bool SetPlaybackRate(float rate) override;
+  void SetPlaybackRate(float rate) override;
   void LogicalPause() override;
   void LogicalResume() override;
   void SetDelegate(Delegate* delegate) override;
@@ -47,10 +58,35 @@ class MultizoneAudioDecoderProxyImpl : public MultizoneAudioDecoderProxy {
   void SetObserver(Observer* observer) override;
 
  private:
+  // CmaProxyHandler::Client overrides:
+  void OnError() override;
+  void OnPipelineStateChange(CmaProxyHandler::PipelineState state) override;
+  void OnBytesDecoded(int64_t decoded_byte_count) override;
+
   // The PTS offset as determined by the receiver of the gRPC endpoint wrapped
   // by this class. This value is updated as new PTS values are received over
   // the IPC.
   int64_t pts_offset_ = std::numeric_limits<int64_t>::min();
+
+  // Number of bytes decoded so far.
+  int64_t bytes_decoded_ = 0;
+
+  // Parameters for the Initialize() call captured in the ctor.
+  const std::string cast_session_id_;
+  const CmaProxyHandler::AudioDecoderOperationMode decoder_mode_;
+
+  // Decoder to which the CmaBackend::AudioDecoder calls should be duplicated
+  // (when appropriate). It is expected to be the AudioDecoder associated with
+  // the "real" CmaBackend, which plays out audio data using the physical
+  // device's hardware. By design, this decoder is always assumed to exist.
+  CmaBackend::AudioDecoder* const downstream_decoder_;
+
+  // This is the local instance representing the "remote" backend. All above
+  // public method calls should call into this instance to proxy the call to
+  // the remote backend.
+  std::unique_ptr<CmaProxyHandler> proxy_handler_;
+
+  SEQUENCE_CHECKER(sequence_checker_);
 };
 
 }  // namespace media