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

google_apis/gcm/base/socket_stream.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/callback.h"
+#include "base/message_loop/message_loop.h"
+#include "net/base/io_buffer.h"
+#include "net/socket/stream_socket.h"
+
+namespace gcm {
+
+namespace {
+
+// TODO(zea): consider having dynamically-sized buffers if this becomes too
+// expensive.
+const uint32 kDefaultBufferSize = 8*1024;
+
+}  // namespace
+
+SocketInputStream::SocketInputStream(net::StreamSocket* socket)
+    : socket_(socket),
+      io_buffer_(new net::IOBuffer(kDefaultBufferSize)),
+      read_buffer_(new net::DrainableIOBuffer(io_buffer_.get(),
+                                              kDefaultBufferSize)),
+      next_pos_(0),
+      last_error_(net::OK),
+      state_(EMPTY),
+      weak_ptr_factory_(this) {
+  DCHECK(socket->IsConnected());
+}
+
+SocketInputStream::~SocketInputStream() {
+}
+
+bool SocketInputStream::Next(const void** data, int* size) {
+  DCHECK_NE(state_, CLOSED);
+  DCHECK_NE(state_, READING);
+
+  if (state_ == EMPTY) {
+    DVLOG(1) << "No unread data remaining, ending read.";
+    return false;
+  }
+
+  DCHECK_EQ(state_, READY)
+      << " Input stream must have pending data before reading.";
+  DCHECK_NE(read_buffer_->BytesConsumed(), next_pos_);

[akalin, 2013/10/10 08:47:40]

don't you want something stronger, i.e. DCHECK_LT(next_pos_,
read_buffer_->BytesConsumed())?

[Nicolas Zea, 2013/10/11 01:14:30]

Done.

+  *data = io_buffer_->data() + next_pos_;
+  *size = read_buffer_->BytesConsumed() - next_pos_;
+  next_pos_ = read_buffer_->BytesConsumed();
+  state_ = EMPTY;
+  DVLOG(1) << "Consuming " << *size << " bytes in input buffer.";
+  return true;
+}
+
+void SocketInputStream::BackUp(int count) {
+  DCHECK(state_ == READY || state_ == EMPTY);
+  DCHECK_GE(count, 0);

[akalin, 2013/10/10 08:47:40]

perhaps change this to _GT? if not, then READY might not be the correct state
below, right

[Nicolas Zea, 2013/10/11 01:14:30]

Done.

+  DCHECK_LE(count, next_pos_);
+
+  next_pos_ -= count;
+  state_ = READY;
+  DVLOG(1) << "Backing up " << count << " bytes in input buffer. "
+           << "Current position now at " << next_pos_
+           << " of " << read_buffer_->BytesConsumed();
+}
+
+bool SocketInputStream::Skip(int count) {
+  NOTIMPLEMENTED();
+  return false;
+}
+
+int64 SocketInputStream::ByteCount() const {
+  DCHECK_NE(state_, CLOSED);
+  DCHECK_NE(state_, READING);
+  return read_buffer_->BytesConsumed() - next_pos_;
+}
+
+void SocketInputStream::Refresh(const base::Closure& callback,
+                                int byte_limit) {
+  DCHECK_NE(state_, CLOSED);
+  DCHECK_NE(state_, READING);
+  DCHECK_GT(byte_limit, 0);
+
+  if (byte_limit > read_buffer_->BytesRemaining()) {
+    NOTREACHED();

[akalin, 2013/10/10 08:47:40]

you want NOTREACHED() << ... since NOTREACHED() turns into LOG(ERROR) for
release

[Nicolas Zea, 2013/10/11 01:14:30]

Done.

+    LOG(ERROR) << "Out of buffer space, closing input stream.";
+    CloseStream(net::ERR_UNEXPECTED, callback);
+    return;
+  }
+
+  if (!socket_->IsConnected()) {
+    LOG(ERROR) << "Socket was disconnected, closing input stream";
+    CloseStream(net::ERR_CONNECTION_CLOSED, callback);
+    return;
+  }
+
+  state_ = READING;
+
+  DVLOG(1) << "Refreshing input stream, limit of " << byte_limit << " bytes.";
+  int result = socket_->Read(
+      read_buffer_,
+      byte_limit,
+      base::Bind(&SocketInputStream::RefreshCompletionCallback,
+                 weak_ptr_factory_.GetWeakPtr(),
+                 callback));
+  DVLOG(1) << "Read returned " << result;
+  if (result != net::ERR_IO_PENDING) {
+    base::MessageLoop::current()->PostTask(

[akalin, 2013/10/10 08:47:40]

forgot to mention this last time. This does simplify the flow, but it does add
latency since even a synchronous return has to wait for the task to run. Are you
sure you're okay with that?

i'd normally just make this return OK || ERR_IO_PENDING and let the caller
handle the OK case and the callback handle the ERR_IO_PENDING case.

[Nicolas Zea, 2013/10/11 01:14:30]

Yeah, I guess I may as well keep it as consistent with net:: convention as
possible. I originally figured it wouldn't matter, but I suppose it helps in the
case of receiving multiple messages at once and processing them with little
latency. Done.

+        FROM_HERE,
+        base::Bind(&SocketInputStream::RefreshCompletionCallback,
+                   weak_ptr_factory_.GetWeakPtr(),
+                   callback,
+                   result));
+  }
+}
+
+void SocketInputStream::RebuildBuffer() {
+  DVLOG(1) << "Resetting input stream, consumed "
+           << next_pos_ << " bytes.";
+  DCHECK_NE(state_, READING);
+  DCHECK_NE(state_, CLOSED);
+
+  int last_read_pos = next_pos_;
+  char* unread_data_ptr = io_buffer_->data() + last_read_pos;

[akalin, 2013/10/10 08:47:40]

use Next() instead of manually calculating?

[Nicolas Zea, 2013/10/11 01:14:30]

Done.

+  int unread_buffer_size = read_buffer_->BytesConsumed() - last_read_pos;
+  ResetInternal();
+
+  if (unread_buffer_size > 0) {
+    read_buffer_->SetOffset(unread_buffer_size);
+    state_ = READY;
+
+    if (last_read_pos != 0) {

[akalin, 2013/10/10 08:47:40]

you can do:

if (unread_data_ptr != io_buffer_->data())

I think, eliminating the need for last_read_pos

In fact, you can probably replace this entire block with:

if (unread_data_ptr != io_buffer_->data()) {
  std::memmove(...);
}
read_buffer_->Consume(unread_buffer_size);
if (unread_buffer_size > 0)
  state_ = READY;

[Nicolas Zea, 2013/10/11 01:14:30]

Done.

+      DVLOG(1) << "Have " << unread_buffer_size
+               << " unread bytes remaining, shifting.";
+      // Move any remaining unread data to the start of the buffer;
+      std::memmove(io_buffer_->data(), unread_data_ptr, unread_buffer_size);
+    } else {
+      DVLOG(1) << "Have " << unread_buffer_size << " unread bytes remaining.";
+    }
+  }
+}
+
+net::Error SocketInputStream::last_error() const {
+  return last_error_;
+}
+
+SocketInputStream::State SocketInputStream::state() const {
+  return state_;
+}
+
+void SocketInputStream::RefreshCompletionCallback(
+    const base::Closure& callback, int result) {
+  DCHECK_EQ(state_, READING);
+  if (state_ == CLOSED) {
+    // An error occured before the completion callback could complete. Ignore
+    // the result.
+    return;
+  }
+
+  if (result < net::OK) {
+    DVLOG(1) << "Failed to refresh socket: " << result;
+    CloseStream(static_cast<net::Error>(result), callback);
+    return;
+  }
+  DCHECK_GT(result, 0);

[akalin, 2013/10/10 08:47:40]

Isn't Read() allowed to return a result of 0? Double-check?

[Nicolas Zea, 2013/10/11 01:14:30]

Done.

+
+  state_ = READY;
+  read_buffer_->DidConsume(result);
+  if (next_pos_ == read_buffer_->BytesConsumed())
+    state_ = EMPTY;
+
+  DVLOG(1) << "Refresh complete with " << result << " new bytes. "
+           << "Current position " << next_pos_
+           << " of " << read_buffer_->BytesConsumed() << ".";
+
+  if (!callback.is_null())
+    callback.Run();
+}
+
+void SocketInputStream::ResetInternal() {
+  weak_ptr_factory_.InvalidateWeakPtrs();  // Invalidate any callbacks.

[akalin, 2013/10/10 08:47:40]

follow declaration order to be parallel to the constructor?

[Nicolas Zea, 2013/10/11 01:14:30]

Done.

+  next_pos_ = 0;
+  state_ = EMPTY;
+  read_buffer_->SetOffset(0);
+
+  last_error_ = net::OK;
+}
+
+void SocketInputStream::CloseStream(net::Error error,
+                                    const base::Closure& callback) {
+  ResetInternal();
+  state_ = CLOSED;
+  last_error_ = error;
+  LOG(ERROR) << "Closing stream with result " << error;
+  if (!callback.is_null())
+    callback.Run();
+}
+
+SocketOutputStream::SocketOutputStream(net::StreamSocket* socket)
+    : socket_(socket),
+      io_buffer_(new net::IOBuffer(kDefaultBufferSize)),
+      write_buffer_(new net::DrainableIOBuffer(io_buffer_.get(),
+                                               kDefaultBufferSize)),
+      buffer_used_(0),
+      state_(EMPTY),
+      weak_ptr_factory_(this) {
+  DCHECK(socket->IsConnected());
+}
+
+SocketOutputStream::~SocketOutputStream() {
+}
+
+bool SocketOutputStream::Next(void** data, int* size) {
+  DCHECK_NE(state_, CLOSED);
+  DCHECK_NE(state_, FLUSHING);
+  if (buffer_used_ == write_buffer_->size())
+    return false;
+
+  *data = write_buffer_->data() + buffer_used_;
+  *size = write_buffer_->size() - buffer_used_;
+  buffer_used_ = write_buffer_->size();
+  state_ = READY;
+  return true;
+}
+
+void SocketOutputStream::BackUp(int count) {
+  DCHECK_GE(count, 0);
+  if (count > buffer_used_)
+    buffer_used_ = 0;
+  buffer_used_ -= count;
+  DVLOG(1) << "Backing up " << count << " bytes in output buffer. "
+           << buffer_used_ << " bytes used.";
+}
+
+int64 SocketOutputStream::ByteCount() const {
+  DCHECK_NE(state_, CLOSED);
+  DCHECK_NE(state_, FLUSHING);
+  return buffer_used_;
+}
+
+void SocketOutputStream::Flush(const base::Closure& callback) {
+  DCHECK_EQ(state_, READY);
+  state_ = FLUSHING;
+
+  if (!socket_->IsConnected()) {
+    LOG(ERROR) << "Socket was disconnected, closing output stream";
+    last_error_ = net::ERR_CONNECTION_CLOSED;
+    state_ = CLOSED;
+    if (!callback.is_null())
+      callback.Run();
+    return;
+  }
+
+  DVLOG(1) << "Flushing " << buffer_used_ << " bytes into socket.";
+  int result = socket_->Write(
+      write_buffer_,
+      buffer_used_,
+      base::Bind(&SocketOutputStream::FlushCompletionCallback,
+                 weak_ptr_factory_.GetWeakPtr(),
+                 callback));
+  DVLOG(1) << "Write returned " << result;
+  if (result != net::ERR_IO_PENDING) {
+    base::MessageLoop::current()->PostTask(
+        FROM_HERE,
+        base::Bind(&SocketOutputStream::FlushCompletionCallback,
+                   weak_ptr_factory_.GetWeakPtr(),
+                   callback,
+                   result));
+  }
+}
+
+SocketOutputStream::State SocketOutputStream::state() const{
+  return state_;
+}
+
+net::Error SocketOutputStream::last_error() const {
+  return last_error_;
+}
+
+void SocketOutputStream::FlushCompletionCallback(
+    const base::Closure& callback, int result) {
+  DCHECK_EQ(state_, FLUSHING);
+  if (result < net::OK) {
+    LOG(ERROR) << "Failed to flush socket.";
+    last_error_ = static_cast<net::Error>(result);
+    state_ = CLOSED;
+    if (!callback.is_null())
+      callback.Run();
+    return;
+  }
+
+  state_ = READY;
+  if (write_buffer_->BytesConsumed() + result < buffer_used_) {
+    DVLOG(1) << "Partial flush complete. Retrying.";
+     // Only a partial write was completed. Flush again to finish the write.
+    write_buffer_->DidConsume(result);
+    Flush(callback);
+    return;
+  }
+
+  DVLOG(1) << "Socket flush complete.";
+  write_buffer_->SetOffset(0);
+  state_ = EMPTY;
+  buffer_used_ = 0;
+  if (!callback.is_null())
+    callback.Run();
+}
+
+}  // namespace gcm