[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/basictypes.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 uint64 kReadDataSize = arraysize(kReadData) - 1;
+const char kReadData2[] = "read_alternate_data";
+const uint64 kReadData2Size = arraysize(kReadData2) - 1;
+const char kWriteData[] = "write_data";
+const uint64 kWriteDataSize = arraysize(kWriteData) - 1;
+
+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,

[akalin, 2013/10/15 00:56:07]

perhaps have DoInputStreamRead() return a std::string with the read data so you
can verify that it matches what you expect? That would catch bugs where the size
isn't changed but the contents are corrupted

[Nicolas Zea, 2013/10/15 20:07:38]

That's what output_buffer is used for. I don't return a string in order to
exersize the zero-copy semantics.

[akalin, 2013/10/15 23:24:15]

I see. What about returning a base::StringPiece?

[Nicolas Zea, 2013/10/16 01:20:40]

Done.

+                                              bool set_limit,

[akalin, 2013/10/15 00:56:07]

you don't seem to use set_limit. Remove?

[Nicolas Zea, 2013/10/15 20:07:38]

Done.

+                                              const void** output_buffer) {
+  uint64 total_bytes_read = 0;
+  const void* initial_buffer = NULL;
+  const void* buffer = NULL;
+  int size = 0;
+
+  do {
+    DCHECK(socket_input_stream_->GetState() == SocketInputStream::EMPTY ||
+           socket_input_stream_->GetState() == SocketInputStream::READY);
+    if (!socket_input_stream_->Next(&buffer, &size))
+      break;
+    total_bytes_read += size;
+    if (initial_buffer) { // Verify the buffer doesn't skip data.

[akalin, 2013/10/15 00:56:07]

two spaces before //

[Nicolas Zea, 2013/10/15 20:07:38]

Done.

+      EXPECT_EQ(static_cast<const uint8*>(initial_buffer) + total_bytes_read,
+                static_cast<const uint8*>(buffer) + size);
+    } else {
+      initial_buffer = buffer;
+    }
+  } while (total_bytes_read < bytes);
+
+  *output_buffer = 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) {

[akalin, 2013/10/15 00:56:07]

avoid depending on kWriteData explicitly and change this function to take a
StringPiece or something?

[Nicolas Zea, 2013/10/15 20:07:38]

Done.

+  DCHECK_EQ(socket_output_stream_->GetState(), SocketOutputStream::EMPTY);
+  uint64 total_bytes_written = 0;
+  void* buffer = NULL;
+  int size = 0;
+  const void* write_src = kWriteData;
+
+  do {
+    if (!socket_output_stream_->Next(&buffer, &size))
+      break;
+    uint64 bytes_to_write = (static_cast<uint64>(size) < bytes ? size : bytes);
+    memcpy(buffer,
+           static_cast<const uint8*>(write_src) + total_bytes_written,
+           bytes_to_write);
+    if (bytes_to_write < static_cast<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;
+  if (socket_output_stream_->Flush(run_loop.QuitClosure()) ==

[akalin, 2013/10/15 23:24:15]

if Next() returns false in the above loop, isn't it incorrect to call Flush()
here?

+          net::ERR_IO_PENDING) {
+    run_loop.Run();
+  }
+
+  return total_bytes_written;

[akalin, 2013/10/15 00:56:07]

Do you need to return total_bytes_written? Perhaps EXPECT_EQ(bytes) here?

[Nicolas Zea, 2013/10/15 20:07:38]

That's not true in the cases where an error is encountered. I think it's cleaner
to let the test itself decide whether to enforce that.

+}
+
+void GCMSocketStreamTest::WaitForData(int msg_size) {
+  while (input_stream()->ByteCount() < msg_size) {
+    base::RunLoop run_loop;
+    if (input_stream()->Refresh(run_loop.QuitClosure(),
+                                msg_size) == net::ERR_IO_PENDING) {

[akalin, 2013/10/15 00:56:07]

don't you mean msg_size - input_stream()->ByteCount()? Test didn't catch this?

[Nicolas Zea, 2013/10/15 20:07:38]

This method is waiting for |msg_size| new bytes, not |msg_size| total bytes
(note that Refresh treats that value as the byte_limit for the Socket::Read
call). That way, I can do WaitForData(kReadDataSize), followed by
WaitForData(kReadData2Size), in order to receive two messages. So I think this
is correct as is.

[akalin, 2013/10/15 23:24:15]

Right, but that's only for the first call to Refresh, right? What I mean is
that, if you're waiting for msg_size new bytes, shouldn't you have something
like:

size_t old_size = input_stream()->ByteCount();
while (input_stream()->ByteCount() < old_size + msg_size) {
  ...->Refresh(..., old_size + msg_size - input_stream()->ByteCount()...)
}

? Perhaps I'm missing something.

[Nicolas Zea, 2013/10/16 01:20:40]

Ah, the thing is that ByteCount is actually BytesConsumed() - next_pos_. In
other words, it's returning the amount of unread data. That said, you're right
that the refresh call is setting too high a limit, it should be msg_size -
ByteCount(). This doesn't really affect correctness, as the input stream is
built to be able to process correctly when there is more data than expected. 

One thing this does make me realize though is that, according to the
documentation, ByteCount is actually supposed to be the amount read via Next(),
not the bytes remaining. As it happens it's not used by CodedInputStream, so
that wasn't caught. I've gone ahead and made that change for consistency though,
and added an UnreadByteCount method for what I actually wanted.

+      run_loop.Run();
+    }
+    if (input_stream()->GetState() == 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,
+                                        kReadDataSize)),
+              WriteList());
+
+  const void* output = NULL;
+  WaitForData(kReadDataSize);
+  ASSERT_EQ(kReadDataSize,
+            DoInputStreamRead(kReadDataSize, false, &output));
+  ASSERT_EQ(std::string(kReadData),
+            std::string(static_cast<const char*>(output), kReadDataSize));
+}
+
+// A read that comes in two parts.
+TEST_F(GCMSocketStreamTest, ReadPartialDataSync) {
+  size_t first_read_len = kReadDataSize/2;
+  size_t second_read_len = kReadDataSize - 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(kReadDataSize);
+  ASSERT_EQ(kReadDataSize,
+            DoInputStreamRead(kReadDataSize, false, &output));
+  ASSERT_EQ(std::string(kReadData),
+            std::string(static_cast<const char*>(output), kReadDataSize));
+}
+
+// A read where no data is available at first (IO_PENDING will be returned).
+TEST_F(GCMSocketStreamTest, ReadAsync) {
+  size_t first_read_len = kReadDataSize/2;
+  size_t second_read_len = kReadDataSize - 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(kReadDataSize);
+  ASSERT_EQ(kReadDataSize,
+            DoInputStreamRead(kReadDataSize, false, &output));
+  ASSERT_EQ(std::string(kReadData),
+            std::string(static_cast<const char*>(output), kReadDataSize));
+}
+
+// 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(),
+                                        long_data.size())),
+              WriteList());
+  const void* output = NULL;
+
+  WaitForData(kReadDataSize);
+  ASSERT_EQ(kReadDataSize,
+            DoInputStreamRead(kReadDataSize, false, &output));
+  ASSERT_EQ(std::string(kReadData, kReadDataSize),
+            std::string(static_cast<const char*>(output), kReadDataSize));
+
+  WaitForData(kReadData2Size);
+  ASSERT_EQ(kReadData2Size,
+            DoInputStreamRead(kReadData2Size, false, &output));
+  ASSERT_EQ(std::string(kReadData2, kReadDataSize),
+            std::string(static_cast<const char*>(output), kReadDataSize));
+}
+
+// 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(),
+                                        long_data.size())),
+              WriteList());
+
+  const void* output = NULL;
+  WaitForData(kReadDataSize);
+  ASSERT_EQ(kReadDataSize,
+            DoInputStreamRead(kReadDataSize, false, &output));
+  ASSERT_EQ(std::string(kReadData, kReadDataSize),
+            std::string(static_cast<const char*>(output), kReadDataSize));
+
+  input_stream()->RebuildBuffer();
+  WaitForData(kReadData2Size);
+  ASSERT_EQ(kReadData2Size,
+            DoInputStreamRead(kReadData2Size, false, &output));
+  ASSERT_EQ(std::string(kReadData2, kReadData2Size),
+            std::string(static_cast<const char*>(output), kReadData2Size));
+}
+
+// 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(kReadDataSize);
+  ASSERT_EQ(SocketInputStream::CLOSED, input_stream()->GetState());
+  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(kReadDataSize);
+  ASSERT_EQ(SocketInputStream::CLOSED, input_stream()->GetState());
+  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,
+                                          kWriteDataSize)));
+  ASSERT_EQ(kWriteDataSize, DoOutputStreamWrite(kWriteDataSize));
+}
+
+// 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(),
+                                      part1.size()));
+  write_list.push_back(net::MockWrite(net::SYNCHRONOUS,
+                                      part2.c_str(),
+                                      part2.size()));
+  BuildSocket(ReadList(), write_list);
+  ASSERT_EQ(kWriteDataSize, DoOutputStreamWrite(kWriteDataSize));
+}
+
+// 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(),
+                                      part1.size()));
+  write_list.push_back(net::MockWrite(net::ASYNC,
+                                      part2.c_str(),
+                                      part2.size()));
+  BuildSocket(ReadList(), write_list);
+  ASSERT_EQ(kWriteDataSize, DoOutputStreamWrite(kWriteDataSize));
+}
+
+// Write a message then read a message.
+TEST_F(GCMSocketStreamTest, WriteThenRead) {
+  BuildSocket(ReadList(1, net::MockRead(net::SYNCHRONOUS,
+                                        kReadData,
+                                        kReadDataSize)),
+              WriteList(1, net::MockWrite(net::SYNCHRONOUS,
+                                          kWriteData,
+                                          kWriteDataSize)));
+
+  ASSERT_EQ(kWriteDataSize, DoOutputStreamWrite(kWriteDataSize));
+
+  const void* output = NULL;
+  WaitForData(kReadDataSize);
+  ASSERT_EQ(kReadDataSize,
+            DoInputStreamRead(kReadDataSize, false, &output));
+  ASSERT_EQ(std::string(kReadData),
+            std::string(static_cast<const char*>(output), kReadDataSize));
+}
+
+// Read a message then write a message.
+TEST_F(GCMSocketStreamTest, ReadThenWrite) {
+  BuildSocket(ReadList(1, net::MockRead(net::SYNCHRONOUS,
+                                        kReadData,
+                                        kReadDataSize)),
+              WriteList(1, net::MockWrite(net::SYNCHRONOUS,
+                                          kWriteData,
+                                          kWriteDataSize)));
+
+  const void* output = NULL;
+  WaitForData(kReadDataSize);
+  ASSERT_EQ(kReadDataSize,
+            DoInputStreamRead(kReadDataSize, false, &output));
+  ASSERT_EQ(std::string(kReadData),
+            std::string(static_cast<const char*>(output), kReadDataSize));
+
+  ASSERT_EQ(kWriteDataSize, DoOutputStreamWrite(kWriteDataSize));
+}
+
+// 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(kWriteDataSize);
+  ASSERT_EQ(SocketOutputStream::CLOSED, output_stream()->GetState());
+  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(kWriteDataSize);
+  ASSERT_EQ(SocketOutputStream::CLOSED, output_stream()->GetState());
+  ASSERT_EQ(net::ERR_CONNECTION_CLOSED, output_stream()->last_error());
+}
+
+}  // namespace
+}  // namespace gcm