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
 199 // A macro runs current member function on |io_message_loop_proxy_| thread.
 200 #define FORWARD_ON_IO_THREAD(param_fn, ...) \
 201   io_message_loop_proxy_->PostTask( \
 202       FROM_HERE, \
 203       base::Bind(&AudioPipelineProxyInternal::param_fn, \
 204                  base::Unretained(proxy_.get()), ##__VA_ARGS__))
 205
 206 AudioPipelineProxy::AudioPipelineProxy(
 207     scoped_refptr<base::MessageLoopProxy> io_message_loop_proxy,
 208     scoped_refptr<MediaChannelProxy> media_channel_proxy)
 209     : io_message_loop_proxy_(io_message_loop_proxy),
 210       proxy_(new AudioPipelineProxyInternal(media_channel_proxy)),
 211       audio_streamer_(new AvStreamerProxy()),
 212       weak_factory_(this) {
 213   DCHECK(io_message_loop_proxy_.get());
 214   weak_this_ = weak_factory_.GetWeakPtr();
 215   thread_checker_.DetachFromThread();
 216 }
 217
 218 AudioPipelineProxy::~AudioPipelineProxy() {
 219   DCHECK(thread_checker_.CalledOnValidThread());
 220   // Release the underlying object on the right thread.
 221   io_message_loop_proxy_->PostTask(
 222       FROM_HERE,
 223       base::Bind(&AudioPipelineProxyInternal::Release, base::Passed(&proxy_)));
 224 }
 225
 226 void AudioPipelineProxy::SetClient(
 227     const AvPipelineClient& client) {
 228   DCHECK(thread_checker_.CalledOnValidThread());
 229   base::Closure pipe_read_cb = ::media::BindToCurrentLoop(
 230       base::Bind(&AudioPipelineProxy::OnPipeRead, weak_this_));
 231   FORWARD_ON_IO_THREAD(SetClient, pipe_read_cb, client);
 232 }
 233
 234 void AudioPipelineProxy::Initialize(
 235     const ::media::AudioDecoderConfig& config,
 236     scoped_ptr<CodedFrameProvider> frame_provider,
 237     const ::media::PipelineStatusCB& status_cb) {
 238   CMALOG(kLogControl) << "AudioPipelineProxy::Initialize";
 239   DCHECK(thread_checker_.CalledOnValidThread());
 240   audio_streamer_->SetCodedFrameProvider(frame_provider.Pass());
 241
 242   AudioPipelineProxyInternal::SharedMemCB shared_mem_cb =
 243       ::media::BindToCurrentLoop(base::Bind(
 244           &AudioPipelineProxy::OnAvPipeCreated, weak_this_,
 245           config, status_cb));
 246   FORWARD_ON_IO_THREAD(CreateAvPipe, shared_mem_cb);
 247 }
 248
 249 void AudioPipelineProxy::OnAvPipeCreated(
 250     const ::media::AudioDecoderConfig& config,
 251     const ::media::PipelineStatusCB& status_cb,
 252     scoped_ptr<base::SharedMemory> shared_memory) {
 253   CMALOG(kLogControl) << "AudioPipelineProxy::OnAvPipeCreated";
 254   DCHECK(thread_checker_.CalledOnValidThread());
 255   if (!shared_memory ||
 256       !shared_memory->Map(kAppAudioBufferSize)) {
 257     status_cb.Run(::media::PIPELINE_ERROR_INITIALIZATION_FAILED);
 258     return;
 259   }
 260   CHECK(shared_memory->memory());
 261
 262   scoped_ptr<MediaMemoryChunk> shared_memory_chunk(
 263       new SharedMemoryChunk(shared_memory.Pass(), kAppAudioBufferSize));
 264   scoped_ptr<MediaMessageFifo> audio_pipe(
 265       new MediaMessageFifo(shared_memory_chunk.Pass(), false));
 266   audio_pipe->ObserveWriteActivity(
 267       base::Bind(&AudioPipelineProxy::OnPipeWrite, weak_this_));
 268
 269   audio_streamer_->SetMediaMessageFifo(audio_pipe.Pass());
 270
 271   // Now proceed to the decoder/renderer initialization.
 272   FORWARD_ON_IO_THREAD(Initialize, config, status_cb);
 273 }
 274
 275 void AudioPipelineProxy::StartFeeding() {
 276   DCHECK(thread_checker_.CalledOnValidThread());
 277   DCHECK(audio_streamer_);
 278   audio_streamer_->Start();
 279 }
 280
 281 void AudioPipelineProxy::Flush(const base::Closure& done_cb) {
 282   DCHECK(thread_checker_.CalledOnValidThread());
 283   DCHECK(audio_streamer_);
 284   audio_streamer_->StopAndFlush(done_cb);
 285 }
 286
 287 void AudioPipelineProxy::Stop() {
 288   DCHECK(thread_checker_.CalledOnValidThread());
 289   if (!audio_streamer_)
 290     return;
 291   audio_streamer_->StopAndFlush(base::Bind(&IgnoreResult));
 292 }
 293
 294 void AudioPipelineProxy::SetVolume(float volume) {
 295   DCHECK(thread_checker_.CalledOnValidThread());
 296   FORWARD_ON_IO_THREAD(SetVolume, volume);
 297 }
 298
 299 void AudioPipelineProxy::OnPipeWrite() {
 300   DCHECK(thread_checker_.CalledOnValidThread());
 301   FORWARD_ON_IO_THREAD(NotifyPipeWrite);
 302 }
 303
 304 void AudioPipelineProxy::OnPipeRead() {
 305   DCHECK(thread_checker_.CalledOnValidThread());
 306   if (audio_streamer_)
 307     audio_streamer_->OnFifoReadEvent();
 308 }
 309
 310 }  // namespace cma
 311 }  // namespace chromecast