[DO NOT REVIEW][DO NOT COMMIT] Implement WebDataProducerHandleImpl.

content/child/web_data_producer_handle_impl_unittest.cc

+// Copyright 2015 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/child/web_data_producer_handle_impl.h"
+
+#include <algorithm>
+#include <deque>
+#include <string>
+#include <vector>
+#include "base/bind.h"
+#include "base/memory/scoped_ptr.h"
+#include "base/message_loop/message_loop.h"
+#include "base/run_loop.h"
+#include "base/synchronization/waitable_event.h"
+#include "base/threading/thread.h"
+#include "mojo/public/cpp/system/data_pipe.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace content {
+
+namespace {
+
+using blink::WebDataProducerHandle;
+
+class WriteDataOperationBase {
+ public:
+  virtual ~WriteDataOperationBase() {}
+  virtual void WriteMore() = 0;
+
+  static const WebDataProducerHandle::Flags kNone =
+      WebDataProducerHandle::FlagNone;
+  static const WebDataProducerHandle::Result kOk = WebDataProducerHandle::Ok;
+  //  static const WebDataProducerHandle::Result kAlreadyClosed =
+  //      WebDataProducerHandle::AlreadyClosed;
+  static const WebDataProducerHandle::Result kShouldWait =
+      WebDataProducerHandle::ShouldWait;
+};
+
+class ClientImpl final : public WebDataProducerHandle::Client {
+ public:
+  explicit ClientImpl(WriteDataOperationBase* operation)
+      : operation_(operation) {}
+
+  void didGetWritable() override {
+    base::MessageLoop::current()->PostTask(
+        FROM_HERE,
+        base::Bind(&WriteDataOperationBase::WriteMore,
+                   base::Unretained(operation_)));
+  }
+
+ private:
+  WriteDataOperationBase* operation_;
+};
+
+class WriteDataOperation : public WriteDataOperationBase {
+ public:
+  typedef WebDataProducerHandle::Result Result;
+  WriteDataOperation(mojo::ScopedDataPipeProducerHandle handle,
+                     const std::deque<char>& input)
+      : handle_(new WebDataProducerHandleImpl(handle.Pass())),
+        input_(input) {}
+
+  Result final_result() const { return final_result_; }
+
+  void WriteMore() override { WriteData(); }
+
+  void WriteData() {
+    if (!client_) {
+      client_.reset(new ClientImpl(this));
+      handle_->registerClient(client_.get());
+    }
+
+    Result rv = kOk;
+
+    while (!input_.empty()) {
+      const size_t size = std::min(static_cast<size_t>(16), input_.size());
+      std::vector<char> buffer;
+      size_t written_size = 0;
+      buffer.insert(buffer.end(), input_.begin(), input_.begin() + size);
+      rv = handle_->write(&buffer[0], size, kNone, &written_size);
+      if (rv != kOk)
+        break;
+      input_.erase(input_.begin(), input_.begin() + written_size);
+    }
+
+    if (rv == kShouldWait) {
+      // Wait a while...
+      return;
+    }
+
+    final_result_ = rv;
+
+    // The operation is done.
+    handle_ = nullptr;
+  }
+
+ private:
+  scoped_ptr<WebDataProducerHandle> handle_;
+  scoped_ptr<WebDataProducerHandle::Client> client_;
+  std::deque<char> input_;
+  Result final_result_;
+};
+
+class TwoPhaseWriteDataOperation : public WriteDataOperationBase {
+ public:
+  typedef WebDataProducerHandle::Result Result;
+  TwoPhaseWriteDataOperation(mojo::ScopedDataPipeProducerHandle handle,
+                             const std::deque<char>& input)
+      : handle_(new WebDataProducerHandleImpl(handle.Pass())),
+        input_(input) {}
+
+  Result final_result() const { return final_result_; }
+
+  void WriteMore() override {
+    WriteData();
+  }
+
+  void WriteData() {
+    if (!client_) {
+      client_.reset(new ClientImpl(this));
+      handle_->registerClient(client_.get());
+    }
+
+    Result rv = kOk;
+    while (!input_.empty()) {
+      void* buffer = nullptr;
+      size_t size;
+      rv = handle_->beginWrite(&buffer, kNone, &size);
+      if (rv != kOk)
+        break;
+      // In order to verify endWrite, we write at most one byte for each time.
+      size_t written_size = std::max(static_cast<size_t>(1), size);
+      std::copy(input_.begin(),
+                input_.begin() + written_size,
+                static_cast<char*>(buffer));
+      rv = handle_->endWrite(written_size);
+      if (rv != kOk)
+        break;
+      input_.erase(input_.begin(), input_.begin() + written_size);
+    }
+
+    if (rv == kShouldWait) {
+      // Wait a while...
+      return;
+    }
+
+    final_result_ = rv;
+
+    // The operation is done.
+    handle_ = nullptr;
+  }
+
+ private:
+  scoped_ptr<WebDataProducerHandle> handle_;
+  scoped_ptr<WebDataProducerHandle::Client> client_;
+  std::deque<char> input_;
+  Result final_result_;
+};
+
+class WebDataProducerHandleImplTest : public ::testing::Test {
+ public:
+  typedef WebDataProducerHandle::Result Result;
+
+  void SetUp() override {
+    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 = 4;
+
+    MojoResult result = mojo::CreateDataPipe(&options, &producer_, &consumer_);
+    ASSERT_EQ(MOJO_RESULT_OK, result);
+  }
+
+  // This function can be blocked if the associated consumer doesn't consume
+  // the data.
+  std::string ProduceData(size_t total_size) {
+    int index = 0;
+    std::string expected;
+    for (size_t i = 0; i < total_size; ++i) {
+      expected += static_cast<char>(index + 'a');
+      index = (37 * index + 11) % 26;
+    }
+    return expected;
+  }
+
+  std::string Read() {
+    std::string result;
+    const MojoWriteDataFlags kNone = MOJO_WRITE_DATA_FLAG_NONE;
+    MojoResult rv;
+    while (true) {
+      char buffer[16];
+      uint32_t size = sizeof(buffer);
+      rv = mojo::ReadDataRaw(consumer_.get(), buffer, &size, kNone);
+      if (rv == MOJO_RESULT_OK) {
+        result.insert(result.end(), buffer, buffer + size);
+      } else if (rv == MOJO_RESULT_FAILED_PRECONDITION) {
+        break;
+      } else if (rv != MOJO_RESULT_SHOULD_WAIT) {
+        // Something is wrong.
+        EXPECT_TRUE(false) << "mojo::ReadDataRaw returns an invalid value.";
+        return "error on reading";
+      }
+    }
+    return result;
+  }
+
+  mojo::ScopedDataPipeProducerHandle producer_;
+  mojo::ScopedDataPipeConsumerHandle consumer_;
+};
+
+TEST_F(WebDataProducerHandleImplTest, WriteData) {
+  std::string expected = ProduceData(24 * 1024);
+  std::deque<char> input(expected.begin(), expected.end());
+
+  auto operation =
+      make_scoped_ptr(new WriteDataOperation(producer_.Pass(), input));
+
+  base::Thread t("DataProducerHandle test thread");
+  ASSERT_TRUE(t.Start());
+
+  t.message_loop()->PostTask(
+      FROM_HERE,
+      base::Bind(&WriteDataOperation::WriteData,
+                 base::Unretained(operation.get())));
+
+  std::string actual = Read();
+  t.Stop();
+
+  EXPECT_EQ(WebDataProducerHandle::Ok, operation->final_result());
+  EXPECT_EQ(expected, actual);
+}
+
+TEST_F(WebDataProducerHandleImplTest, TwoPhaseWriteData) {
+  std::string expected = ProduceData(24 * 1024);
+  std::deque<char> input(expected.begin(), expected.end());
+
+  auto operation =
+      make_scoped_ptr(new TwoPhaseWriteDataOperation(producer_.Pass(), input));
+
+  base::Thread t("DataProducerHandle test thread");
+  ASSERT_TRUE(t.Start());
+
+  t.message_loop()->PostTask(
+      FROM_HERE,
+      base::Bind(&TwoPhaseWriteDataOperation::WriteData,
+                 base::Unretained(operation.get())));
+
+  std::string actual = Read();
+
+  t.Stop();
+
+  EXPECT_EQ(WebDataProducerHandle::Ok, operation->final_result());
+  EXPECT_EQ(expected, actual);
+}
+
+TEST_F(WebDataProducerHandleImplTest, WriteToClosedPipe) {
+  auto handle =
+      make_scoped_ptr(new WebDataProducerHandleImpl(producer_.Pass()));
+  const auto kNone = WebDataProducerHandleImpl::FlagNone;
+  const auto kAlreadyClosed = WebDataProducerHandleImpl::AlreadyClosed;
+
+  size_t written_size = 0;
+  void* buffer = nullptr;
+  size_t available = 0;
+
+  consumer_.reset();
+
+  EXPECT_EQ(kAlreadyClosed, handle->write(nullptr, 0, kNone, &written_size));
+  EXPECT_EQ(kAlreadyClosed, handle->beginWrite(&buffer, kNone, &available));
+}
+
+}  // namespace
+
+}  // namespace content