[Chromecast] Add CmaMediaRendererFactory and IPC proxy components

chromecast/renderer/media/audio_pipeline_proxy.cc

+// Copyright 2014 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/renderer/media/audio_pipeline_proxy.h"
+
+#include "base/bind.h"
+#include "base/callback_helpers.h"
+#include "base/memory/shared_memory.h"
+#include "base/message_loop/message_loop.h"
+#include "base/threading/thread_checker.h"
+#include "chromecast/common/media/cma_ipc_common.h"
+#include "chromecast/common/media/cma_messages.h"
+#include "chromecast/common/media/shared_memory_chunk.h"
+#include "chromecast/media/cma/base/buffering_defs.h"
+#include "chromecast/media/cma/base/cma_logging.h"
+#include "chromecast/media/cma/base/coded_frame_provider.h"
+#include "chromecast/media/cma/ipc/media_message_fifo.h"
+#include "chromecast/media/cma/ipc_streamer/av_streamer_proxy.h"
+#include "chromecast/media/cma/pipeline/av_pipeline_client.h"
+#include "chromecast/renderer/media/cma_message_filter_proxy.h"
+#include "chromecast/renderer/media/media_channel_proxy.h"
+#include "media/base/bind_to_current_loop.h"
+#include "media/base/pipeline_status.h"
+
+namespace chromecast {
+namespace cma {
+
+namespace {
+
+void Noop() {

[gunsch, 2014/12/20 22:41:33]

probably rename to something like "IgnoreResult"

[erickung1, 2014/12/21 11:10:46]

Done.

+}
+
+}  // namespace
+
+// AudioPipelineProxyInternal -
+// This class is not thread safe and should run on the same thread
+// as the media channel proxy.
+class AudioPipelineProxyInternal {
+ public:
+  typedef base::Callback<void(scoped_ptr<base::SharedMemory>)> SharedMemCB;
+
+  static void Release(scoped_ptr<AudioPipelineProxyInternal> proxy);
+
+  explicit AudioPipelineProxyInternal(
+      scoped_refptr<MediaChannelProxy> media_channel_proxy);
+  virtual ~AudioPipelineProxyInternal();
+
+  // Notify the other side (browser process) of some activity on the audio pipe.
+  // TODO(damienv): either send an IPC message or write a byte on the

[gunsch, 2014/12/20 22:41:33]

we probably shouldn't be adding any TODO(damienv) into our code :) please check
the CL for these and either resolve or remove if they're no longer relevant

[erickung1, 2014/12/21 11:10:46]

Change to my name first. I checked the code and it's a ACK message sending back
to browser side once the work is done. So far it's working fine without sending
other byte. I will probably leave as is for now.

+  // SyncSocket.
+  void NotifyPipeWrite();
+
+  // These functions are almost a one to one correspondence with AudioPipeline
+  // but this is an internal class and there is no reason to derive from
+  // AudioPipeline.
+  void SetClient(const base::Closure& pipe_read_cb,
+                 const chromecast::media::AvPipelineClient& client);
+  void CreateAvPipe(const SharedMemCB& shared_mem_cb);
+  void Initialize(const ::media::AudioDecoderConfig& config,
+                  const ::media::PipelineStatusCB& status_cb);
+  void SetVolume(float volume);
+
+ private:
+  void Shutdown();
+
+  // Callbacks for CmaMessageFilterHost::AudioDelegate.
+  void OnAvPipeCreated(bool status,
+                       base::SharedMemoryHandle shared_mem_handle,
+                       base::FileDescriptor socket);
+  void OnStateChanged(::media::PipelineStatus status);
+
+  base::ThreadChecker thread_checker_;
+
+  scoped_refptr<MediaChannelProxy> media_channel_proxy_;
+
+  // Store the callback for a pending state transition.
+  ::media::PipelineStatusCB status_cb_;
+
+  SharedMemCB shared_mem_cb_;
+
+  DISALLOW_COPY_AND_ASSIGN(AudioPipelineProxyInternal);

[gunsch, 2014/12/20 22:41:33]

include base/macros.h

[erickung1, 2014/12/21 11:10:46]

probably don't need it since audio_pipeline_proxy.h includes base/macros.h
already?

+};
+
+// static
+void AudioPipelineProxyInternal::Release(
+    scoped_ptr<AudioPipelineProxyInternal> proxy) {
+  proxy->Shutdown();
+}
+
+AudioPipelineProxyInternal::AudioPipelineProxyInternal(
+    scoped_refptr<MediaChannelProxy> media_channel_proxy)
+    : media_channel_proxy_(media_channel_proxy) {
+  DCHECK(media_channel_proxy.get());
+
+  // Creation can be done on a different thread.
+  thread_checker_.DetachFromThread();
+}
+
+AudioPipelineProxyInternal::~AudioPipelineProxyInternal() {
+}
+
+void AudioPipelineProxyInternal::Shutdown() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  // Remove any callback on AudioPipelineProxyInternal.
+  media_channel_proxy_->SetAudioDelegate(
+      CmaMessageFilterProxy::AudioDelegate());
+}
+
+void AudioPipelineProxyInternal::NotifyPipeWrite() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  // TODO(damienv): An alternative way would be to use a dedicated socket for
+  // this event.
+  bool success = media_channel_proxy_->Send(scoped_ptr<IPC::Message>(
+      new CmaHostMsg_NotifyPipeWrite(
+          media_channel_proxy_->GetId(), chromecast::media::kAudioTrackId)));
+  VLOG_IF(4, !success) << "Sending msg failed";
+}
+
+void AudioPipelineProxyInternal::SetClient(
+    const base::Closure& pipe_read_cb,
+    const chromecast::media::AvPipelineClient& client) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  CmaMessageFilterProxy::AudioDelegate delegate;
+  delegate.av_pipe_cb =
+      base::Bind(&AudioPipelineProxyInternal::OnAvPipeCreated,
+                 base::Unretained(this));
+  delegate.state_changed_cb =
+      base::Bind(&AudioPipelineProxyInternal::OnStateChanged,
+                 base::Unretained(this));
+  delegate.pipe_read_cb = pipe_read_cb;
+  delegate.client = client;
+  bool success = media_channel_proxy_->SetAudioDelegate(delegate);
+  CHECK(success);
+}
+
+void AudioPipelineProxyInternal::CreateAvPipe(
+    const SharedMemCB& shared_mem_cb) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(shared_mem_cb_.is_null());
+  bool success = media_channel_proxy_->Send(scoped_ptr<IPC::Message>(
+      new CmaHostMsg_CreateAvPipe(
+          media_channel_proxy_->GetId(), chromecast::media::kAudioTrackId,
+          chromecast::media::kAppAudioBufferSize)));
+  if (!success) {
+    shared_mem_cb.Run(scoped_ptr<base::SharedMemory>());
+    return;
+  }
+  shared_mem_cb_ = shared_mem_cb;
+}
+
+void AudioPipelineProxyInternal::OnAvPipeCreated(
+    bool success,
+    base::SharedMemoryHandle shared_mem_handle,
+    base::FileDescriptor socket) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(!shared_mem_cb_.is_null());
+  if (!success) {
+    shared_mem_cb_.Run(scoped_ptr<base::SharedMemory>());
+    return;
+  }
+
+  CHECK(base::SharedMemory::IsHandleValid(shared_mem_handle));
+  shared_mem_cb_.Run(scoped_ptr<base::SharedMemory>(
+      new base::SharedMemory(shared_mem_handle, false)));
+}
+
+#define DEFINE_STATE_TRANSITION_1(param_fn, param_msg, param_type1) \

