Add mojo <--> net IO adapters.

mojo/services/network/url_loader_impl.cc

Index: mojo/services/network/url_loader_impl.cc
diff --git a/mojo/services/network/url_loader_impl.cc b/mojo/services/network/url_loader_impl.cc
index 1be5ee2bad48b0572b00c04b075958add072ccc4..3bf1285918728389bf9b96954bbe69a9d3b06284 100644
--- a/mojo/services/network/url_loader_impl.cc
+++ b/mojo/services/network/url_loader_impl.cc
@@ -20,8 +20,6 @@
 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());
@@ -99,59 +97,6 @@ class UploadDataPipeElementReader : public net::UploadElementReader {
 
 }  // 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),
@@ -312,44 +257,32 @@ void URLLoaderImpl::OnResponseBodyStreamReady(MojoResult result) {
   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 = pending_write_->BeginWrite(&num_bytes);
+  MojoResult result = NetToMojoPendingBuffer::BeginWrite(
+      &response_body_stream_, &pending_write_, &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());
+  scoped_refptr<net::IOBuffer> buf(new NetToMojoIOBuffer(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;
-
   if (url_request_->status().is_io_pending()) {
     // Wait for OnReadCompleted.
   } else if (url_request_->status().is_success() && bytes_read > 0) {