chromecast/renderer/media/media_pipeline_proxy.cc

   1 // Copyright 2014 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "chromecast/renderer/media/media_pipeline_proxy.h"
   6
   7 #include "base/bind.h"
   8 #include "base/callback_helpers.h"
   9 #include "base/location.h"
  10 #include "base/logging.h"
  11 #include "base/single_thread_task_runner.h"
  12 #include "chromecast/common/media/cma_messages.h"
  13 #include "chromecast/media/cma/base/coded_frame_provider.h"
  14 #include "chromecast/renderer/media/audio_pipeline_proxy.h"
  15 #include "chromecast/renderer/media/media_channel_proxy.h"
  16 #include "chromecast/renderer/media/video_pipeline_proxy.h"
  17
  18 namespace chromecast {
  19 namespace media {
  20
  21 // MediaPipelineProxyInternal -
  22 // This class is not thread safe and should run on the same thread
  23 // as the media channel proxy.
  24 class MediaPipelineProxyInternal {
  25  public:
  26   static void Release(scoped_ptr<MediaPipelineProxyInternal> proxy);
  27
  28   explicit MediaPipelineProxyInternal(
  29       scoped_refptr<MediaChannelProxy> media_channel_proxy);
  30   virtual ~MediaPipelineProxyInternal();
  31
  32   void SetClient(const MediaPipelineClient& client);
  33   void SetCdm(int render_frame_id, int cdm_id);
  34   void StartPlayingFrom(const base::TimeDelta& time);
  35   void Flush(const ::media::PipelineStatusCB& status_cb);
  36   void Stop();
  37   void SetPlaybackRate(double playback_rate);
  38
  39  private:
  40   void Shutdown();
  41
  42   // Callbacks for CmaMessageFilterHost::MediaDelegate.
  43   void OnStateChanged(::media::PipelineStatus status);
  44
  45   base::ThreadChecker thread_checker_;
  46
  47   scoped_refptr<MediaChannelProxy> media_channel_proxy_;
  48
  49   MediaPipelineClient client_;
  50
  51   // Store the callback for a pending state transition.
  52   ::media::PipelineStatusCB status_cb_;
  53
  54   DISALLOW_COPY_AND_ASSIGN(MediaPipelineProxyInternal);
  55 };
  56
  57 // static
  58 void MediaPipelineProxyInternal::Release(
  59     scoped_ptr<MediaPipelineProxyInternal> proxy) {
  60   proxy->Shutdown();
  61 }
  62
  63 MediaPipelineProxyInternal::MediaPipelineProxyInternal(
  64     scoped_refptr<MediaChannelProxy> media_channel_proxy)
  65     : media_channel_proxy_(media_channel_proxy) {
  66   DCHECK(media_channel_proxy.get());
  67
  68   // Creation can be done on a different thread.
  69   thread_checker_.DetachFromThread();
  70 }
  71
  72 MediaPipelineProxyInternal::~MediaPipelineProxyInternal() {
  73 }
  74
  75 void MediaPipelineProxyInternal::Shutdown() {
  76   DCHECK(thread_checker_.CalledOnValidThread());
  77
  78   // Remove any callback on VideoPipelineProxyInternal.
  79   media_channel_proxy_->SetMediaDelegate(
  80       CmaMessageFilterProxy::MediaDelegate());
  81 }
  82
  83 void MediaPipelineProxyInternal::SetClient(
  84     const MediaPipelineClient& client) {
  85   DCHECK(thread_checker_.CalledOnValidThread());
  86   DCHECK(!client.error_cb.is_null());
  87   DCHECK(!client.buffering_state_cb.is_null());
  88   client_ = client;
  89
  90   CmaMessageFilterProxy::MediaDelegate delegate;
  91   delegate.state_changed_cb =
  92       base::Bind(&MediaPipelineProxyInternal::OnStateChanged,
  93                  base::Unretained(this));
  94   delegate.client = client;
  95   bool success = media_channel_proxy_->SetMediaDelegate(delegate);
  96   CHECK(success);
  97 }
  98
  99 void MediaPipelineProxyInternal::SetCdm(int render_frame_id, int cdm_id) {
 100   DCHECK(thread_checker_.CalledOnValidThread());
 101   bool success = media_channel_proxy_->Send(scoped_ptr<IPC::Message>(
 102       new CmaHostMsg_SetCdm(media_channel_proxy_->GetId(),
 103                             render_frame_id,
 104                             cdm_id)));
 105   LOG_IF(ERROR, !success) << "Failed to send SetCdm=" << cdm_id;
 106 }
 107
 108 void MediaPipelineProxyInternal::Flush(
 109     const ::media::PipelineStatusCB& status_cb) {
 110   DCHECK(thread_checker_.CalledOnValidThread());
 111   bool success = media_channel_proxy_->Send(scoped_ptr<IPC::Message>(
 112       new CmaHostMsg_Flush(media_channel_proxy_->GetId())));
 113   if (!success) {
 114     status_cb.Run(::media::PIPELINE_ERROR_ABORT);
 115     return;
 116   }
 117   DCHECK(status_cb_.is_null());
 118   status_cb_ = status_cb;
 119 }
 120
 121 void MediaPipelineProxyInternal::Stop() {
 122   DCHECK(thread_checker_.CalledOnValidThread());
 123   bool success = media_channel_proxy_->Send(scoped_ptr<IPC::Message>(
 124       new CmaHostMsg_Stop(media_channel_proxy_->GetId())));
 125   if (!success)
 126     client_.error_cb.Run(::media::PIPELINE_ERROR_ABORT);
 127 }
 128
 129 void MediaPipelineProxyInternal::StartPlayingFrom(const base::TimeDelta& time) {
 130   DCHECK(thread_checker_.CalledOnValidThread());
 131   bool success = media_channel_proxy_->Send(scoped_ptr<IPC::Message>(
 132       new CmaHostMsg_StartPlayingFrom(
 133           media_channel_proxy_->GetId(), time)));
 134   if (!success)
 135     client_.error_cb.Run(::media::PIPELINE_ERROR_ABORT);
 136 }
 137
 138 void MediaPipelineProxyInternal::SetPlaybackRate(double playback_rate) {
 139   DCHECK(thread_checker_.CalledOnValidThread());
 140   media_channel_proxy_->Send(scoped_ptr<IPC::Message>(
 141       new CmaHostMsg_SetPlaybackRate(
 142           media_channel_proxy_->GetId(), playback_rate)));
 143 }
 144
 145 void MediaPipelineProxyInternal::OnStateChanged(
 146     ::media::PipelineStatus status) {
 147   DCHECK(thread_checker_.CalledOnValidThread());
 148   DCHECK(!status_cb_.is_null());
 149   base::ResetAndReturn(&status_cb_).Run(status);
 150 }
 151
 152 // A macro runs current member function on |io_task_runner_| thread.
 153 #define FORWARD_ON_IO_THREAD(param_fn, ...)                        \
 154   io_task_runner_->PostTask(                                       \
 155       FROM_HERE, base::Bind(&MediaPipelineProxyInternal::param_fn, \
 156                             base::Unretained(proxy_.get()), ##__VA_ARGS__))
 157
 158 MediaPipelineProxy::MediaPipelineProxy(
 159     int render_frame_id,
 160     scoped_refptr<base::SingleThreadTaskRunner> io_task_runner,
 161     LoadType load_type)
 162     : io_task_runner_(io_task_runner),
 163       render_frame_id_(render_frame_id),
 164       media_channel_proxy_(new MediaChannelProxy),
 165       proxy_(new MediaPipelineProxyInternal(media_channel_proxy_)),
 166       has_audio_(false),
 167       has_video_(false),
 168       audio_pipeline_(
 169           new AudioPipelineProxy(io_task_runner, media_channel_proxy_)),
 170       video_pipeline_(
 171           new VideoPipelineProxy(io_task_runner, media_channel_proxy_)),
 172       weak_factory_(this) {
 173   weak_this_ = weak_factory_.GetWeakPtr();
 174   io_task_runner_->PostTask(
 175       FROM_HERE,
 176       base::Bind(&MediaChannelProxy::Open, media_channel_proxy_, load_type));
 177   thread_checker_.DetachFromThread();
 178 }
 179
 180 MediaPipelineProxy::~MediaPipelineProxy() {
 181   io_task_runner_->PostTask(
 182       FROM_HERE,
 183       base::Bind(&MediaPipelineProxyInternal::Release, base::Passed(&proxy_)));
 184   io_task_runner_->PostTask(
 185       FROM_HERE, base::Bind(&MediaChannelProxy::Close, media_channel_proxy_));
 186 }
 187
 188 void MediaPipelineProxy::SetClient(
 189     const MediaPipelineClient& client) {
 190   DCHECK(thread_checker_.CalledOnValidThread());
 191   FORWARD_ON_IO_THREAD(SetClient, client);
 192 }
 193
 194 void MediaPipelineProxy::SetCdm(int cdm_id) {
 195   DCHECK(thread_checker_.CalledOnValidThread());
 196   FORWARD_ON_IO_THREAD(SetCdm, render_frame_id_, cdm_id);
 197 }
 198
 199 AudioPipeline* MediaPipelineProxy::GetAudioPipeline() const {
 200   return audio_pipeline_.get();
 201 }
 202
 203 VideoPipeline* MediaPipelineProxy::GetVideoPipeline() const {
 204   return video_pipeline_.get();
 205 }
 206
 207 void MediaPipelineProxy::InitializeAudio(
 208     const ::media::AudioDecoderConfig& config,
 209     scoped_ptr<CodedFrameProvider> frame_provider,
 210     const ::media::PipelineStatusCB& status_cb) {
 211   DCHECK(thread_checker_.CalledOnValidThread());
 212   has_audio_ = true;
 213   audio_pipeline_->Initialize(config, frame_provider.Pass(), status_cb);
 214 }
 215
 216 void MediaPipelineProxy::InitializeVideo(
 217     const std::vector<::media::VideoDecoderConfig>& configs,
 218     scoped_ptr<CodedFrameProvider> frame_provider,
 219     const ::media::PipelineStatusCB& status_cb) {
 220   DCHECK(thread_checker_.CalledOnValidThread());
 221   has_video_ = true;
 222   video_pipeline_->Initialize(configs, frame_provider.Pass(), status_cb);
 223 }
 224
 225 void MediaPipelineProxy::StartPlayingFrom(base::TimeDelta time) {
 226   DCHECK(thread_checker_.CalledOnValidThread());
 227   if (has_audio_)
 228     audio_pipeline_->StartFeeding();
 229   if (has_video_)
 230     video_pipeline_->StartFeeding();
 231   FORWARD_ON_IO_THREAD(StartPlayingFrom, time);
 232 }
 233
 234 void MediaPipelineProxy::Flush(const ::media::PipelineStatusCB& status_cb) {
 235   DCHECK(thread_checker_.CalledOnValidThread());
 236   DCHECK(has_audio_ || has_video_);
 237
 238   ::media::SerialRunner::Queue bound_fns;
 239   if (has_audio_) {
 240     bound_fns.Push(base::Bind(&AudioPipelineProxy::Flush,
 241                               base::Unretained(audio_pipeline_.get())));
 242   }
 243   if (has_video_) {
 244     bound_fns.Push(base::Bind(&VideoPipelineProxy::Flush,
 245                               base::Unretained(video_pipeline_.get())));
 246   }
 247   ::media::PipelineStatusCB cb =
 248       base::Bind(&MediaPipelineProxy::OnProxyFlushDone, weak_this_, status_cb);
 249   pending_flush_callbacks_ = ::media::SerialRunner::Run(bound_fns, cb);
 250 }
 251
 252 void MediaPipelineProxy::OnProxyFlushDone(
 253     const ::media::PipelineStatusCB& status_cb,
 254     ::media::PipelineStatus status) {
 255   DCHECK(thread_checker_.CalledOnValidThread());
 256   DCHECK_EQ(status, ::media::PIPELINE_OK);
 257   pending_flush_callbacks_.reset();
 258   FORWARD_ON_IO_THREAD(Flush, status_cb);
 259 }
 260
 261 void MediaPipelineProxy::Stop() {
 262   DCHECK(thread_checker_.CalledOnValidThread());
 263   DCHECK(has_audio_ || has_video_);
 264
 265   // Cancel pending flush callbacks since we are about to stop/shutdown
 266   // audio/video pipelines. This will ensure A/V Flush won't happen in
 267   // stopped state.
 268   pending_flush_callbacks_.reset();
 269
 270   if (has_audio_)
 271     audio_pipeline_->Stop();
 272   if (has_video_)
 273     video_pipeline_->Stop();
 274
 275   FORWARD_ON_IO_THREAD(Stop);
 276 }
 277
 278 void MediaPipelineProxy::SetPlaybackRate(double playback_rate) {
 279   DCHECK(thread_checker_.CalledOnValidThread());
 280   FORWARD_ON_IO_THREAD(SetPlaybackRate, playback_rate);
 281 }
 282
 283 }  // namespace cma
 284 }  // namespace chromecast