device/serial/data_receiver.cc - platform/external/chromium_org - Git at Google

 // Copyright 2014 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 "device/serial/data_receiver.h"

 #include <limits>

 #include "base/bind.h"
 #include "base/message_loop/message_loop.h"
 #include "device/serial/async_waiter.h"

 namespace device {

 // Represents a receive that is not yet fulfilled.
 class DataReceiver::PendingReceive {
  public:
   PendingReceive(DataReceiver* receiver,
                  const ReceiveDataCallback& callback,
                  const ReceiveErrorCallback& error_callback,
                  int32_t fatal_error_value);

   // Dispatches |data| to |receive_callback_|.
   void DispatchData(const void* data, uint32_t num_bytes);

   // Reports |error| to |receive_error_callback_| if it is an appropriate time.
   // Returns whether it dispatched |error|.
   bool DispatchError(DataReceiver::PendingError* error,
                      uint32_t bytes_received);

   // Reports |fatal_error_value_| to |receive_error_callback_|.
   void DispatchFatalError();

  private:
   class Buffer;

   // Invoked when the user is finished with the ReadOnlyBuffer provided to
   // |receive_callback_|.
   void Done(uint32_t num_bytes);

   // The DataReceiver that owns this.
   DataReceiver* receiver_;

   // The callback to dispatch data.
   ReceiveDataCallback receive_callback_;

   // The callback to report errors.
   ReceiveErrorCallback receive_error_callback_;

   // The error value to report when DispatchFatalError() is called.
   const int32_t fatal_error_value_;

   // True if the user owns a buffer passed to |receive_callback_| as part of
   // DispatchData().
   bool buffer_in_use_;
 };

 // A ReadOnlyBuffer implementation that provides a view of a data pipe owned by
 // a DataReceiver.
 class DataReceiver::PendingReceive::Buffer : public ReadOnlyBuffer {
  public:
   Buffer(scoped_refptr<DataReceiver> pipe,
          PendingReceive* receive,
          const char* buffer,
          uint32_t buffer_size);
   virtual ~Buffer();

   // ReadOnlyBuffer overrides.
   virtual const char* GetData() OVERRIDE;
   virtual uint32_t GetSize() OVERRIDE;
   virtual void Done(uint32_t bytes_consumed) OVERRIDE;
   virtual void DoneWithError(uint32_t bytes_consumed, int32_t error) OVERRIDE;

  private:
   // The DataReceiver whose data pipe we are providing a view.
   scoped_refptr<DataReceiver> receiver_;

   // The PendingReceive to which this buffer has been created in response.
   PendingReceive* pending_receive_;

   const char* buffer_;
   uint32_t buffer_size_;
 };

 // Represents an error received from the DataSource.
 struct DataReceiver::PendingError {
   PendingError(uint32_t offset, int32_t error)
       : offset(offset), error(error), dispatched(false) {}

   // The location within the data stream where the error occurred.
   const uint32_t offset;

   // The value of the error that occurred.
   const int32_t error;

   // Whether the error has been dispatched to the user.
   bool dispatched;
 };

 DataReceiver::DataReceiver(mojo::InterfacePtr<serial::DataSource> source,
                            uint32_t buffer_size,
                            int32_t fatal_error_value)
     : source_(source.Pass()),
       fatal_error_value_(fatal_error_value),
       bytes_received_(0),
       shut_down_(false),
       weak_factory_(this) {
   MojoCreateDataPipeOptions options = {
       sizeof(options), MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE, 1, buffer_size,
   };
   mojo::ScopedDataPipeProducerHandle remote_handle;
   MojoResult result = mojo::CreateDataPipe(&options, &remote_handle, &handle_);
   DCHECK_EQ(MOJO_RESULT_OK, result);
   source_->Init(remote_handle.Pass());
   source_.set_client(this);
 }

 bool DataReceiver::Receive(const ReceiveDataCallback& callback,
                            const ReceiveErrorCallback& error_callback) {
   DCHECK(!callback.is_null() && !error_callback.is_null());
   if (pending_receive_ || shut_down_)
     return false;
   // When the DataSource encounters an error, it pauses transmission. When the
   // user starts a new receive following notification of the error (via
   // |error_callback| of the previous Receive call) of the error we can tell the
   // DataSource to resume transmission of data.
   if (pending_error_ && pending_error_->dispatched) {
     source_->Resume();
     pending_error_.reset();
   }

   pending_receive_.reset(
       new PendingReceive(this, callback, error_callback, fatal_error_value_));
   base::MessageLoop::current()->PostTask(
       FROM_HERE,
       base::Bind(&DataReceiver::ReceiveInternal, weak_factory_.GetWeakPtr()));
   return true;
 }

 DataReceiver::~DataReceiver() {
   ShutDown();
 }

 void DataReceiver::OnError(uint32_t offset, int32_t error) {
   if (shut_down_)
     return;

   if (pending_error_) {
     // When OnError is called by the DataSource, transmission of data is
     // suspended. Thus we shouldn't receive another call to OnError until we
     // have fully dealt with the error and called Resume to resume transmission
     // (see Receive()). Under normal operation we should never get here, but if
     // we do (e.g. in the case of a hijacked service process) just shut down.
     ShutDown();
     return;
   }
   pending_error_.reset(new PendingError(offset, error));
   if (pending_receive_ &&
       pending_receive_->DispatchError(pending_error_.get(), bytes_received_)) {
     pending_receive_.reset();
     waiter_.reset();
   }
 }

 void DataReceiver::OnConnectionError() {
   ShutDown();
 }

 void DataReceiver::Done(uint32_t bytes_consumed) {
   if (shut_down_)
     return;

   DCHECK(pending_receive_);
   MojoResult result = mojo::EndReadDataRaw(handle_.get(), bytes_consumed);
   DCHECK_EQ(MOJO_RESULT_OK, result);
   pending_receive_.reset();
   bytes_received_ += bytes_consumed;
 }

 void DataReceiver::OnDoneWaiting(MojoResult result) {
   DCHECK(pending_receive_ && !shut_down_ && waiter_);
   waiter_.reset();
   if (result != MOJO_RESULT_OK) {
     ShutDown();
     return;
   }
   ReceiveInternal();
 }

 void DataReceiver::ReceiveInternal() {
   if (shut_down_)
     return;
   DCHECK(pending_receive_);
   if (pending_error_ &&
       pending_receive_->DispatchError(pending_error_.get(), bytes_received_)) {
     pending_receive_.reset();
     waiter_.reset();
     return;
   }

   const void* data;
   uint32_t num_bytes = std::numeric_limits<uint32_t>::max();
   MojoResult result = mojo::BeginReadDataRaw(
       handle_.get(), &data, &num_bytes, MOJO_READ_DATA_FLAG_NONE);
   if (result == MOJO_RESULT_OK) {
     if (!CheckErrorNotInReadRange(num_bytes)) {
       ShutDown();
       return;
     }

     pending_receive_->DispatchData(data, num_bytes);
     return;
   }
   if (result == MOJO_RESULT_SHOULD_WAIT) {
     waiter_.reset(new AsyncWaiter(
         handle_.get(),
         MOJO_HANDLE_SIGNAL_READABLE,
         base::Bind(&DataReceiver::OnDoneWaiting, weak_factory_.GetWeakPtr())));
     return;
   }
   ShutDown();
 }

 bool DataReceiver::CheckErrorNotInReadRange(uint32_t num_bytes) {
   DCHECK(pending_receive_);
   if (!pending_error_)
     return true;

   DCHECK_NE(bytes_received_, pending_error_->offset);
   DCHECK_NE(num_bytes, 0u);
   uint32_t potential_bytes_received = bytes_received_ + num_bytes;
   // bytes_received_ can overflow so we must consider two cases:
   //   1. Both |bytes_received_| and |pending_error_->offset| have overflowed an
   //      equal number of times. In this case, |potential_bytes_received| must
   //      be in the range (|bytes_received|, |pending_error_->offset|]. Below
   //      this range can only occur if |bytes_received_| overflows before
   //      |pending_error_->offset|. Above can only occur if |bytes_received_|
   //      overtakes |pending_error_->offset|.
   //   2. |pending_error_->offset| has overflowed once more than
   //      |bytes_received_|. In this case, |potential_bytes_received| must not
   //      be in the range (|pending_error_->offset|, |bytes_received_|].
   if ((bytes_received_ < pending_error_->offset &&
        (potential_bytes_received > pending_error_->offset ||
         potential_bytes_received <= bytes_received_)) ||
       (bytes_received_ > pending_error_->offset &&
        potential_bytes_received > pending_error_->offset &&
        potential_bytes_received <= bytes_received_)) {
     return false;
   }
   return true;
 }

 void DataReceiver::ShutDown() {
   shut_down_ = true;
   if (pending_receive_)
     pending_receive_->DispatchFatalError();
   pending_error_.reset();
   waiter_.reset();
 }

 DataReceiver::PendingReceive::PendingReceive(
     DataReceiver* receiver,
     const ReceiveDataCallback& callback,
     const ReceiveErrorCallback& error_callback,
     int32_t fatal_error_value)
     : receiver_(receiver),
       receive_callback_(callback),
       receive_error_callback_(error_callback),
       fatal_error_value_(fatal_error_value),
       buffer_in_use_(false) {
 }

 void DataReceiver::PendingReceive::DispatchData(const void* data,
                                                 uint32_t num_bytes) {
   DCHECK(!buffer_in_use_);
   buffer_in_use_ = true;
   receive_callback_.Run(scoped_ptr<ReadOnlyBuffer>(
       new Buffer(receiver_, this, static_cast<const char*>(data), num_bytes)));
 }

 bool DataReceiver::PendingReceive::DispatchError(PendingError* error,
                                                  uint32_t bytes_received) {
   DCHECK(!error->dispatched);
   if (buffer_in_use_ || bytes_received != error->offset)
     return false;

   error->dispatched = true;
   receive_error_callback_.Run(error->error);
   return true;
 }

 void DataReceiver::PendingReceive::DispatchFatalError() {
   receive_error_callback_.Run(fatal_error_value_);
 }

 void DataReceiver::PendingReceive::Done(uint32_t bytes_consumed) {
   DCHECK(buffer_in_use_);
   buffer_in_use_ = false;
   receiver_->Done(bytes_consumed);
 }

 DataReceiver::PendingReceive::Buffer::Buffer(
     scoped_refptr<DataReceiver> receiver,
     PendingReceive* receive,
     const char* buffer,
     uint32_t buffer_size)
     : receiver_(receiver),
       pending_receive_(receive),
       buffer_(buffer),
       buffer_size_(buffer_size) {
 }

 DataReceiver::PendingReceive::Buffer::~Buffer() {
   if (pending_receive_)
     pending_receive_->Done(0);
 }

 const char* DataReceiver::PendingReceive::Buffer::GetData() {
   return buffer_;
 }

 uint32_t DataReceiver::PendingReceive::Buffer::GetSize() {
   return buffer_size_;
 }

 void DataReceiver::PendingReceive::Buffer::Done(uint32_t bytes_consumed) {
   pending_receive_->Done(bytes_consumed);
   pending_receive_ = NULL;
   receiver_ = NULL;
   buffer_ = NULL;
   buffer_size_ = 0;
 }

 void DataReceiver::PendingReceive::Buffer::DoneWithError(
     uint32_t bytes_consumed,
     int32_t error) {
   Done(bytes_consumed);
 }

 }  // namespace device
	// Copyright 2014 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 "device/serial/data_receiver.h"

	#include <limits>

	#include "base/bind.h"
	#include "base/message_loop/message_loop.h"
	#include "device/serial/async_waiter.h"

	namespace device {

	// Represents a receive that is not yet fulfilled.
	class DataReceiver::PendingReceive {
	public:
	PendingReceive(DataReceiver* receiver,
	const ReceiveDataCallback& callback,
	const ReceiveErrorCallback& error_callback,
	int32_t fatal_error_value);

	// Dispatches \|data\| to \|receive_callback_\|.
	void DispatchData(const void* data, uint32_t num_bytes);

	// Reports \|error\| to \|receive_error_callback_\| if it is an appropriate time.
	// Returns whether it dispatched \|error\|.
	bool DispatchError(DataReceiver::PendingError* error,
	uint32_t bytes_received);

	// Reports \|fatal_error_value_\| to \|receive_error_callback_\|.
	void DispatchFatalError();

	private:
	class Buffer;

	// Invoked when the user is finished with the ReadOnlyBuffer provided to
	// \|receive_callback_\|.
	void Done(uint32_t num_bytes);

	// The DataReceiver that owns this.
	DataReceiver* receiver_;

	// The callback to dispatch data.
	ReceiveDataCallback receive_callback_;

	// The callback to report errors.
	ReceiveErrorCallback receive_error_callback_;

	// The error value to report when DispatchFatalError() is called.
	const int32_t fatal_error_value_;

	// True if the user owns a buffer passed to \|receive_callback_\| as part of
	// DispatchData().
	bool buffer_in_use_;
	};

	// A ReadOnlyBuffer implementation that provides a view of a data pipe owned by
	// a DataReceiver.
	class DataReceiver::PendingReceive::Buffer : public ReadOnlyBuffer {
	public:
	Buffer(scoped_refptr<DataReceiver> pipe,
	PendingReceive* receive,
	const char* buffer,
	uint32_t buffer_size);
	virtual ~Buffer();

	// ReadOnlyBuffer overrides.
	virtual const char* GetData() OVERRIDE;
	virtual uint32_t GetSize() OVERRIDE;
	virtual void Done(uint32_t bytes_consumed) OVERRIDE;
	virtual void DoneWithError(uint32_t bytes_consumed, int32_t error) OVERRIDE;

	private:
	// The DataReceiver whose data pipe we are providing a view.
	scoped_refptr<DataReceiver> receiver_;

	// The PendingReceive to which this buffer has been created in response.
	PendingReceive* pending_receive_;

	const char* buffer_;
	uint32_t buffer_size_;
	};

	// Represents an error received from the DataSource.
	struct DataReceiver::PendingError {
	PendingError(uint32_t offset, int32_t error)
	: offset(offset), error(error), dispatched(false) {}

	// The location within the data stream where the error occurred.
	const uint32_t offset;

	// The value of the error that occurred.
	const int32_t error;

	// Whether the error has been dispatched to the user.
	bool dispatched;
	};

	DataReceiver::DataReceiver(mojo::InterfacePtr<serial::DataSource> source,
	uint32_t buffer_size,
	int32_t fatal_error_value)
	: source_(source.Pass()),
	fatal_error_value_(fatal_error_value),
	bytes_received_(0),
	shut_down_(false),
	weak_factory_(this) {
	MojoCreateDataPipeOptions options = {
	sizeof(options), MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE, 1, buffer_size,
	};
	mojo::ScopedDataPipeProducerHandle remote_handle;
	MojoResult result = mojo::CreateDataPipe(&options, &remote_handle, &handle_);
	DCHECK_EQ(MOJO_RESULT_OK, result);
	source_->Init(remote_handle.Pass());
	source_.set_client(this);
	}

	bool DataReceiver::Receive(const ReceiveDataCallback& callback,
	const ReceiveErrorCallback& error_callback) {
	DCHECK(!callback.is_null() && !error_callback.is_null());
	if (pending_receive_ \|\| shut_down_)
	return false;
	// When the DataSource encounters an error, it pauses transmission. When the
	// user starts a new receive following notification of the error (via
	// \|error_callback\| of the previous Receive call) of the error we can tell the
	// DataSource to resume transmission of data.
	if (pending_error_ && pending_error_->dispatched) {
	source_->Resume();
	pending_error_.reset();
	}

	pending_receive_.reset(
	new PendingReceive(this, callback, error_callback, fatal_error_value_));
	base::MessageLoop::current()->PostTask(
	FROM_HERE,
	base::Bind(&DataReceiver::ReceiveInternal, weak_factory_.GetWeakPtr()));
	return true;
	}

	DataReceiver::~DataReceiver() {
	ShutDown();
	}

	void DataReceiver::OnError(uint32_t offset, int32_t error) {
	if (shut_down_)
	return;

	if (pending_error_) {
	// When OnError is called by the DataSource, transmission of data is
	// suspended. Thus we shouldn't receive another call to OnError until we
	// have fully dealt with the error and called Resume to resume transmission
	// (see Receive()). Under normal operation we should never get here, but if
	// we do (e.g. in the case of a hijacked service process) just shut down.
	ShutDown();
	return;
	}
	pending_error_.reset(new PendingError(offset, error));
	if (pending_receive_ &&
	pending_receive_->DispatchError(pending_error_.get(), bytes_received_)) {
	pending_receive_.reset();
	waiter_.reset();
	}
	}

	void DataReceiver::OnConnectionError() {
	ShutDown();
	}

	void DataReceiver::Done(uint32_t bytes_consumed) {
	if (shut_down_)
	return;

	DCHECK(pending_receive_);
	MojoResult result = mojo::EndReadDataRaw(handle_.get(), bytes_consumed);
	DCHECK_EQ(MOJO_RESULT_OK, result);
	pending_receive_.reset();
	bytes_received_ += bytes_consumed;
	}

	void DataReceiver::OnDoneWaiting(MojoResult result) {
	DCHECK(pending_receive_ && !shut_down_ && waiter_);
	waiter_.reset();
	if (result != MOJO_RESULT_OK) {
	ShutDown();
	return;
	}
	ReceiveInternal();
	}

	void DataReceiver::ReceiveInternal() {
	if (shut_down_)
	return;
	DCHECK(pending_receive_);
	if (pending_error_ &&
	pending_receive_->DispatchError(pending_error_.get(), bytes_received_)) {
	pending_receive_.reset();
	waiter_.reset();
	return;
	}

	const void* data;
	uint32_t num_bytes = std::numeric_limits<uint32_t>::max();
	MojoResult result = mojo::BeginReadDataRaw(
	handle_.get(), &data, &num_bytes, MOJO_READ_DATA_FLAG_NONE);
	if (result == MOJO_RESULT_OK) {
	if (!CheckErrorNotInReadRange(num_bytes)) {
	ShutDown();
	return;
	}

	pending_receive_->DispatchData(data, num_bytes);
	return;
	}
	if (result == MOJO_RESULT_SHOULD_WAIT) {
	waiter_.reset(new AsyncWaiter(
	handle_.get(),
	MOJO_HANDLE_SIGNAL_READABLE,
	base::Bind(&DataReceiver::OnDoneWaiting, weak_factory_.GetWeakPtr())));
	return;
	}
	ShutDown();
	}

	bool DataReceiver::CheckErrorNotInReadRange(uint32_t num_bytes) {
	DCHECK(pending_receive_);
	if (!pending_error_)
	return true;

	DCHECK_NE(bytes_received_, pending_error_->offset);
	DCHECK_NE(num_bytes, 0u);
	uint32_t potential_bytes_received = bytes_received_ + num_bytes;
	// bytes_received_ can overflow so we must consider two cases:
	// 1. Both \|bytes_received_\| and \|pending_error_->offset\| have overflowed an
	// equal number of times. In this case, \|potential_bytes_received\| must
	// be in the range (\|bytes_received\|, \|pending_error_->offset\|]. Below
	// this range can only occur if \|bytes_received_\| overflows before
	// \|pending_error_->offset\|. Above can only occur if \|bytes_received_\|
	// overtakes \|pending_error_->offset\|.
	// 2. \|pending_error_->offset\| has overflowed once more than
	// \|bytes_received_\|. In this case, \|potential_bytes_received\| must not
	// be in the range (\|pending_error_->offset\|, \|bytes_received_\|].
	if ((bytes_received_ < pending_error_->offset &&
	(potential_bytes_received > pending_error_->offset \|\|
	potential_bytes_received <= bytes_received_)) \|\|
	(bytes_received_ > pending_error_->offset &&
	potential_bytes_received > pending_error_->offset &&
	potential_bytes_received <= bytes_received_)) {
	return false;
	}
	return true;
	}

	void DataReceiver::ShutDown() {
	shut_down_ = true;
	if (pending_receive_)
	pending_receive_->DispatchFatalError();
	pending_error_.reset();
	waiter_.reset();
	}

	DataReceiver::PendingReceive::PendingReceive(
	DataReceiver* receiver,
	const ReceiveDataCallback& callback,
	const ReceiveErrorCallback& error_callback,
	int32_t fatal_error_value)
	: receiver_(receiver),
	receive_callback_(callback),
	receive_error_callback_(error_callback),
	fatal_error_value_(fatal_error_value),
	buffer_in_use_(false) {
	}

	void DataReceiver::PendingReceive::DispatchData(const void* data,
	uint32_t num_bytes) {
	DCHECK(!buffer_in_use_);
	buffer_in_use_ = true;
	receive_callback_.Run(scoped_ptr<ReadOnlyBuffer>(
	new Buffer(receiver_, this, static_cast<const char*>(data), num_bytes)));
	}

	bool DataReceiver::PendingReceive::DispatchError(PendingError* error,
	uint32_t bytes_received) {
	DCHECK(!error->dispatched);
	if (buffer_in_use_ \|\| bytes_received != error->offset)
	return false;

	error->dispatched = true;
	receive_error_callback_.Run(error->error);
	return true;
	}

	void DataReceiver::PendingReceive::DispatchFatalError() {
	receive_error_callback_.Run(fatal_error_value_);
	}

	void DataReceiver::PendingReceive::Done(uint32_t bytes_consumed) {
	DCHECK(buffer_in_use_);
	buffer_in_use_ = false;
	receiver_->Done(bytes_consumed);
	}

	DataReceiver::PendingReceive::Buffer::Buffer(
	scoped_refptr<DataReceiver> receiver,
	PendingReceive* receive,
	const char* buffer,
	uint32_t buffer_size)
	: receiver_(receiver),
	pending_receive_(receive),
	buffer_(buffer),
	buffer_size_(buffer_size) {
	}

	DataReceiver::PendingReceive::Buffer::~Buffer() {
	if (pending_receive_)
	pending_receive_->Done(0);
	}

	const char* DataReceiver::PendingReceive::Buffer::GetData() {
	return buffer_;
	}

	uint32_t DataReceiver::PendingReceive::Buffer::GetSize() {
	return buffer_size_;
	}

	void DataReceiver::PendingReceive::Buffer::Done(uint32_t bytes_consumed) {
	pending_receive_->Done(bytes_consumed);
	pending_receive_ = NULL;
	receiver_ = NULL;
	buffer_ = NULL;
	buffer_size_ = 0;
	}

	void DataReceiver::PendingReceive::Buffer::DoneWithError(
	uint32_t bytes_consumed,
	int32_t error) {
	Done(bytes_consumed);
	}

	} // namespace device