Add MessagePipe interface. Use it in ChannelDispatcherBase.

remoting/protocol/stream_message_pipe_adapter.cc

Index: remoting/protocol/stream_message_pipe_adapter.cc
diff --git a/remoting/protocol/stream_message_pipe_adapter.cc b/remoting/protocol/stream_message_pipe_adapter.cc
new file mode 100644
index 0000000000000000000000000000000000000000..773a4043e1b9daae2baac48c2ebf862bc0d87f5e
--- /dev/null
+++ b/remoting/protocol/stream_message_pipe_adapter.cc
@@ -0,0 +1,57 @@
+// 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.
+
+#include "remoting/protocol/stream_message_pipe_adapter.h"
+
+#include <utility>
+
+#include "base/bind.h"
+#include "remoting/base/compound_buffer.h"
+#include "remoting/protocol/message_serialization.h"
+#include "remoting/protocol/p2p_stream_socket.h"
+
+namespace remoting {
+namespace protocol {
+
+StreamMessagePipeAdapter::StreamMessagePipeAdapter() {}
+StreamMessagePipeAdapter::~StreamMessagePipeAdapter() {}
+
+void StreamMessagePipeAdapter::Initialize(scoped_ptr<P2PStreamSocket> socket,
+                                          const ErrorCallback& error_callback) {
+  DCHECK(socket);
+  DCHECK(!error_callback.is_null());
+
+  socket_ = std::move(socket);
+  error_callback_ = error_callback;
+
+  writer_.Start(
+      base::Bind(&P2PStreamSocket::Write, base::Unretained(socket_.get())),
+      base::Bind(&StreamMessagePipeAdapter::CloseOnError,
+                 base::Unretained(this)));
+}
+
+void StreamMessagePipeAdapter::StartReceiving(
+    const MessageReceivedCallback& callback) {
+  reader_.StartReading(socket_.get(), callback,
+                       base::Bind(&StreamMessagePipeAdapter::CloseOnError,
+                                  base::Unretained(this)));
+}
+
+void StreamMessagePipeAdapter::Send(google::protobuf::MessageLite* message,
+                                    const base::Closure& done) {
+  if (closed_)
+    return;

[Jamie, 2016/02/01 22:48:20]

Call the error callback here? Otherwise the caller has no way of knowing that
done() will never be invoked. Alternatively, add a bool return if you don't want
to change the semantics of the error callback.

[Sergey Ulanov, 2016/02/02 21:16:42]

I think just dropping the callback here is the right thing to do because the
message was never sent. Send() could return result, but it's not very useful as
the callers have nothing to do with the result. Normally once a channel gets to
the error state the connection will be torn down. In my next CL I'm going to
remove ChannelDispatcherBase::EventHandler::OnChannelError(), so all channel
errors will be handled using IceTransport::EventHandler::OnIceTransportError().
It doesn't make sense to have both as they both used to handle transport-level
errors.

[Jamie, 2016/02/02 22:43:10]

Based on off-line discussion, it sounds like there are other, asynchronous,
situations that could result in the callback not being invoked. That being the
case, I agree that returning a bool is not the right thing to do. Can you update
the comment on MessagePipe::Send to make it clear that the callback will not be
invoked if the message is never sent?

[Sergey Ulanov, 2016/02/03 18:45:26]

Done.

+
+  writer_.Write(SerializeAndFrameMessage(*message), done);
+}
+
+void StreamMessagePipeAdapter::CloseOnError(int error) {
+  // Stop reading after any error.
+  closed_ = true;
+
+  error_callback_.Run(error);
+}
+
+}  // namespace protocol
+}  // namespace remoting