Mojo: Armed Watchers

media/mojo/common/mojo_decoder_buffer_converter.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/common/mojo_decoder_buffer_converter.h"
 
 #include <memory>
 
 #include "base/logging.h"
 #include "base/memory/ptr_util.h"
@@ skipping 46 matching lines @@
   DVLOG(1) << __func__;
   std::unique_ptr<mojo::DataPipe> data_pipe = CreateDataPipe(type);
   *producer_handle = std::move(data_pipe->producer_handle);
   return base::WrapUnique(
       new MojoDecoderBufferReader(std::move(data_pipe->consumer_handle)));
 }
 
 MojoDecoderBufferReader::MojoDecoderBufferReader(
     mojo::ScopedDataPipeConsumerHandle consumer_handle)
     : consumer_handle_(std::move(consumer_handle)),
-      pipe_watcher_(FROM_HERE),
+      pipe_watcher_(FROM_HERE, mojo::SimpleWatcher::ArmingPolicy::MANUAL),
       bytes_read_(0) {
   DVLOG(1) << __func__;
 
   MojoResult result =
-      pipe_watcher_.Start(consumer_handle_.get(), MOJO_HANDLE_SIGNAL_READABLE,
+      pipe_watcher_.Watch(consumer_handle_.get(), MOJO_HANDLE_SIGNAL_READABLE,
                           base::Bind(&MojoDecoderBufferReader::OnPipeReadable,
                                      base::Unretained(this)));
   if (result != MOJO_RESULT_OK) {
     DVLOG(1) << __func__
              << ": Failed to start watching the pipe. result=" << result;
     consumer_handle_.reset();
+  } else {
+    pipe_watcher_.ArmOrNotify();
   }
 }
 
 MojoDecoderBufferReader::~MojoDecoderBufferReader() {
   DVLOG(1) << __func__;
 }
 
 void MojoDecoderBufferReader::ReadDecoderBuffer(
     mojom::DecoderBufferPtr mojo_buffer,
     ReadCB read_cb) {
@@ skipping 63 matching lines @@
 
   uint32_t num_bytes = buffer_size - bytes_read_;
   DCHECK_GT(num_bytes, 0u);
 
   MojoResult result = ReadDataRaw(consumer_handle_.get(),
                                   media_buffer_->writable_data() + bytes_read_,
                                   &num_bytes, MOJO_WRITE_DATA_FLAG_NONE);
 
   if (IsPipeReadWriteError(result)) {
     OnPipeError(result);
-  } else if (result == MOJO_RESULT_OK) {
-    DCHECK_GT(num_bytes, 0u);
-    bytes_read_ += num_bytes;
-    if (bytes_read_ == buffer_size) {
-      DCHECK(read_cb_);
-      bytes_read_ = 0;
-      std::move(read_cb_).Run(std::move(media_buffer_));
+  } else {
+    pipe_watcher_.ArmOrNotify();
+    if (result == MOJO_RESULT_OK) {
+      DCHECK_GT(num_bytes, 0u);
+      bytes_read_ += num_bytes;
+      if (bytes_read_ == buffer_size) {
+        DCHECK(read_cb_);
+        bytes_read_ = 0;
+        std::move(read_cb_).Run(std::move(media_buffer_));
+      }
     }
   }
 }
 
 // MojoDecoderBufferWriter
 
 // static
 std::unique_ptr<MojoDecoderBufferWriter> MojoDecoderBufferWriter::Create(
     DemuxerStream::Type type,
     mojo::ScopedDataPipeConsumerHandle* consumer_handle) {
   DVLOG(1) << __func__;
   std::unique_ptr<mojo::DataPipe> data_pipe = CreateDataPipe(type);
   *consumer_handle = std::move(data_pipe->consumer_handle);
   return base::WrapUnique(
       new MojoDecoderBufferWriter(std::move(data_pipe->producer_handle)));
 }
 
 MojoDecoderBufferWriter::MojoDecoderBufferWriter(
     mojo::ScopedDataPipeProducerHandle producer_handle)
     : producer_handle_(std::move(producer_handle)),
-      pipe_watcher_(FROM_HERE),
+      pipe_watcher_(FROM_HERE, mojo::SimpleWatcher::ArmingPolicy::MANUAL),
       bytes_written_(0) {
   DVLOG(1) << __func__;
 
   MojoResult result =
-      pipe_watcher_.Start(producer_handle_.get(), MOJO_HANDLE_SIGNAL_WRITABLE,
+      pipe_watcher_.Watch(producer_handle_.get(), MOJO_HANDLE_SIGNAL_WRITABLE,
                           base::Bind(&MojoDecoderBufferWriter::OnPipeWritable,
                                      base::Unretained(this)));
   if (result != MOJO_RESULT_OK) {
     DVLOG(1) << __func__
              << ": Failed to start watching the pipe. result=" << result;
     producer_handle_.reset();
+  } else {
+    pipe_watcher_.ArmOrNotify();
   }
 }
 
 MojoDecoderBufferWriter::~MojoDecoderBufferWriter() {
   DVLOG(1) << __func__;
 }
 
 mojom::DecoderBufferPtr MojoDecoderBufferWriter::WriteDecoderBuffer(
     const scoped_refptr<DecoderBuffer>& media_buffer) {
   DVLOG(3) << __func__;
@@ skipping 55 matching lines @@
 
   uint32_t num_bytes = buffer_size - bytes_written_;
   DCHECK_GT(num_bytes, 0u);
 
   MojoResult result = WriteDataRaw(producer_handle_.get(),
                                    media_buffer_->data() + bytes_written_,
                                    &num_bytes, MOJO_WRITE_DATA_FLAG_NONE);
 
   if (IsPipeReadWriteError(result)) {
     OnPipeError(result);
-  } else if (result == MOJO_RESULT_OK) {
-    DCHECK_GT(num_bytes, 0u);
-    bytes_written_ += num_bytes;
-    if (bytes_written_ == buffer_size) {
-      media_buffer_ = nullptr;
-      bytes_written_ = 0;
+  } else {
+    pipe_watcher_.ArmOrNotify();
+    if (result == MOJO_RESULT_OK) {
+      DCHECK_GT(num_bytes, 0u);
+      bytes_written_ += num_bytes;
+      if (bytes_written_ == buffer_size) {
+        media_buffer_ = nullptr;
+        bytes_written_ = 0;
+      }
     }
   }
 
   return result;
 }
 
 }  // namespace media