Mojo: Armed Watchers

content/browser/loader/mojo_async_resource_handler.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 "content/browser/loader/mojo_async_resource_handler.h"
 
 #include <algorithm>
 #include <utility>
 #include <vector>
 
@@ skipping 103 matching lines @@
 
 MojoAsyncResourceHandler::MojoAsyncResourceHandler(
     net::URLRequest* request,
     ResourceDispatcherHostImpl* rdh,
     mojom::URLLoaderAssociatedRequest mojo_request,
     mojom::URLLoaderClientPtr url_loader_client,
     ResourceType resource_type)
     : ResourceHandler(request),
       rdh_(rdh),
       binding_(this, std::move(mojo_request)),
-      handle_watcher_(FROM_HERE),
+      handle_watcher_(FROM_HERE, mojo::SimpleWatcher::ArmingPolicy::MANUAL),
       url_loader_client_(std::move(url_loader_client)),
       weak_factory_(this) {
   DCHECK(url_loader_client_);
   InitializeResourceBufferConstants();
   // This unretained pointer is safe, because |binding_| is owned by |this| and
   // the callback will never be called after |this| is destroyed.
   binding_.set_connection_error_handler(
       base::Bind(&MojoAsyncResourceHandler::Cancel, base::Unretained(this)));
 
   if (IsResourceTypeFrame(resource_type)) {
@@ skipping 103 matching lines @@
     options.flags = MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE;
     options.element_num_bytes = 1;
     options.capacity_num_bytes = g_allocation_size;
     mojo::DataPipe data_pipe(options);
 
     DCHECK(data_pipe.producer_handle.is_valid());
     DCHECK(data_pipe.consumer_handle.is_valid());
 
     response_body_consumer_handle_ = std::move(data_pipe.consumer_handle);
     shared_writer_ = new SharedWriter(std::move(data_pipe.producer_handle));
-    handle_watcher_.Start(shared_writer_->writer(), MOJO_HANDLE_SIGNAL_WRITABLE,
+    handle_watcher_.Watch(shared_writer_->writer(), MOJO_HANDLE_SIGNAL_WRITABLE,
                           base::Bind(&MojoAsyncResourceHandler::OnWritable,
                                      base::Unretained(this)));
+    handle_watcher_.ArmOrNotify();
 
     bool defer = false;
     scoped_refptr<net::IOBufferWithSize> buffer;
     if (!AllocateWriterIOBuffer(&buffer, &defer)) {
       controller->CancelWithError(net::ERR_INSUFFICIENT_RESOURCES);
       return;
     }
     if (!defer) {
       if (static_cast<size_t>(buffer->size()) >= kMinAllocationSize) {
         *buf = buffer_ = buffer;
@@ skipping 120 matching lines @@
 void MojoAsyncResourceHandler::SetAllocationSizeForTesting(size_t size) {
   g_allocation_size = size;
 }
 
 MojoResult MojoAsyncResourceHandler::BeginWrite(void** data,
                                                 uint32_t* available) {
   MojoResult result = mojo::BeginWriteDataRaw(
       shared_writer_->writer(), data, available, MOJO_WRITE_DATA_FLAG_NONE);
   if (result == MOJO_RESULT_OK)
     *available = std::min(*available, static_cast<uint32_t>(kMaxChunkSize));
+  else if (result == MOJO_RESULT_SHOULD_WAIT)
+    handle_watcher_.ArmOrNotify();
   return result;
 }
 
 MojoResult MojoAsyncResourceHandler::EndWrite(uint32_t written) {
-  return mojo::EndWriteDataRaw(shared_writer_->writer(), written);
+  MojoResult result = mojo::EndWriteDataRaw(shared_writer_->writer(), written);
+  if (result == MOJO_RESULT_OK)
+    handle_watcher_.ArmOrNotify();
+  return result;
 }
 
 net::IOBufferWithSize* MojoAsyncResourceHandler::GetResponseMetadata(
     net::URLRequest* request) {
   return request->response_info().metadata.get();
 }
 
 void MojoAsyncResourceHandler::OnResponseCompleted(
     const net::URLRequestStatus& status,
     std::unique_ptr<ResourceController> controller) {
@@ skipping 172 matching lines @@
       base::Bind(&MojoAsyncResourceHandler::OnUploadProgressACK,
                  weak_factory_.GetWeakPtr()));
 }
 
 void MojoAsyncResourceHandler::OnUploadProgressACK() {
   if (upload_progress_tracker_)
     upload_progress_tracker_->OnAckReceived();
 }
 
 }  // namespace content