Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(79)

Unified Diff: google_apis/gcm/base/socket_stream.cc

Issue 23684017: [GCM] Initial work to set up directory structure and introduce socket integration (Closed) Base URL: svn://svn.chromium.org/chrome/trunk/src
Patch Set: Address comments Created 7 years, 2 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View side-by-side diff with in-line comments
Download patch
Index: google_apis/gcm/base/socket_stream.cc
diff --git a/google_apis/gcm/base/socket_stream.cc b/google_apis/gcm/base/socket_stream.cc
new file mode 100644
index 0000000000000000000000000000000000000000..db2b084fec77646dcf2d84b8d4c512834c3a8c9b
--- /dev/null
+++ b/google_apis/gcm/base/socket_stream.cc
@@ -0,0 +1,318 @@
+// 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
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 u
Nicolas Zea 2013/10/09 00:44:22 ResetInternal builds a new drainable_io_buffer_, w
akalin 2013/10/10 08:47:40 forgot to address this? (now that we're not buildi
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
akalin 2013/10/09 22:04:28 Huh, are you sure? I'm looking at DrainableIOBuffe
Nicolas Zea 2013/10/09 23:24:49 Huh, you're right. I completely misread that code
+ 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

Powered by Google App Engine
This is Rietveld 408576698