Add data pipe wrappers to be used to implement serial receive.

device/serial/data_receiver.cc

+// 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 {
+
+class DataReceiver::Buffer : public ReadOnlyBuffer {
+ public:
+  Buffer(scoped_refptr<DataReceiver> pipe,
+         const char* buffer,
+         uint32_t buffer_size);
+  virtual ~Buffer();
+  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:
+  scoped_refptr<DataReceiver> pipe_;
+  const char* buffer_;
+  uint32_t buffer_size_;
+};
+
+DataReceiver::DataReceiver(mojo::InterfacePtr<serial::DataSource> source,
+                           uint32_t buffer_size,
+                           int32_t connection_error_value)
+    : source_(source.Pass()),
+      connection_error_value_(connection_error_value),
+      bytes_received_(0),
+      pending_error_(false),
+      error_offset_(0),
+      error_(0),
+      state_(STATE_IDLE),
+      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);
+}
+
+// This is part of the public interface so can be called while we are in any
+// state.
+bool DataReceiver::Receive(const ReceiveDataCallback& callback,
+                           const ReceiveErrorCallback& error_callback) {
+  DCHECK(!callback.is_null() && !error_callback.is_null());
+  if (state_ == STATE_PAUSED) {
+    source_->Resume();
+    state_ = STATE_IDLE;
+  }
+  if (state_ != STATE_IDLE)
+    return false;
+
+  state_ = STATE_WAITING_FOR_DATA;
+  receive_callback_ = callback;
+  receive_error_callback_ = error_callback;
+  base::MessageLoop::current()->PostTask(
+      FROM_HERE,
+      base::Bind(&DataReceiver::ReceiveInternal, weak_factory_.GetWeakPtr()));
+  return true;
+}
+
+DataReceiver::~DataReceiver() {
+  DCHECK(state_ == STATE_IDLE || state_ == STATE_PAUSED ||
+         state_ == STATE_WAITING_FOR_DATA || state_ == STATE_SHUT_DOWN);
+  if (state_ == STATE_WAITING_FOR_DATA) {
+    state_ = STATE_SHUT_DOWN;
+    DispatchError(connection_error_value_);
+  }
+}
+
+void DataReceiver::Done(uint32_t bytes_consumed) {
+  DCHECK(state_ == STATE_WAITING_FOR_BUFFER || state_ == STATE_SHUT_DOWN);
+  if (state_ == STATE_SHUT_DOWN)
+    return;
+
+  MojoResult result = mojo::EndReadDataRaw(handle_.get(), bytes_consumed);
+  DCHECK_EQ(MOJO_RESULT_OK, result);
+  bytes_received_ += bytes_consumed;

[raymes, 2014/08/06 00:13:36]

We should make sure that this never exceeds error_offset_ under normal
operation. 

[Sam McNally, 2014/08/06 03:51:30]

Done.

+  state_ = STATE_IDLE;
+}
+
+void DataReceiver::OnDoneWaiting(MojoResult result) {
+  DCHECK_EQ(state_, STATE_WAITING_FOR_DATA);

[raymes, 2014/08/06 00:13:36]

could we be shutdown here? I'm not sure?

[Sam McNally, 2014/08/06 03:51:30]

No, when waiting is cancelled when shutting down.

+  if (result != MOJO_RESULT_OK) {
+    state_ = STATE_SHUT_DOWN;
+    DispatchError(connection_error_value_);
+    return;
+  }
+  ReceiveInternal();
+}
+
+// This is part of the DataSourceClient interface so can be called while we are
+// in any state.
+void DataReceiver::OnError(uint32_t bytes_since_last_error, int32_t error) {

[raymes, 2014/08/06 00:13:36]

We should know that we aren't in the paused state here and also that
pending_error isn't already true. This would only happen if the other side of
the connection didn't behave as expected, so we could bail in that case.

[Sam McNally, 2014/08/06 03:51:30]

Done.

+  pending_error_ = true;
+  error_ = error;
+  error_offset_ = bytes_since_last_error;
+  if (state_ == STATE_WAITING_FOR_DATA)
+    ReceiveInternal();
+}
+
+// This is invoked in the case of a connection error so can be called while we
+// are in any state.
+void DataReceiver::OnConnectionError() {
+  State old_state = state_;
+  state_ = STATE_SHUT_DOWN;
+  if (old_state == STATE_WAITING_FOR_DATA)
+    DispatchError(connection_error_value_);
+  waiter_.reset();
+}
+
+void DataReceiver::ReceiveInternal() {
+  DCHECK_EQ(state_, STATE_WAITING_FOR_DATA);
+  if (pending_error_ && bytes_received_ == error_offset_) {
+    pending_error_ = false;
+    error_offset_ = 0;
+    state_ = STATE_PAUSED;
+    DispatchError(error_);
+    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) {
+    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;
+  }
+  waiter_.reset();
+  state_ = STATE_SHUT_DOWN;
+  DispatchError(connection_error_value_);
+}
+
+void DataReceiver::DispatchData(const void* data, uint32_t num_bytes) {
+  DCHECK(state_ == STATE_WAITING_FOR_DATA);
+  state_ = STATE_WAITING_FOR_BUFFER;
+  ReceiveDataCallback callback = receive_callback_;
+  receive_callback_.Reset();
+  receive_error_callback_.Reset();
+  callback.Run(scoped_ptr<ReadOnlyBuffer>(
+      new Buffer(this, static_cast<const char*>(data), num_bytes)));
+}
+
+void DataReceiver::DispatchError(int32_t error) {
+  DCHECK(state_ == STATE_PAUSED || state_ == STATE_SHUT_DOWN);
+  waiter_.reset();
+  ReceiveErrorCallback callback = receive_error_callback_;
+  receive_callback_.Reset();
+  receive_error_callback_.Reset();
+  callback.Run(error);
+}
+
+DataReceiver::Buffer::Buffer(scoped_refptr<DataReceiver> pipe,
+                             const char* buffer,
+                             uint32_t buffer_size)
+    : pipe_(pipe), buffer_(buffer), buffer_size_(buffer_size) {
+}
+
+DataReceiver::Buffer::~Buffer() {
+  if (pipe_)
+    pipe_->Done(0);
+}
+
+const char* DataReceiver::Buffer::GetData() {
+  return buffer_;
+}
+
+uint32_t DataReceiver::Buffer::GetSize() {
+  return buffer_size_;
+}
+
+void DataReceiver::Buffer::Done(uint32_t bytes_consumed) {
+  pipe_->Done(bytes_consumed);
+  pipe_ = NULL;
+  buffer_ = NULL;
+  buffer_size_ = 0;
+}
+
+void DataReceiver::Buffer::DoneWithError(uint32_t bytes_consumed,
+                                         int32_t error) {
+  Done(bytes_consumed);
+}
+
+}  // namespace device