Add mojo <--> net IO adapters.

mojo/services/network/url_loader_impl.cc

 // Copyright 2014 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 "mojo/services/network/url_loader_impl.h"
 
 #include "base/memory/scoped_vector.h"
 #include "base/message_loop/message_loop.h"
 #include "mojo/common/common_type_converters.h"
 #include "mojo/services/network/net_adapters.h"
 #include "mojo/services/network/network_context.h"
 #include "net/base/io_buffer.h"
 #include "net/base/load_flags.h"
 #include "net/base/upload_bytes_element_reader.h"
 #include "net/base/upload_data_stream.h"
 #include "net/http/http_response_headers.h"
 #include "net/url_request/redirect_info.h"
 #include "net/url_request/url_request_context.h"
 
 namespace mojo {
 namespace {
 
-const uint32_t kMaxReadSize = 64 * 1024;
-
 // Generates an URLResponsePtr from the response state of a net::URLRequest.
 URLResponsePtr MakeURLResponse(const net::URLRequest* url_request) {
   URLResponsePtr response(URLResponse::New());
   response->url = String::From(url_request->url());
 
   const net::HttpResponseHeaders* headers = url_request->response_headers();
   if (headers) {
     response->status_code = headers->response_code();
     response->status_line = headers->GetStatusLine();
 
@@ skipping 57 matching lines @@
  private:
   ScopedDataPipeConsumerHandle pipe_;
   uint32_t num_bytes_;
   uint32_t offset_;
 
   DISALLOW_COPY_AND_ASSIGN(UploadDataPipeElementReader);
 };
 
 }  // namespace
 
-// Keeps track of a pending two-phase write on a DataPipeProducerHandle.
-class URLLoaderImpl::PendingWriteToDataPipe :
-    public base::RefCountedThreadSafe<PendingWriteToDataPipe> {
- public:
-  explicit PendingWriteToDataPipe(ScopedDataPipeProducerHandle handle)
-      : handle_(handle.Pass()),
-        buffer_(NULL) {
-  }
-
-  MojoResult BeginWrite(uint32_t* num_bytes) {
-    MojoResult result = BeginWriteDataRaw(handle_.get(), &buffer_, num_bytes,
-                                          MOJO_WRITE_DATA_FLAG_NONE);
-    if (*num_bytes > kMaxReadSize)
-      *num_bytes = kMaxReadSize;
-
-    return result;
-  }
-
-  ScopedDataPipeProducerHandle Complete(uint32_t num_bytes) {
-    EndWriteDataRaw(handle_.get(), num_bytes);
-    buffer_ = NULL;
-    return handle_.Pass();
-  }
-
-  char* buffer() { return static_cast<char*>(buffer_); }
-
- private:
-  friend class base::RefCountedThreadSafe<PendingWriteToDataPipe>;
-
-  ~PendingWriteToDataPipe() {
-    if (handle_.is_valid())
-      EndWriteDataRaw(handle_.get(), 0);
-  }
-
-  ScopedDataPipeProducerHandle handle_;
-  void* buffer_;
-
-  DISALLOW_COPY_AND_ASSIGN(PendingWriteToDataPipe);
-};
-
-// Takes ownership of a pending two-phase write on a DataPipeProducerHandle,
-// and makes its buffer available as a net::IOBuffer.
-class URLLoaderImpl::DependentIOBuffer : public net::WrappedIOBuffer {
- public:
-  DependentIOBuffer(PendingWriteToDataPipe* pending_write)
-      : net::WrappedIOBuffer(pending_write->buffer()),
-        pending_write_(pending_write) {
-  }
- private:
-  virtual ~DependentIOBuffer() {}
-  scoped_refptr<PendingWriteToDataPipe> pending_write_;
-};
-
 URLLoaderImpl::URLLoaderImpl(NetworkContext* context)
     : context_(context),
       response_body_buffer_size_(0),
       auto_follow_redirects_(true),
       weak_ptr_factory_(this) {
 }
 
 URLLoaderImpl::~URLLoaderImpl() {
 }
 
@@ skipping 140 matching lines @@
   Callback<void(URLResponsePtr)> callback;
   std::swap(callback_, callback);
   callback.Run(response.Pass());
 }
 
 void URLLoaderImpl::OnResponseBodyStreamReady(MojoResult result) {
   // TODO(darin): Handle a bad |result| value.
   ReadMore();
 }
 
-void URLLoaderImpl::WaitToReadMore() {
-  handle_watcher_.Start(response_body_stream_.get(),
-                        MOJO_HANDLE_SIGNAL_WRITABLE,
-                        MOJO_DEADLINE_INDEFINITE,
-                        base::Bind(&URLLoaderImpl::OnResponseBodyStreamReady,
-                                   weak_ptr_factory_.GetWeakPtr()));
-}
-
 void URLLoaderImpl::ReadMore() {
   DCHECK(!pending_write_.get());
 
-  pending_write_ = new PendingWriteToDataPipe(response_body_stream_.Pass());
+  uint32_t num_bytes;
+  MojoResult result = NetToMojoPendingBuffer::BeginWrite(
+      &response_body_stream_, &pending_write_, &num_bytes);
 
-  uint32_t num_bytes;
-  MojoResult result = pending_write_->BeginWrite(&num_bytes);
   if (result == MOJO_RESULT_SHOULD_WAIT) {
     // The pipe is full. We need to wait for it to have more space.
-    response_body_stream_ = pending_write_->Complete(num_bytes);
-    pending_write_ = NULL;
-    WaitToReadMore();
+    handle_watcher_.Start(response_body_stream_.get(),
+                          MOJO_HANDLE_SIGNAL_WRITABLE,
+                          MOJO_DEADLINE_INDEFINITE,
+                          base::Bind(&URLLoaderImpl::OnResponseBodyStreamReady,
+                                     weak_ptr_factory_.GetWeakPtr()));
     return;
-  }
-  if (result != MOJO_RESULT_OK) {
+  } else if (result != MOJO_RESULT_OK) {
     // The response body stream is in a bad state. Bail.
     // TODO(darin): How should this be communicated to our client?
     return;
   }
   CHECK_GT(static_cast<uint32_t>(std::numeric_limits<int>::max()), num_bytes);
 
-  scoped_refptr<net::IOBuffer> buf =
-      new DependentIOBuffer(pending_write_.get());
-
   int bytes_read;
-  url_request_->Read(buf.get(), static_cast<int>(num_bytes), &bytes_read);
-
-  // Drop our reference to the buffer.
-  buf = NULL;
+  url_request_->Read(new NetToMojoIOBuffer(pending_write_.get()),

[yzshen1, 2014/10/10 21:19:00]

This may leak the buffer, because Read() may not add/release ref (e.g., when
fails and returns early).

[brettw, 2014/10/10 22:36:59]

Ooh, good catch!

+                     static_cast<int>(num_bytes), &bytes_read);
 
   if (url_request_->status().is_io_pending()) {
     // Wait for OnReadCompleted.
   } else if (url_request_->status().is_success() && bytes_read > 0) {
     DidRead(static_cast<uint32_t>(bytes_read), true);
   } else {
     pending_write_->Complete(0);
     pending_write_ = NULL;  // This closes the data pipe.
   }
 }
 
 void URLLoaderImpl::DidRead(uint32_t num_bytes, bool completed_synchronously) {
   DCHECK(url_request_->status().is_success());
 
   response_body_stream_ = pending_write_->Complete(num_bytes);
   pending_write_ = NULL;
 
   if (completed_synchronously) {
     base::MessageLoop::current()->PostTask(
         FROM_HERE,
         base::Bind(&URLLoaderImpl::ReadMore, weak_ptr_factory_.GetWeakPtr()));
   } else {
     ReadMore();
   }
 }
 
 }  // namespace mojo