[gunsch, 2014/12/20 22:41:33]

This #define is only used once; this would be clearer if you just unrolled it.

[erickung1, 2014/12/21 11:10:46]

Done.

+void AudioPipelineProxyInternal::param_fn( \
+    const param_type1& arg1, \
+    const ::media::PipelineStatusCB& status_cb) { \
+  DCHECK(thread_checker_.CalledOnValidThread()); \
+  bool success = media_channel_proxy_->Send(scoped_ptr<IPC::Message>( \
+      new param_msg( \
+          media_channel_proxy_->GetId(), chromecast::media::kAudioTrackId, \
+          arg1))); \
+  if (!success) { \
+    status_cb.Run( \
+        ::media::PIPELINE_ERROR_INITIALIZATION_FAILED); \
+    return; \
+  } \
+  DCHECK(status_cb_.is_null()); \
+  status_cb_ = status_cb; \
+}
+
+DEFINE_STATE_TRANSITION_1(Initialize, CmaHostMsg_AudioInitialize,
+                          ::media::AudioDecoderConfig)
+
+void AudioPipelineProxyInternal::SetVolume(float volume) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  media_channel_proxy_->Send(scoped_ptr<IPC::Message>(
+      new CmaHostMsg_SetVolume(
+          media_channel_proxy_->GetId(),
+          chromecast::media::kAudioTrackId, volume)));
+}
+
+void AudioPipelineProxyInternal::OnStateChanged(
+    ::media::PipelineStatus status) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(!status_cb_.is_null());
+  base::ResetAndReturn(&status_cb_).Run(status);
+}
+
+
+#define FORWARD_ON_IO_0(param_fn) \

[gunsch, 2014/12/20 22:41:33]

This #define is also only used once, just unroll it

alternately, can you combine all the #defines with ##__VA_ARGS__ ?

[erickung1, 2014/12/21 11:10:46]

Done.

