Mojo: Armed Watchers

media/mojo/common/mojo_decoder_buffer_converter.cc

Index: media/mojo/common/mojo_decoder_buffer_converter.cc
diff --git a/media/mojo/common/mojo_decoder_buffer_converter.cc b/media/mojo/common/mojo_decoder_buffer_converter.cc
index 32080d3af4458ff28707a811c51bec9e38035b8f..d47e5f56a67644b34719a05660029b385dd34a8a 100644
--- a/media/mojo/common/mojo_decoder_buffer_converter.cc
+++ b/media/mojo/common/mojo_decoder_buffer_converter.cc
@@ -64,18 +64,20 @@ std::unique_ptr<MojoDecoderBufferReader> MojoDecoderBufferReader::Create(
 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();
   }
 }
 
@@ -159,13 +161,16 @@ void MojoDecoderBufferReader::ReadDecoderBufferData() {
 
   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_));
+      }
     }
   }
 }
@@ -186,18 +191,20 @@ std::unique_ptr<MojoDecoderBufferWriter> MojoDecoderBufferWriter::Create(
 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();
   }
 }
 
@@ -273,12 +280,15 @@ MojoResult MojoDecoderBufferWriter::WriteDecoderBufferData() {
 
   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;
+      }
     }
   }