chromecast/renderer/media/audio_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/audio_pipeline_proxy.h"
   6
   7 #include "base/bind.h"
   8 #include "base/callback_helpers.h"
   9 #include "base/memory/shared_memory.h"
  10 #include "base/message_loop/message_loop.h"
  11 #include "base/threading/thread_checker.h"
  12 #include "chromecast/common/media/cma_messages.h"
  13 #include "chromecast/common/media/shared_memory_chunk.h"
  14 #include "chromecast/media/cma/base/buffering_defs.h"
  15 #include "chromecast/media/cma/base/cma_logging.h"
  16 #include "chromecast/media/cma/base/coded_frame_provider.h"
  17 #include "chromecast/media/cma/ipc/media_message_fifo.h"
  18 #include "chromecast/media/cma/ipc_streamer/av_streamer_proxy.h"
  19 #include "chromecast/media/cma/pipeline/av_pipeline_client.h"
  20 #include "chromecast/renderer/media/cma_message_filter_proxy.h"
  21 #include "chromecast/renderer/media/media_channel_proxy.h"
  22 #include "media/base/bind_to_current_loop.h"
  23 #include "media/base/pipeline_status.h"
  24
  25 namespace chromecast {
  26 namespace media {
  27
  28 namespace {
  29
  30 void IgnoreResult() {
  31 }
  32
  33 }  // namespace
  34
  35 // AudioPipelineProxyInternal -
  36 // This class is not thread safe and should run on the same thread
  37 // as the media channel proxy.
  38 class AudioPipelineProxyInternal {
  39  public:
  40   typedef base::Callback<void(scoped_ptr<base::SharedMemory>)> SharedMemCB;
  41
  42   static void Release(scoped_ptr<AudioPipelineProxyInternal> proxy);
  43
  44   explicit AudioPipelineProxyInternal(
  45       scoped_refptr<MediaChannelProxy> media_channel_proxy);
  46   virtual ~AudioPipelineProxyInternal();
  47
  48   // Notify the other side (browser process) of some activity on the audio pipe.
  49   // TODO(erickung): either send an IPC message or write a byte on the
  50   // SyncSocket.
  51   void NotifyPipeWrite();
  52
  53   // These functions are almost a one to one correspondence with AudioPipeline
  54   // but this is an internal class and there is no reason to derive from
  55   // AudioPipeline.
  56   void SetClient(const base::Closure& pipe_read_cb,
  57                  const AvPipelineClient& client);
  58   void CreateAvPipe(const SharedMemCB& shared_mem_cb);
  59   void Initialize(const ::media::AudioDecoderConfig& config,
  60                   const ::media::PipelineStatusCB& status_cb);
  61   void SetVolume(float volume);
  62
  63  private:
  64   void Shutdown();
  65
  66   // Callbacks for CmaMessageFilterHost::AudioDelegate.
  67   void OnAvPipeCreated(bool status,
  68                        base::SharedMemoryHandle shared_mem_handle,
  69                        base::FileDescriptor socket);
  70   void OnStateChanged(::media::PipelineStatus status);
  71
  72   base::ThreadChecker thread_checker_;
  73
  74   scoped_refptr<MediaChannelProxy> media_channel_proxy_;
  75
  76   // Store the callback for a pending state transition.
  77   ::media::PipelineStatusCB status_cb_;
  78
  79   SharedMemCB shared_mem_cb_;
  80
  81   DISALLOW_COPY_AND_ASSIGN(AudioPipelineProxyInternal);
  82 };
  83
  84 // static
  85 void AudioPipelineProxyInternal::Release(
  86     scoped_ptr<AudioPipelineProxyInternal> proxy) {
  87   proxy->Shutdown();
  88 }
  89
  90 AudioPipelineProxyInternal::AudioPipelineProxyInternal(
  91     scoped_refptr<MediaChannelProxy> media_channel_proxy)
  92     : media_channel_proxy_(media_channel_proxy) {
  93   DCHECK(media_channel_proxy.get());
  94
  95   // Creation can be done on a different thread.
  96   thread_checker_.DetachFromThread();
  97 }
  98
  99 AudioPipelineProxyInternal::~AudioPipelineProxyInternal() {
 100 }
 101
 102 void AudioPipelineProxyInternal::Shutdown() {
 103   DCHECK(thread_checker_.CalledOnValidThread());
 104
 105   // Remove any callback on AudioPipelineProxyInternal.
 106   media_channel_proxy_->SetAudioDelegate(
 107       CmaMessageFilterProxy::AudioDelegate());
 108 }
 109
 110 void AudioPipelineProxyInternal::NotifyPipeWrite() {
 111   DCHECK(thread_checker_.CalledOnValidThread());
 112
 113   // TODO(erickung): An alternative way would be to use a dedicated socket for
 114   // this event.
 115   bool success = media_channel_proxy_->Send(scoped_ptr<IPC::Message>(
 116       new CmaHostMsg_NotifyPipeWrite(
 117           media_channel_proxy_->GetId(), kAudioTrackId)));
 118   VLOG_IF(4, !success) << "Sending msg failed";
 119 }
 120
 121 void AudioPipelineProxyInternal::SetClient(
 122     const base::Closure& pipe_read_cb,
 123     const AvPipelineClient& client) {
 124   DCHECK(thread_checker_.CalledOnValidThread());
 125
 126   CmaMessageFilterProxy::AudioDelegate delegate;
 127   delegate.av_pipe_cb =
 128       base::Bind(&AudioPipelineProxyInternal::OnAvPipeCreated,
 129                  base::Unretained(this));
 130   delegate.state_changed_cb =
 131       base::Bind(&AudioPipelineProxyInternal::OnStateChanged,
 132                  base::Unretained(this));
 133   delegate.pipe_read_cb = pipe_read_cb;
 134   delegate.client = client;
 135   bool success = media_channel_proxy_->SetAudioDelegate(delegate);
 136   CHECK(success);
 137 }
 138
 139 void AudioPipelineProxyInternal::CreateAvPipe(
 140     const SharedMemCB& shared_mem_cb) {
 141   DCHECK(thread_checker_.CalledOnValidThread());
 142   DCHECK(shared_mem_cb_.is_null());
 143   bool success = media_channel_proxy_->Send(scoped_ptr<IPC::Message>(
 144       new CmaHostMsg_CreateAvPipe(
 145           media_channel_proxy_->GetId(), kAudioTrackId, kAppAudioBufferSize)));
 146   if (!success) {
 147     shared_mem_cb.Run(scoped_ptr<base::SharedMemory>());
 148     return;
 149   }
 150   shared_mem_cb_ = shared_mem_cb;
 151 }
 152
 153 void AudioPipelineProxyInternal::OnAvPipeCreated(
 154     bool success,
 155     base::SharedMemoryHandle shared_mem_handle,
 156     base::FileDescriptor socket) {
 157   DCHECK(thread_checker_.CalledOnValidThread());
 158   DCHECK(!shared_mem_cb_.is_null());
 159   if (!success) {
 160     shared_mem_cb_.Run(scoped_ptr<base::SharedMemory>());
 161     return;
 162   }
 163
 164   CHECK(base::SharedMemory::IsHandleValid(shared_mem_handle));
 165   shared_mem_cb_.Run(scoped_ptr<base::SharedMemory>(
 166       new base::SharedMemory(shared_mem_handle, false)));
 167 }
 168
 169 void AudioPipelineProxyInternal::Initialize(
 170     const ::media::AudioDecoderConfig& config,
 171     const ::media::PipelineStatusCB& status_cb) {
 172   DCHECK(thread_checker_.CalledOnValidThread());
 173   bool success = media_channel_proxy_->Send(scoped_ptr<IPC::Message>(
 174       new CmaHostMsg_AudioInitialize(
 175           media_channel_proxy_->GetId(), kAudioTrackId, config)));
 176   if (!success) {
 177     status_cb.Run( ::media::PIPELINE_ERROR_INITIALIZATION_FAILED);
 178     return;
 179   }
 180   DCHECK(status_cb_.is_null());
 181   status_cb_ = status_cb;
 182 }
 183
 184 void AudioPipelineProxyInternal::SetVolume(float volume) {
 185   DCHECK(thread_checker_.CalledOnValidThread());
 186   media_channel_proxy_->Send(scoped_ptr<IPC::Message>(
 187       new CmaHostMsg_SetVolume(media_channel_proxy_->GetId(),
 188                                kAudioTrackId, volume)));
 189 }
 190
 191 void AudioPipelineProxyInternal::OnStateChanged(
 192     ::media::PipelineStatus status) {
 193   DCHECK(thread_checker_.CalledOnValidThread());
 194   DCHECK(!status_cb_.is_null());
 195   base::ResetAndReturn(&status_cb_).Run(status);
 196 }
 197
 198 // A macro runs current member function on |io_task_runner_| thread.
 199 #define FORWARD_ON_IO_THREAD(param_fn, ...)                        \
 200   io_task_runner_->PostTask(                                       \
 201       FROM_HERE, base::Bind(&AudioPipelineProxyInternal::param_fn, \
 202                             base::Unretained(proxy_.get()), ##__VA_ARGS__))
 203
 204 AudioPipelineProxy::AudioPipelineProxy(
 205     scoped_refptr<base::SingleThreadTaskRunner> io_task_runner,
 206     scoped_refptr<MediaChannelProxy> media_channel_proxy)
 207     : io_task_runner_(io_task_runner),
 208       proxy_(new AudioPipelineProxyInternal(media_channel_proxy)),
 209       audio_streamer_(new AvStreamerProxy()),
 210       weak_factory_(this) {
 211   DCHECK(io_task_runner_.get());
 212   weak_this_ = weak_factory_.GetWeakPtr();
 213   thread_checker_.DetachFromThread();
 214 }
 215
 216 AudioPipelineProxy::~AudioPipelineProxy() {
 217   DCHECK(thread_checker_.CalledOnValidThread());
 218   // Release the underlying object on the right thread.
 219   io_task_runner_->PostTask(
 220       FROM_HERE,
 221       base::Bind(&AudioPipelineProxyInternal::Release, base::Passed(&proxy_)));
 222 }
 223
 224 void AudioPipelineProxy::SetClient(
 225     const AvPipelineClient& client) {
 226   DCHECK(thread_checker_.CalledOnValidThread());
 227   base::Closure pipe_read_cb = ::media::BindToCurrentLoop(
 228       base::Bind(&AudioPipelineProxy::OnPipeRead, weak_this_));
 229   FORWARD_ON_IO_THREAD(SetClient, pipe_read_cb, client);
 230 }
 231
 232 void AudioPipelineProxy::Initialize(
 233     const ::media::AudioDecoderConfig& config,
 234     scoped_ptr<CodedFrameProvider> frame_provider,
 235     const ::media::PipelineStatusCB& status_cb) {
 236   CMALOG(kLogControl) << "AudioPipelineProxy::Initialize";
 237   DCHECK(thread_checker_.CalledOnValidThread());
 238   audio_streamer_->SetCodedFrameProvider(frame_provider.Pass());
 239
 240   AudioPipelineProxyInternal::SharedMemCB shared_mem_cb =
 241       ::media::BindToCurrentLoop(base::Bind(
 242           &AudioPipelineProxy::OnAvPipeCreated, weak_this_,
 243           config, status_cb));
 244   FORWARD_ON_IO_THREAD(CreateAvPipe, shared_mem_cb);
 245 }
 246
 247 void AudioPipelineProxy::OnAvPipeCreated(
 248     const ::media::AudioDecoderConfig& config,
 249     const ::media::PipelineStatusCB& status_cb,
 250     scoped_ptr<base::SharedMemory> shared_memory) {
 251   CMALOG(kLogControl) << "AudioPipelineProxy::OnAvPipeCreated";
 252   DCHECK(thread_checker_.CalledOnValidThread());
 253   if (!shared_memory ||
 254       !shared_memory->Map(kAppAudioBufferSize)) {
 255     status_cb.Run(::media::PIPELINE_ERROR_INITIALIZATION_FAILED);
 256     return;
 257   }
 258   CHECK(shared_memory->memory());
 259
 260   scoped_ptr<MediaMemoryChunk> shared_memory_chunk(
 261       new SharedMemoryChunk(shared_memory.Pass(), kAppAudioBufferSize));
 262   scoped_ptr<MediaMessageFifo> audio_pipe(
 263       new MediaMessageFifo(shared_memory_chunk.Pass(), false));
 264   audio_pipe->ObserveWriteActivity(
 265       base::Bind(&AudioPipelineProxy::OnPipeWrite, weak_this_));
 266
 267   audio_streamer_->SetMediaMessageFifo(audio_pipe.Pass());
 268
 269   // Now proceed to the decoder/renderer initialization.
 270   FORWARD_ON_IO_THREAD(Initialize, config, status_cb);
 271 }
 272
 273 void AudioPipelineProxy::StartFeeding() {
 274   DCHECK(thread_checker_.CalledOnValidThread());
 275   DCHECK(audio_streamer_);
 276   audio_streamer_->Start();
 277 }
 278
 279 void AudioPipelineProxy::Flush(const base::Closure& done_cb) {
 280   DCHECK(thread_checker_.CalledOnValidThread());
 281   DCHECK(audio_streamer_);
 282   audio_streamer_->StopAndFlush(done_cb);
 283 }
 284
 285 void AudioPipelineProxy::Stop() {
 286   DCHECK(thread_checker_.CalledOnValidThread());
 287   if (!audio_streamer_)
 288     return;
 289   audio_streamer_->StopAndFlush(base::Bind(&IgnoreResult));
 290 }
 291
 292 void AudioPipelineProxy::SetVolume(float volume) {
 293   DCHECK(thread_checker_.CalledOnValidThread());
 294   FORWARD_ON_IO_THREAD(SetVolume, volume);
 295 }
 296
 297 void AudioPipelineProxy::OnPipeWrite() {
 298   DCHECK(thread_checker_.CalledOnValidThread());
 299   FORWARD_ON_IO_THREAD(NotifyPipeWrite);
 300 }
 301
 302 void AudioPipelineProxy::OnPipeRead() {
 303   DCHECK(thread_checker_.CalledOnValidThread());
 304   if (audio_streamer_)
 305     audio_streamer_->OnFifoReadEvent();
 306 }
 307
 308 }  // namespace cma
 309 }  // namespace chromecast