[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::IOBufferWithSize(kDefaultBufferSize)),

[akalin, 2013/10/07 23:00:08]

plain IOBuffer here?

[Nicolas Zea, 2013/10/09 00:44:22]

Done.

+      drainable_io_buffer_(new net::DrainableIOBuffer(io_buffer_.get(),
+                                                      kDefaultBufferSize)),
+      buffer_read_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(drainable_io_buffer_->BytesConsumed(), buffer_read_pos_);
+  *data = io_buffer_->data() + buffer_read_pos_;
+  *size = drainable_io_buffer_->BytesConsumed() - buffer_read_pos_;
+  buffer_read_pos_ = drainable_io_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);
+  DCHECK_LE(count, buffer_read_pos_);
+
+  buffer_read_pos_ -= count;
+  state_ = READY;
+  DVLOG(1) << "Backing up " << count << " bytes in input buffer. "
+           << "Current position now at " << buffer_read_pos_
+           << " of " << drainable_io_buffer_->BytesConsumed();
+}
+
+bool SocketInputStream::Skip(int count) {
+  NOTIMPLEMENTED();
+  return false;
+}
+
+int64 SocketInputStream::ByteCount() const {
+  DCHECK_NE(state_, CLOSED);
+  DCHECK_NE(state_, READING);
+  return drainable_io_buffer_->BytesConsumed() - buffer_read_pos_;
+}
+
+void SocketInputStream::Refresh(const base::Closure& callback,
+                                int byte_limit) {
+  DCHECK_NE(state_, CLOSED);
+  DCHECK_NE(state_, READING);
+  DCHECK_GT(byte_limit, 0);
+  DCHECK_LE(byte_limit, drainable_io_buffer_->BytesRemaining());
+
+  if (drainable_io_buffer_->BytesRemaining() < byte_limit) {

[akalin, 2013/10/07 23:00:08]

can you reverse the order of this comparison? (to make it easier to compare to
the above dcheck)

wait, this is the same check as the DCHECK above, right? Why not:

if (byte_limit > BytesRemaining()) {
  NOTREACHED() << ...;
  ...
}

[Nicolas Zea, 2013/10/09 00:44:22]

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(
+      drainable_io_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(
+        FROM_HERE,
+        base::Bind(&SocketInputStream::RefreshCompletionCallback,
+                   weak_ptr_factory_.GetWeakPtr(),
+                   callback,
+                   result));
+  }
+}
+
+void SocketInputStream::RebuildBuffer() {
+  DVLOG(1) << "Resetting input stream, consumed "
+           << buffer_read_pos_ << " bytes.";
+  DCHECK_NE(state_, READING);
+  DCHECK_NE(state_, CLOSED);
+
+  int last_read_pos = buffer_read_pos_;
+  char* unread_data_ptr = io_buffer_->data() + last_read_pos;
+  int unread_buffer_size =
+      drainable_io_buffer_->BytesConsumed() - last_read_pos;
+  ResetInternal();
+
+  if (unread_buffer_size > 0) {
+    drainable_io_buffer_->SetOffset(unread_buffer_size);

[akalin, 2013/10/07 23:00:08]

i feel like you want to call SetOffset even when unread_buffer_size == 0, since
the client can still call BackUp, right?

[Nicolas Zea, 2013/10/09 00:44:22]

ResetInternal builds a new drainable_io_buffer_, which has an initial offset of
0 already.

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

forgot to address this? (now that we're not building a new one anymore)

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

n/a in newest version I believe.

+    state_ = READY;
+
+    if (last_read_pos != 0) {
+      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);
+
+  state_ = READY;
+  drainable_io_buffer_->DidConsume(result);
+  if (buffer_read_pos_ == drainable_io_buffer_->BytesConsumed())
+    state_ = EMPTY;
+
+  DVLOG(1) << "Refresh complete with " << result << " new bytes. "
+           << "Current position " << buffer_read_pos_
+           << " of " << drainable_io_buffer_->BytesConsumed() << ".";
+
+  if (!callback.is_null())
+    callback.Run();
+}
+
+void SocketInputStream::ResetInternal() {
+  weak_ptr_factory_.InvalidateWeakPtrs();  // Invalidate any callbacks.
+  buffer_read_pos_ = 0;
+  state_ = EMPTY;
+
+  last_error_ = net::OK;
+
+  // Reset the offset by creating a new one. Note that DrainableIOBuffers don't
+  // actually allocate their own buffer memory like normal IOBuffers. This will
+  // just reset the pointers to point to the beginning of io_buffer_'s data.
+  drainable_io_buffer_ = new net::DrainableIOBuffer(io_buffer_.get(),

[akalin, 2013/10/07 23:00:08]

use setoffset here?

[Nicolas Zea, 2013/10/09 00:44:22]

SetOffset (and DrainableIOBuffer in general) only allows you to move forward in
offset. We want to go back to 0 here.

[akalin, 2013/10/09 22:04:28]

Huh, are you sure? I'm looking at DrainableIOBuffer::SetOffset:

void DrainableIOBuffer::SetOffset(int bytes) {
  DCHECK_GE(bytes, 0);
  DCHECK_LE(bytes, size_);
  used_ = bytes;
  data_ = base_->data() + used_;
}

and it looks like you can set it to an arbitrary point

[Nicolas Zea, 2013/10/09 23:24:49]

Huh, you're right. I completely misread that code somehow. Previous comments
addressed.

+                                                    kDefaultBufferSize);
+}
+
+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::IOBufferWithSize(kDefaultBufferSize)),
+      drainable_io_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_ == drainable_io_buffer_->size())
+    return false;
+
+  *data = drainable_io_buffer_->data() + buffer_used_;
+  *size = drainable_io_buffer_->size() - buffer_used_;
+  buffer_used_ = drainable_io_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(
+      drainable_io_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 (drainable_io_buffer_->BytesConsumed() + result < buffer_used_) {
+    DVLOG(1) << "Partial flush complete. Retrying.";
+     // Only a partial write was completed. Flush again to finish the write.
+    drainable_io_buffer_->DidConsume(result);
+    Flush(callback);
+    return;
+  }
+
+  DVLOG(1) << "Socket flush complete.";
+  drainable_io_buffer_ = new net::DrainableIOBuffer(io_buffer_.get(),
+                                                    kDefaultBufferSize);
+  state_ = EMPTY;
+  buffer_used_ = 0;
+  if (!callback.is_null())
+    callback.Run();
+}
+
+}  // namespace gcm