+  io_message_loop_proxy_->PostTask( \
+      FROM_HERE, \
+      base::Bind(&AudioPipelineProxyInternal::param_fn, \
+                 base::Unretained(proxy_.get())))
+#define FORWARD_ON_IO_1(param_fn, param_arg1) \
+  io_message_loop_proxy_->PostTask( \
+      FROM_HERE, \
+      base::Bind(&AudioPipelineProxyInternal::param_fn, \
+                 base::Unretained(proxy_.get()), (param_arg1)))
+#define FORWARD_ON_IO_2(param_fn, param_arg1, param_arg2) \
+  io_message_loop_proxy_->PostTask( \
+      FROM_HERE, \
+      base::Bind(&AudioPipelineProxyInternal::param_fn, \
+                 base::Unretained(proxy_.get()), (param_arg1), (param_arg2)))
+
+AudioPipelineProxy::AudioPipelineProxy(
+    scoped_refptr<base::MessageLoopProxy> io_message_loop_proxy,
+    scoped_refptr<MediaChannelProxy> media_channel_proxy)
+  : io_message_loop_proxy_(io_message_loop_proxy),
+    proxy_(new AudioPipelineProxyInternal(media_channel_proxy)),
+    audio_streamer_(new chromecast::media::AvStreamerProxy()),
+    weak_factory_(this) {
+  DCHECK(io_message_loop_proxy_.get());
+  weak_this_ = weak_factory_.GetWeakPtr();
+  thread_checker_.DetachFromThread();
+}
+
+AudioPipelineProxy::~AudioPipelineProxy() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  // Release the underlying object on the right thread.
+  io_message_loop_proxy_->PostTask(
+      FROM_HERE,
+      base::Bind(&AudioPipelineProxyInternal::Release, base::Passed(&proxy_)));
+}
+
+void AudioPipelineProxy::SetClient(
+    const chromecast::media::AvPipelineClient& client) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  base::Closure pipe_read_cb = ::media::BindToCurrentLoop(
+      base::Bind(&AudioPipelineProxy::OnPipeRead, weak_this_));
+  FORWARD_ON_IO_2(SetClient, pipe_read_cb, client);
+}
+
+void AudioPipelineProxy::Initialize(
+    const ::media::AudioDecoderConfig& config,
+    scoped_ptr<chromecast::media::CodedFrameProvider> frame_provider,
+    const ::media::PipelineStatusCB& status_cb) {
+  CMALOG(chromecast::media::kLogControl) << "AudioPipelineProxy::Initialize";
+  DCHECK(thread_checker_.CalledOnValidThread());
+  audio_streamer_->SetCodedFrameProvider(frame_provider.Pass());
+
+  AudioPipelineProxyInternal::SharedMemCB shared_mem_cb =
+      ::media::BindToCurrentLoop(base::Bind(
+          &AudioPipelineProxy::OnAvPipeCreated, weak_this_,
+          config, status_cb));
+  FORWARD_ON_IO_1(CreateAvPipe, shared_mem_cb);
+}
+
+void AudioPipelineProxy::OnAvPipeCreated(
+    const ::media::AudioDecoderConfig& config,
+    const ::media::PipelineStatusCB& status_cb,
+    scoped_ptr<base::SharedMemory> shared_memory) {
+  CMALOG(chromecast::media::kLogControl)
+      << "AudioPipelineProxy::OnAvPipeCreated";
+  DCHECK(thread_checker_.CalledOnValidThread());
+  if (!shared_memory ||
+      !shared_memory->Map(chromecast::media::kAppAudioBufferSize)) {
+    status_cb.Run(::media::PIPELINE_ERROR_INITIALIZATION_FAILED);
+    return;
+  }
+  CHECK(shared_memory->memory());
+
+  scoped_ptr<chromecast::media::MediaMemoryChunk> shared_memory_chunk(
+      new chromecast::media::SharedMemoryChunk(
+          shared_memory.Pass(), chromecast::media::kAppAudioBufferSize));
+  scoped_ptr<chromecast::media::MediaMessageFifo> audio_pipe(
+      new chromecast::media::MediaMessageFifo(
+          shared_memory_chunk.Pass(), false));
+  audio_pipe->ObserveWriteActivity(
+      base::Bind(&AudioPipelineProxy::OnPipeWrite, weak_this_));
+
+  audio_streamer_->SetMediaMessageFifo(audio_pipe.Pass());
+
+  // Now proceed to the decoder/renderer initialization.
+  FORWARD_ON_IO_2(Initialize, config, status_cb);
+}
+
+void AudioPipelineProxy::StartFeeding() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(audio_streamer_);
+  audio_streamer_->Start();
+}
+
+void AudioPipelineProxy::Flush(const base::Closure& done_cb) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(audio_streamer_);
+  audio_streamer_->StopAndFlush(done_cb);
+}
+
+void AudioPipelineProxy::Stop() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  if (!audio_streamer_)
+    return;
+  audio_streamer_->StopAndFlush(base::Bind(&Noop));
+}
+
+void AudioPipelineProxy::SetVolume(float volume) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  FORWARD_ON_IO_1(SetVolume, volume);
+}
+
+void AudioPipelineProxy::OnPipeWrite() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  FORWARD_ON_IO_0(NotifyPipeWrite);
+}
+
+void AudioPipelineProxy::OnPipeRead() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  if (audio_streamer_)
+    audio_streamer_->OnFifoReadEvent();
+}
+
+}  // namespace cma
+}  // namespace chromecast
+