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

google_apis/gcm/base/socket_stream_unittest.cc

Index: google_apis/gcm/base/socket_stream_unittest.cc
diff --git a/google_apis/gcm/base/socket_stream_unittest.cc b/google_apis/gcm/base/socket_stream_unittest.cc
new file mode 100644
index 0000000000000000000000000000000000000000..d81f70239a825d22c691def2f8047115bc37b445
--- /dev/null
+++ b/google_apis/gcm/base/socket_stream_unittest.cc
@@ -0,0 +1,401 @@
+// 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 "base/test/test_timeouts.h"
+#include "google_apis/gcm/base/socket_stream_test_data_provider.h"
+#include "net/socket/socket_test_util.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace gcm {
+namespace {
+
+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();
+
+  void PumpLoop();
+
+  void OpenConnection();
+  void ResetInputStream();
+  void ResetOutputStream();
+
+  base::MessageLoop* message_loop() { return &message_loop_; };
+  SocketStreamTestDataProvider* data_provider() { return &data_provider_; }
+  SocketInputStream* input_stream() { return socket_input_stream_.get(); }
+  SocketOutputStream* output_stream() { return socket_output_stream_.get(); }
+  net::StreamSocket* socket() { return socket_.get(); }
+
+  // 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);
+
+ private:
+  void ConnectCallback(int result);
+
+  // SocketStreams and their data provider.
+  SocketStreamTestDataProvider 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);
+
+  socket_factory_.AddSocketDataProvider(&data_provider_);
+
+  OpenConnection();
+  ResetInputStream();
+  ResetOutputStream();
+}
+
+GCMSocketStreamTest::~GCMSocketStreamTest() {}
+
+void GCMSocketStreamTest::PumpLoop() {
+  base::RunLoop run_loop;
+  run_loop.RunUntilIdle();
+}
+
+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(TestTimeouts::tiny_timeout(),
+                            socket_.get()));
+}
+
+void GCMSocketStreamTest::ResetOutputStream() {
+  DCHECK(socket_.get());
+  socket_output_stream_.reset(new SocketOutputStream(socket_.get()));
+}
+
+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;
+
+  if (set_limit) {
+    socket_input_stream_->SetLimit(socket_input_stream_->GetCurrentPosition() +
+                                   bytes);
+  }
+
+  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;
+}
+
+// A read where all data is already available.
+TEST_F(GCMSocketStreamTest, ReadPendingData) {
+  data_provider()->AddRead(net::MockRead(kReadData));
+  const void* output = NULL;
+  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, ReadPartialData) {
+  size_t first_read_len = strlen(kReadData)/2;
+  size_t second_read_len = strlen(kReadData) - first_read_len;
+  data_provider()->AddRead(
+      net::MockRead(net::ASYNC, kReadData, first_read_len));
+  data_provider()->AddRead(
+      net::MockRead(net::ASYNC, &kReadData[first_read_len], second_read_len));
+  const void* output = NULL;
+  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, ReadNoData) {
+  size_t first_read_len = strlen(kReadData)/2;
+  size_t second_read_len = strlen(kReadData) - first_read_len;
+  message_loop()->PostTask(
+      FROM_HERE,
+      base::Bind(&SocketStreamTestDataProvider::AddRead,
+                 base::Unretained(data_provider()),
+                 net::MockRead(net::ASYNC, kReadData, first_read_len)));
+  message_loop()->PostTask(
+      FROM_HERE,
+      base::Bind(&SocketStreamTestDataProvider::AddRead,
+                 base::Unretained(data_provider()),
+                 net::MockRead(net::ASYNC,
+                               &kReadData[first_read_len],
+                               second_read_len)));
+  const void* output = NULL;
+  ASSERT_EQ(strlen(kReadData),
+            DoInputStreamRead(strlen(kReadData), false, &output));
+  ASSERT_EQ(std::string(kReadData),
+            std::string((const char*)output, strlen(kReadData)));
+}
+
+// A read where a second message arrives at the same time. The second message
+// is then read.
+TEST_F(GCMSocketStreamTest, TwoReadsAtOnce) {
+  std::string long_data = std::string(kReadData) + std::string(kReadData2);
+  data_provider()->AddRead(net::MockRead(long_data.c_str()));
+  const void* output = NULL;
+  ASSERT_EQ(strlen(kReadData),
+            DoInputStreamRead(strlen(kReadData), false, &output));
+  ASSERT_EQ(std::string(kReadData, strlen(kReadData)),
+            std::string((const char*)output, strlen(kReadData)));
+  ASSERT_EQ(strlen(kReadData2),
+            DoInputStreamRead(strlen(kReadData2), false, &output));
+  ASSERT_EQ(std::string(kReadData2, strlen(kReadData)),
+            std::string((const char*)output, strlen(kReadData)));
+}
+
+// A read where a second message arrives at the same time. The input stream
+// is reset before the second message is read.
+TEST_F(GCMSocketStreamTest, TwoReadsAtOnceWithReset) {
+  std::string long_data = std::string(kReadData) + std::string(kReadData2);
+  data_provider()->AddRead(net::MockRead(long_data.c_str()));
+  const void* output = NULL;
+  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()->Reset();
+  ASSERT_EQ(strlen(kReadData2),
+            DoInputStreamRead(strlen(kReadData2), false, &output));
+  ASSERT_EQ(std::string(kReadData2, strlen(kReadData2)),
+            std::string((const char*)output, strlen(kReadData2)));
+}
+
+// A read where a second message arrives at the same time. The second message
+// is then read. A limit is set on each read.
+TEST_F(GCMSocketStreamTest, TwoReadsAtOnceWithLimit) {
+  std::string long_data = std::string(kReadData) + std::string(kReadData2);
+  data_provider()->AddRead(net::MockRead(net::ASYNC,
+                                         long_data.c_str(),
+                                         long_data.size()));
+  const void* output = NULL;
+  ASSERT_EQ(strlen(kReadData),
+            DoInputStreamRead(strlen(kReadData), true, &output));
+  ASSERT_EQ(std::string(kReadData, strlen(kReadData)),
+            std::string((const char*)output, strlen(kReadData)));
+  ASSERT_EQ(strlen(kReadData2),
+            DoInputStreamRead(strlen(kReadData2), true, &output));
+  ASSERT_EQ(std::string(kReadData2, strlen(kReadData2)),
+            std::string((const char*)output, strlen(kReadData2)));
+}
+
+// A read where a second message arrives at the same time, but a reset happens
+// between reads. A limit is set on each read.
+TEST_F(GCMSocketStreamTest, TwoReadsAtOnceWithLimitAndReset) {
+  std::string long_data = std::string(kReadData) + std::string(kReadData2);
+  data_provider()->AddRead(net::MockRead(long_data.c_str()));
+  const void* output = NULL;
+  ASSERT_EQ(strlen(kReadData),
+            DoInputStreamRead(strlen(kReadData), true, &output));
+  ASSERT_EQ(std::string(kReadData, strlen(kReadData)),
+            std::string((const char*)output, strlen(kReadData)));
+  input_stream()->Reset();
+  ASSERT_EQ(strlen(kReadData2),
+            DoInputStreamRead(strlen(kReadData2), true, &output));
+  ASSERT_EQ(std::string(kReadData2, strlen(kReadData2)),
+            std::string((const char*)output, strlen(kReadData2)));
+}
+
+// Simulate a read timing out.
+TEST_F(GCMSocketStreamTest, ReadTimeout) {
+  int result = net::ERR_TIMED_OUT;
+  std::string original_data = std::string(kReadData);
+  std::string partial_data = original_data.substr(original_data.size()/2);
+  const void* output = NULL;
+
+  // Send part of the data.
+  data_provider()->AddRead(net::MockRead(net::ASYNC,
+                                         partial_data.c_str(),
+                                         partial_data.length()));
+  ASSERT_EQ(partial_data.length(),
+            DoInputStreamRead(original_data.length(), true, &output));
+  ASSERT_EQ(SocketInputStream::CLOSED, input_stream()->state());
+  ASSERT_EQ(result, input_stream()->last_error());
+
+  // Send the rest of the data. Should have no effect.
+  data_provider()->AddRead(
+      net::MockRead(net::ASYNC,
+                    &original_data.c_str()[partial_data.size()],
+                    original_data.size() - partial_data.size()));
+  ASSERT_EQ(SocketInputStream::CLOSED, input_stream()->state());
+  ASSERT_EQ(result, input_stream()->last_error());
+}
+
+// Simulate a read that is aborted.
+TEST_F(GCMSocketStreamTest, ReadError) {
+  int result = net::ERR_ABORTED;
+  const void* output = NULL;
+  data_provider()->AddRead(net::MockRead(net::SYNCHRONOUS, result));
+  ASSERT_EQ(0U, DoInputStreamRead(strlen(kReadData), true, &output));
+  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) {
+  socket()->Disconnect();
+  const void* output = NULL;
+  ASSERT_EQ(0U, DoInputStreamRead(strlen(kReadData), true, &output));
+  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) {
+  data_provider()->AddWrite(net::MockWrite(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);
+  data_provider()->AddWrite(net::MockWrite(part1.c_str()));
+  data_provider()->AddWrite(net::MockWrite(part2.c_str()));
+  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);
+  message_loop()->PostTask(
+      FROM_HERE,
+      base::Bind(&SocketStreamTestDataProvider::AddWrite,
+                 base::Unretained(data_provider()),
+                 net::MockWrite(part1.c_str())));
+  message_loop()->PostTask(
+      FROM_HERE,
+      base::Bind(&SocketStreamTestDataProvider::AddWrite,
+                 base::Unretained(data_provider()),
+                 net::MockWrite(part2.c_str())));
+  ASSERT_EQ(strlen(kWriteData), DoOutputStreamWrite(strlen(kWriteData)));
+}
+
+// Write a message then read a message.
+TEST_F(GCMSocketStreamTest, WriteThenRead) {
+  data_provider()->AddWrite(net::MockWrite(kWriteData));
+  data_provider()->AddRead(net::MockRead(kReadData));
+  ASSERT_EQ(strlen(kWriteData), DoOutputStreamWrite(strlen(kWriteData)));
+  const void* output = NULL;
+  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) {
+  data_provider()->AddRead(net::MockRead(kReadData));
+  data_provider()->AddWrite(net::MockWrite(kWriteData));
+  const void* output = NULL;
+  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;
+  data_provider()->AddWrite(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) {
+  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