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

device/serial/data_source_sender.h

+// 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.
+
+#ifndef DEVICE_SERIAL_DATA_SOURCE_SENDER_H_
+#define DEVICE_SERIAL_DATA_SOURCE_SENDER_H_
+
+#include "base/callback.h"
+#include "base/memory/ref_counted.h"
+#include "device/serial/buffer.h"
+#include "device/serial/data_stream.mojom.h"
+#include "mojo/public/cpp/system/data_pipe.h"
+
+namespace device {
+
+class AsyncWaiter;
+
+// A DataSourceSender is an interface between a source of data and a data pipe.
+class DataSourceSender : public base::RefCounted<DataSourceSender>,
+                         public mojo::InterfaceImpl<serial::DataSource> {
+ public:
+  typedef base::Callback<void(scoped_ptr<WritableBuffer>)> ReadyCallback;
+  typedef base::Callback<void()> ErrorCallback;
+
+  // Constructs a DataSourceSender. Whenever the pipe is ready for writing, the
+  // |ready_callback| will be called with the WritableBuffer to be filled.
+  // |ready_callback| will not be called again until the previous WritableBuffer
+  // is destroyed. If a connection error occurs, |error_callback| will be
+  // called and the DataSourceSender will act as if Shutdown() had been called.
+  DataSourceSender(const ReadyCallback& ready_callback,
+                   const ErrorCallback& error_callback);
+
+  // Shuts down this DataSourceSender. After shut down, |ready_callback| and
+  // |error_callback| will never be called.
+  void Shutdown();
+
+ private:
+  class Buffer;
+  friend class base::RefCounted<DataSourceSender>;
+
+  enum State {
+    STATE_UNINITIALIZED,
+    STATE_WAITING_FOR_SPACE,
+    STATE_WAITING_FOR_BUFFER,
+    STATE_PAUSED,
+    STATE_SHUT_DOWN,
+  };
+
+  virtual ~DataSourceSender();
+
+  // mojo::InterfaceImpl<serial::DataSourceSender> overrides.
+  virtual void Init(mojo::ScopedDataPipeProducerHandle handle) OVERRIDE;
+  virtual void Resume() OVERRIDE;
+  virtual void OnConnectionError() OVERRIDE;
+
+  void StartWaiting();
+  void OnDoneWaiting(MojoResult result);
+
+  void Done(uint32_t bytes_produced);
+  void DoneWithError(uint32_t bytes_produced, int32_t error);
+  bool HandleMojoResult(MojoResult result);
+
+  mojo::ScopedDataPipeProducerHandle handle_;
+  const ReadyCallback ready_callback_;
+  const ErrorCallback error_callback_;
+  scoped_ptr<AsyncWaiter> waiter_;
+  State state_;
+  uint32_t bytes_since_last_error_;
+
+  DISALLOW_COPY_AND_ASSIGN(DataSourceSender);
+};
+
+}  // namespace device
+
+#endif  // DEVICE_SERIAL_DATA_SOURCE_SENDER_H_