Simplify data channel creation logic in WebRTC-based protocol

Previously WebrtcDataStreamAdapter was implementing MessageChannelFactory,
which means that it was always creating channels asynchronously and in OPEN
state. Refactored it to create channels in CONNECTING state. Significantly
simplifies channel setup code as it's always synchronous. It's also possible
to reject incoming channels before they are connected and handle channels
rejected by the peer before being fully connected.

Review-Url: https://codereview.chromium.org/2164163002
Cr-Commit-Position: refs/heads/master@{#407046}

remoting/protocol/channel_dispatcher_base.cc

@@ -42,7 +42,11 @@ void ChannelDispatcherBase::OnChannelReady(
   channel_factory_ = nullptr;
   message_pipe_ = std::move(message_pipe);
   message_pipe_->Start(this);
+}
 
+void ChannelDispatcherBase::OnMessagePipeOpen() {
+  DCHECK(!is_connected_);
+  is_connected_ = true;
   event_handler_->OnChannelInitialized(this);
 }
 
@@ -52,6 +56,7 @@ void ChannelDispatcherBase::OnMessageReceived(
 }
 
 void ChannelDispatcherBase::OnMessagePipeClosed() {
+  is_connected_ = false;
   event_handler_->OnChannelClosed(this);
 }
 

remoting/protocol/channel_dispatcher_base.h

@@ -57,7 +57,7 @@ class ChannelDispatcherBase : public MessagePipe::EventHandler {
   const std::string& channel_name() { return channel_name_; }
 
   // Returns true if the channel is currently connected.
-  bool is_connected() { return message_pipe() != nullptr; }
+  bool is_connected() { return is_connected_; }
 
  protected:
   explicit ChannelDispatcherBase(const char* channel_name);
@@ -71,12 +71,14 @@ class ChannelDispatcherBase : public MessagePipe::EventHandler {
   void OnChannelReady(std::unique_ptr<MessagePipe> message_pipe);
 
   // MessagePipe::EventHandler interface.
+  void OnMessagePipeOpen() override;
   void OnMessageReceived(std::unique_ptr<CompoundBuffer> message) override;
   void OnMessagePipeClosed() override;
 
   std::string channel_name_;
   MessageChannelFactory* channel_factory_ = nullptr;
   EventHandler* event_handler_ = nullptr;
+  bool is_connected_ = false;
 
   std::unique_ptr<MessagePipe> message_pipe_;
 

remoting/protocol/connection_tester.cc

@@ -273,7 +273,9 @@ MessagePipeConnectionTester::~MessagePipeConnectionTester() {}
 
 void MessagePipeConnectionTester::RunAndCheckResults() {
   host_pipe_->Start(this);
+}
 
+void MessagePipeConnectionTester::OnMessagePipeOpen() {
   for (int i = 0; i < message_count_; ++i) {
     std::unique_ptr<VideoPacket> message(new VideoPacket());
     message->mutable_data()->resize(message_size_);

remoting/protocol/connection_tester.h

@@ -121,6 +121,7 @@ class MessagePipeConnectionTester : public MessagePipe::EventHandler {
 
  protected:
   // MessagePipe::EventHandler interface.
+  void OnMessagePipeOpen() override;
   void OnMessageReceived(std::unique_ptr<CompoundBuffer> message) override;
   void OnMessagePipeClosed() override;
 

remoting/protocol/message_pipe.h

@@ -26,6 +26,9 @@ class MessagePipe {
  public:
   class EventHandler {
    public:
+    // Called when the channel is open.
+    virtual void OnMessagePipeOpen() = 0;
+
     // Called when a message is received.
     virtual void OnMessageReceived(std::unique_ptr<CompoundBuffer> message) = 0;
 
@@ -38,8 +41,9 @@ class MessagePipe {
 
   virtual ~MessagePipe() {}
 
-  // Starts the channel. |event_handler| will be called to notify when a message
-  // is received or the pipe is closed.
+  // Starts the channel. Must be called immediately after MessagePipe is
+  // created. |event_handler| will be notified when state of the pipe changes or
+  // when a message is received.
   virtual void Start(EventHandler* event_handler) = 0;
 
   // Sends a message. |done| is called when the message has been sent to the

remoting/protocol/stream_message_pipe_adapter.cc

@@ -42,6 +42,7 @@ void StreamMessagePipeAdapter::Start(EventHandler* event_handler) {
                                   base::Unretained(event_handler)),
                        base::Bind(&StreamMessagePipeAdapter::CloseOnError,
                                   base::Unretained(this)));
+  event_handler->OnMessagePipeOpen();
 }
 
 void StreamMessagePipeAdapter::Send(google::protobuf::MessageLite* message,

remoting/protocol/webrtc_connection_to_client.cc

@@ -155,10 +155,11 @@ void WebrtcConnectionToClient::OnSessionStateChange(Session::State state) {
 }
 
 void WebrtcConnectionToClient::OnWebrtcTransportConnecting() {
-  // Create outgoing control channel by initializing |control_dispatcher_|.
-  // |event_dispatcher_| is initialized later because event channel is expected
-  // to be created by the client.
-  control_dispatcher_->Init(transport_->outgoing_channel_factory(), this);
+  // Create outgoing control channel. |event_dispatcher_| is initialized later
+  // because event channel is expected to be created by the client.
+  control_dispatcher_->Init(
+      transport_->CreateOutgoingChannel(control_dispatcher_->channel_name()),
+      this);
 }
 
 void WebrtcConnectionToClient::OnWebrtcTransportConnected() {

remoting/protocol/webrtc_connection_to_host.cc

@@ -104,7 +104,9 @@ void WebrtcConnectionToHost::OnSessionStateChange(Session::State state) {
 
 void WebrtcConnectionToHost::OnWebrtcTransportConnecting() {
   event_dispatcher_.reset(new ClientEventDispatcher());
-  event_dispatcher_->Init(transport_->outgoing_channel_factory(), this);
+  event_dispatcher_->Init(
+      transport_->CreateOutgoingChannel(event_dispatcher_->channel_name()),
+      this);
 }
 
 void WebrtcConnectionToHost::OnWebrtcTransportConnected() {}

remoting/protocol/webrtc_data_stream_adapter.cc

@@ -21,13 +21,14 @@
 namespace remoting {
 namespace protocol {
 
-class WebrtcDataStreamAdapter::Channel : public MessagePipe,
-                                         public webrtc::DataChannelObserver {
- public:
-  explicit Channel(WebrtcDataStreamAdapter* adapter);
-  ~Channel() override;
+namespace {
 
-  void Start(rtc::scoped_refptr<webrtc::DataChannelInterface> channel);
+class WebrtcDataChannel : public MessagePipe,
+                          public webrtc::DataChannelObserver {
+ public:
+  explicit WebrtcDataChannel(
+      rtc::scoped_refptr<webrtc::DataChannelInterface> channel);
+  ~WebrtcDataChannel() override;
 
   std::string name() { return channel_->label(); }
 
@@ -47,52 +48,38 @@ class WebrtcDataStreamAdapter::Channel : public MessagePipe,
 
   void OnClosed();
 
-  // |adapter_| owns channels while they are being connected.
-  WebrtcDataStreamAdapter* adapter_;
-
   rtc::scoped_refptr<webrtc::DataChannelInterface> channel_;
 
   EventHandler* event_handler_ = nullptr;
 
   State state_ = State::CONNECTING;
 
-  DISALLOW_COPY_AND_ASSIGN(Channel);
+  DISALLOW_COPY_AND_ASSIGN(WebrtcDataChannel);
 };
 
-WebrtcDataStreamAdapter::Channel::Channel(WebrtcDataStreamAdapter* adapter)
-    : adapter_(adapter) {}
+WebrtcDataChannel::WebrtcDataChannel(
+    rtc::scoped_refptr<webrtc::DataChannelInterface> channel)
+    : channel_(channel) {
+  channel_->RegisterObserver(this);
+  DCHECK_EQ(channel_->state(), webrtc::DataChannelInterface::kConnecting);
+}
 
-WebrtcDataStreamAdapter::Channel::~Channel() {
+WebrtcDataChannel::~WebrtcDataChannel() {
   if (channel_) {
     channel_->UnregisterObserver();
     channel_->Close();
   }
 }
 
-void WebrtcDataStreamAdapter::Channel::Start(
-    rtc::scoped_refptr<webrtc::DataChannelInterface> channel) {
-  DCHECK(!channel_);
-
-  channel_ = channel;
-  channel_->RegisterObserver(this);
-
-   if (channel_->state() == webrtc::DataChannelInterface::kOpen) {
-    OnConnected();
-  } else {
-    DCHECK_EQ(channel_->state(), webrtc::DataChannelInterface::kConnecting);
-  }
-}
-
-void WebrtcDataStreamAdapter::Channel::Start(EventHandler* event_handler) {
+void WebrtcDataChannel::Start(EventHandler* event_handler) {
   DCHECK(!event_handler_);
   DCHECK(event_handler);
 
   event_handler_ = event_handler;
 }
 
-void WebrtcDataStreamAdapter::Channel::Send(
-    google::protobuf::MessageLite* message,
-    const base::Closure& done) {
+void WebrtcDataChannel::Send(google::protobuf::MessageLite* message,
+                             const base::Closure& done) {
   DCHECK(state_ == State::OPEN);
 
   rtc::CopyOnWriteBuffer buffer;
@@ -110,14 +97,19 @@ void WebrtcDataStreamAdapter::Channel::Send(
     done.Run();
 }
 
-void WebrtcDataStreamAdapter::Channel::OnStateChange() {
+void WebrtcDataChannel::OnStateChange() {
   switch (channel_->state()) {
     case webrtc::DataChannelInterface::kOpen:
-      OnConnected();
+      DCHECK(state_ == State::CONNECTING);
+      state_ = State::OPEN;
+      event_handler_->OnMessagePipeOpen();
       break;
 
     case webrtc::DataChannelInterface::kClosing:
-      OnClosed();
+      if (state_ != State::CLOSED) {
+        state_ = State::CLOSED;
+        event_handler_->OnMessagePipeClosed();
+      }
       break;
 
     case webrtc::DataChannelInterface::kConnecting:
@@ -126,8 +118,7 @@ void WebrtcDataStreamAdapter::Channel::OnStateChange() {
   }
 }
 
-void WebrtcDataStreamAdapter::Channel::OnMessage(
-    const webrtc::DataBuffer& rtc_buffer) {
+void WebrtcDataChannel::OnMessage(const webrtc::DataBuffer& rtc_buffer) {
   if (state_ != State::OPEN) {
     LOG(ERROR) << "Dropping a message received when the channel is not open.";
     return;
@@ -140,114 +131,24 @@ void WebrtcDataStreamAdapter::Channel::OnMessage(
   event_handler_->OnMessageReceived(std::move(buffer));
 }
 
-void WebrtcDataStreamAdapter::Channel::OnConnected() {
-  DCHECK(state_ == State::CONNECTING);
-  state_ = State::OPEN;
-  WebrtcDataStreamAdapter* adapter = adapter_;
-  adapter_ = nullptr;
-  adapter->OnChannelConnected(this);
-}
-
-void WebrtcDataStreamAdapter::Channel::OnClosed() {
-  switch (state_) {
-    case State::CONNECTING:
-      state_ = State::CLOSED;
-      LOG(WARNING) << "Channel " << channel_->label()
-                   << " was closed before it's connected.";
-      adapter_->OnChannelError();
-      return;
-
-    case State::OPEN:
-      state_ = State::CLOSED;
-      event_handler_->OnMessagePipeClosed();
-      return;
-
-    case State::CLOSED:
-      break;
-  }
-}
-
-struct WebrtcDataStreamAdapter::PendingChannel {
-  PendingChannel(std::unique_ptr<Channel> channel,
-                 const ChannelCreatedCallback& connected_callback)
-      : channel(std::move(channel)), connected_callback(connected_callback) {}
-  PendingChannel(PendingChannel&& other)
-      : channel(std::move(other.channel)),
-        connected_callback(std::move(other.connected_callback)) {}
-  PendingChannel& operator=(PendingChannel&& other) {
-    channel = std::move(other.channel);
-    connected_callback = std::move(other.connected_callback);
-    return *this;
-  }
-
-  std::unique_ptr<Channel> channel;
-  ChannelCreatedCallback connected_callback;
-};
+}  // namespace
 
 WebrtcDataStreamAdapter::WebrtcDataStreamAdapter(
-    const ErrorCallback& error_callback)
-    : error_callback_(error_callback) {}
+    rtc::scoped_refptr<webrtc::PeerConnectionInterface> peer_connection)
+    : peer_connection_(peer_connection) {}
+WebrtcDataStreamAdapter::~WebrtcDataStreamAdapter() {}
 
-WebrtcDataStreamAdapter::~WebrtcDataStreamAdapter() {
-  DCHECK(pending_channels_.empty());
-}
-
-void WebrtcDataStreamAdapter::Initialize(
-    rtc::scoped_refptr<webrtc::PeerConnectionInterface> peer_connection) {
-  peer_connection_ = peer_connection;
-}
-
-void WebrtcDataStreamAdapter::WrapIncomingDataChannel(
-    rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel,
-    const ChannelCreatedCallback& callback) {
-  AddPendingChannel(data_channel, callback);
-}
-
-void WebrtcDataStreamAdapter::CreateChannel(
-    const std::string& name,
-    const ChannelCreatedCallback& callback) {
+std::unique_ptr<MessagePipe> WebrtcDataStreamAdapter::CreateOutgoingChannel(
+    const std::string& name) {
   webrtc::DataChannelInit config;
   config.reliable = true;
-  AddPendingChannel(peer_connection_->CreateDataChannel(name, &config),
-                    callback);
-}
-
-void WebrtcDataStreamAdapter::CancelChannelCreation(const std::string& name) {
-  auto it = pending_channels_.find(name);
-  DCHECK(it != pending_channels_.end());
-  pending_channels_.erase(it);
-}
-
-void WebrtcDataStreamAdapter::AddPendingChannel(
-    rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel,
-    const ChannelCreatedCallback& callback) {
-  DCHECK(peer_connection_);
-  DCHECK(pending_channels_.find(data_channel->label()) ==
-         pending_channels_.end());
-
-  Channel* channel = new Channel(this);
-  pending_channels_.insert(
-      std::make_pair(data_channel->label(),
-                     PendingChannel(base::WrapUnique(channel), callback)));
-
-  channel->Start(data_channel);
-}
-
-void WebrtcDataStreamAdapter::OnChannelConnected(Channel* channel) {
-  auto it = pending_channels_.find(channel->name());
-  DCHECK(it != pending_channels_.end());
-
-  PendingChannel pending_channel = std::move(it->second);
-  pending_channels_.erase(it);
-
-  // Once the channel is connected its ownership  is passed to the
-  // |connected_callback|.
-  pending_channel.connected_callback.Run(std::move(pending_channel.channel));
+  return base::WrapUnique(new WebrtcDataChannel(
+      peer_connection_->CreateDataChannel(name, &config)));
 }
 
-void WebrtcDataStreamAdapter::OnChannelError() {
-  pending_channels_.clear();
-  error_callback_.Run(CHANNEL_CONNECTION_ERROR);
+std::unique_ptr<MessagePipe> WebrtcDataStreamAdapter::WrapIncomingDataChannel(
+    rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel) {
+  return base::WrapUnique(new WebrtcDataChannel(data_channel));
 }
 
 }  // namespace protocol

remoting/protocol/webrtc_data_stream_adapter.h

@@ -5,6 +5,7 @@
 #ifndef REMOTING_PROTOCOL_WEBRTC_DATA_STREAM_ADAPTER_H_
 #define REMOTING_PROTOCOL_WEBRTC_DATA_STREAM_ADAPTER_H_
 
+#include <memory>
 #include <string>
 
 #include "base/callback.h"
@@ -21,50 +22,25 @@ class PeerConnectionInterface;
 namespace remoting {
 namespace protocol {
 
-// WebrtcDataStreamAdapter is a MessageChannelFactory that creates channels that
-// send and receive messages over PeerConnection data channels.
-class WebrtcDataStreamAdapter : public MessageChannelFactory {
+// WebrtcDataStreamAdapter wraps MessagePipe for WebRTC data channels.
+class WebrtcDataStreamAdapter {
  public:
   typedef base::Callback<void(ErrorCode)> ErrorCallback;
 
-  explicit WebrtcDataStreamAdapter(const ErrorCallback& error_callback);
-  ~WebrtcDataStreamAdapter() override;
-
-  // Initializes the adapter for |peer_connection|. If |outgoing| is set to true
-  // all channels will be created as outgoing. Otherwise CreateChannel() will
-  // wait for the other end to create connection.
-  void Initialize(
+  explicit WebrtcDataStreamAdapter(
       rtc::scoped_refptr<webrtc::PeerConnectionInterface> peer_connection);
+  ~WebrtcDataStreamAdapter();
 
-  // Called by WebrtcTransport.
-  void WrapIncomingDataChannel(
-      rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel,
-      const ChannelCreatedCallback& callback);
+  // Creates outgoing data channel.
+  std::unique_ptr<MessagePipe> CreateOutgoingChannel(const std::string& name);
 
-  // MessageChannelFactory interface.
-  void CreateChannel(const std::string& name,
-                     const ChannelCreatedCallback& callback) override;
-  void CancelChannelCreation(const std::string& name) override;
+  // Creates incoming data channel.
+  std::unique_ptr<MessagePipe> WrapIncomingDataChannel(
+      rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel);
 
  private:
-  class Channel;
-  friend class Channel;
-
-  struct PendingChannel;
-
-  void AddPendingChannel(
-      rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel,
-      const ChannelCreatedCallback& callback);
-
-  void OnChannelConnected(Channel* channel);
-  void OnChannelError();
-
-  ErrorCallback error_callback_;
-
   rtc::scoped_refptr<webrtc::PeerConnectionInterface> peer_connection_;
 
-  std::map<std::string, PendingChannel> pending_channels_;
-
   DISALLOW_COPY_AND_ASSIGN(WebrtcDataStreamAdapter);
 };
 

remoting/protocol/webrtc_transport.cc

@@ -152,6 +152,11 @@ WebrtcTransport::WebrtcTransport(
 
 WebrtcTransport::~WebrtcTransport() {}
 
+std::unique_ptr<MessagePipe> WebrtcTransport::CreateOutgoingChannel(
+    const std::string& name) {
+  return data_stream_adapter_->CreateOutgoingChannel(name);
+}
+
 void WebrtcTransport::Start(
     Authenticator* authenticator,
     SendTransportInfoCallback send_transport_info_callback) {
@@ -192,9 +197,7 @@ void WebrtcTransport::Start(
   peer_connection_ = peer_connection_factory_->CreatePeerConnection(
       rtc_config, &constraints, std::move(port_allocator), nullptr, this);
 
-  data_stream_adapter_.reset(new WebrtcDataStreamAdapter(
-      base::Bind(&WebrtcTransport::Close, base::Unretained(this))));
-  data_stream_adapter_->Initialize(peer_connection_);
+  data_stream_adapter_.reset(new WebrtcDataStreamAdapter(peer_connection_));
 
   event_handler_->OnWebrtcTransportConnecting();
 
@@ -434,10 +437,9 @@ void WebrtcTransport::OnRemoveStream(
 void WebrtcTransport::OnDataChannel(
     rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel) {
   DCHECK(thread_checker_.CalledOnValidThread());
-  data_stream_adapter_->WrapIncomingDataChannel(
-      data_channel,
-      base::Bind(&EventHandler::OnWebrtcTransportIncomingDataChannel,
-                 base::Unretained(event_handler_), data_channel->label()));
+  event_handler_->OnWebrtcTransportIncomingDataChannel(
+      data_channel->label(),
+      data_stream_adapter_->WrapIncomingDataChannel(data_channel));
 }
 
 void WebrtcTransport::OnRenegotiationNeeded() {

remoting/protocol/webrtc_transport.h

@@ -29,7 +29,6 @@ namespace remoting {
 namespace protocol {
 
 class TransportContext;
-class MessageChannelFactory;
 class MessagePipe;
 
 class WebrtcTransport : public Transport,
@@ -81,11 +80,10 @@ class WebrtcTransport : public Transport,
     return video_encoder_factory_;
   }
 
-  // Factory for outgoing data channels. Must be used only after the transport
-  // is connected.
-  MessageChannelFactory* outgoing_channel_factory() {
-    return data_stream_adapter_.get();
-  }
+  // Creates outgoing data channel. The channel is created in CONNECTING state.
+  // The caller must wait for OnMessagePipeOpen() notification before sending
+  // any messages.
+  std::unique_ptr<MessagePipe> CreateOutgoingChannel(const std::string& name);
 
   // Transport interface.
   void Start(Authenticator* authenticator,

remoting/protocol/webrtc_transport_unittest.cc

@@ -14,10 +14,11 @@
 #include "net/base/io_buffer.h"
 #include "net/url_request/url_request_context_getter.h"
 #include "remoting/base/compound_buffer.h"
-#include "remoting/protocol/connection_tester.h"
+#include "remoting/proto/event.pb.h"
 #include "remoting/protocol/fake_authenticator.h"
 #include "remoting/protocol/message_channel_factory.h"
 #include "remoting/protocol/message_pipe.h"
+#include "remoting/protocol/message_serialization.h"
 #include "remoting/protocol/network_settings.h"
 #include "remoting/protocol/transport_context.h"
 #include "remoting/signaling/fake_signal_strategy.h"
@@ -97,15 +98,32 @@ class TestMessagePipeEventHandler : public MessagePipe::EventHandler {
   TestMessagePipeEventHandler() {}
   ~TestMessagePipeEventHandler() override {}
 
+  void set_open_callback(const base::Closure& callback) {
+    open_callback_ = callback;
+  }
+  void set_message_callback(const base::Closure& callback) {
+    message_callback_ = callback;
+  }
   void set_closed_callback(const base::Closure& callback) {
     closed_callback_ = callback;
   }
 
+  bool is_open() { return is_open_; }
+  const std::list<std::unique_ptr<CompoundBuffer>>& received_messages() {
+    return received_messages_;
+  }
+
   // MessagePipe::EventHandler interface.
+  void OnMessagePipeOpen() override {
+    is_open_ = true;
+    if (!open_callback_.is_null())
+      open_callback_.Run();
+  }
   void OnMessageReceived(std::unique_ptr<CompoundBuffer> message) override {
-    NOTREACHED();
+    received_messages_.push_back(std::move(message));
+    if (!message_callback_.is_null())
+      message_callback_.Run();
   }
-
   void OnMessagePipeClosed() override {
     if (!closed_callback_.is_null()) {
       closed_callback_.Run();
@@ -115,8 +133,13 @@ class TestMessagePipeEventHandler : public MessagePipe::EventHandler {
   }
 
  private:
+  bool is_open_ = false;
+  base::Closure open_callback_;
+  base::Closure message_callback_;
   base::Closure closed_callback_;
 
+  std::list<std::unique_ptr<CompoundBuffer>> received_messages_;
+
   DISALLOW_COPY_AND_ASSIGN(TestMessagePipeEventHandler);
 };
 
@@ -220,22 +243,24 @@ class WebrtcTransportTest : public testing::Test {
   }
 
   void CreateHostDataStream() {
-    host_transport_->outgoing_channel_factory()->CreateChannel(
-        kChannelName, base::Bind(&WebrtcTransportTest::OnHostChannelCreated,
-                                 base::Unretained(this)));
+    host_message_pipe_ = host_transport_->CreateOutgoingChannel(kChannelName);
+    host_message_pipe_->Start(&host_message_pipe_event_handler_);
+    host_message_pipe_event_handler_.set_open_callback(base::Bind(
+        &WebrtcTransportTest::OnHostChannelConnected, base::Unretained(this)));
   }
 
   void OnIncomingChannel(const std::string& name,
                          std::unique_ptr<MessagePipe> pipe) {
     EXPECT_EQ(kChannelName, name);
     client_message_pipe_ = std::move(pipe);
-    if (run_loop_ && host_message_pipe_)
+    client_message_pipe_->Start(&client_message_pipe_event_handler_);
+
+    if (run_loop_ && host_message_pipe_event_handler_.is_open())
       run_loop_->Quit();
   }
 
-  void OnHostChannelCreated(std::unique_ptr<MessagePipe> pipe) {
-    host_message_pipe_ = std::move(pipe);
-    if (run_loop_ && client_message_pipe_)
+  void OnHostChannelConnected() {
+    if (run_loop_ && client_message_pipe_event_handler_.is_open())
       run_loop_->Quit();
   }
 
@@ -283,6 +308,7 @@ class WebrtcTransportTest : public testing::Test {
   std::unique_ptr<FakeAuthenticator> client_authenticator_;
 
   std::unique_ptr<MessagePipe> client_message_pipe_;
+  TestMessagePipeEventHandler client_message_pipe_event_handler_;
   std::unique_ptr<MessagePipe> host_message_pipe_;
   TestMessagePipeEventHandler host_message_pipe_event_handler_;
 
@@ -324,12 +350,20 @@ TEST_F(WebrtcTransportTest, DataStream) {
   EXPECT_TRUE(client_message_pipe_);
   EXPECT_TRUE(host_message_pipe_);
 
-  const int kMessageSize = 1024;
-  const int kMessages = 100;
-  MessagePipeConnectionTester tester(host_message_pipe_.get(),
-                                     client_message_pipe_.get(), kMessageSize,
-                                     kMessages);
-  tester.RunAndCheckResults();
+  TextEvent message;
+  message.set_text("Hello");
+  host_message_pipe_->Send(&message, base::Closure());
+
+  run_loop_.reset(new base::RunLoop());
+  client_message_pipe_event_handler_.set_message_callback(
+      base::Bind(&base::RunLoop::Quit, base::Unretained(run_loop_.get())));
+  run_loop_->Run();
+
+  ASSERT_EQ(1U, client_message_pipe_event_handler_.received_messages().size());
+
+  std::unique_ptr<TextEvent> received_message = ParseMessage<TextEvent>(
+      client_message_pipe_event_handler_.received_messages().front().get());
+  EXPECT_EQ(message.text(), received_message->text());
 }
 
 // Verify that data streams can be created after connection has been initiated.
@@ -366,7 +400,6 @@ TEST_F(WebrtcTransportTest, TerminateDataChannel) {
 
   // Expect that the channel is closed on the host side once the client closes
   // the channel.
-  host_message_pipe_->Start(&host_message_pipe_event_handler_);
   host_message_pipe_event_handler_.set_closed_callback(base::Bind(
       &WebrtcTransportTest::OnHostChannelClosed, base::Unretained(this)));