Audio/video data streaming over shared memory.

chromecast/media/cma/ipc_streamer/av_streamer_proxy.cc

Index: chromecast/media/cma/ipc_streamer/av_streamer_proxy.cc
diff --git a/chromecast/media/cma/ipc_streamer/av_streamer_proxy.cc b/chromecast/media/cma/ipc_streamer/av_streamer_proxy.cc
new file mode 100644
index 0000000000000000000000000000000000000000..de8dd54f6b3ae673203aedd885cfef60c84d2c8a
--- /dev/null
+++ b/chromecast/media/cma/ipc_streamer/av_streamer_proxy.cc
@@ -0,0 +1,200 @@
+// 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/media/cma/ipc_streamer/av_streamer_proxy.h"
+
+#include "base/bind.h"
+#include "base/location.h"
+#include "base/message_loop/message_loop_proxy.h"
+#include "chromecast/media/cma/base/coded_frame_provider.h"
+#include "chromecast/media/cma/base/decoder_buffer_base.h"
+#include "chromecast/media/cma/ipc/media_memory_chunk.h"
+#include "chromecast/media/cma/ipc/media_message.h"
+#include "chromecast/media/cma/ipc/media_message_fifo.h"
+#include "chromecast/media/cma/ipc/media_message_type.h"
+#include "chromecast/media/cma/ipc_streamer/audio_decoder_config_marshaller.h"
+#include "chromecast/media/cma/ipc_streamer/decoder_buffer_base_marshaller.h"
+#include "chromecast/media/cma/ipc_streamer/video_decoder_config_marshaller.h"
+
+namespace chromecast {
+namespace media {
+
+AvStreamerProxy::AvStreamerProxy()
+    : is_running_(false),
+      pending_read_(false),
+      pending_av_data_(false),
+      weak_factory_(this),
+      weak_this_(weak_factory_.GetWeakPtr()) {
+  thread_checker_.DetachFromThread();
+}
+
+AvStreamerProxy::~AvStreamerProxy() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+}
+
+void AvStreamerProxy::SetCodedFrameProvider(
+    scoped_ptr<CodedFrameProvider> frame_provider) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(!frame_provider_);
+  frame_provider_.reset(frame_provider.release());
+}
+
+void AvStreamerProxy::SetMediaMessageFifo(
+    scoped_ptr<MediaMessageFifo> fifo) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(!fifo_);
+  fifo_.reset(fifo.release());
+}
+
+void AvStreamerProxy::Start() {
+  DCHECK(!is_running_);
+
+  is_running_ = true;
+  RequestBufferIfNeeded();
+}
+
+void AvStreamerProxy::StopAndFlush(const base::Closure& done_cb) {
+  is_running_ = false;
+
+  pending_av_data_ = false;
+  pending_audio_config_ = ::media::AudioDecoderConfig();
+  pending_video_config_ = ::media::VideoDecoderConfig();
+  pending_buffer_ = scoped_refptr<DecoderBufferBase>();
+
+  pending_read_ = false;
+  frame_provider_->Flush(done_cb);
+}
+
+void AvStreamerProxy::OnFifoReadEvent() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  // Some enough space might have been released
+  // to accommodate the pending data.
+  if (pending_av_data_)
+    ProcessPendingData();
+}
+
+void AvStreamerProxy::RequestBufferIfNeeded() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  if (!is_running_ || pending_read_ || pending_av_data_)
+    return;
+
+  // |frame_provider_| is assumed to run on the same message loop.
+  // Add a BindToCurrentLoop if that's not the case in the future.
+  pending_read_ = true;
+  frame_provider_->Read(base::Bind(&AvStreamerProxy::OnNewBuffer, weak_this_));
+}
+
+void AvStreamerProxy::OnNewBuffer(
+    const scoped_refptr<DecoderBufferBase>& buffer,
+    const ::media::AudioDecoderConfig& audio_config,
+    const ::media::VideoDecoderConfig& video_config) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  pending_read_ = false;
+
+  if (buffer->end_of_stream())
+    is_running_ = false;
+
+  DCHECK(!pending_av_data_);
+  pending_av_data_ = true;
+
+  pending_buffer_ = buffer;
+  pending_audio_config_ = audio_config;
+  pending_video_config_ = video_config;
+
+  ProcessPendingData();
+}
+
+void AvStreamerProxy::ProcessPendingData() {
+  if (pending_audio_config_.IsValidConfig()) {
+    if (!SendAudioDecoderConfig(pending_audio_config_))
+      return;
+    pending_audio_config_ = ::media::AudioDecoderConfig();
+  }
+
+  if (pending_video_config_.IsValidConfig()) {
+    if (!SendVideoDecoderConfig(pending_video_config_))
+      return;
+    pending_video_config_ = ::media::VideoDecoderConfig();
+  }
+
+  if (pending_buffer_.get()) {
+    if (!SendBuffer(pending_buffer_))
+      return;
+    pending_buffer_ = scoped_refptr<DecoderBufferBase>();
+  }
+
+  pending_av_data_ = false;
+  base::MessageLoopProxy::current()->PostTask(
+      FROM_HERE,
+      base::Bind(&AvStreamerProxy::RequestBufferIfNeeded, weak_this_));
+}
+
+bool AvStreamerProxy::SendAudioDecoderConfig(
+    const ::media::AudioDecoderConfig& config) {
+  // Create a dummy message to calculate first the message size.
+  scoped_ptr<MediaMessage> dummy_msg(
+      MediaMessage::CreateDummyMessage(AudioConfigMediaMsg));
+  AudioDecoderConfigMarshaller::Write(config, dummy_msg.get());
+
+  // Create the real message and write the actual content.
+  scoped_ptr<MediaMessage> msg(
+      MediaMessage::CreateMessage(
+          AudioConfigMediaMsg,
+          base::Bind(&MediaMessageFifo::ReserveMemory,
+                     base::Unretained(fifo_.get())),
+          dummy_msg->content_size()));
+  if (!msg)
+    return false;
+
+  AudioDecoderConfigMarshaller::Write(config, msg.get());
+  return true;
+}
+
+bool AvStreamerProxy::SendVideoDecoderConfig(
+    const ::media::VideoDecoderConfig& config) {
+  // Create a dummy message to calculate first the message size.
+  scoped_ptr<MediaMessage> dummy_msg(
+      MediaMessage::CreateDummyMessage(VideoConfigMediaMsg));
+  VideoDecoderConfigMarshaller::Write(config, dummy_msg.get());
+
+  // Create the real message and write the actual content.
+  scoped_ptr<MediaMessage> msg(
+      MediaMessage::CreateMessage(
+          VideoConfigMediaMsg,
+          base::Bind(&MediaMessageFifo::ReserveMemory,
+                     base::Unretained(fifo_.get())),
+          dummy_msg->content_size()));
+  if (!msg)
+    return false;
+
+  VideoDecoderConfigMarshaller::Write(config, msg.get());
+  return true;
+}
+
+bool AvStreamerProxy::SendBuffer(
+    const scoped_refptr<DecoderBufferBase>& buffer) {
+  // Create a dummy message to calculate first the message size.
+  scoped_ptr<MediaMessage> dummy_msg(
+      MediaMessage::CreateDummyMessage(FrameMediaMsg));
+  DecoderBufferBaseMarshaller::Write(buffer, dummy_msg.get());
+
+  // Create the real message and write the actual content.
+  scoped_ptr<MediaMessage> msg(
+      MediaMessage::CreateMessage(
+          FrameMediaMsg,
+          base::Bind(&MediaMessageFifo::ReserveMemory,
+                     base::Unretained(fifo_.get())),
+          dummy_msg->content_size()));
+  if (!msg)
+    return false;
+
+  DecoderBufferBaseMarshaller::Write(buffer, msg.get());
+  return true;
+}
+
+}  // namespace media
+}  // namespace chromecast