Mojo: Armed Watchers

chrome/browser/media/cast_remoting_sender.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 "chrome/browser/media/cast_remoting_sender.h"
 
 #include <algorithm>
 #include <map>
 
 #include "base/bind.h"
@@ skipping 74 matching lines @@
                 media::cast::RtpPayloadType::REMOTE_AUDIO),
       logging_flush_interval_(logging_flush_interval),
       frame_event_cb_(cb),
       clock_(new base::DefaultTickClock()),
       binding_(this),
       max_ack_delay_(kMaxAckDelay),
       last_sent_frame_id_(media::cast::FrameId::first() - 1),
       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.
   DCHECK_CURRENTLY_ON(BrowserThread::IO);
 
   CastRemotingSender*& pointer_in_map = g_sender_map.Get()[rtp_stream_id_];
   DCHECK(!pointer_in_map);
   pointer_in_map = this;
 
   transport_->InitializeStream(
@@ skipping 51 matching lines @@
     DLOG(ERROR) << "Attempt to bind to CastRemotingSender a second time (id="
                 << rtp_stream_id << ")!";
     error_callback.Run();
     return;
   }
 
   DCHECK(sender->error_callback_.is_null());
   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_NEW_DATA_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_);
 }
 
 void CastRemotingSender::OnReceivedRtt(base::TimeDelta round_trip_time) {
   DCHECK_GT(round_trip_time, base::TimeDelta());
   current_round_trip_time_ = round_trip_time;
   max_ack_delay_ = 2 * std::max(current_round_trip_time_, base::TimeDelta()) +
       kReceiverProcessTime;
   max_ack_delay_ = std::min(max_ack_delay_, kMaxAckDelay);
@@ skipping 190 matching lines @@
     }
 
     // If the data is to be discarded, do a data pipe read with the DISCARD flag
     // set.
     if (discard_data) {
       const MojoResult result = mojo::ReadDataRaw(
           pipe_.get(), nullptr, &size,
           MOJO_READ_DATA_FLAG_DISCARD | MOJO_READ_DATA_FLAG_ALL_OR_NONE);
       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 << ')';
       break;
     }
 
     // If |total_payload_size| has changed, resize the data string. If it has
     // not changed, the following statement will be a no-op.
     next_frame_data_.resize(total_payload_size);
 
     const MojoResult result = mojo::ReadDataRaw(
         pipe_.get(), base::string_as_array(&next_frame_data_) + offset, &size,
         MOJO_READ_DATA_FLAG_ALL_OR_NONE);
     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);
 
   // If this point is reached, there was a fatal error. Shut things down and run
   // the error callback.
   pipe_watcher_.Cancel();
   pipe_.reset();
   binding_.Close();
   error_callback_.Run();
@@ skipping 125 matching lines @@
     frame_event_cb_.Run(frame_events);
   }
 
   base::ThreadTaskRunnerHandle::Get()->PostDelayedTask(
       FROM_HERE, base::Bind(&CastRemotingSender::SendFrameEvents,
                             weak_factory_.GetWeakPtr()),
       logging_flush_interval_);
 }
 
 }  // namespace cast