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