Add MessagePipe interface. Use it in ChannelDispatcherBase.

remoting/protocol/stream_message_pipe_adapter.h

Index: remoting/protocol/stream_message_pipe_adapter.h
diff --git a/remoting/protocol/stream_message_pipe_adapter.h b/remoting/protocol/stream_message_pipe_adapter.h
new file mode 100644
index 0000000000000000000000000000000000000000..1de218f04662cc40032d04ed15d032891d9f9551
--- /dev/null
+++ b/remoting/protocol/stream_message_pipe_adapter.h
@@ -0,0 +1,54 @@
+// Copyright 2016 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 REMOTING_PROTOCOL_STREAM_MESSAGE_PIPE_ADAPTER_H_
+#define REMOTING_PROTOCOL_STREAM_MESSAGE_PIPE_ADAPTER_H_
+
+#include "remoting/protocol/message_pipe.h"
+
+#include "base/callback.h"
+#include "remoting/base/buffered_socket_writer.h"
+#include "remoting/protocol/message_reader.h"
+
+namespace remoting {
+namespace protocol {
+
+class P2PStreamSocket;
+
+// MessagePipe implementation that sends and receives messages over a
+// P2PStreamChannel.
+class StreamMessagePipeAdapter : public MessagePipe {
+ public:
+  typedef base::Callback<void(int)> ErrorCallback;
+
+  StreamMessagePipeAdapter();
+  ~StreamMessagePipeAdapter() override;
+
+  // Starts reading from |socket|.
+  void Initialize(scoped_ptr<P2PStreamSocket> socket,
+                  const ErrorCallback& error_callback);
+
+  // MessagePipe interface.
+  void StartReceiving(const MessageReceivedCallback& callback) override;
+  void Send(google::protobuf::MessageLite* message,
+            const base::Closure& done) override;
+
+ private:
+  void CloseOnError(int error);
+
+  scoped_ptr<P2PStreamSocket> socket_;
+  ErrorCallback error_callback_;
+
+  bool closed_ = false;
+
+  MessageReader reader_;
+  BufferedSocketWriter writer_;
+
+  DISALLOW_COPY_AND_ASSIGN(StreamMessagePipeAdapter);
+};
+
+}  // namespace protocol
+}  // namespace remoting
+
+#endif  // REMOTING_PROTOCOL_STREAM_MESSAGE_PIPE_ADAPTER_H_