Use data pipe for encoded buffers in MojoAudioDecoder

Create data pipe, connect the client and the service and
serialize/deserialize DecoderBuffers for this pipe.

The propagation of decoded AudioBuffers is not implemented.

BUG=542910

Review URL: https://codereview.chromium.org/1823443002

Cr-Commit-Position: refs/heads/master@{#382438}

5 files changed, 95 insertions, 7 deletions

diff --git a/media/mojo/interfaces/audio_decoder.mojom b/media/mojo/interfaces/audio_decoder.mojom
index bb4f059..70a2755 100644
--- a/media/mojo/interfaces/audio_decoder.mojom
+++ b/media/mojo/interfaces/audio_decoder.mojom
@@ -21,6 +21,9 @@ interface AudioDecoder {
   Initialize(AudioDecoderClient client, AudioDecoderConfig config, int32 cdm_id)
      => (bool success, bool needs_bitstream_conversion);
 
+  // Established data connection. Should be called before Decode().
+  SetDataSource(handle<data_pipe_consumer> receive_pipe);
+
   // Sends the |buffer| to the underlying codec. Should be called only after
   // Initialize() succeeds. The callback with the status is called after the
   // decoder has accepted corresponding DecoderBuffer, indicating that the
diff --git a/media/mojo/services/mojo_audio_decoder.cc b/media/mojo/services/mojo_audio_decoder.cc
index ee6fe3c..0e9ea511 100644
--- a/media/mojo/services/mojo_audio_decoder.cc
+++ b/media/mojo/services/mojo_audio_decoder.cc
@@ -11,6 +11,7 @@
 #include "base/logging.h"
 #include "base/single_thread_task_runner.h"
 #include "base/thread_task_runner_handle.h"
+#include "media/base/audio_buffer.h"
 #include "media/base/cdm_context.h"
 #include "media/mojo/common/media_type_converters.h"
 
@@ -65,6 +66,7 @@ void MojoAudioDecoder::Initialize(const AudioDecoderConfig& config,
       base::Bind(&MojoAudioDecoder::OnConnectionError, base::Unretained(this)));
 
   init_cb_ = init_cb;
+  output_cb_ = output_cb;
 
   // Using base::Unretained(this) is safe because |this| owns |remote_decoder_|,
   // and the callback won't be dispatched if |remote_decoder_| is destroyed.
@@ -87,10 +89,8 @@ void MojoAudioDecoder::Decode(const scoped_refptr<DecoderBuffer>& buffer,
   DCHECK(decode_cb_.is_null());
   decode_cb_ = decode_cb;
 
-  // This code won't work because |buffer| is not serialized.
-  // TODO(timav): Serialize DecodeBuffer into data pipe.
   remote_decoder_->Decode(
-      interfaces::DecoderBuffer::From(buffer),
+      TransferDecoderBuffer(buffer),
       base::Bind(&MojoAudioDecoder::OnDecodeStatus, base::Unretained(this)));
 }
 
@@ -126,7 +126,7 @@ void MojoAudioDecoder::OnBufferDecoded(interfaces::AudioBufferPtr buffer) {
   DVLOG(1) << __FUNCTION__;
   DCHECK(task_runner_->BelongsToCurrentThread());
 
-  NOTIMPLEMENTED();
+  output_cb_.Run(buffer.To<scoped_refptr<AudioBuffer>>());
 }
 
 void MojoAudioDecoder::OnConnectionError() {
@@ -153,6 +153,9 @@ void MojoAudioDecoder::OnInitialized(bool status,
 
   needs_bitstream_conversion_ = needs_bitstream_conversion;
 
+  if (status)
+    CreateDataPipe();
+
   task_runner_->PostTask(FROM_HERE, base::Bind(init_cb_, status));
 }
 
@@ -190,4 +193,38 @@ void MojoAudioDecoder::OnResetDone() {
   base::ResetAndReturn(&reset_cb_).Run();
 }
 
+void MojoAudioDecoder::CreateDataPipe() {
+  MojoCreateDataPipeOptions options;
+  options.struct_size = sizeof(MojoCreateDataPipeOptions);
+  options.flags = MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE;
+  options.element_num_bytes = 1;
+  // TODO(timav): Consider capacity calculation based on AudioDecoderConfig.
+  options.capacity_num_bytes = 512 * 1024;
+
+  mojo::DataPipe write_pipe(options);
+
+  // Keep producer end.
+  producer_handle_ = std::move(write_pipe.producer_handle);
+
+  // Pass consumer end to |remote_decoder_|.
+  remote_decoder_->SetDataSource(std::move(write_pipe.consumer_handle));
+}
+
+interfaces::DecoderBufferPtr MojoAudioDecoder::TransferDecoderBuffer(
+    const scoped_refptr<DecoderBuffer>& media_buffer) {
+  interfaces::DecoderBufferPtr buffer =
+      interfaces::DecoderBuffer::From(media_buffer);
+  if (media_buffer->end_of_stream())
+    return buffer;
+
+  // Serialize the data section of the DecoderBuffer into our pipe.
+  uint32_t num_bytes = base::checked_cast<uint32_t>(media_buffer->data_size());
+  DCHECK_GT(num_bytes, 0u);
+  CHECK_EQ(WriteDataRaw(producer_handle_.get(), media_buffer->data(),
+                        &num_bytes, MOJO_READ_DATA_FLAG_ALL_OR_NONE),
+           MOJO_RESULT_OK);
+  CHECK_EQ(num_bytes, static_cast<uint32_t>(media_buffer->data_size()));
+  return buffer;
+}
+
 }  // namespace media
diff --git a/media/mojo/services/mojo_audio_decoder.h b/media/mojo/services/mojo_audio_decoder.h
index 6bfae6e..3f0d58f 100644
--- a/media/mojo/services/mojo_audio_decoder.h
+++ b/media/mojo/services/mojo_audio_decoder.h
@@ -10,6 +10,7 @@
 #include "media/base/audio_decoder.h"
 #include "media/mojo/interfaces/audio_decoder.mojom.h"
 #include "mojo/public/cpp/bindings/binding.h"
+#include "mojo/public/cpp/system/data_pipe.h"
 
 namespace base {
 class SingleThreadTaskRunner;
@@ -52,15 +53,29 @@ class MojoAudioDecoder : public AudioDecoder,
   // called when |remote_decoder_| finished Reset() sequence.
   void OnResetDone();
 
+  // A helper method that creates data pipe and sets the data connection to
+  // the service.
+  void CreateDataPipe();
+
+  // A helper method to serialize the data section of DecoderBuffer into pipe.
+  interfaces::DecoderBufferPtr TransferDecoderBuffer(
+      const scoped_refptr<DecoderBuffer>& media_buffer);
+
   scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
 
   interfaces::AudioDecoderPtr remote_decoder_;
 
+  // DataPipe for serializing the data section of DecoderBuffer.
+  mojo::ScopedDataPipeProducerHandle producer_handle_;
+
   // Binding for AudioDecoderClient, bound to the |task_runner_|.
   mojo::Binding<AudioDecoderClient> binding_;
 
-  // We call these callbacks to pass the information to the pipeline.
+  // We call the following callbacks to pass the information to the pipeline.
+  // |output_cb_| is permanent while other three are called only once,
+  // |decode_cb_| and |reset_cb_| are replaced by every by Decode() and Reset().
   InitCB init_cb_;
+  OutputCB output_cb_;
   DecodeCB decode_cb_;
   base::Closure reset_cb_;
 
diff --git a/media/mojo/services/mojo_audio_decoder_service.cc b/media/mojo/services/mojo_audio_decoder_service.cc
index 51fdef5..96c38a0 100644
--- a/media/mojo/services/mojo_audio_decoder_service.cc
+++ b/media/mojo/services/mojo_audio_decoder_service.cc
@@ -50,6 +50,12 @@ void MojoAudioDecoderService::Initialize(
       base::Bind(&MojoAudioDecoderService::OnAudioBufferReady, weak_this_));
 }
 
+void MojoAudioDecoderService::SetDataSource(
+    mojo::ScopedDataPipeConsumerHandle receive_pipe) {
+  DVLOG(1) << __FUNCTION__;
+  consumer_handle_ = std::move(receive_pipe);
+}
+
 void MojoAudioDecoderService::Decode(interfaces::DecoderBufferPtr buffer,
                                      const DecodeCallback& callback) {
   DVLOG(3) << __FUNCTION__;
@@ -99,7 +105,9 @@ void MojoAudioDecoderService::OnResetDone(const ResetCallback& callback) {
 void MojoAudioDecoderService::OnAudioBufferReady(
     const scoped_refptr<AudioBuffer>& audio_buffer) {
   DVLOG(1) << __FUNCTION__;
-  NOTIMPLEMENTED();
+
+  // TODO(timav): Use DataPipe.
+  client_->OnBufferDecoded(interfaces::AudioBuffer::From(audio_buffer));
 }
 
 scoped_refptr<DecoderBuffer> MojoAudioDecoderService::ReadDecoderBuffer(
@@ -107,7 +115,26 @@ scoped_refptr<DecoderBuffer> MojoAudioDecoderService::ReadDecoderBuffer(
   scoped_refptr<DecoderBuffer> media_buffer(
       buffer.To<scoped_refptr<DecoderBuffer>>());
 
-  NOTIMPLEMENTED();
+  if (media_buffer->end_of_stream())
+    return media_buffer;
+
+  // Wait for the data to become available in the DataPipe.
+  MojoHandleSignalsState state;
+  CHECK_EQ(MOJO_RESULT_OK,
+           MojoWait(consumer_handle_.get().value(), MOJO_HANDLE_SIGNAL_READABLE,
+                    MOJO_DEADLINE_INDEFINITE, &state));
+  CHECK_EQ(MOJO_HANDLE_SIGNAL_READABLE, state.satisfied_signals);
+
+  // Read the inner data for the DecoderBuffer from our DataPipe.
+  uint32_t bytes_to_read =
+      base::checked_cast<uint32_t>(media_buffer->data_size());
+  DCHECK_GT(bytes_to_read, 0u);
+  uint32_t bytes_read = bytes_to_read;
+  CHECK_EQ(ReadDataRaw(consumer_handle_.get(), media_buffer->writable_data(),
+                       &bytes_read, MOJO_READ_DATA_FLAG_ALL_OR_NONE),
+           MOJO_RESULT_OK);
+  CHECK_EQ(bytes_to_read, bytes_read);
+
   return media_buffer;
 }
 
diff --git a/media/mojo/services/mojo_audio_decoder_service.h b/media/mojo/services/mojo_audio_decoder_service.h
index 756e19c..558d0eb 100644
--- a/media/mojo/services/mojo_audio_decoder_service.h
+++ b/media/mojo/services/mojo_audio_decoder_service.h
@@ -11,6 +11,7 @@
 #include "media/base/audio_decoder.h"
 #include "media/mojo/interfaces/audio_decoder.mojom.h"
 #include "mojo/public/cpp/bindings/strong_binding.h"
+#include "mojo/public/cpp/system/data_pipe.h"
 
 namespace media {
 
@@ -28,6 +29,8 @@ class MojoAudioDecoderService : public interfaces::AudioDecoder {
                   int32_t cdm_id,
                   const InitializeCallback& callback) final;
 
+  void SetDataSource(mojo::ScopedDataPipeConsumerHandle receive_pipe) final;
+
   void Decode(interfaces::DecoderBufferPtr buffer,
               const DecodeCallback& callback) final;
 
@@ -55,6 +58,9 @@ class MojoAudioDecoderService : public interfaces::AudioDecoder {
   // communication channel, i.e. the pipe.
   mojo::StrongBinding<interfaces::AudioDecoder> binding_;
 
+  // DataPipe for serializing the data section of DecoderBuffer.
+  mojo::ScopedDataPipeConsumerHandle consumer_handle_;
+
   // The AudioDecoder that does actual decoding work.
   scoped_ptr<media::AudioDecoder> decoder_;