Mojo: Armed Watchers

chrome/browser/media/cast_remoting_sender.cc

Index: chrome/browser/media/cast_remoting_sender.cc
diff --git a/chrome/browser/media/cast_remoting_sender.cc b/chrome/browser/media/cast_remoting_sender.cc
index 00de02537bbeafe4c4f4ca00c7e7e0e197bb1207..e0ad59804c4efd3797eabec7e119bf9446676282 100644
--- a/chrome/browser/media/cast_remoting_sender.cc
+++ b/chrome/browser/media/cast_remoting_sender.cc
@@ -92,7 +92,7 @@ CastRemotingSender::CastRemotingSender(
       latest_acked_frame_id_(media::cast::FrameId::first() - 1),
       duplicate_ack_counter_(0),
       input_queue_discards_remaining_(0),
-      pipe_watcher_(FROM_HERE),
+      pipe_watcher_(FROM_HERE, mojo::SimpleWatcher::ArmingPolicy::MANUAL),
       flow_restart_pending_(true),
       weak_factory_(this) {
   // Confirm this constructor is running on the IO BrowserThread.
@@ -164,10 +164,10 @@ void CastRemotingSender::FindAndBind(
   sender->error_callback_ = error_callback;
 
   sender->pipe_ = std::move(pipe);
-  sender->pipe_watcher_.Start(
-      sender->pipe_.get(), MOJO_HANDLE_SIGNAL_READABLE,
-      base::Bind(&CastRemotingSender::ProcessInputQueue,
-                 base::Unretained(sender)));
+  sender->pipe_watcher_.Watch(sender->pipe_.get(), MOJO_HANDLE_SIGNAL_READABLE,
+                              base::Bind(&CastRemotingSender::ProcessInputQueue,
+                                         base::Unretained(sender)));
+  sender->pipe_watcher_.ArmOrNotify();
   sender->binding_.Bind(std::move(request));
   sender->binding_.set_connection_error_handler(sender->error_callback_);
 }
@@ -375,11 +375,14 @@ bool CastRemotingSender::TryConsumeDataChunk(uint32_t offset, uint32_t size,
     if (discard_data) {
       const MojoResult result = mojo::ReadDataRaw(
           pipe_.get(), nullptr, &size,
-          MOJO_READ_DATA_FLAG_DISCARD | MOJO_READ_DATA_FLAG_ALL_OR_NONE);
+          MOJO_READ_DATA_FLAG_DISCARD | MOJO_READ_DATA_FLAG_ALL_OR_NONE |
+              MOJO_READ_DATA_FLAG_CLEAR_SIGNAL);
       if (result == MOJO_RESULT_OK)
         return true;  // Successfully discarded data.
-      if (result == MOJO_RESULT_OUT_OF_RANGE)
+      if (result == MOJO_RESULT_OUT_OF_RANGE) {
+        pipe_watcher_.ArmOrNotify();
         return false;  // Retry later.
+      }
       LOG(ERROR) << SENDER_SSRC
                  << "Unexpected result when discarding from data pipe ("
                  << result << ')';
@@ -392,11 +395,13 @@ bool CastRemotingSender::TryConsumeDataChunk(uint32_t offset, uint32_t size,
 
     const MojoResult result = mojo::ReadDataRaw(
         pipe_.get(), base::string_as_array(&next_frame_data_) + offset, &size,
-        MOJO_READ_DATA_FLAG_ALL_OR_NONE);
+        MOJO_READ_DATA_FLAG_ALL_OR_NONE | MOJO_READ_DATA_FLAG_CLEAR_SIGNAL);
     if (result == MOJO_RESULT_OK)
       return true;  // Successfully consumed data.
-    if (result == MOJO_RESULT_OUT_OF_RANGE)
+    if (result == MOJO_RESULT_OUT_OF_RANGE) {
+      pipe_watcher_.ArmOrNotify();
       return false;  // Retry later.
+    }
     LOG(ERROR)
         << SENDER_SSRC << "Read from data pipe failed (" << result << ')';
   } while (false);