Mojo: Armed Watchers

media/remoting/demuxer_stream_adapter.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/remoting/demuxer_stream_adapter.h"
 
 #include "base/base64.h"
 #include "base/bind.h"
 #include "base/callback_helpers.h"
 #include "media/base/bind_to_current_loop.h"
@@ skipping 39 matching lines @@
       demuxer_stream_(demuxer_stream),
       type_(demuxer_stream ? demuxer_stream->type() : DemuxerStream::UNKNOWN),
       error_callback_(error_callback),
       remote_callback_handle_(RpcBroker::kInvalidHandle),
       read_until_callback_handle_(RpcBroker::kInvalidHandle),
       read_until_count_(0),
       last_count_(0),
       pending_flush_(false),
       current_pending_frame_offset_(0),
       pending_frame_is_eos_(false),
-      write_watcher_(FROM_HERE),
+      write_watcher_(FROM_HERE, mojo::SimpleWatcher::ArmingPolicy::MANUAL),
       media_status_(DemuxerStream::kOk),
       producer_handle_(std::move(producer_handle)),
       bytes_written_to_pipe_(0),
       request_buffer_weak_factory_(this),
       weak_factory_(this) {
   DCHECK(main_task_runner_);
   DCHECK(media_task_runner_);
   DCHECK(media_task_runner_->BelongsToCurrentThread());
   DCHECK(demuxer_stream);
   DCHECK(!error_callback.is_null());
@@ skipping 124 matching lines @@
                           ? audio_config_.AsHumanReadableString()
                           : video_config_.AsHumanReadableString())
                   << '}';
   main_task_runner_->PostTask(
       FROM_HERE, base::Bind(&RpcBroker::SendMessageToRemote, rpc_broker_,
                             base::Passed(&rpc)));
 
   // Starts Mojo watcher.
   if (!write_watcher_.IsWatching()) {
     DEMUXER_VLOG(2) << "Start Mojo data pipe watcher";
-    write_watcher_.Start(producer_handle_.get(), MOJO_HANDLE_SIGNAL_WRITABLE,
+    write_watcher_.Watch(producer_handle_.get(), MOJO_HANDLE_SIGNAL_WRITABLE,
                          base::Bind(&DemuxerStreamAdapter::TryWriteData,
                                     weak_factory_.GetWeakPtr()));
+    write_watcher_.ArmOrNotify();
   }
 }
 
 void DemuxerStreamAdapter::ReadUntil(std::unique_ptr<pb::RpcMessage> message) {
   DCHECK(media_task_runner_->BelongsToCurrentThread());
   DCHECK(message);
   if (!message->has_demuxerstream_readuntil_rpc()) {
     DEMUXER_VLOG(1) << "Missing required DemuxerStreamReadUntil struct in RPC";
     OnFatalError(RPC_INVALID);
     return;
@@ skipping 101 matching lines @@
   if (!stream_sender_ || !producer_handle_.is_valid()) {
     DEMUXER_VLOG(1) << "Ignore since data pipe stream sender is invalid";
     return;
   }
 
   uint32_t num_bytes = pending_frame_.size() - current_pending_frame_offset_;
   MojoResult mojo_result =
       WriteDataRaw(producer_handle_.get(),
                    pending_frame_.data() + current_pending_frame_offset_,
                    &num_bytes, MOJO_WRITE_DATA_FLAG_NONE);
-  if (mojo_result != MOJO_RESULT_OK) {
-    if (mojo_result != MOJO_RESULT_SHOULD_WAIT) {
-      DEMUXER_VLOG(1) << "Pipe was closed unexpectedly (or a bug). result:"
-                      << mojo_result;
-      OnFatalError(MOJO_PIPE_ERROR);
-    }
+  if (mojo_result != MOJO_RESULT_OK && mojo_result != MOJO_RESULT_SHOULD_WAIT) {
+    DEMUXER_VLOG(1) << "Pipe was closed unexpectedly (or a bug). result:"
+                    << mojo_result;
+    OnFatalError(MOJO_PIPE_ERROR);
     return;
   }
 
+  write_watcher_.ArmOrNotify();
+  if (mojo_result != MOJO_RESULT_OK)
+    return;
+
   stream_sender_->ConsumeDataChunk(current_pending_frame_offset_, num_bytes,
                                    pending_frame_.size());
   current_pending_frame_offset_ += num_bytes;
   bytes_written_to_pipe_ += num_bytes;
 
   // Checks if all buffer was written to browser process.
   if (current_pending_frame_offset_ != pending_frame_.size()) {
     // Returns and wait for mojo watcher to notify to write more data.
     return;
   }
@@ skipping 86 matching lines @@
   if (write_watcher_.IsWatching()) {
     DEMUXER_VLOG(2) << "Cancel mojo data pipe watcher";
     write_watcher_.Cancel();
   }
 
   base::ResetAndReturn(&error_callback_).Run(stop_trigger);
 }
 
 }  // namespace remoting
 }  // namespace media