Use data pipe for encoded buffers in MojoAudioDecoder

media/mojo/services/mojo_audio_decoder.cc

 // Copyright 2016 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 "media/mojo/services/mojo_audio_decoder.h"
 
 #include "base/bind.h"
 #include "base/bind_helpers.h"
 #include "base/callback_helpers.h"
 #include "base/location.h"
 #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"
 
 namespace media {
 
 MojoAudioDecoder::MojoAudioDecoder(
     scoped_refptr<base::SingleThreadTaskRunner> task_runner,
     interfaces::AudioDecoderPtr remote_decoder)
     : task_runner_(task_runner),
       remote_decoder_(std::move(remote_decoder)),
@@ skipping 34 matching lines @@
     return;
   }
 
   // Otherwise, set an error handler to catch the connection error.
   // Using base::Unretained(this) is safe because |this| owns |remote_decoder_|,
   // and the error handler can't be invoked once |remote_decoder_| is destroyed.
   remote_decoder_.set_connection_error_handler(
       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.
   remote_decoder_->Initialize(
       binding_.CreateInterfacePtrAndBind(),
       interfaces::AudioDecoderConfig::From(config), cdm_id,
       base::Bind(&MojoAudioDecoder::OnInitialized, base::Unretained(this)));
 }
 
 void MojoAudioDecoder::Decode(const scoped_refptr<DecoderBuffer>& buffer,
                               const DecodeCB& decode_cb) {
   DVLOG(3) << __FUNCTION__;
   DCHECK(task_runner_->BelongsToCurrentThread());
 
   if (has_connection_error_) {
     task_runner_->PostTask(FROM_HERE, base::Bind(decode_cb, kDecodeError));
     return;
   }
 
   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)));
 }
 
 void MojoAudioDecoder::Reset(const base::Closure& closure) {
   DVLOG(2) << __FUNCTION__;
   DCHECK(task_runner_->BelongsToCurrentThread());
 
   if (has_connection_error_) {
     if (!decode_cb_.is_null()) {
       task_runner_->PostTask(
@@ skipping 15 matching lines @@
   DVLOG(1) << __FUNCTION__;
   DCHECK(task_runner_->BelongsToCurrentThread());
 
   return needs_bitstream_conversion_;
 }
 
 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() {
   DVLOG(1) << __FUNCTION__;
   DCHECK(task_runner_->BelongsToCurrentThread());
 
   has_connection_error_ = true;
 
   if (!init_cb_.is_null()) {
     base::ResetAndReturn(&init_cb_).Run(false);
     return;
   }
 
   if (!decode_cb_.is_null())
     base::ResetAndReturn(&decode_cb_).Run(kDecodeError);
   if (!reset_cb_.is_null())
     base::ResetAndReturn(&reset_cb_).Run();
 }
 
 void MojoAudioDecoder::OnInitialized(bool status,
                                      bool needs_bitstream_conversion) {
   DVLOG(1) << __FUNCTION__ << ": status:" << status;
   DCHECK(task_runner_->BelongsToCurrentThread());
 
   needs_bitstream_conversion_ = needs_bitstream_conversion;
 
+  if (status)
+    CreateDataPipe();
+
   task_runner_->PostTask(FROM_HERE, base::Bind(init_cb_, status));
 }
 
 static media::AudioDecoder::Status ConvertDecodeStatus(
     interfaces::AudioDecoder::DecodeStatus status) {
   switch (status) {
     case interfaces::AudioDecoder::DecodeStatus::OK:
       return media::AudioDecoder::kOk;
     case interfaces::AudioDecoder::DecodeStatus::ABORTED:
       return media::AudioDecoder::kAborted;
@@ skipping 17 matching lines @@
   DVLOG(1) << __FUNCTION__;
   DCHECK(task_runner_->BelongsToCurrentThread());
 
   // For pending decodes OnDecodeStatus() should arrive before OnResetDone().
   DCHECK(decode_cb_.is_null());
 
   DCHECK(!reset_cb_.is_null());
   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