google_apis/gcm/base/socket_stream.cc - chromium/src - Git at Google

 // 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 <stddef.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 size_t kDefaultBufferSize = 8*1024;

 }  // namespace

 SocketInputStream::SocketInputStream(mojo::ScopedDataPipeConsumerHandle stream)
     : stream_(std::move(stream)),
       stream_watcher_(FROM_HERE, mojo::SimpleWatcher::ArmingPolicy::MANUAL),
       read_size_(0),
       io_buffer_(base::MakeRefCounted<net::IOBuffer>(kDefaultBufferSize)),
       read_buffer_(
           base::MakeRefCounted<net::DrainableIOBuffer>(io_buffer_,
                                                        kDefaultBufferSize)),
       next_pos_(0),
       last_error_(net::OK),
       weak_ptr_factory_(this) {
   stream_watcher_.Watch(
       stream_.get(),
       MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
       MOJO_TRIGGER_CONDITION_SIGNALS_SATISFIED,
       base::BindRepeating(&SocketInputStream::ReadMore,
                           base::Unretained(this)));
 }

 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);
   // TODO(zea): investigating crbug.com/409985
   CHECK_GT(count, 0);
   CHECK_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_t SocketInputStream::ByteCount() const {
   DCHECK_NE(GetState(), CLOSED);
   DCHECK_NE(GetState(), READING);
   return next_pos_;
 }

 int SocketInputStream::UnreadByteCount() const {
   DCHECK_NE(GetState(), CLOSED);
   DCHECK_NE(GetState(), READING);
   return read_buffer_->BytesConsumed() - next_pos_;
 }

 net::Error SocketInputStream::Refresh(base::OnceClosure callback,
                                       int byte_limit) {
   DCHECK(!read_callback_);
   DCHECK_NE(GetState(), CLOSED);
   DCHECK_NE(GetState(), READING);
   DCHECK_GT(byte_limit, 0);

   if (byte_limit > read_buffer_->BytesRemaining()) {
     LOG(ERROR) << "Out of buffer space, closing input stream.";
     CloseStream(net::ERR_FILE_TOO_BIG);
     return net::OK;
   }

   read_size_ = byte_limit;
   read_callback_ = std::move(callback);
   stream_watcher_.ArmOrNotify();
   last_error_ = net::ERR_IO_PENDING;
   return net::ERR_IO_PENDING;
 }

 void SocketInputStream::ReadMore(
     MojoResult result,
     const mojo::HandleSignalsState& /* ignored */) {
   DCHECK(read_callback_);
   DCHECK_NE(0u, read_size_);

   uint32_t num_bytes = read_size_;
   if (result == MOJO_RESULT_OK) {
     DVLOG(1) << "Refreshing input stream, limit of " << num_bytes << " bytes.";
     result = stream_->ReadData(read_buffer_->data(), &num_bytes,
                                MOJO_READ_DATA_FLAG_NONE);
     DVLOG(1) << "Read returned mojo result" << result;
   }

   if (result == MOJO_RESULT_SHOULD_WAIT) {
     stream_watcher_.ArmOrNotify();
     return;
   }

   read_size_ = 0;
   if (result != MOJO_RESULT_OK) {
     CloseStream(net::ERR_FAILED);
     std::move(read_callback_).Run();
     return;
   }

   // If an EOF has been received, close the stream.
   if (result == MOJO_RESULT_OK && num_bytes == 0) {
     CloseStream(net::ERR_CONNECTION_CLOSED);
     std::move(read_callback_).Run();
     return;
   }

   // If an error occurred before the completion callback could complete, ignore
   // the result.
   if (GetState() == CLOSED)
     return;

   last_error_ = net::OK;
   read_buffer_->DidConsume(num_bytes);
   // TODO(zea): investigating crbug.com/409985
   CHECK_GT(UnreadByteCount(), 0);

   DVLOG(1) << "Refresh complete with " << num_bytes << " new bytes. "
            << "Current position " << next_pos_ << " of "
            << read_buffer_->BytesConsumed() << ".";

   std::move(read_callback_).Run();
 }

 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 = nullptr;
   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);
   // TODO(zea): investigating crbug.com/409985
   CHECK_GE(UnreadByteCount(), 0);
 }

 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::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) {
   DCHECK_LT(error, net::ERR_IO_PENDING);
   ResetInternal();
   last_error_ = error;
   LOG(ERROR) << "Closing stream with result " << error;
 }

 SocketOutputStream::SocketOutputStream(
     mojo::ScopedDataPipeProducerHandle stream)
     : stream_(std::move(stream)),
       stream_watcher_(FROM_HERE, mojo::SimpleWatcher::ArmingPolicy::MANUAL),
       io_buffer_(
           base::MakeRefCounted<net::IOBufferWithSize>(kDefaultBufferSize)),
       next_pos_(0),
       last_error_(net::OK),
       weak_ptr_factory_(this) {
   stream_watcher_.Watch(
       stream_.get(),
       MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
       MOJO_TRIGGER_CONDITION_SIGNALS_SATISFIED,
       base::BindRepeating(&SocketOutputStream::WriteMore,
                           base::Unretained(this)));
 }

 SocketOutputStream::~SocketOutputStream() {
 }

 bool SocketOutputStream::Next(void** data, int* size) {
   DCHECK_NE(GetState(), CLOSED);
   DCHECK_NE(GetState(), FLUSHING);
   if (next_pos_ == io_buffer_->size())
     return false;

   *data = io_buffer_->data() + next_pos_;
   *size = io_buffer_->size() - next_pos_;
   next_pos_ = io_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_t SocketOutputStream::ByteCount() const {
   DCHECK_NE(GetState(), CLOSED);
   DCHECK_NE(GetState(), FLUSHING);
   return next_pos_;
 }

 net::Error SocketOutputStream::Flush(base::OnceClosure callback) {
   DCHECK(!write_callback_);
   DCHECK_EQ(GetState(), READY);

   if (!write_buffer_) {
     write_buffer_ = base::MakeRefCounted<net::DrainableIOBuffer>(
         io_buffer_.get(), next_pos_);
   }

   last_error_ = net::ERR_IO_PENDING;
   stream_watcher_.ArmOrNotify();
   write_callback_ = std::move(callback);
   return net::ERR_IO_PENDING;
 }

 void SocketOutputStream::WriteMore(MojoResult result,
                                    const mojo::HandleSignalsState& state) {
   DCHECK(write_callback_);
   DCHECK(write_buffer_);

   uint32_t num_bytes = write_buffer_->BytesRemaining();
   DVLOG(1) << "Flushing " << num_bytes << " bytes into socket.";
   if (result == MOJO_RESULT_OK) {
     result = stream_->WriteData(write_buffer_->data(), &num_bytes,
                                 MOJO_WRITE_DATA_FLAG_NONE);
   }
   if (result == MOJO_RESULT_SHOULD_WAIT) {
     stream_watcher_.ArmOrNotify();
     return;
   }
   if (result != MOJO_RESULT_OK) {
     LOG(ERROR) << "Failed to flush socket.";
     last_error_ = net::ERR_FAILED;
     std::move(write_callback_).Run();
     return;
   }
   DVLOG(1) << "Wrote  " << num_bytes;
   // If an error occurred before the completion callback could complete, ignore
   // the result.
   if (GetState() == CLOSED)
     return;

   DCHECK_GE(num_bytes, 0u);
   last_error_ = net::OK;
   write_buffer_->DidConsume(num_bytes);
   if (write_buffer_->BytesRemaining() > 0) {
     DVLOG(1) << "Partial flush complete. Retrying.";
     // Only a partial write was completed. Flush again to finish the write.
     Flush(std::move(write_callback_));
     return;
   }
   DVLOG(1) << "Socket flush complete.";
   write_buffer_ = nullptr;
   next_pos_ = 0;
   std::move(write_callback_).Run();
 }

 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_;
 }

 }  // namespace gcm
	// 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 <stddef.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 size_t kDefaultBufferSize = 8*1024;

	} // namespace

	SocketInputStream::SocketInputStream(mojo::ScopedDataPipeConsumerHandle stream)
	: stream_(std::move(stream)),
	stream_watcher_(FROM_HERE, mojo::SimpleWatcher::ArmingPolicy::MANUAL),
	read_size_(0),
	io_buffer_(base::MakeRefCounted<net::IOBuffer>(kDefaultBufferSize)),
	read_buffer_(
	base::MakeRefCounted<net::DrainableIOBuffer>(io_buffer_,
	kDefaultBufferSize)),
	next_pos_(0),
	last_error_(net::OK),
	weak_ptr_factory_(this) {
	stream_watcher_.Watch(
	stream_.get(),
	MOJO_HANDLE_SIGNAL_READABLE \| MOJO_HANDLE_SIGNAL_PEER_CLOSED,
	MOJO_TRIGGER_CONDITION_SIGNALS_SATISFIED,
	base::BindRepeating(&SocketInputStream::ReadMore,
	base::Unretained(this)));
	}

	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);
	// TODO(zea): investigating crbug.com/409985
	CHECK_GT(count, 0);
	CHECK_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_t SocketInputStream::ByteCount() const {
	DCHECK_NE(GetState(), CLOSED);
	DCHECK_NE(GetState(), READING);
	return next_pos_;
	}

	int SocketInputStream::UnreadByteCount() const {
	DCHECK_NE(GetState(), CLOSED);
	DCHECK_NE(GetState(), READING);
	return read_buffer_->BytesConsumed() - next_pos_;
	}

	net::Error SocketInputStream::Refresh(base::OnceClosure callback,
	int byte_limit) {
	DCHECK(!read_callback_);
	DCHECK_NE(GetState(), CLOSED);
	DCHECK_NE(GetState(), READING);
	DCHECK_GT(byte_limit, 0);

	if (byte_limit > read_buffer_->BytesRemaining()) {
	LOG(ERROR) << "Out of buffer space, closing input stream.";
	CloseStream(net::ERR_FILE_TOO_BIG);
	return net::OK;
	}

	read_size_ = byte_limit;
	read_callback_ = std::move(callback);
	stream_watcher_.ArmOrNotify();
	last_error_ = net::ERR_IO_PENDING;
	return net::ERR_IO_PENDING;
	}

	void SocketInputStream::ReadMore(
	MojoResult result,
	const mojo::HandleSignalsState& /* ignored */) {
	DCHECK(read_callback_);
	DCHECK_NE(0u, read_size_);

	uint32_t num_bytes = read_size_;
	if (result == MOJO_RESULT_OK) {
	DVLOG(1) << "Refreshing input stream, limit of " << num_bytes << " bytes.";
	result = stream_->ReadData(read_buffer_->data(), &num_bytes,
	MOJO_READ_DATA_FLAG_NONE);
	DVLOG(1) << "Read returned mojo result" << result;
	}

	if (result == MOJO_RESULT_SHOULD_WAIT) {
	stream_watcher_.ArmOrNotify();
	return;
	}

	read_size_ = 0;
	if (result != MOJO_RESULT_OK) {
	CloseStream(net::ERR_FAILED);
	std::move(read_callback_).Run();
	return;
	}

	// If an EOF has been received, close the stream.
	if (result == MOJO_RESULT_OK && num_bytes == 0) {
	CloseStream(net::ERR_CONNECTION_CLOSED);
	std::move(read_callback_).Run();
	return;
	}

	// If an error occurred before the completion callback could complete, ignore
	// the result.
	if (GetState() == CLOSED)
	return;

	last_error_ = net::OK;
	read_buffer_->DidConsume(num_bytes);
	// TODO(zea): investigating crbug.com/409985
	CHECK_GT(UnreadByteCount(), 0);

	DVLOG(1) << "Refresh complete with " << num_bytes << " new bytes. "
	<< "Current position " << next_pos_ << " of "
	<< read_buffer_->BytesConsumed() << ".";

	std::move(read_callback_).Run();
	}

	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 = nullptr;
	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);
	// TODO(zea): investigating crbug.com/409985
	CHECK_GE(UnreadByteCount(), 0);
	}

	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::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) {
	DCHECK_LT(error, net::ERR_IO_PENDING);
	ResetInternal();
	last_error_ = error;
	LOG(ERROR) << "Closing stream with result " << error;
	}

	SocketOutputStream::SocketOutputStream(
	mojo::ScopedDataPipeProducerHandle stream)
	: stream_(std::move(stream)),
	stream_watcher_(FROM_HERE, mojo::SimpleWatcher::ArmingPolicy::MANUAL),
	io_buffer_(
	base::MakeRefCounted<net::IOBufferWithSize>(kDefaultBufferSize)),
	next_pos_(0),
	last_error_(net::OK),
	weak_ptr_factory_(this) {
	stream_watcher_.Watch(
	stream_.get(),
	MOJO_HANDLE_SIGNAL_WRITABLE \| MOJO_HANDLE_SIGNAL_PEER_CLOSED,
	MOJO_TRIGGER_CONDITION_SIGNALS_SATISFIED,
	base::BindRepeating(&SocketOutputStream::WriteMore,
	base::Unretained(this)));
	}

	SocketOutputStream::~SocketOutputStream() {
	}

	bool SocketOutputStream::Next(void** data, int* size) {
	DCHECK_NE(GetState(), CLOSED);
	DCHECK_NE(GetState(), FLUSHING);
	if (next_pos_ == io_buffer_->size())
	return false;

	*data = io_buffer_->data() + next_pos_;
	*size = io_buffer_->size() - next_pos_;
	next_pos_ = io_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_t SocketOutputStream::ByteCount() const {
	DCHECK_NE(GetState(), CLOSED);
	DCHECK_NE(GetState(), FLUSHING);
	return next_pos_;
	}

	net::Error SocketOutputStream::Flush(base::OnceClosure callback) {
	DCHECK(!write_callback_);
	DCHECK_EQ(GetState(), READY);

	if (!write_buffer_) {
	write_buffer_ = base::MakeRefCounted<net::DrainableIOBuffer>(
	io_buffer_.get(), next_pos_);
	}

	last_error_ = net::ERR_IO_PENDING;
	stream_watcher_.ArmOrNotify();
	write_callback_ = std::move(callback);
	return net::ERR_IO_PENDING;
	}

	void SocketOutputStream::WriteMore(MojoResult result,
	const mojo::HandleSignalsState& state) {
	DCHECK(write_callback_);
	DCHECK(write_buffer_);

	uint32_t num_bytes = write_buffer_->BytesRemaining();
	DVLOG(1) << "Flushing " << num_bytes << " bytes into socket.";
	if (result == MOJO_RESULT_OK) {
	result = stream_->WriteData(write_buffer_->data(), &num_bytes,
	MOJO_WRITE_DATA_FLAG_NONE);
	}
	if (result == MOJO_RESULT_SHOULD_WAIT) {
	stream_watcher_.ArmOrNotify();
	return;
	}
	if (result != MOJO_RESULT_OK) {
	LOG(ERROR) << "Failed to flush socket.";
	last_error_ = net::ERR_FAILED;
	std::move(write_callback_).Run();
	return;
	}
	DVLOG(1) << "Wrote " << num_bytes;
	// If an error occurred before the completion callback could complete, ignore
	// the result.
	if (GetState() == CLOSED)
	return;

	DCHECK_GE(num_bytes, 0u);
	last_error_ = net::OK;
	write_buffer_->DidConsume(num_bytes);
	if (write_buffer_->BytesRemaining() > 0) {
	DVLOG(1) << "Partial flush complete. Retrying.";
	// Only a partial write was completed. Flush again to finish the write.
	Flush(std::move(write_callback_));
	return;
	}
	DVLOG(1) << "Socket flush complete.";
	write_buffer_ = nullptr;
	next_pos_ = 0;
	std::move(write_callback_).Run();
	}

	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_;
	}

	} // namespace gcm