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

device/serial/data_pipe_receiver.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_PIPE_RECEIVER_H_
+#define DEVICE_SERIAL_DATA_PIPE_RECEIVER_H_
+
+#include "base/callback.h"
+#include "base/memory/ref_counted.h"
+#include "base/memory/weak_ptr.h"
+#include "device/serial/buffer.h"
+#include "device/serial/data_pipe.mojom.h"
+#include "mojo/public/cpp/system/data_pipe.h"
+
+namespace device {
+
+class AsyncWaiter;
+
+// A wrapper around the consumer end of a data pipe.
+class DataPipeReceiver : public base::RefCounted<DataPipeReceiver>,
+                         public serial::DataPipeReceiver,
+                         public mojo::ErrorHandler {
+ public:
+  typedef base::Callback<void(scoped_ptr<ReadOnlyBuffer>)> ReceiveDataCallback;
+  typedef base::Callback<void(int32_t error)> ReceiveErrorCallback;
+
+  // Constructs a DataPipeReceiver to receive data from |producer|, using a data
+  // pipe with a buffer size of |buffer_size|, with connections errors reported

[raymes, 2014/08/05 06:26:45]

connections->connection

[Sam McNally, 2014/08/05 07:26:33]

Done.

+  // as |connection_error_value|.
+  DataPipeReceiver(mojo::InterfacePtr<serial::DataPipeProducer> producer,
+                   uint32_t buffer_size,
+                   int32_t connection_error_value);
+
+  // Begins a receive operation. If this returns true, exactly one of |callback|
+  // and |error_callback| will eventually be called. If there is already a
+  // receive in progress, |callback| or |error_callback| is invalid or there has
+  // been a connection error, this will have no effect and return false.
+  bool Receive(const ReceiveDataCallback& callback,
+               const ReceiveErrorCallback& error_callback);
+
+ private:
+  class Buffer;
+  friend class base::RefCounted<DataPipeReceiver>;
+
+  enum State {
+    STATE_IDLE,
+    STATE_WAITING_FOR_DATA,
+    STATE_WAITING_FOR_BUFFER,
+    STATE_PAUSED,
+    STATE_SHUT_DOWN,
+  };
+
+  virtual ~DataPipeReceiver();
+  void Done(uint32_t bytes_consumed);
+  void OnDoneWaiting(MojoResult result);
+
+  void ReceiveInternal();
+  void RetryReceive();
+  void DispatchData(const void* data, uint32_t num_bytes);
+  void DispatchError(int32_t error);
+
+  // serial::DataPipeReceiver override.
+  virtual void OnError(uint32_t bytes_since_last_error, int32_t error) OVERRIDE;
+
+  // mojo::ErrorHandler override.
+  virtual void OnConnectionError() OVERRIDE;
+
+  mojo::InterfacePtr<serial::DataPipeProducer> producer_;
+  const int32_t connection_error_value_;
+  mojo::ScopedDataPipeConsumerHandle handle_;
+  ReceiveDataCallback receive_callback_;
+  ReceiveErrorCallback receive_error_callback_;
+  uint32_t bytes_since_last_error_;
+  bool pending_error_;
+  uint32_t error_offset_;
+  int32_t error_;
+  State state_;
+  scoped_ptr<AsyncWaiter> waiter_;
+  base::WeakPtrFactory<DataPipeReceiver> weak_factory_;
+
+  DISALLOW_COPY_AND_ASSIGN(DataPipeReceiver);
+};
+
+}  // namespace device
+
+#endif  // DEVICE_SERIAL_DATA_PIPE_RECEIVER_H_