Add MessagePipe interface. Use it in ChannelDispatcherBase.

remoting/protocol/channel_dispatcher_base.cc

 // Copyright (c) 2012 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/channel_dispatcher_base.h"
 
 #include <utility>
 
 #include "base/bind.h"
 #include "remoting/protocol/p2p_stream_socket.h"
@@ skipping 23 matching lines @@
 }
 
 void ChannelDispatcherBase::OnChannelReady(
     scoped_ptr<P2PStreamSocket> socket) {
   if (!socket.get()) {
     event_handler_->OnChannelError(this, CHANNEL_CONNECTION_ERROR);
     return;
   }
 
   channel_factory_ = nullptr;
-  channel_ = std::move(socket);
-  writer_.Start(
-      base::Bind(&P2PStreamSocket::Write, base::Unretained(channel_.get())),
-      base::Bind(&ChannelDispatcherBase::OnReadWriteFailed,
-                 base::Unretained(this)));
-  reader_.StartReading(channel_.get(),
-                       base::Bind(&ChannelDispatcherBase::OnIncomingMessage,
-                                  base::Unretained(this)),
-                       base::Bind(&ChannelDispatcherBase::OnReadWriteFailed,
-                                  base::Unretained(this)));
+  message_pipe_.Initialize(
+      std::move(socket),
+      base::Bind(&ChannelDispatcherBase::OnPipeError, base::Unretained(this)));
+  message_pipe_.StartReceiving(base::Bind(
+      &ChannelDispatcherBase::OnIncomingMessage, base::Unretained(this)));
 
   event_handler_->OnChannelInitialized(this);
 }
 
 void ChannelDispatcherBase::OnIncomingMessage(
     scoped_ptr<CompoundBuffer> message) {
   // By default incoming message are not expected. Child classes override
   // OnIncomingMessage() if the want to handle incoming messages.
   LOG(ERROR) << "Received unexpected message on " << channel_name();
 }
 
-void ChannelDispatcherBase::OnReadWriteFailed(int error) {
+void ChannelDispatcherBase::OnPipeError(int error) {
   event_handler_->OnChannelError(this, CHANNEL_CONNECTION_ERROR);
 }
 
 }  // namespace protocol
 }  // namespace remoting