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

device/serial/data_source_sender.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_source_sender.h"
+
+#include <limits>
+
+#include "base/bind.h"
+#include "base/message_loop/message_loop.h"
+#include "device/serial/async_waiter.h"
+
+namespace device {
+
+// Represents a send that is not yet fulfilled.
+class DataSourceSender::PendingSend {
+ public:
+  PendingSend(DataSourceSender* sender, const ReadyCallback& callback);
+
+  // Asynchronously fills |data| with up to |num_bytes| of data. Following this,
+  // one of Done() and DoneWithError() will be called with the result.
+  void GetData(void* data, uint32_t num_bytes);
+
+ private:
+  class Buffer;
+  // Reports a successful write of |bytes_written|.
+  void Done(uint32_t bytes_written);
+
+  // Reports a partially successful or unsuccessful write of |bytes_written|
+  // with an error of |error|.
+  void DoneWithError(uint32_t bytes_written, int32_t error);
+
+  // The DataSourceSender that owns this.
+  DataSourceSender* sender_;
+
+  // The callback to call to get data.
+  ReadyCallback callback_;
+
+  // Whether the buffer specified by GetData() has been passed to |callback_|,
+  // but has not yet called Done() or DoneWithError().
+  bool buffer_in_use_;
+};
+
+// A Writable implementation that provides a view of a data pipe owned by a
+// DataSourceSender.
+class DataSourceSender::PendingSend::Buffer : public WritableBuffer {
+ public:
+  Buffer(scoped_refptr<DataSourceSender> sender,
+         PendingSend* send,
+         char* buffer,
+         uint32_t buffer_size);
+  virtual ~Buffer();
+
+  // WritableBuffer overrides.
+  virtual char* GetData() OVERRIDE;
+  virtual uint32_t GetSize() OVERRIDE;
+  virtual void Done(uint32_t bytes_written) OVERRIDE;
+  virtual void DoneWithError(uint32_t bytes_written, int32_t error) OVERRIDE;
+
+ private:
+  // The DataSourceSender whose data pipe we are providing a view.
+  scoped_refptr<DataSourceSender> sender_;
+
+  // The PendingSend to which this buffer has been created in response.
+  PendingSend* pending_send_;
+
+  char* buffer_;
+  uint32_t buffer_size_;
+};
+
+DataSourceSender::DataSourceSender(const ReadyCallback& ready_callback,
+                                   const ErrorCallback& error_callback)
+    : ready_callback_(ready_callback),
+      error_callback_(error_callback),
+      bytes_sent_(0),
+      shut_down_(false) {
+  DCHECK(!ready_callback.is_null() && !error_callback.is_null());
+}
+
+void DataSourceSender::ShutDown() {
+  shut_down_ = true;
+  waiter_.reset();
+  ready_callback_.Reset();
+  error_callback_.Reset();
+}
+
+DataSourceSender::~DataSourceSender() {
+  DCHECK(shut_down_);
+}
+
+void DataSourceSender::Init(mojo::ScopedDataPipeProducerHandle handle) {
+  // This should never occur. |handle_| is only valid and |pending_send_| is
+  // only set after Init is called. Receiving an invalid |handle| from the
+  // client is also unrecoverable.
+  if (pending_send_ || handle_.is_valid() || !handle.is_valid() || shut_down_) {
+    DispatchFatalError();
+    return;
+  }
+  handle_ = handle.Pass();
+  pending_send_.reset(new PendingSend(this, ready_callback_));
+  StartWaiting();
+}
+
+void DataSourceSender::Resume() {
+  if (pending_send_ || !handle_.is_valid()) {
+    DispatchFatalError();
+    return;
+  }
+
+  pending_send_.reset(new PendingSend(this, ready_callback_));
+  StartWaiting();
+}
+
+void DataSourceSender::OnConnectionError() {
+  DispatchFatalError();
+}
+
+void DataSourceSender::StartWaiting() {
+  DCHECK(pending_send_ && !waiter_);
+  waiter_.reset(
+      new AsyncWaiter(handle_.get(),
+                      MOJO_HANDLE_SIGNAL_WRITABLE,
+                      base::Bind(&DataSourceSender::OnDoneWaiting, this)));
+}
+
+void DataSourceSender::OnDoneWaiting(MojoResult result) {
+  DCHECK(pending_send_ && !shut_down_ && waiter_);
+  waiter_.reset();
+  if (result != MOJO_RESULT_OK) {
+    DispatchFatalError();
+    return;
+  }
+  void* data = NULL;
+  uint32_t num_bytes = std::numeric_limits<uint32_t>::max();
+  result = mojo::BeginWriteDataRaw(
+      handle_.get(), &data, &num_bytes, MOJO_WRITE_DATA_FLAG_NONE);
+  if (result != MOJO_RESULT_OK) {
+    DispatchFatalError();
+    return;
+  }
+  pending_send_->GetData(static_cast<char*>(data), num_bytes);
+}
+
+void DataSourceSender::Done(uint32_t bytes_written) {
+  DoneInternal(bytes_written);
+  if (!shut_down_)
+    StartWaiting();
+}
+
+void DataSourceSender::DoneWithError(uint32_t bytes_written, int32_t error) {
+  DoneInternal(bytes_written);
+  pending_send_.reset();
+  if (!shut_down_)
+    client()->OnError(bytes_sent_, error);
+  // We don't call StartWaiting here so we don't send any additional data until
+  // Resume() is called.
+}
+
+void DataSourceSender::DoneInternal(uint32_t bytes_written) {
+  DCHECK(pending_send_);
+  if (shut_down_)
+    return;
+
+  bytes_sent_ += bytes_written;
+  MojoResult result = mojo::EndWriteDataRaw(handle_.get(), bytes_written);
+  if (result != MOJO_RESULT_OK) {
+    DispatchFatalError();
+    return;
+  }
+}
+
+void DataSourceSender::DispatchFatalError() {
+  if (shut_down_)
+    return;
+
+  error_callback_.Run();
+  ShutDown();
+}
+
+DataSourceSender::PendingSend::PendingSend(DataSourceSender* sender,
+                                           const ReadyCallback& callback)
+    : sender_(sender), callback_(callback), buffer_in_use_(false) {
+}
+
+void DataSourceSender::PendingSend::GetData(void* data, uint32_t num_bytes) {
+  DCHECK(!buffer_in_use_);
+  buffer_in_use_ = true;
+  callback_.Run(scoped_ptr<WritableBuffer>(
+      new Buffer(sender_, this, static_cast<char*>(data), num_bytes)));
+}
+
+void DataSourceSender::PendingSend::Done(uint32_t bytes_written) {
+  DCHECK(buffer_in_use_);
+  buffer_in_use_ = false;
+  sender_->Done(bytes_written);
+}
+
+void DataSourceSender::PendingSend::DoneWithError(uint32_t bytes_written,
+                                                  int32_t error) {
+  DCHECK(buffer_in_use_);
+  buffer_in_use_ = false;
+  sender_->DoneWithError(bytes_written, error);
+}
+
+DataSourceSender::PendingSend::Buffer::Buffer(
+    scoped_refptr<DataSourceSender> sender,
+    PendingSend* send,
+    char* buffer,
+    uint32_t buffer_size)
+    : sender_(sender),
+      pending_send_(send),
+      buffer_(buffer),
+      buffer_size_(buffer_size) {
+}
+
+DataSourceSender::PendingSend::Buffer::~Buffer() {
+  if (sender_)
+    pending_send_->Done(0);
+}
+
+char* DataSourceSender::PendingSend::Buffer::GetData() {
+  return buffer_;
+}
+
+uint32_t DataSourceSender::PendingSend::Buffer::GetSize() {
+  return buffer_size_;
+}
+
+void DataSourceSender::PendingSend::Buffer::Done(uint32_t bytes_written) {
+  DCHECK(sender_);
+  pending_send_->Done(bytes_written);
+  sender_ = NULL;
+  pending_send_ = NULL;
+  buffer_ = NULL;
+  buffer_size_ = 0;
+}
+
+void DataSourceSender::PendingSend::Buffer::DoneWithError(
+    uint32_t bytes_written,
+    int32_t error) {
+  DCHECK(sender_);
+  pending_send_->DoneWithError(bytes_written, error);
+  sender_ = NULL;
+  pending_send_ = NULL;
+  buffer_ = NULL;
+  buffer_size_ = 0;
+}
+
+}  // namespace device