PROTOTYPE: PlzNavigate: use a Mojo data pipe to stream navigation data to the renderer.

content/browser/loader/mojo_stream_writer.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_stream_writer.h"
+
+#include "base/bind.h"
+#include "base/guid.h"
+#include "base/numerics/safe_conversions.h"
+#include "content/browser/streams/stream.h"
+#include "content/browser/streams/stream_registry.h"
+#include "content/public/browser/resource_controller.h"
+#include "net/base/io_buffer.h"
+#include "url/gurl.h"
+#include "url/url_constants.h"
+
+namespace content {
+
+MojoStreamWriter::MojoStreamWriter()
+    : controller_(nullptr), buffer_(nullptr), buffer_size_(-1) {}
+
+MojoStreamWriter::~MojoStreamWriter() {
+  if (data_producer_handle_.is_valid())
+    Finalize();
+}
+
+void MojoStreamWriter::InitializeStream(
+    mojo::ScopedDataPipeConsumerHandle* data_consumer_handle) {
+  LOG(ERROR) << "MojoStreamWriter@" << this << " InitializeStream";
+  DCHECK(!data_producer_handle_.is_valid());
+
+  // TODO(carlosk): why not have a default constructor for this struct???
+  // The user having to default initialize it is bad.
+  MojoCreateDataPipeOptions options;
+  options.struct_size = sizeof(MojoCreateDataPipeOptions);
+  options.flags = MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE;
+  options.element_num_bytes = 1;
+  options.capacity_num_bytes = kReadBufSize;
+
+  mojo::DataPipe data_pipe(options);
+  data_producer_handle_ = std::move(data_pipe.producer_handle);
+  *data_consumer_handle = std::move(data_pipe.consumer_handle);
+
+  // TODO(carlosk): We're creating the buffer ahead of time because:
+  // - I'm guessing that by using mojo::(Begin|End)WriteDataRaw we get access to
+  // the final memory position that has to be written to and avoid temporary
+  // copies.
+  // - And because of the mismatch of the Mojo and this class' APIs in regards
+  // to when the information about the need to defer further writes is available
+  // from Mojo and when deferring is allowed by the calls.
+  // TODO(carlosk): Should make sure MOJO_WRITE_DATA_FLAG_ALL_OR_NONE is in fact
+  // needed here and in the Begin* calls below.
+  // TODO(carlosk): need to properly handle Mojo error cases (MojoResult). Here
+  // and everywhere else.
+  buffer_size_ = kReadBufSize;
+  CHECK_EQ(
+      mojo::BeginWriteDataRaw(data_producer_handle_.get(), &buffer_,
+                              &buffer_size_, MOJO_WRITE_DATA_FLAG_ALL_OR_NONE),
+      MOJO_RESULT_OK);
+}
+
+void MojoStreamWriter::OnWillRead(scoped_refptr<net::IOBuffer>* buf,
+                                  int* buf_size,
+                                  int min_size) {
+  LOG(ERROR) << "MojoStreamWriter@" << this << " OnWillRead";
+
+  DCHECK(buf);
+  DCHECK(buf_size > 0);
+  DCHECK_LE(min_size, base::checked_cast<int>(kReadBufSize));
+  // TODO(carlosk): using a net::WrappedIOBuffer might be unsafe! The buffer's
+  // life cycle must be better understood and we must be certain there will not
+  // be cases of user after free.
+  *buf = new net::WrappedIOBuffer(static_cast<char*>(buffer_));
+  *buf_size = base::checked_cast<int>(buffer_size_);
+  buffer_ = nullptr;
+  buffer_size_ = -1;
+}
+
+void MojoStreamWriter::OnReadCompleted(int bytes_read, bool* defer) {
+  CHECK(data_producer_handle_.is_valid());
+  LOG(ERROR) << "MojoStreamWriter@" << this
+             << " OnReadCompleted: bytes_read = " << bytes_read;
+  if (!bytes_read)
+    return;
+
+  // We have more data to read.
+  // DCHECK(read_buffer_.get());
+
+  // TODO(carlosk): double check if there's need for the equivalent of
+  // StreamWriter's immediate_mode for Mojo Data Pipes.
+  CHECK_EQ(mojo::EndWriteDataRaw(data_producer_handle_.get(), bytes_read),
+           MOJO_RESULT_OK);
+
+  // TODO(carlosk): hack to comply with the "deferring" interface (further
+  // explained in constructor).
+  buffer_size_ = kReadBufSize;
+  MojoResult result =
+      mojo::BeginWriteDataRaw(data_producer_handle_.get(), &buffer_,
+                              &buffer_size_, MOJO_WRITE_DATA_FLAG_ALL_OR_NONE);
+
+  if (result == MOJO_RESULT_SHOULD_WAIT) {
+    handle_watcher_.Start(data_producer_handle_.get(),
+                          MOJO_HANDLE_SIGNAL_WRITABLE, MOJO_DEADLINE_INDEFINITE,
+                          base::Bind(&MojoStreamWriter::OnSpaceAvailable,
+                                     base::Unretained(this)));
+    *defer = true;
+    LOG(ERROR) << "MojoStreamWriter@" << this
+               << " OnReadCompleted - deferring!";
+  } else {
+    // TODO(carlosk); Must handle errors!
+    CHECK_EQ(result, MOJO_RESULT_OK);
+  }
+}
+
+void MojoStreamWriter::Finalize() {
+  DCHECK(data_producer_handle_.is_valid());
+  if (buffer_ != nullptr) {
+    CHECK_EQ(mojo::EndWriteDataRaw(data_producer_handle_.get(), 0),
+             MOJO_RESULT_OK);
+    buffer_size_ = -1;
+  }
+  data_producer_handle_.reset();
+}
+
+void MojoStreamWriter::OnSpaceAvailable(MojoResult result) {
+  LOG(ERROR) << "MojoStreamWriter@" << this << " OnSpaceAvailable - resuming!";
+  LOG(ERROR) << "MojoStreamWriter@ data_producer_handle_.is_valid(): "
+             << data_producer_handle_.is_valid();
+  LOG(ERROR) << "MojoStreamWriter@ result: " << result;
+  if (result == MOJO_RESULT_OK) {
+    CHECK(data_producer_handle_.is_valid());
+    controller_->Resume();
+    return;
+  }
+  controller_->Cancel();
+}
+
+}  // namespace content