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

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

Issue 29273008: Revert r229533 "[GCM] Initial work to set up directory structure and introduce socket integration" (Closed) Base URL: svn://svn.chromium.org/chrome/trunk/src
Patch Set: 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
« no previous file with comments | « google_apis/gcm/base/socket_stream.h ('k') | google_apis/gcm/base/socket_stream_unittest.cc » ('j') | no next file with comments »
Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
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
deleted file mode 100644
index 1a0b29d8d075ea7db249a94cd464d7cf771cabad..0000000000000000000000000000000000000000
--- a/google_apis/gcm/base/socket_stream.cc
+++ /dev/null
@@ -1,332 +0,0 @@
-// Copyright 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/bind.h"
-#include "base/callback.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),
- weak_ptr_factory_(this) {
- DCHECK(socket->IsConnected());
-}
-
-SocketInputStream::~SocketInputStream() {
-}
-
-bool SocketInputStream::Next(const void** data, int* size) {
- if (GetState() != EMPTY && GetState() != READY) {
- NOTREACHED() << "Invalid input stream read attempt.";
- return false;
- }
-
- if (GetState() == EMPTY) {
- DVLOG(1) << "No unread data remaining, ending read.";
- return false;
- }
-
- DCHECK_EQ(GetState(), READY)
- << " Input stream must have pending data before reading.";
- DCHECK_LT(next_pos_, read_buffer_->BytesConsumed());
- *data = io_buffer_->data() + next_pos_;
- *size = UnreadByteCount();
- next_pos_ = read_buffer_->BytesConsumed();
- DVLOG(1) << "Consuming " << *size << " bytes in input buffer.";
- return true;
-}
-
-void SocketInputStream::BackUp(int count) {
- DCHECK(GetState() == READY || GetState() == EMPTY);
- DCHECK_GT(count, 0);
- DCHECK_LE(count, next_pos_);
-
- next_pos_ -= count;
- 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(GetState(), CLOSED);
- DCHECK_NE(GetState(), READING);
- return next_pos_;
-}
-
-size_t SocketInputStream::UnreadByteCount() const {
- DCHECK_NE(GetState(), CLOSED);
- DCHECK_NE(GetState(), READING);
- return read_buffer_->BytesConsumed() - next_pos_;
-}
-
-net::Error SocketInputStream::Refresh(const base::Closure& callback,
- int byte_limit) {
- DCHECK_NE(GetState(), CLOSED);
- DCHECK_NE(GetState(), READING);
- DCHECK_GT(byte_limit, 0);
-
- if (byte_limit > read_buffer_->BytesRemaining()) {
- NOTREACHED() << "Out of buffer space, closing input stream.";
- CloseStream(net::ERR_UNEXPECTED, base::Closure());
- return net::OK;
- }
-
- if (!socket_->IsConnected()) {
- LOG(ERROR) << "Socket was disconnected, closing input stream";
- CloseStream(net::ERR_CONNECTION_CLOSED, base::Closure());
- return net::OK;
- }
-
- 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) {
- last_error_ = net::ERR_IO_PENDING;
- return net::ERR_IO_PENDING;
- }
-
- RefreshCompletionCallback(base::Closure(), result);
- return net::OK;
-}
-
-void SocketInputStream::RebuildBuffer() {
- DVLOG(1) << "Rebuilding input stream, consumed "
- << next_pos_ << " bytes.";
- DCHECK_NE(GetState(), READING);
- DCHECK_NE(GetState(), CLOSED);
-
- int unread_data_size = 0;
- const void* unread_data_ptr = NULL;
- Next(&unread_data_ptr, &unread_data_size);
- ResetInternal();
-
- if (unread_data_ptr != io_buffer_->data()) {
- DVLOG(1) << "Have " << unread_data_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_data_size);
- } else {
- DVLOG(1) << "Have " << unread_data_size << " unread bytes remaining.";
- }
- read_buffer_->DidConsume(unread_data_size);
-}
-
-net::Error SocketInputStream::last_error() const {
- return last_error_;
-}
-
-SocketInputStream::State SocketInputStream::GetState() const {
- if (last_error_ < net::ERR_IO_PENDING)
- return CLOSED;
-
- if (last_error_ == net::ERR_IO_PENDING)
- return READING;
-
- DCHECK_EQ(last_error_, net::OK);
- if (read_buffer_->BytesConsumed() == next_pos_)
- return EMPTY;
-
- return READY;
-}
-
-void SocketInputStream::RefreshCompletionCallback(
- const base::Closure& callback, int result) {
- // If an error occurred before the completion callback could complete, ignore
- // the result.
- if (GetState() == CLOSED)
- return;
-
- // Result == 0 implies EOF, which is treated as an error.
- if (result == 0)
- result = net::ERR_CONNECTION_CLOSED;
-
- DCHECK_NE(result, net::ERR_IO_PENDING);
-
- if (result < net::OK) {
- DVLOG(1) << "Failed to refresh socket: " << result;
- CloseStream(static_cast<net::Error>(result), callback);
- return;
- }
-
- DCHECK_GT(result, 0);
- last_error_ = net::OK;
- read_buffer_->DidConsume(result);
-
- 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() {
- read_buffer_->SetOffset(0);
- next_pos_ = 0;
- last_error_ = net::OK;
- weak_ptr_factory_.InvalidateWeakPtrs(); // Invalidate any callbacks.
-}
-
-void SocketInputStream::CloseStream(net::Error error,
- const base::Closure& callback) {
- DCHECK_LT(error, net::ERR_IO_PENDING);
- ResetInternal();
- 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)),
- next_pos_(0),
- last_error_(net::OK),
- weak_ptr_factory_(this) {
- DCHECK(socket->IsConnected());
-}
-
-SocketOutputStream::~SocketOutputStream() {
-}
-
-bool SocketOutputStream::Next(void** data, int* size) {
- DCHECK_NE(GetState(), CLOSED);
- DCHECK_NE(GetState(), FLUSHING);
- if (next_pos_ == write_buffer_->size())
- return false;
-
- *data = write_buffer_->data() + next_pos_;
- *size = write_buffer_->size() - next_pos_;
- next_pos_ = write_buffer_->size();
- return true;
-}
-
-void SocketOutputStream::BackUp(int count) {
- DCHECK_GE(count, 0);
- if (count > next_pos_)
- next_pos_ = 0;
- next_pos_ -= count;
- DVLOG(1) << "Backing up " << count << " bytes in output buffer. "
- << next_pos_ << " bytes used.";
-}
-
-int64 SocketOutputStream::ByteCount() const {
- DCHECK_NE(GetState(), CLOSED);
- DCHECK_NE(GetState(), FLUSHING);
- return next_pos_;
-}
-
-net::Error SocketOutputStream::Flush(const base::Closure& callback) {
- DCHECK_EQ(GetState(), READY);
-
- if (!socket_->IsConnected()) {
- LOG(ERROR) << "Socket was disconnected, closing output stream";
- last_error_ = net::ERR_CONNECTION_CLOSED;
- return net::OK;
- }
-
- DVLOG(1) << "Flushing " << next_pos_ << " bytes into socket.";
- int result = socket_->Write(
- write_buffer_,
- next_pos_,
- base::Bind(&SocketOutputStream::FlushCompletionCallback,
- weak_ptr_factory_.GetWeakPtr(),
- callback));
- DVLOG(1) << "Write returned " << result;
- if (result == net::ERR_IO_PENDING) {
- last_error_ = net::ERR_IO_PENDING;
- return net::ERR_IO_PENDING;
- }
-
- FlushCompletionCallback(base::Closure(), result);
- return net::OK;
-}
-
-SocketOutputStream::State SocketOutputStream::GetState() const{
- if (last_error_ < net::ERR_IO_PENDING)
- return CLOSED;
-
- if (last_error_ == net::ERR_IO_PENDING)
- return FLUSHING;
-
- DCHECK_EQ(last_error_, net::OK);
- if (next_pos_ == 0)
- return EMPTY;
-
- return READY;
-}
-
-net::Error SocketOutputStream::last_error() const {
- return last_error_;
-}
-
-void SocketOutputStream::FlushCompletionCallback(
- const base::Closure& callback, int result) {
- // If an error occurred before the completion callback could complete, ignore
- // the result.
- if (GetState() == CLOSED)
- return;
-
- // Result == 0 implies EOF, which is treated as an error.
- if (result == 0)
- result = net::ERR_CONNECTION_CLOSED;
-
- DCHECK_NE(result, net::ERR_IO_PENDING);
-
- if (result < net::OK) {
- LOG(ERROR) << "Failed to flush socket.";
- last_error_ = static_cast<net::Error>(result);
- if (!callback.is_null())
- callback.Run();
- return;
- }
-
- DCHECK_GT(result, net::OK);
- last_error_ = net::OK;
-
- if (write_buffer_->BytesConsumed() + result < next_pos_) {
- 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);
- next_pos_ = 0;
- if (!callback.is_null())
- callback.Run();
-}
-
-} // namespace gcm
« no previous file with comments | « google_apis/gcm/base/socket_stream.h ('k') | google_apis/gcm/base/socket_stream_unittest.cc » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698