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 |