Revert of Remove Client= from device/serial/data_stream.mojom. (patchset #1 id:20001 of https://codereview.chromium.org/889283002/)

Reason for revert:
Switching to two pipes broke FIFO assumption, causing DataSinkTest.CancelDuringError and SerialConnectionTest.Cancel to become flaky

Original issue's description:
> Remove Client= from device/serial/data_stream.mojom.
>
> BUG=451321
>
> Committed: https://crrev.com/bc78a2320b50eba4fe397f1f838a82ad30b16cd6
> Cr-Commit-Position: refs/heads/master@{#314231}

TBR=rockot@chromium.org
NOPRESUBMIT=true
NOTREECHECKS=true
NOTRY=true
BUG=451321

Review URL: https://codereview.chromium.org/873293006

Cr-Commit-Position: refs/heads/master@{#314524}

21 files changed, 91 insertions, 161 deletions

diff --git a/device/serial/data_receiver.cc b/device/serial/data_receiver.cc
index 2f238e3..fdc23ca 100644
--- a/device/serial/data_receiver.cc
+++ b/device/serial/data_receiver.cc
@@ -107,19 +107,16 @@ struct DataReceiver::DataFrame {
   bool dispatched;
 };
 
-DataReceiver::DataReceiver(
-    mojo::InterfacePtr<serial::DataSource> source,
-    mojo::InterfaceRequest<serial::DataSourceClient> client,
-    uint32_t buffer_size,
-    int32_t fatal_error_value)
+DataReceiver::DataReceiver(mojo::InterfacePtr<serial::DataSource> source,
+                           uint32_t buffer_size,
+                           int32_t fatal_error_value)
     : source_(source.Pass()),
-      client_(this, client.Pass()),
       fatal_error_value_(fatal_error_value),
       shut_down_(false),
       weak_factory_(this) {
+  source_.set_client(this);
   source_.set_error_handler(this);
   source_->Init(buffer_size);
-  client_.set_error_handler(this);
 }
 
 bool DataReceiver::Receive(const ReceiveDataCallback& callback,
diff --git a/device/serial/data_receiver.h b/device/serial/data_receiver.h
index 95be640..942bb5c 100644
--- a/device/serial/data_receiver.h
+++ b/device/serial/data_receiver.h
@@ -29,7 +29,6 @@ class DataReceiver : public base::RefCounted<DataReceiver>,
   // size of |buffer_size|, with connection errors reported as
   // |fatal_error_value|.
   DataReceiver(mojo::InterfacePtr<serial::DataSource> source,
-               mojo::InterfaceRequest<serial::DataSourceClient> client,
                uint32_t buffer_size,
                int32_t fatal_error_value);
 
@@ -73,7 +72,6 @@ class DataReceiver : public base::RefCounted<DataReceiver>,
 
   // The control connection to the data source.
   mojo::InterfacePtr<serial::DataSource> source_;
-  mojo::Binding<serial::DataSourceClient> client_;
 
   // The error value to report in the event of a fatal error.
   const int32_t fatal_error_value_;
diff --git a/device/serial/data_sender.cc b/device/serial/data_sender.cc
index bea8eb8..18b1a43 100644
--- a/device/serial/data_sender.cc
+++ b/device/serial/data_sender.cc
@@ -62,17 +62,15 @@ class DataSender::PendingSend {
 };
 
 DataSender::DataSender(mojo::InterfacePtr<serial::DataSink> sink,
-                       mojo::InterfaceRequest<serial::DataSinkClient> client,
                        uint32_t buffer_size,
                        int32_t fatal_error_value)
     : sink_(sink.Pass()),
-      client_(this, client.Pass()),
       fatal_error_value_(fatal_error_value),
       available_buffer_capacity_(buffer_size),
       shut_down_(false) {
   sink_.set_error_handler(this);
+  sink_.set_client(this);
   sink_->Init(buffer_size);
-  client_.set_error_handler(this);
 }
 
 DataSender::~DataSender() {
diff --git a/device/serial/data_sender.h b/device/serial/data_sender.h
index 0b5ea09..f4db8bc 100644
--- a/device/serial/data_sender.h
+++ b/device/serial/data_sender.h
@@ -27,7 +27,6 @@ class DataSender : public serial::DataSinkClient, public mojo::ErrorHandler {
   // Constructs a DataSender to send data to |sink|, using a buffer size of
   // |buffer_size|, with connection errors reported as |fatal_error_value|.
   DataSender(mojo::InterfacePtr<serial::DataSink> sink,
-             mojo::InterfaceRequest<serial::DataSinkClient> client,
              uint32_t buffer_size,
              int32_t fatal_error_value);
 
@@ -74,7 +73,6 @@ class DataSender : public serial::DataSinkClient, public mojo::ErrorHandler {
 
   // The control connection to the data sink.
   mojo::InterfacePtr<serial::DataSink> sink_;
-  mojo::Binding<serial::DataSinkClient> client_;
 
   // The error value to report in the event of a fatal error.
   const int32_t fatal_error_value_;
diff --git a/device/serial/data_sink_receiver.cc b/device/serial/data_sink_receiver.cc
index 18dd569..170f8b2 100644
--- a/device/serial/data_sink_receiver.cc
+++ b/device/serial/data_sink_receiver.cc
@@ -61,15 +61,10 @@ class DataSinkReceiver::DataFrame {
   uint32_t offset_;
 };
 
-DataSinkReceiver::DataSinkReceiver(
-    mojo::InterfaceRequest<serial::DataSink> request,
-    mojo::InterfacePtr<serial::DataSinkClient> client,
-    const ReadyCallback& ready_callback,
-    const CancelCallback& cancel_callback,
-    const ErrorCallback& error_callback)
-    : binding_(this, request.Pass()),
-      client_(client.Pass()),
-      ready_callback_(ready_callback),
+DataSinkReceiver::DataSinkReceiver(const ReadyCallback& ready_callback,
+                                   const CancelCallback& cancel_callback,
+                                   const ErrorCallback& error_callback)
+    : ready_callback_(ready_callback),
       cancel_callback_(cancel_callback),
       error_callback_(error_callback),
       flush_pending_(false),
@@ -78,8 +73,6 @@ DataSinkReceiver::DataSinkReceiver(
       available_buffer_capacity_(0),
       shut_down_(false),
       weak_factory_(this) {
-  binding_.set_error_handler(this);
-  client_.set_error_handler(this);
 }
 
 void DataSinkReceiver::ShutDown() {
@@ -158,7 +151,7 @@ void DataSinkReceiver::RunReadyCallback() {
 void DataSinkReceiver::Done(uint32_t bytes_read) {
   if (!DoneInternal(bytes_read))
     return;
-  client_->ReportBytesSent(bytes_read);
+  client()->ReportBytesSent(bytes_read);
   if (!pending_data_buffers_.empty()) {
     base::MessageLoop::current()->PostTask(
         FROM_HERE,
@@ -191,8 +184,9 @@ void DataSinkReceiver::ReportBytesSentAndError(uint32_t bytes_read,
   // When we encounter an error, we must discard the data from any send buffers
   // transmitted by the DataSinkClient before it receives this error.
   flush_pending_ = true;
-  client_->ReportBytesSentAndError(
-      bytes_read, error,
+  client()->ReportBytesSentAndError(
+      bytes_read,
+      error,
       base::Bind(&DataSinkReceiver::DoFlush, weak_factory_.GetWeakPtr()));
 }
 
diff --git a/device/serial/data_sink_receiver.h b/device/serial/data_sink_receiver.h
index 0709f90..2126cdd 100644
--- a/device/serial/data_sink_receiver.h
+++ b/device/serial/data_sink_receiver.h
@@ -18,8 +18,7 @@
 namespace device {
 
 class DataSinkReceiver : public base::RefCounted<DataSinkReceiver>,
-                         public serial::DataSink,
-                         public mojo::ErrorHandler {
+                         public mojo::InterfaceImpl<serial::DataSink> {
  public:
   typedef base::Callback<void(scoped_ptr<ReadOnlyBuffer>)> ReadyCallback;
   typedef base::Callback<void(int32_t error)> CancelCallback;
@@ -32,9 +31,7 @@ class DataSinkReceiver : public base::RefCounted<DataSinkReceiver>,
   // and the DataSinkReceiver will act as if ShutDown() had been called. If
   // |cancel_callback| is valid, it will be called when the DataSinkClient
   // requests cancellation of the in-progress read.
-  DataSinkReceiver(mojo::InterfaceRequest<serial::DataSink> request,
-                   mojo::InterfacePtr<serial::DataSinkClient> client,
-                   const ReadyCallback& ready_callback,
+  DataSinkReceiver(const ReadyCallback& ready_callback,
                    const CancelCallback& cancel_callback,
                    const ErrorCallback& error_callback);
 
@@ -78,9 +75,6 @@ class DataSinkReceiver : public base::RefCounted<DataSinkReceiver>,
   // Reports a fatal error to the client and shuts down.
   void DispatchFatalError();
 
-  mojo::Binding<serial::DataSink> binding_;
-  mojo::InterfacePtr<serial::DataSinkClient> client_;
-
   // The callback to call when there is data ready to read.
   const ReadyCallback ready_callback_;
 
diff --git a/device/serial/data_sink_unittest.cc b/device/serial/data_sink_unittest.cc
index 4108692..2484362 100644
--- a/device/serial/data_sink_unittest.cc
+++ b/device/serial/data_sink_unittest.cc
@@ -38,16 +38,13 @@ class DataSinkTest : public testing::Test {
   void SetUp() override {
     message_loop_.reset(new base::MessageLoop);
     mojo::InterfacePtr<serial::DataSink> sink_handle;
-    mojo::InterfacePtr<serial::DataSinkClient> sink_client_handle;
-    mojo::InterfaceRequest<serial::DataSinkClient> sink_client_request =
-        mojo::GetProxy(&sink_client_handle);
-    sink_receiver_ = new DataSinkReceiver(
-        mojo::GetProxy(&sink_handle), sink_client_handle.Pass(),
-        base::Bind(&DataSinkTest::OnDataToRead, base::Unretained(this)),
-        base::Bind(&DataSinkTest::OnCancel, base::Unretained(this)),
-        base::Bind(&DataSinkTest::OnError, base::Unretained(this)));
-    sender_.reset(new DataSender(sink_handle.Pass(), sink_client_request.Pass(),
-                                 kBufferSize, kFatalError));
+    sink_receiver_ = mojo::WeakBindToProxy(
+        new DataSinkReceiver(
+            base::Bind(&DataSinkTest::OnDataToRead, base::Unretained(this)),
+            base::Bind(&DataSinkTest::OnCancel, base::Unretained(this)),
+            base::Bind(&DataSinkTest::OnError, base::Unretained(this))),
+        &sink_handle);
+    sender_.reset(new DataSender(sink_handle.Pass(), kBufferSize, kFatalError));
   }
 
   void TearDown() override {
diff --git a/device/serial/data_source_sender.cc b/device/serial/data_source_sender.cc
index 97029dd..f9cc4f1 100644
--- a/device/serial/data_source_sender.cc
+++ b/device/serial/data_source_sender.cc
@@ -71,22 +71,15 @@ class DataSourceSender::PendingSend::Buffer : public WritableBuffer {
   uint32_t buffer_size_;
 };
 
-DataSourceSender::DataSourceSender(
-    mojo::InterfaceRequest<serial::DataSource> source,
-    mojo::InterfacePtr<serial::DataSourceClient> client,
-    const ReadyCallback& ready_callback,
-    const ErrorCallback& error_callback)
-    : binding_(this, source.Pass()),
-      client_(client.Pass()),
-      ready_callback_(ready_callback),
+DataSourceSender::DataSourceSender(const ReadyCallback& ready_callback,
+                                   const ErrorCallback& error_callback)
+    : ready_callback_(ready_callback),
       error_callback_(error_callback),
       available_buffer_capacity_(0),
       paused_(false),
       shut_down_(false),
       weak_factory_(this) {
   DCHECK(!ready_callback.is_null() && !error_callback.is_null());
-  binding_.set_error_handler(this);
-  client_.set_error_handler(this);
 }
 
 void DataSourceSender::ShutDown() {
@@ -145,7 +138,7 @@ void DataSourceSender::DoneWithError(const std::vector<char>& data,
                                      int32_t error) {
   DoneInternal(data);
   if (!shut_down_)
-    client_->OnError(error);
+    client()->OnError(error);
   paused_ = true;
   // We don't call GetMoreData here so we don't send any additional data until
   // Resume() is called.
@@ -160,7 +153,7 @@ void DataSourceSender::DoneInternal(const std::vector<char>& data) {
   if (!data.empty()) {
     mojo::Array<uint8_t> data_to_send(data.size());
     std::copy(data.begin(), data.end(), &data_to_send[0]);
-    client_->OnData(data_to_send.Pass());
+    client()->OnData(data_to_send.Pass());
   }
   pending_send_.reset();
 }
diff --git a/device/serial/data_source_sender.h b/device/serial/data_source_sender.h
index a8c2c63..3a90240 100644
--- a/device/serial/data_source_sender.h
+++ b/device/serial/data_source_sender.h
@@ -19,8 +19,7 @@ namespace device {
 // A DataSourceSender is an interface between a source of data and a
 // DataSourceClient.
 class DataSourceSender : public base::RefCounted<DataSourceSender>,
-                         public serial::DataSource,
-                         public mojo::ErrorHandler {
+                         public mojo::InterfaceImpl<serial::DataSource> {
  public:
   typedef base::Callback<void(scoped_ptr<WritableBuffer>)> ReadyCallback;
   typedef base::Callback<void()> ErrorCallback;
@@ -30,9 +29,7 @@ class DataSourceSender : public base::RefCounted<DataSourceSender>,
   // |ready_callback| will not be called again until the previous WritableBuffer
   // is destroyed. If a connection error occurs, |error_callback| will be
   // called and the DataSourceSender will act as if ShutDown() had been called.
-  DataSourceSender(mojo::InterfaceRequest<serial::DataSource> source,
-                   mojo::InterfacePtr<serial::DataSourceClient> client,
-                   const ReadyCallback& ready_callback,
+  DataSourceSender(const ReadyCallback& ready_callback,
                    const ErrorCallback& error_callback);
 
   // Shuts down this DataSourceSender. After shut down, |ready_callback| and
@@ -68,9 +65,6 @@ class DataSourceSender : public base::RefCounted<DataSourceSender>,
   // Reports a fatal error to the client and shuts down.
   void DispatchFatalError();
 
-  mojo::Binding<serial::DataSource> binding_;
-  mojo::InterfacePtr<serial::DataSourceClient> client_;
-
   // The callback to call when the client is ready for more data.
   ReadyCallback ready_callback_;
 
diff --git a/device/serial/data_source_unittest.cc b/device/serial/data_source_unittest.cc
index b725f35..71dfa81 100644
--- a/device/serial/data_source_unittest.cc
+++ b/device/serial/data_source_unittest.cc
@@ -31,18 +31,12 @@ class DataSourceTest : public testing::Test {
   void SetUp() override {
     message_loop_.reset(new base::MessageLoop);
     mojo::InterfacePtr<serial::DataSource> source_sender_handle;
-    mojo::InterfacePtr<serial::DataSourceClient> source_sender_client_handle;
-    mojo::InterfaceRequest<serial::DataSourceClient>
-        source_sender_client_request =
-            mojo::GetProxy(&source_sender_client_handle);
-    source_sender_ = new DataSourceSender(
-        mojo::GetProxy(&source_sender_handle),
-        source_sender_client_handle.Pass(),
-        base::Bind(&DataSourceTest::CanWriteData, base::Unretained(this)),
-        base::Bind(&DataSourceTest::OnError, base::Unretained(this)));
-    receiver_ =
-        new DataReceiver(source_sender_handle.Pass(),
-                         source_sender_client_request.Pass(), 100, kFatalError);
+    source_sender_ = mojo::WeakBindToProxy(
+        new DataSourceSender(
+            base::Bind(&DataSourceTest::CanWriteData, base::Unretained(this)),
+            base::Bind(&DataSourceTest::OnError, base::Unretained(this))),
+        &source_sender_handle);
+    receiver_ = new DataReceiver(source_sender_handle.Pass(), 100, kFatalError);
   }
 
   void TearDown() override {
diff --git a/device/serial/data_stream.mojom b/device/serial/data_stream.mojom
index e547806..a288eb2 100644
--- a/device/serial/data_stream.mojom
+++ b/device/serial/data_stream.mojom
@@ -4,6 +4,7 @@
 
 module device.serial;
 
+[Client=DataSourceClient]
 interface DataSource {
   // Initializes this DataSource with the amount of data its client will
   // buffer.
@@ -26,6 +27,7 @@ interface DataSourceClient {
   OnData(array<uint8> data);
 };
 
+[Client=DataSinkClient]
 interface DataSink {
   // Initializes this DataSink with the amount of data it is expected to
   // buffer.
diff --git a/device/serial/data_stream_serialization.mojom b/device/serial/data_stream_serialization.mojom
index de580d3..259be28 100644
--- a/device/serial/data_stream_serialization.mojom
+++ b/device/serial/data_stream_serialization.mojom
@@ -10,8 +10,6 @@ import "data_stream.mojom";
 struct SerializedDataSender {
   // The control channel to the DataSink to which this DataSender sends data.
   DataSink sink;
-  // DataSinkClient&
-  handle<message_pipe> sink_client;
 
   // The error to report for sends in progress when a fatal error occurs.
   int32 fatal_error_value;
@@ -34,8 +32,6 @@ struct SerializedDataReceiver {
   // The control channel to the DataSource from which this DataReceiver receives
   // data.
   DataSource source;
-  // DataSourceClient&
-  handle<message_pipe> source_client;
 
   // The error to report for a receive in progress when a fatal error occurs.
   int32 fatal_error_value;
diff --git a/device/serial/serial.mojom b/device/serial/serial.mojom
index 877d1e2..6714a94 100644
--- a/device/serial/serial.mojom
+++ b/device/serial/serial.mojom
@@ -94,9 +94,7 @@ interface SerialService {
           ConnectionOptions? options,
           Connection& connection,
           DataSink& sink,
-          DataSinkClient sink_client,
-          DataSource& source,
-          DataSourceClient source_client);
+          DataSource& source);
 };
 
 interface Connection {
diff --git a/device/serial/serial_connection.cc b/device/serial/serial_connection.cc
index 04cccbe..0d3e417 100644
--- a/device/serial/serial_connection.cc
+++ b/device/serial/serial_connection.cc
@@ -15,19 +15,20 @@ namespace device {
 SerialConnection::SerialConnection(
     scoped_refptr<SerialIoHandler> io_handler,
     mojo::InterfaceRequest<serial::DataSink> sink,
-    mojo::InterfacePtr<serial::DataSinkClient> sink_client,
-    mojo::InterfaceRequest<serial::DataSource> source,
-    mojo::InterfacePtr<serial::DataSourceClient> source_client)
+    mojo::InterfaceRequest<serial::DataSource> source)
     : io_handler_(io_handler) {
-  receiver_ = new DataSinkReceiver(
-      sink.Pass(), sink_client.Pass(),
-      base::Bind(&SerialConnection::OnSendPipeReady, base::Unretained(this)),
-      base::Bind(&SerialConnection::OnSendCancelled, base::Unretained(this)),
-      base::Bind(base::DoNothing));
-  sender_ = new DataSourceSender(
-      source.Pass(), source_client.Pass(),
-      base::Bind(&SerialConnection::OnReceivePipeReady, base::Unretained(this)),
-      base::Bind(base::DoNothing));
+  receiver_ = mojo::WeakBindToRequest(
+      new DataSinkReceiver(base::Bind(&SerialConnection::OnSendPipeReady,
+                                      base::Unretained(this)),
+                           base::Bind(&SerialConnection::OnSendCancelled,
+                                      base::Unretained(this)),
+                           base::Bind(base::DoNothing)),
+      &sink);
+  sender_ = mojo::WeakBindToRequest(
+      new DataSourceSender(base::Bind(&SerialConnection::OnReceivePipeReady,
+                                      base::Unretained(this)),
+                           base::Bind(base::DoNothing)),
+      &source);
 }
 
 SerialConnection::~SerialConnection() {
diff --git a/device/serial/serial_connection.h b/device/serial/serial_connection.h
index 6143c8b..d0b940f 100644
--- a/device/serial/serial_connection.h
+++ b/device/serial/serial_connection.h
@@ -22,9 +22,7 @@ class SerialConnection : public mojo::InterfaceImpl<serial::Connection> {
  public:
   SerialConnection(scoped_refptr<SerialIoHandler> io_handler,
                    mojo::InterfaceRequest<serial::DataSink> sink,
-                   mojo::InterfacePtr<serial::DataSinkClient> sink_client,
-                   mojo::InterfaceRequest<serial::DataSource> source,
-                   mojo::InterfacePtr<serial::DataSourceClient> source_client);
+                   mojo::InterfaceRequest<serial::DataSource> source);
   ~SerialConnection() override;
 
   // mojo::InterfaceImpl<serial::Connection> overrides.
diff --git a/device/serial/serial_connection_factory.cc b/device/serial/serial_connection_factory.cc
index 9beb33ae..6b01490 100644
--- a/device/serial/serial_connection_factory.cc
+++ b/device/serial/serial_connection_factory.cc
@@ -37,9 +37,7 @@ class SerialConnectionFactory::ConnectTask
               serial::ConnectionOptionsPtr options,
               mojo::InterfaceRequest<serial::Connection> connection_request,
               mojo::InterfaceRequest<serial::DataSink> sink,
-              mojo::InterfacePtr<serial::DataSinkClient> sink_client,
-              mojo::InterfaceRequest<serial::DataSource> source,
-              mojo::InterfacePtr<serial::DataSourceClient> source_client);
+              mojo::InterfaceRequest<serial::DataSource> source);
   void Run();
 
  private:
@@ -53,9 +51,7 @@ class SerialConnectionFactory::ConnectTask
   serial::ConnectionOptionsPtr options_;
   mojo::InterfaceRequest<serial::Connection> connection_request_;
   mojo::InterfaceRequest<serial::DataSink> sink_;
-  mojo::InterfacePtr<serial::DataSinkClient> sink_client_;
   mojo::InterfaceRequest<serial::DataSource> source_;
-  mojo::InterfacePtr<serial::DataSourceClient> source_client_;
   scoped_refptr<SerialIoHandler> io_handler_;
 
   DISALLOW_COPY_AND_ASSIGN(ConnectTask);
@@ -73,12 +69,13 @@ void SerialConnectionFactory::CreateConnection(
     serial::ConnectionOptionsPtr options,
     mojo::InterfaceRequest<serial::Connection> connection_request,
     mojo::InterfaceRequest<serial::DataSink> sink,
-    mojo::InterfacePtr<serial::DataSinkClient> sink_client,
-    mojo::InterfaceRequest<serial::DataSource> source,
-    mojo::InterfacePtr<serial::DataSourceClient> source_client) {
-  scoped_refptr<ConnectTask> task(new ConnectTask(
-      this, path, options.Pass(), connection_request.Pass(), sink.Pass(),
-      sink_client.Pass(), source.Pass(), source_client.Pass()));
+    mojo::InterfaceRequest<serial::DataSource> source) {
+  scoped_refptr<ConnectTask> task(new ConnectTask(this,
+                                                  path,
+                                                  options.Pass(),
+                                                  connection_request.Pass(),
+                                                  sink.Pass(),
+                                                  source.Pass()));
   task->Run();
 }
 
@@ -91,17 +88,13 @@ SerialConnectionFactory::ConnectTask::ConnectTask(
     serial::ConnectionOptionsPtr options,
     mojo::InterfaceRequest<serial::Connection> connection_request,
     mojo::InterfaceRequest<serial::DataSink> sink,
-    mojo::InterfacePtr<serial::DataSinkClient> sink_client,
-    mojo::InterfaceRequest<serial::DataSource> source,
-    mojo::InterfacePtr<serial::DataSourceClient> source_client)
+    mojo::InterfaceRequest<serial::DataSource> source)
     : factory_(factory),
       path_(path),
       options_(options.Pass()),
       connection_request_(connection_request.Pass()),
       sink_(sink.Pass()),
-      sink_client_(sink_client.Pass()),
-      source_(source.Pass()),
-      source_client_(source_client.Pass()) {
+      source_(source.Pass()) {
   if (!options_) {
     options_ = serial::ConnectionOptions::New();
   }
@@ -131,8 +124,7 @@ void SerialConnectionFactory::ConnectTask::OnConnected(bool success) {
   }
 
   mojo::BindToRequest(
-      new SerialConnection(io_handler_, sink_.Pass(), sink_client_.Pass(),
-                           source_.Pass(), source_client_.Pass()),
+      new SerialConnection(io_handler_, sink_.Pass(), source_.Pass()),
       &connection_request_);
 }
 
diff --git a/device/serial/serial_connection_factory.h b/device/serial/serial_connection_factory.h
index 2e1dcca..4200dca 100644
--- a/device/serial/serial_connection_factory.h
+++ b/device/serial/serial_connection_factory.h
@@ -32,9 +32,7 @@ class SerialConnectionFactory
       serial::ConnectionOptionsPtr options,
       mojo::InterfaceRequest<serial::Connection> connection_request,
       mojo::InterfaceRequest<serial::DataSink> sink,
-      mojo::InterfacePtr<serial::DataSinkClient> sink_client,
-      mojo::InterfaceRequest<serial::DataSource> source,
-      mojo::InterfacePtr<serial::DataSourceClient> source_client);
+      mojo::InterfaceRequest<serial::DataSource> source);
 
  private:
   friend class base::RefCountedThreadSafe<SerialConnectionFactory>;
diff --git a/device/serial/serial_connection_unittest.cc b/device/serial/serial_connection_unittest.cc
index 53d5303..151ed77 100644
--- a/device/serial/serial_connection_unittest.cc
+++ b/device/serial/serial_connection_unittest.cc
@@ -74,23 +74,17 @@ class SerialConnectionTest : public testing::Test, public mojo::ErrorHandler {
             scoped_ptr<SerialDeviceEnumerator>(new FakeSerialDeviceEnumerator)),
         &service);
     service.set_error_handler(this);
-    mojo::InterfacePtr<serial::DataSink> sink;
-    mojo::InterfacePtr<serial::DataSinkClient> sink_client;
-    mojo::InterfaceRequest<serial::DataSinkClient> sink_client_request =
-        mojo::GetProxy(&sink_client);
-    mojo::InterfacePtr<serial::DataSource> source;
-    mojo::InterfacePtr<serial::DataSourceClient> source_client;
-    mojo::InterfaceRequest<serial::DataSourceClient> source_client_request =
-        mojo::GetProxy(&source_client);
-    service->Connect("device", serial::ConnectionOptions::New(),
-                     mojo::GetProxy(&connection_), mojo::GetProxy(&sink),
-                     sink_client.Pass(), mojo::GetProxy(&source),
-                     source_client.Pass());
-    sender_.reset(new DataSender(sink.Pass(), sink_client_request.Pass(),
-                                 kBufferSize, serial::SEND_ERROR_DISCONNECTED));
-    receiver_ =
-        new DataReceiver(source.Pass(), source_client_request.Pass(),
-                         kBufferSize, serial::RECEIVE_ERROR_DISCONNECTED);
+    mojo::InterfacePtr<serial::DataSink> consumer;
+    mojo::InterfacePtr<serial::DataSource> producer;
+    service->Connect("device",
+                     serial::ConnectionOptions::New(),
+                     mojo::GetProxy(&connection_),
+                     mojo::GetProxy(&consumer),
+                     mojo::GetProxy(&producer));
+    sender_.reset(new DataSender(
+        consumer.Pass(), kBufferSize, serial::SEND_ERROR_DISCONNECTED));
+    receiver_ = new DataReceiver(
+        producer.Pass(), kBufferSize, serial::RECEIVE_ERROR_DISCONNECTED);
     connection_.set_error_handler(this);
     connection_->GetInfo(
         base::Bind(&SerialConnectionTest::StoreInfo, base::Unretained(this)));
diff --git a/device/serial/serial_service_impl.cc b/device/serial/serial_service_impl.cc
index 6fd6ce0..fceb448 100644
--- a/device/serial/serial_service_impl.cc
+++ b/device/serial/serial_service_impl.cc
@@ -61,14 +61,14 @@ void SerialServiceImpl::Connect(
     serial::ConnectionOptionsPtr options,
     mojo::InterfaceRequest<serial::Connection> connection_request,
     mojo::InterfaceRequest<serial::DataSink> sink,
-    mojo::InterfacePtr<serial::DataSinkClient> sink_client,
-    mojo::InterfaceRequest<serial::DataSource> source,
-    mojo::InterfacePtr<serial::DataSourceClient> source_client) {
+    mojo::InterfaceRequest<serial::DataSource> source) {
   if (!IsValidPath(path))
     return;
-  connection_factory_->CreateConnection(
-      path, options.Pass(), connection_request.Pass(), sink.Pass(),
-      sink_client.Pass(), source.Pass(), source_client.Pass());
+  connection_factory_->CreateConnection(path,
+                                        options.Pass(),
+                                        connection_request.Pass(),
+                                        sink.Pass(),
+                                        source.Pass());
 }
 
 SerialDeviceEnumerator* SerialServiceImpl::GetDeviceEnumerator() {
diff --git a/device/serial/serial_service_impl.h b/device/serial/serial_service_impl.h
index 8e6ce3b..0caf00e 100644
--- a/device/serial/serial_service_impl.h
+++ b/device/serial/serial_service_impl.h
@@ -36,14 +36,11 @@ class SerialServiceImpl : public mojo::InterfaceImpl<serial::SerialService> {
   void GetDevices(
       const mojo::Callback<void(mojo::Array<serial::DeviceInfoPtr>)>& callback)
       override;
-  void Connect(
-      const mojo::String& path,
-      serial::ConnectionOptionsPtr options,
-      mojo::InterfaceRequest<serial::Connection> connection_request,
-      mojo::InterfaceRequest<serial::DataSink> sink,
-      mojo::InterfacePtr<serial::DataSinkClient> sink_client,
-      mojo::InterfaceRequest<serial::DataSource> source,
-      mojo::InterfacePtr<serial::DataSourceClient> source_client) override;
+  void Connect(const mojo::String& path,
+               serial::ConnectionOptionsPtr options,
+               mojo::InterfaceRequest<serial::Connection> connection_request,
+               mojo::InterfaceRequest<serial::DataSink> sink,
+               mojo::InterfaceRequest<serial::DataSource> source) override;
 
  private:
   SerialDeviceEnumerator* GetDeviceEnumerator();
diff --git a/device/serial/serial_service_unittest.cc b/device/serial/serial_service_unittest.cc
index 103fb20..7f8beb2 100644
--- a/device/serial/serial_service_unittest.cc
+++ b/device/serial/serial_service_unittest.cc
@@ -84,15 +84,12 @@ class SerialServiceTest : public testing::Test, public mojo::ErrorHandler {
         &service);
     mojo::InterfacePtr<serial::Connection> connection;
     mojo::InterfacePtr<serial::DataSink> sink;
-    mojo::InterfacePtr<serial::DataSinkClient> sink_client;
-    mojo::GetProxy(&sink_client);
     mojo::InterfacePtr<serial::DataSource> source;
-    mojo::InterfacePtr<serial::DataSourceClient> source_client;
-    mojo::GetProxy(&source_client);
-    service->Connect(path, serial::ConnectionOptions::New(),
-                     mojo::GetProxy(&connection), mojo::GetProxy(&sink),
-                     sink_client.Pass(), mojo::GetProxy(&source),
-                     source_client.Pass());
+    service->Connect(path,
+                     serial::ConnectionOptions::New(),
+                     mojo::GetProxy(&connection),
+                     mojo::GetProxy(&sink),
+                     mojo::GetProxy(&source));
     connection.set_error_handler(this);
     expecting_error_ = !expecting_success;
     connection->GetInfo(