[GCM] Initial work to set up directory structure and introduce socket integration

google_apis/gcm/base/socket_stream_unittest.cc

+// Copyright (c) 2013 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 "google_apis/gcm/base/socket_stream.h"
+
+#include "base/bind.h"
+#include "base/memory/scoped_ptr.h"
+#include "base/run_loop.h"
+#include "net/socket/socket_test_util.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace gcm {
+namespace {
+
+typedef std::vector<net::MockRead> ReadList;
+typedef std::vector<net::MockWrite> WriteList;
+
+const char kReadData[] = "read_data";
+const char kReadData2[] = "read_alternate_data";
+const char kWriteData[] = "write_data";
+
+class GCMSocketStreamTest : public testing::Test {
+ public:
+  GCMSocketStreamTest();
+  virtual ~GCMSocketStreamTest();
+
+  // Build a socket with the expected reads and writes.
+  void BuildSocket(const ReadList& read_list, const WriteList& write_list);
+
+  // Pump the message loop until idle.
+  void PumpLoop();
+
+  // Simulates a google::protobuf::io::CodedInputStream read.
+  uint64 DoInputStreamRead(uint64 bytes,
+                           bool set_limit,
+                           const void** output_buffer);
+  // Simulates a google::protobuf::io::CodedOutputStream write.
+  uint64 DoOutputStreamWrite(uint64 bytes);
+
+  // Synchronous Refresh wrapper.
+  void WaitForData(int msg_size);
+
+  base::MessageLoop* message_loop() { return &message_loop_; };
+  net::DelayedSocketData* data_provider() { return data_provider_.get(); }
+  SocketInputStream* input_stream() { return socket_input_stream_.get(); }
+  SocketOutputStream* output_stream() { return socket_output_stream_.get(); }
+  net::StreamSocket* socket() { return socket_.get(); }
+
+ private:
+  void OpenConnection();
+  void ResetInputStream();
+  void ResetOutputStream();
+
+  void ConnectCallback(int result);
+
+  // SocketStreams and their data providers.
+  ReadList mock_reads_;
+  WriteList mock_writes_;
+  scoped_ptr<net::DelayedSocketData> data_provider_;
+  scoped_ptr<SocketInputStream> socket_input_stream_;
+  scoped_ptr<SocketOutputStream> socket_output_stream_;
+
+  // net:: components.
+  scoped_ptr<net::StreamSocket> socket_;
+  net::MockClientSocketFactory socket_factory_;
+  net::AddressList address_list_;
+
+  base::MessageLoopForIO message_loop_;
+};
+
+GCMSocketStreamTest::GCMSocketStreamTest() {
+  net::IPAddressNumber ip_number;
+  net::ParseIPLiteralToNumber("127.0.0.1", &ip_number);
+  address_list_ = net::AddressList::CreateFromIPAddress(ip_number, 5228);
+}
+
+GCMSocketStreamTest::~GCMSocketStreamTest() {}
+
+void GCMSocketStreamTest::BuildSocket(const ReadList& read_list,
+                                      const WriteList& write_list) {
+  mock_reads_ = read_list;
+  mock_writes_ = write_list;
+  data_provider_.reset(
+      new net::DelayedSocketData(0,
+                                 &(mock_reads_[0]), mock_reads_.size(),
+                                 &(mock_writes_[0]), mock_writes_.size()));
+  socket_factory_.AddSocketDataProvider(data_provider_.get());
+  OpenConnection();
+  ResetInputStream();
+  ResetOutputStream();
+}
+
+void GCMSocketStreamTest::PumpLoop() {
+  base::RunLoop run_loop;
+  run_loop.RunUntilIdle();
+}
+
+uint64 GCMSocketStreamTest::DoInputStreamRead(uint64 bytes,
+                                              bool set_limit,
+                                              const void** output_buffer) {
+  uint64 total_bytes_read = 0;
+  const char* initial_buffer = NULL;
+  const char* buffer = NULL;
+  int size = 0;
+
+  do {
+    DCHECK(socket_input_stream_->state() == SocketInputStream::EMPTY ||
+           socket_input_stream_->state() == SocketInputStream::READY);
+    if (!socket_input_stream_->Next((const void **)&buffer, &size))
+      break;
+    total_bytes_read += size;
+    if (initial_buffer != NULL)  // Verify the buffer doesn't skip data.
+      EXPECT_EQ(initial_buffer + total_bytes_read, buffer + size);
+    else
+      initial_buffer = buffer;
+  } while (total_bytes_read < bytes);
+
+  *output_buffer = (const void*)initial_buffer;
+
+  if (total_bytes_read > bytes) {
+    socket_input_stream_->BackUp(total_bytes_read - bytes);
+    total_bytes_read = bytes;
+  }
+
+  return total_bytes_read;
+}
+
+uint64 GCMSocketStreamTest::DoOutputStreamWrite(uint64 bytes) {
+  DCHECK_EQ(socket_output_stream_->state(), SocketOutputStream::EMPTY);
+  uint64 total_bytes_written = 0;
+  char* buffer = NULL;
+  int size = 0;
+  const char* write_src = kWriteData;
+
+  do {
+    if (!socket_output_stream_->Next((void **)&buffer, &size))
+      break;
+    uint64 bytes_to_write = ((uint64)size < bytes ? size : bytes);
+    memcpy(buffer, &write_src[total_bytes_written], bytes_to_write);
+    if (bytes_to_write < (uint64)size)
+      socket_output_stream_->BackUp(size - bytes_to_write);
+    total_bytes_written += bytes_to_write;
+  } while (total_bytes_written < bytes);
+
+  base::RunLoop run_loop;
+  socket_output_stream_->Flush(run_loop.QuitClosure());
+  run_loop.Run();
+
+  return total_bytes_written;
+}
+
+void ReadCallback(base::RunLoop* run_loop) {
+  run_loop->Quit();
+}
+
+void GCMSocketStreamTest::WaitForData(int msg_size) {
+  while (input_stream()->ByteCount() < msg_size) {
+    base::RunLoop run_loop;
+    input_stream()->Refresh(base::Bind(&ReadCallback,
+                                        base::Unretained(&run_loop)),
+                            msg_size);
+    run_loop.Run();
+    if (input_stream()->state() == SocketInputStream::CLOSED)
+      return;
+  }
+}
+
+void GCMSocketStreamTest::OpenConnection() {
+  socket_ = socket_factory_.CreateTransportClientSocket(
+      address_list_, NULL, net::NetLog::Source());
+  socket_->Connect(
+    base::Bind(&GCMSocketStreamTest::ConnectCallback,
+               base::Unretained(this)));
+  PumpLoop();
+}
+
+void GCMSocketStreamTest::ConnectCallback(int result) {}
+
+void GCMSocketStreamTest::ResetInputStream() {
+  DCHECK(socket_.get());
+  socket_input_stream_.reset(
+      new SocketInputStream(socket_.get()));
+}
+
+void GCMSocketStreamTest::ResetOutputStream() {
+  DCHECK(socket_.get());
+  socket_output_stream_.reset(new SocketOutputStream(socket_.get()));
+}
+
+// A read where all data is already available.
+TEST_F(GCMSocketStreamTest, ReadDataSync) {
+  BuildSocket(ReadList(1, net::MockRead(net::SYNCHRONOUS, kReadData)),
+              WriteList());
+
+  const void* output = NULL;
+  WaitForData(strlen(kReadData));
+  ASSERT_EQ(strlen(kReadData),
+            DoInputStreamRead(strlen(kReadData), false, &output));
+  ASSERT_EQ(std::string(kReadData),
+            std::string((const char*)output, strlen(kReadData)));
+}
+
+// A read that comes in two parts.
+TEST_F(GCMSocketStreamTest, ReadPartialDataSync) {
+  size_t first_read_len = strlen(kReadData)/2;
+  size_t second_read_len = strlen(kReadData) - first_read_len;
+  ReadList read_list;
+  read_list.push_back(
+      net::MockRead(net::SYNCHRONOUS,
+                    kReadData, first_read_len));
+  read_list.push_back(
+      net::MockRead(net::SYNCHRONOUS,
+                    &kReadData[first_read_len], second_read_len));
+  BuildSocket(read_list, WriteList());
+
+  const void* output = NULL;
+  WaitForData(strlen(kReadData));
+  ASSERT_EQ(strlen(kReadData),
+            DoInputStreamRead(strlen(kReadData), false, &output));
+  ASSERT_EQ(std::string(kReadData),
+            std::string((const char*)output, strlen(kReadData)));
+}
+
+// A read where no data is available at first (IO_PENDING will be returned).
+TEST_F(GCMSocketStreamTest, ReadAsync) {
+  size_t first_read_len = strlen(kReadData)/2;
+  size_t second_read_len = strlen(kReadData) - first_read_len;
+  ReadList read_list;
+  read_list.push_back(
+      net::MockRead(net::SYNCHRONOUS, net::ERR_IO_PENDING));
+  read_list.push_back(
+      net::MockRead(net::ASYNC, kReadData, first_read_len));
+  read_list.push_back(
+      net::MockRead(net::ASYNC, &kReadData[first_read_len], second_read_len));
+  BuildSocket(read_list, WriteList());
+
+  const void* output = NULL;
+  base::MessageLoop::current()->PostTask(
+      FROM_HERE,
+      base::Bind(&net::DelayedSocketData::ForceNextRead,
+                 base::Unretained(data_provider())));
+  WaitForData(strlen(kReadData));
+  ASSERT_EQ(strlen(kReadData),
+            DoInputStreamRead(strlen(kReadData), false, &output));
+  ASSERT_EQ(std::string(kReadData),
+            std::string((const char*)output, strlen(kReadData)));
+}
+
+// Simulate two packets arriving at once. Read them in two separate calls.
+TEST_F(GCMSocketStreamTest, TwoReadsAtOnce) {
+  std::string long_data = std::string(kReadData) + std::string(kReadData2);
+  BuildSocket(ReadList(1, net::MockRead(net::SYNCHRONOUS, long_data.c_str())),
+              WriteList());
+  const void* output = NULL;
+
+  WaitForData(strlen(kReadData));
+  ASSERT_EQ(strlen(kReadData),
+            DoInputStreamRead(strlen(kReadData), false, &output));
+  ASSERT_EQ(std::string(kReadData, strlen(kReadData)),
+            std::string((const char*)output, strlen(kReadData)));
+
+  WaitForData(strlen(kReadData2));
+  ASSERT_EQ(strlen(kReadData2),
+            DoInputStreamRead(strlen(kReadData2), false, &output));
+  ASSERT_EQ(std::string(kReadData2, strlen(kReadData)),
+            std::string((const char*)output, strlen(kReadData)));
+}
+
+// Simulate two packets arriving at once. Read them in two calls separated
+// by a Rebuild.
+TEST_F(GCMSocketStreamTest, TwoReadsAtOnceWithRebuild) {
+  std::string long_data = std::string(kReadData) + std::string(kReadData2);
+  BuildSocket(ReadList(1, net::MockRead(net::SYNCHRONOUS, long_data.c_str())),
+              WriteList());
+
+  const void* output = NULL;
+  WaitForData(strlen(kReadData));
+  ASSERT_EQ(strlen(kReadData),
+            DoInputStreamRead(strlen(kReadData), false, &output));
+  ASSERT_EQ(std::string(kReadData, strlen(kReadData)),
+            std::string((const char*)output, strlen(kReadData)));
+
+  input_stream()->RebuildBuffer();
+  WaitForData(strlen(kReadData2));
+  ASSERT_EQ(strlen(kReadData2),
+            DoInputStreamRead(strlen(kReadData2), false, &output));
+  ASSERT_EQ(std::string(kReadData2, strlen(kReadData2)),
+            std::string((const char*)output, strlen(kReadData2)));
+}
+
+// Simulate a read that is aborted.
+TEST_F(GCMSocketStreamTest, ReadError) {
+  int result = net::ERR_ABORTED;
+  BuildSocket(ReadList(1, net::MockRead(net::SYNCHRONOUS, result)),
+              WriteList());
+
+  WaitForData(strlen(kReadData));
+  ASSERT_EQ(SocketInputStream::CLOSED, input_stream()->state());
+  ASSERT_EQ(result, input_stream()->last_error());
+}
+
+// Simulate a read after the connection is closed.
+TEST_F(GCMSocketStreamTest, ReadDisconnected) {
+  BuildSocket(ReadList(), WriteList());
+  socket()->Disconnect();
+  WaitForData(strlen(kReadData));
+  ASSERT_EQ(SocketInputStream::CLOSED, input_stream()->state());
+  ASSERT_EQ(net::ERR_CONNECTION_CLOSED, input_stream()->last_error());
+}
+
+// Write a full message in one go.
+TEST_F(GCMSocketStreamTest, WriteFull) {
+  BuildSocket(ReadList(),
+              WriteList(1, net::MockWrite(net::SYNCHRONOUS, kWriteData)));
+  ASSERT_EQ(strlen(kWriteData), DoOutputStreamWrite(strlen(kWriteData)));
+}
+
+// Write a message in two go's.
+TEST_F(GCMSocketStreamTest, WritePartial) {
+  std::string full_data = std::string(kWriteData);
+  std::string part1 = full_data.substr(0, full_data.length()/2);
+  std::string part2 = full_data.substr(full_data.length()/2);
+  WriteList write_list;
+  write_list.push_back(net::MockWrite(net::SYNCHRONOUS, part1.c_str()));
+  write_list.push_back(net::MockWrite(net::SYNCHRONOUS, part2.c_str()));
+  BuildSocket(ReadList(), write_list);
+  ASSERT_EQ(strlen(kWriteData), DoOutputStreamWrite(strlen(kWriteData)));
+}
+
+// Write a message completely asynchronously (returns IO_PENDING before
+// finishing the write in two go's).
+TEST_F(GCMSocketStreamTest, WriteNone) {
+  std::string full_data = std::string(kWriteData);
+  std::string part1 = full_data.substr(0, full_data.length()/2);
+  std::string part2 = full_data.substr(full_data.length()/2);
+  WriteList write_list;
+  write_list.push_back(net::MockWrite(net::ASYNC, part1.c_str()));
+  write_list.push_back(net::MockWrite(net::ASYNC, part2.c_str()));
+  BuildSocket(ReadList(), write_list);
+  ASSERT_EQ(strlen(kWriteData), DoOutputStreamWrite(strlen(kWriteData)));
+}
+
+// Write a message then read a message.
+TEST_F(GCMSocketStreamTest, WriteThenRead) {
+  BuildSocket(ReadList(1, net::MockRead(kReadData)),
+              WriteList(1, net::MockWrite(kWriteData)));
+
+  ASSERT_EQ(strlen(kWriteData), DoOutputStreamWrite(strlen(kWriteData)));
+
+  const void* output = NULL;
+  WaitForData(strlen(kReadData));
+  ASSERT_EQ(strlen(kReadData),
+            DoInputStreamRead(strlen(kReadData), false, &output));
+  ASSERT_EQ(std::string(kReadData),
+            std::string((const char*)output, strlen(kReadData)));
+}
+
+// Read a message then write a message.
+TEST_F(GCMSocketStreamTest, ReadThenWrite) {
+  BuildSocket(ReadList(1, net::MockRead(kReadData)),
+              WriteList(1, net::MockWrite(kWriteData)));
+
+  const void* output = NULL;
+  WaitForData(strlen(kReadData));
+  ASSERT_EQ(strlen(kReadData),
+            DoInputStreamRead(strlen(kReadData), false, &output));
+  ASSERT_EQ(std::string(kReadData),
+            std::string((const char*)output, strlen(kReadData)));
+
+  ASSERT_EQ(strlen(kWriteData), DoOutputStreamWrite(strlen(kWriteData)));
+}
+
+// Simulate a write that gets aborted.
+TEST_F(GCMSocketStreamTest, WriteError) {
+  int result = net::ERR_ABORTED;
+  BuildSocket(ReadList(),
+              WriteList(1, net::MockWrite(net::SYNCHRONOUS, result)));
+  DoOutputStreamWrite(strlen(kWriteData));
+  ASSERT_EQ(SocketOutputStream::CLOSED, output_stream()->state());
+  ASSERT_EQ(result, output_stream()->last_error());
+}
+
+// Simulate a write after the connection is closed.
+TEST_F(GCMSocketStreamTest, WriteDisconnected) {
+  BuildSocket(ReadList(), WriteList());
+  socket()->Disconnect();
+  DoOutputStreamWrite(strlen(kWriteData));
+  ASSERT_EQ(SocketOutputStream::CLOSED, output_stream()->state());
+  ASSERT_EQ(net::ERR_CONNECTION_CLOSED, output_stream()->last_error());
+}
+
+}  // namespace
+}  // namespace gcm