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

device/serial/data_source_sender.cc

Index: device/serial/data_source_sender.cc
diff --git a/device/serial/data_source_sender.cc b/device/serial/data_source_sender.cc
new file mode 100644
index 0000000000000000000000000000000000000000..eacde5713d92d67b9f32d5008e0415dd3c2669f0
--- /dev/null
+++ b/device/serial/data_source_sender.cc
@@ -0,0 +1,168 @@
+// 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_source_sender.h"
+
+#include <limits>
+
+#include "base/bind.h"
+#include "base/message_loop/message_loop.h"
+#include "device/serial/async_waiter.h"
+
+namespace device {
+
+class DataSourceSender::Buffer : public WritableBuffer {
+ public:
+  Buffer(scoped_refptr<DataSourceSender> pipe,
+         char* buffer,
+         uint32_t buffer_size);
+  virtual ~Buffer();
+  virtual char* GetData() OVERRIDE;
+  virtual uint32_t GetSize() OVERRIDE;
+  virtual void Done(uint32_t bytes_produced) OVERRIDE;
+  virtual void DoneWithError(uint32_t bytes_produced, int32_t error) OVERRIDE;
+
+ private:
+  scoped_refptr<DataSourceSender> pipe_;
+  char* buffer_;
+  uint32_t buffer_size_;
+};
+
+DataSourceSender::DataSourceSender(const ReadyCallback& ready_callback,
+                                   const ErrorCallback& error_callback)
+    : ready_callback_(ready_callback),
+      error_callback_(error_callback),
+      state_(STATE_UNINITIALIZED),
+      bytes_sent_(0) {
+  DCHECK(!ready_callback.is_null() && !error_callback.is_null());
+}
+
+// This is part of the public interface so can be called while we are in any
+// state.
+void DataSourceSender::Shutdown() {
+  state_ = STATE_SHUT_DOWN;
+  waiter_.reset();
+}
+
+DataSourceSender::~DataSourceSender() {
+  DCHECK_EQ(state_, STATE_SHUT_DOWN);
+}
+
+// This is part of the DataSource interface so can be called while we are in any
+// state.
+void DataSourceSender::Init(mojo::ScopedDataPipeProducerHandle handle) {
+  if (state_ != STATE_UNINITIALIZED)
+    return;
+
+  handle_ = handle.Pass();
+  StartWaiting();
+}
+
+// This is part of the DataSource interface so can be called while we are in any
+// state.
+void DataSourceSender::Resume() {
+  if (state_ != STATE_PAUSED)
+    return;
+
+  StartWaiting();
+}
+
+// This is invoked in the case of a connection error so can be called while we
+// are in any state.
+void DataSourceSender::OnConnectionError() {
+  HandleMojoResult(MOJO_RESULT_CANCELLED);
+}
+
+void DataSourceSender::StartWaiting() {
+  state_ = STATE_WAITING_FOR_SPACE;
+  waiter_.reset(
+      new AsyncWaiter(handle_.get(),
+                      MOJO_HANDLE_SIGNAL_WRITABLE,
+                      base::Bind(&DataSourceSender::OnDoneWaiting, this)));
+}
+
+void DataSourceSender::OnDoneWaiting(MojoResult result) {
+  DCHECK(state_ == STATE_WAITING_FOR_SPACE);
+  if (!HandleMojoResult(result))
+    return;
+  void* data = NULL;
+  uint32_t num_bytes = std::numeric_limits<uint32_t>::max();
+  result = mojo::BeginWriteDataRaw(
+      handle_.get(), &data, &num_bytes, MOJO_READ_DATA_FLAG_NONE);
+  if (!HandleMojoResult(result))
+    return;
+  state_ = STATE_WAITING_FOR_BUFFER;
+  ready_callback_.Run(scoped_ptr<WritableBuffer>(
+      new Buffer(this, static_cast<char*>(data), num_bytes)));
+}
+
+void DataSourceSender::Done(uint32_t bytes_produced) {
+  DCHECK(state_ == STATE_WAITING_FOR_BUFFER || state_ == STATE_PAUSED ||
+         state_ == STATE_SHUT_DOWN);
+  bytes_sent_ += bytes_produced;
+  MojoResult result = mojo::EndWriteDataRaw(handle_.get(), bytes_produced);
+  if (!HandleMojoResult(result))
+    return;
+  if (state_ == STATE_WAITING_FOR_BUFFER)
+    StartWaiting();
+}
+
+void DataSourceSender::DoneWithError(uint32_t bytes_produced, int32_t error) {
+  DCHECK(state_ == STATE_WAITING_FOR_BUFFER || state_ == STATE_SHUT_DOWN);
+  if (state_ == STATE_WAITING_FOR_BUFFER)
+    state_ = STATE_PAUSED;
+  Done(bytes_produced);
+  client()->OnError(bytes_sent_, error);
+}
+
+bool DataSourceSender::HandleMojoResult(MojoResult result) {
+  DCHECK(state_ != STATE_UNINITIALIZED);
+  if (result == MOJO_RESULT_OK)
+    return true;
+
+  if (state_ == STATE_SHUT_DOWN)
+    return false;
+
+  Shutdown();
+  error_callback_.Run();
+  return false;
+}
+
+DataSourceSender::Buffer::Buffer(scoped_refptr<DataSourceSender> pipe,
+                                 char* buffer,
+                                 uint32_t buffer_size)
+    : pipe_(pipe), buffer_(buffer), buffer_size_(buffer_size) {
+}
+
+DataSourceSender::Buffer::~Buffer() {
+  if (pipe_)
+    pipe_->Done(0);
+}
+
+char* DataSourceSender::Buffer::GetData() {
+  return buffer_;
+}
+
+uint32_t DataSourceSender::Buffer::GetSize() {
+  return buffer_size_;
+}
+
+void DataSourceSender::Buffer::Done(uint32_t bytes_produced) {
+  DCHECK(pipe_);
+  pipe_->Done(bytes_produced);
+  pipe_ = NULL;
+  buffer_ = NULL;
+  buffer_size_ = 0;
+}
+
+void DataSourceSender::Buffer::DoneWithError(uint32_t bytes_produced,
+                                             int32_t error) {
+  DCHECK(pipe_);
+  pipe_->DoneWithError(bytes_produced, error);
+  pipe_ = NULL;
+  buffer_ = NULL;
+  buffer_size_ = 0;
+}
+
+}  // namespace device