Implement the host side of serial connection I/O on data pipe.

This change adds DataSource and DataSink interface request parameters to
SerialService::Connect, passes them through to SerialConnection
instances via SerialConnectionFactory; the SerialConnection now uses a
DataSourceSender and a DataSinkReceiver for implementing serial receive
and send, respectively.

BUG=389016

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

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

device/serial/serial.mojom

@@ -2,6 +2,8 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
+import "data_stream.mojom"
+
 module device.serial {
 
 struct DeviceInfo {
@@ -83,12 +85,16 @@ interface SerialService {
   GetDevices() => (DeviceInfo?[]? devices);
 
   // Creates a |Connection| to |path| with options specified by |options|,
-  // returning it via |connection|. This will fail and |connection| will not be
-  // usable if |path| does not specify a valid serial device or there is an
-  // error connecting to or configuring the connection.
+  // returning it via |connection|. Sending and receiving data over this
+  // connection is handled by |sink| and |source|, respectively. This will fail
+  // and |connection| will not be usable if |path| does not specify a valid
+  // serial device or there is an error connecting to or configuring the
+  // connection.
   Connect(string? path,
           ConnectionOptions? options,
-          Connection&? connection);
+          Connection&? connection,
+          DataSink&? sink,
+          DataSource&? source);
 };
 
 interface Connection {

device/serial/serial_connection.cc

@@ -4,15 +4,38 @@
 
 #include "device/serial/serial_connection.h"
 
+#include "base/bind.h"
+#include "device/serial/buffer.h"
+#include "device/serial/data_sink_receiver.h"
+#include "device/serial/data_source_sender.h"
 #include "device/serial/serial_io_handler.h"
 
 namespace device {
 
-SerialConnection::SerialConnection(scoped_refptr<SerialIoHandler> io_handler)
+SerialConnection::SerialConnection(
+    scoped_refptr<SerialIoHandler> io_handler,
+    mojo::InterfaceRequest<serial::DataSink> sink,
+    mojo::InterfaceRequest<serial::DataSource> source)
     : io_handler_(io_handler) {
+  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() {
+  receiver_->ShutDown();
+  sender_->ShutDown();
+  io_handler_->CancelRead(serial::RECEIVE_ERROR_DISCONNECTED);
+  io_handler_->CancelWrite(serial::SEND_ERROR_DISCONNECTED);
 }
 
 void SerialConnection::GetInfo(
@@ -41,4 +64,16 @@ void SerialConnection::Flush(const mojo::Callback<void(bool)>& callback) {
   callback.Run(io_handler_->Flush());
 }
 
+void SerialConnection::OnSendCancelled(int32_t error) {
+  io_handler_->CancelWrite(static_cast<serial::SendError>(error));
+}
+
+void SerialConnection::OnSendPipeReady(scoped_ptr<ReadOnlyBuffer> buffer) {
+  io_handler_->Write(buffer.Pass());
+}
+
+void SerialConnection::OnReceivePipeReady(scoped_ptr<WritableBuffer> buffer) {
+  io_handler_->Read(buffer.Pass());
+}
+
 }  // namespace device

device/serial/serial_connection.h

@@ -6,16 +6,23 @@
 #define DEVICE_SERIAL_SERIAL_CONNECTION_H_
 
 #include "base/memory/ref_counted.h"
+#include "base/memory/scoped_ptr.h"
 #include "device/serial/serial.mojom.h"
 #include "mojo/public/cpp/bindings/interface_impl.h"
 
 namespace device {
 
+class DataSinkReceiver;
+class DataSourceSender;
+class ReadOnlyBuffer;
 class SerialIoHandler;
+class WritableBuffer;
 
 class SerialConnection : public mojo::InterfaceImpl<serial::Connection> {
  public:
-  explicit SerialConnection(scoped_refptr<SerialIoHandler> io_handler);
+  SerialConnection(scoped_refptr<SerialIoHandler> io_handler,
+                   mojo::InterfaceRequest<serial::DataSink> sink,
+                   mojo::InterfaceRequest<serial::DataSource> source);
   virtual ~SerialConnection();
 
   // mojo::InterfaceImpl<serial::Connection> overrides.
@@ -31,7 +38,13 @@ class SerialConnection : public mojo::InterfaceImpl<serial::Connection> {
   virtual void Flush(const mojo::Callback<void(bool)>& callback) OVERRIDE;
 
  private:
+  void OnSendPipeReady(scoped_ptr<ReadOnlyBuffer> buffer);
+  void OnSendCancelled(int32_t error);
+  void OnReceivePipeReady(scoped_ptr<WritableBuffer> buffer);
+
   scoped_refptr<SerialIoHandler> io_handler_;
+  scoped_refptr<DataSinkReceiver> receiver_;
+  scoped_refptr<DataSourceSender> sender_;
 
   DISALLOW_COPY_AND_ASSIGN(SerialConnection);
 };

device/serial/serial_connection_factory.cc

@@ -35,7 +35,9 @@ class SerialConnectionFactory::ConnectTask
   ConnectTask(scoped_refptr<SerialConnectionFactory> factory,
               const std::string& path,
               serial::ConnectionOptionsPtr options,
-              mojo::InterfaceRequest<serial::Connection> connection_request);
+              mojo::InterfaceRequest<serial::Connection> connection_request,
+              mojo::InterfaceRequest<serial::DataSink> sink,
+              mojo::InterfaceRequest<serial::DataSource> source);
   void Run();
 
  private:
@@ -48,6 +50,8 @@ class SerialConnectionFactory::ConnectTask
   const std::string path_;
   serial::ConnectionOptionsPtr options_;
   mojo::InterfaceRequest<serial::Connection> connection_request_;
+  mojo::InterfaceRequest<serial::DataSink> sink_;
+  mojo::InterfaceRequest<serial::DataSource> source_;
   scoped_refptr<SerialIoHandler> io_handler_;
 
   DISALLOW_COPY_AND_ASSIGN(ConnectTask);
@@ -63,9 +67,15 @@ SerialConnectionFactory::SerialConnectionFactory(
 void SerialConnectionFactory::CreateConnection(
     const std::string& path,
     serial::ConnectionOptionsPtr options,
-    mojo::InterfaceRequest<serial::Connection> connection_request) {
-  scoped_refptr<ConnectTask> task(
-      new ConnectTask(this, path, options.Pass(), connection_request.Pass()));
+    mojo::InterfaceRequest<serial::Connection> connection_request,
+    mojo::InterfaceRequest<serial::DataSink> sink,
+    mojo::InterfaceRequest<serial::DataSource> source) {
+  scoped_refptr<ConnectTask> task(new ConnectTask(this,
+                                                  path,
+                                                  options.Pass(),
+                                                  connection_request.Pass(),
+                                                  sink.Pass(),
+                                                  source.Pass()));
   task->Run();
 }
 
@@ -76,11 +86,15 @@ SerialConnectionFactory::ConnectTask::ConnectTask(
     scoped_refptr<SerialConnectionFactory> factory,
     const std::string& path,
     serial::ConnectionOptionsPtr options,
-    mojo::InterfaceRequest<serial::Connection> connection_request)
+    mojo::InterfaceRequest<serial::Connection> connection_request,
+    mojo::InterfaceRequest<serial::DataSink> sink,
+    mojo::InterfaceRequest<serial::DataSource> source)
     : factory_(factory),
       path_(path),
       options_(options.Pass()),
-      connection_request_(connection_request.Pass()) {
+      connection_request_(connection_request.Pass()),
+      sink_(sink.Pass()),
+      source_(source.Pass()) {
 }
 
 void SerialConnectionFactory::ConnectTask::Run() {
@@ -108,7 +122,9 @@ void SerialConnectionFactory::ConnectTask::OnConnected(bool success) {
   FillDefaultConnectionOptions(options_.get());
   if (!io_handler_->ConfigurePort(*options_))
     return;
-  mojo::BindToRequest(new SerialConnection(io_handler_), &connection_request_);
+  mojo::BindToRequest(
+      new SerialConnection(io_handler_, sink_.Pass(), source_.Pass()),
+      &connection_request_);
 }
 
 }  // namespace device

device/serial/serial_connection_factory.h

@@ -10,6 +10,7 @@
 #include "base/callback.h"
 #include "base/memory/ref_counted.h"
 #include "base/message_loop/message_loop_proxy.h"
+#include "device/serial/data_stream.mojom.h"
 #include "device/serial/serial.mojom.h"
 #include "mojo/public/cpp/bindings/interface_request.h"
 
@@ -29,7 +30,9 @@ class SerialConnectionFactory
   void CreateConnection(
       const std::string& path,
       serial::ConnectionOptionsPtr options,
-      mojo::InterfaceRequest<serial::Connection> connection_request);
+      mojo::InterfaceRequest<serial::Connection> connection_request,
+      mojo::InterfaceRequest<serial::DataSink> sink,
+      mojo::InterfaceRequest<serial::DataSource> source);
 
  private:
   friend class base::RefCountedThreadSafe<SerialConnectionFactory>;

device/serial/serial_connection_unittest.cc

@@ -2,10 +2,17 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
+#include <string>
+
 #include "base/bind.h"
 #include "base/message_loop/message_loop.h"
 #include "base/run_loop.h"
+#include "base/strings/string_piece.h"
+#include "device/serial/data_receiver.h"
+#include "device/serial/data_sender.h"
+#include "device/serial/data_stream.mojom.h"
 #include "device/serial/serial.mojom.h"
+#include "device/serial/serial_connection.h"
 #include "device/serial/serial_service_impl.h"
 #include "device/serial/test_serial_io_handler.h"
 #include "mojo/public/cpp/bindings/error_handler.h"
@@ -29,7 +36,31 @@ class FakeSerialDeviceEnumerator : public SerialDeviceEnumerator {
 
 class SerialConnectionTest : public testing::Test, public mojo::ErrorHandler {
  public:
-  SerialConnectionTest() : connected_(false), success_(false) {}
+  enum Event {
+    EVENT_NONE,
+    EVENT_GOT_INFO,
+    EVENT_SET_OPTIONS,
+    EVENT_GOT_CONTROL_SIGNALS,
+    EVENT_SET_CONTROL_SIGNALS,
+    EVENT_FLUSHED,
+    EVENT_DATA_AT_IO_HANDLER,
+    EVENT_DATA_SENT,
+    EVENT_SEND_ERROR,
+    EVENT_DATA_RECEIVED,
+    EVENT_RECEIVE_ERROR,
+    EVENT_CANCEL_COMPLETE,
+    EVENT_ERROR,
+  };
+
+  static const uint32_t kBufferSize;
+
+  SerialConnectionTest()
+      : connected_(false),
+        success_(false),
+        bytes_sent_(0),
+        send_error_(serial::SEND_ERROR_NONE),
+        receive_error_(serial::RECEIVE_ERROR_NONE),
+        expected_event_(EVENT_NONE) {}
 
   virtual void SetUp() OVERRIDE {
     message_loop_.reset(new base::MessageLoop);
@@ -43,39 +74,69 @@ class SerialConnectionTest : public testing::Test, public mojo::ErrorHandler {
             scoped_ptr<SerialDeviceEnumerator>(new FakeSerialDeviceEnumerator)),
         &service);
     service.set_error_handler(this);
-    service->Connect(
-        "device", serial::ConnectionOptions::New(), mojo::Get(&connection_));
+    mojo::InterfacePtr<serial::DataSink> consumer;
+    mojo::InterfacePtr<serial::DataSource> producer;
+    service->Connect("device",
+                     serial::ConnectionOptions::New(),
+                     mojo::Get(&connection_),
+                     mojo::Get(&consumer),
+                     mojo::Get(&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)));
-    RunMessageLoop();
+    WaitForEvent(EVENT_GOT_INFO);
     ASSERT_TRUE(io_handler_);
   }
 
   void StoreInfo(serial::ConnectionInfoPtr options) {
     info_ = options.Pass();
-    StopMessageLoop();
+    EventReceived(EVENT_GOT_INFO);
   }
 
   void StoreControlSignals(serial::DeviceControlSignalsPtr signals) {
     signals_ = signals.Pass();
-    StopMessageLoop();
+    EventReceived(EVENT_GOT_CONTROL_SIGNALS);
   }
 
-  void StoreSuccess(bool success) {
+  void StoreSuccess(Event event_to_report, bool success) {
     success_ = success;
-    StopMessageLoop();
+    EventReceived(event_to_report);
+  }
+
+  void Send(const base::StringPiece& data) {
+    ASSERT_TRUE(sender_->Send(
+        data,
+        base::Bind(&SerialConnectionTest::OnDataSent, base::Unretained(this)),
+        base::Bind(&SerialConnectionTest::OnSendError,
+                   base::Unretained(this))));
+  }
+
+  void Receive() {
+    ASSERT_TRUE(
+        receiver_->Receive(base::Bind(&SerialConnectionTest::OnDataReceived,
+                                      base::Unretained(this)),
+                           base::Bind(&SerialConnectionTest::OnReceiveError,
+                                      base::Unretained(this))));
   }
 
-  void RunMessageLoop() {
-    run_loop_.reset(new base::RunLoop);
-    run_loop_->Run();
+  void WaitForEvent(Event event) {
+    expected_event_ = event;
+    base::RunLoop run_loop;
+    stop_run_loop_ = run_loop.QuitClosure();
+    run_loop.Run();
   }
 
-  void StopMessageLoop() {
+  void EventReceived(Event event) {
+    if (event != expected_event_)
+      return;
+    expected_event_ = EVENT_NONE;
     ASSERT_TRUE(message_loop_);
-    ASSERT_TRUE(run_loop_);
-    message_loop_->PostTask(FROM_HERE, run_loop_->QuitClosure());
+    ASSERT_TRUE(!stop_run_loop_.is_null());
+    message_loop_->PostTask(FROM_HERE, stop_run_loop_);
   }
 
   scoped_refptr<SerialIoHandler> CreateIoHandler() {
@@ -83,8 +144,32 @@ class SerialConnectionTest : public testing::Test, public mojo::ErrorHandler {
     return io_handler_;
   }
 
+  void OnDataSent(uint32_t bytes_sent) {
+    bytes_sent_ += bytes_sent;
+    send_error_ = serial::SEND_ERROR_NONE;
+    EventReceived(EVENT_DATA_SENT);
+  }
+
+  void OnSendError(uint32_t bytes_sent, int32_t error) {
+    bytes_sent_ += bytes_sent;
+    send_error_ = static_cast<serial::SendError>(error);
+    EventReceived(EVENT_SEND_ERROR);
+  }
+
+  void OnDataReceived(scoped_ptr<ReadOnlyBuffer> buffer) {
+    data_received_ += std::string(buffer->GetData(), buffer->GetSize());
+    buffer->Done(buffer->GetSize());
+    receive_error_ = serial::RECEIVE_ERROR_NONE;
+    EventReceived(EVENT_DATA_RECEIVED);
+  }
+
+  void OnReceiveError(int32_t error) {
+    receive_error_ = static_cast<serial::ReceiveError>(error);
+    EventReceived(EVENT_RECEIVE_ERROR);
+  }
+
   virtual void OnConnectionError() OVERRIDE {
-    StopMessageLoop();
+    EventReceived(EVENT_ERROR);
     FAIL() << "Connection error";
   }
 
@@ -93,16 +178,25 @@ class SerialConnectionTest : public testing::Test, public mojo::ErrorHandler {
   serial::DeviceControlSignalsPtr signals_;
   bool connected_;
   bool success_;
+  int bytes_sent_;
+  serial::SendError send_error_;
+  serial::ReceiveError receive_error_;
+  std::string data_received_;
+  Event expected_event_;
 
   scoped_ptr<base::MessageLoop> message_loop_;
-  scoped_ptr<base::RunLoop> run_loop_;
+  base::Closure stop_run_loop_;
   mojo::InterfacePtr<serial::Connection> connection_;
+  scoped_ptr<DataSender> sender_;
+  scoped_refptr<DataReceiver> receiver_;
   scoped_refptr<TestSerialIoHandler> io_handler_;
 
  private:
   DISALLOW_COPY_AND_ASSIGN(SerialConnectionTest);
 };
 
+const uint32_t SerialConnectionTest::kBufferSize = 10;
+
 TEST_F(SerialConnectionTest, GetInfo) {
   // |info_| is filled in during SetUp().
   ASSERT_TRUE(info_);
@@ -119,10 +213,11 @@ TEST_F(SerialConnectionTest, SetOptions) {
   options->data_bits = serial::DATA_BITS_SEVEN;
   options->has_cts_flow_control = true;
   options->cts_flow_control = true;
-  connection_->SetOptions(
-      options.Pass(),
-      base::Bind(&SerialConnectionTest::StoreSuccess, base::Unretained(this)));
-  RunMessageLoop();
+  connection_->SetOptions(options.Pass(),
+                          base::Bind(&SerialConnectionTest::StoreSuccess,
+                                     base::Unretained(this),
+                                     EVENT_SET_OPTIONS));
+  WaitForEvent(EVENT_SET_OPTIONS);
   ASSERT_TRUE(success_);
   serial::ConnectionInfo* info = io_handler_->connection_info();
   EXPECT_EQ(12345u, info->bitrate);
@@ -139,7 +234,7 @@ TEST_F(SerialConnectionTest, GetControlSignals) {
   signals->dcd = true;
   signals->dsr = true;
 
-  RunMessageLoop();
+  WaitForEvent(EVENT_GOT_CONTROL_SIGNALS);
   ASSERT_TRUE(signals_);
   EXPECT_TRUE(signals_->dcd);
   EXPECT_FALSE(signals_->cts);
@@ -154,10 +249,11 @@ TEST_F(SerialConnectionTest, SetControlSignals) {
   signals->has_rts = true;
   signals->rts = true;
 
-  connection_->SetControlSignals(
-      signals.Pass(),
-      base::Bind(&SerialConnectionTest::StoreSuccess, base::Unretained(this)));
-  RunMessageLoop();
+  connection_->SetControlSignals(signals.Pass(),
+                                 base::Bind(&SerialConnectionTest::StoreSuccess,
+                                            base::Unretained(this),
+                                            EVENT_SET_CONTROL_SIGNALS));
+  WaitForEvent(EVENT_SET_CONTROL_SIGNALS);
   ASSERT_TRUE(success_);
   EXPECT_TRUE(io_handler_->dtr());
   EXPECT_TRUE(io_handler_->rts());
@@ -165,17 +261,67 @@ TEST_F(SerialConnectionTest, SetControlSignals) {
 
 TEST_F(SerialConnectionTest, Flush) {
   ASSERT_EQ(0, io_handler_->flushes());
-  connection_->Flush(
-      base::Bind(&SerialConnectionTest::StoreSuccess, base::Unretained(this)));
-  RunMessageLoop();
+  connection_->Flush(base::Bind(&SerialConnectionTest::StoreSuccess,
+                                base::Unretained(this),
+                                EVENT_FLUSHED));
+  WaitForEvent(EVENT_FLUSHED);
   ASSERT_TRUE(success_);
   EXPECT_EQ(1, io_handler_->flushes());
 }
 
 TEST_F(SerialConnectionTest, Disconnect) {
   connection_.reset();
-  message_loop_.reset();
+  io_handler_->set_send_callback(base::Bind(base::DoNothing));
+  ASSERT_NO_FATAL_FAILURE(Send("data"));
+  WaitForEvent(EVENT_SEND_ERROR);
+  EXPECT_EQ(serial::SEND_ERROR_DISCONNECTED, send_error_);
+  EXPECT_EQ(0, bytes_sent_);
+  ASSERT_NO_FATAL_FAILURE(Receive());
+  WaitForEvent(EVENT_RECEIVE_ERROR);
+  EXPECT_EQ(serial::RECEIVE_ERROR_DISCONNECTED, receive_error_);
+  EXPECT_EQ("", data_received_);
   EXPECT_TRUE(io_handler_->HasOneRef());
 }
 
+TEST_F(SerialConnectionTest, Echo) {
+  ASSERT_NO_FATAL_FAILURE(Send("data"));
+  WaitForEvent(EVENT_DATA_SENT);
+  EXPECT_EQ(serial::SEND_ERROR_NONE, send_error_);
+  EXPECT_EQ(4, bytes_sent_);
+  ASSERT_NO_FATAL_FAILURE(Receive());
+  WaitForEvent(EVENT_DATA_RECEIVED);
+  EXPECT_EQ("data", data_received_);
+  EXPECT_EQ(serial::RECEIVE_ERROR_NONE, receive_error_);
+}
+
+TEST_F(SerialConnectionTest, Cancel) {
+  // To test that cancels are correctly passed to the IoHandler, we need a send
+  // to be in progress because otherwise, the DataSinkReceiver would handle the
+  // cancel internally.
+  io_handler_->set_send_callback(
+      base::Bind(&SerialConnectionTest::EventReceived,
+                 base::Unretained(this),
+                 EVENT_DATA_AT_IO_HANDLER));
+  ASSERT_NO_FATAL_FAILURE(Send("something else"));
+  WaitForEvent(EVENT_DATA_AT_IO_HANDLER);
+  EXPECT_EQ(0, bytes_sent_);
+
+  ASSERT_TRUE(sender_->Cancel(serial::SEND_ERROR_TIMEOUT,
+                              base::Bind(&SerialConnectionTest::EventReceived,
+                                         base::Unretained(this),
+                                         EVENT_CANCEL_COMPLETE)));
+
+  WaitForEvent(EVENT_CANCEL_COMPLETE);
+  EXPECT_EQ(serial::SEND_ERROR_TIMEOUT, send_error_);
+
+  ASSERT_NO_FATAL_FAILURE(Send("data"));
+  WaitForEvent(EVENT_DATA_SENT);
+  EXPECT_EQ(serial::SEND_ERROR_NONE, send_error_);
+  EXPECT_EQ(4, bytes_sent_);
+  ASSERT_NO_FATAL_FAILURE(Receive());
+  WaitForEvent(EVENT_DATA_RECEIVED);
+  EXPECT_EQ("data", data_received_);
+  EXPECT_EQ(serial::RECEIVE_ERROR_NONE, receive_error_);
+}
+
 }  // namespace device

device/serial/serial_service_impl.cc

@@ -55,11 +55,16 @@ void SerialServiceImpl::GetDevices(
 void SerialServiceImpl::Connect(
     const mojo::String& path,
     serial::ConnectionOptionsPtr options,
-    mojo::InterfaceRequest<serial::Connection> connection_request) {
+    mojo::InterfaceRequest<serial::Connection> connection_request,
+    mojo::InterfaceRequest<serial::DataSink> sink,
+    mojo::InterfaceRequest<serial::DataSource> source) {
   if (!IsValidPath(path))
     return;
-  connection_factory_->CreateConnection(
-      path, options.Pass(), connection_request.Pass());
+  connection_factory_->CreateConnection(path,
+                                        options.Pass(),
+                                        connection_request.Pass(),
+                                        sink.Pass(),
+                                        source.Pass());
 }
 
 SerialDeviceEnumerator* SerialServiceImpl::GetDeviceEnumerator() {
@@ -78,10 +83,4 @@ bool SerialServiceImpl::IsValidPath(const mojo::String& path) {
   return false;
 }
 
-void SerialServiceImpl::OnConnected(
-    const mojo::Callback<void(serial::ConnectionInfoPtr)>& callback,
-    serial::ConnectionInfoPtr result) {
-  callback.Run(result.Pass());
-}
-
 }  // namespace device

device/serial/serial_service_impl.h

@@ -7,6 +7,7 @@
 
 #include "base/memory/scoped_ptr.h"
 #include "base/message_loop/message_loop_proxy.h"
+#include "device/serial/data_stream.mojom.h"
 #include "device/serial/serial.mojom.h"
 #include "device/serial/serial_connection_factory.h"
 #include "device/serial/serial_device_enumerator.h"
@@ -35,14 +36,13 @@ class SerialServiceImpl : public mojo::InterfaceImpl<serial::SerialService> {
   virtual void Connect(
       const mojo::String& path,
       serial::ConnectionOptionsPtr options,
-      mojo::InterfaceRequest<serial::Connection> connection_request) OVERRIDE;
+      mojo::InterfaceRequest<serial::Connection> connection_request,
+      mojo::InterfaceRequest<serial::DataSink> sink,
+      mojo::InterfaceRequest<serial::DataSource> source) OVERRIDE;
 
  private:
   SerialDeviceEnumerator* GetDeviceEnumerator();
   bool IsValidPath(const mojo::String& path);
-  void OnConnected(
-      const mojo::Callback<void(serial::ConnectionInfoPtr)>& callback,
-      serial::ConnectionInfoPtr result);
 
   scoped_ptr<SerialDeviceEnumerator> device_enumerator_;
   scoped_refptr<SerialConnectionFactory> connection_factory_;

device/serial/serial_service_unittest.cc

@@ -27,10 +27,6 @@ class FakeSerialDeviceEnumerator : public SerialDeviceEnumerator {
 
 class FailToOpenIoHandler : public TestSerialIoHandler {
  public:
-  static scoped_refptr<SerialIoHandler> Create() {
-    return new FailToOpenIoHandler;
-  }
-
   virtual void Open(const std::string& port,
                     const OpenCompleteCallback& callback) OVERRIDE {
     callback.Run(false);
@@ -44,7 +40,7 @@ class FailToOpenIoHandler : public TestSerialIoHandler {
 
 class SerialServiceTest : public testing::Test, public mojo::ErrorHandler {
  public:
-  SerialServiceTest() {}
+  SerialServiceTest() : connected_(false), expecting_error_(false) {}
 
   void StoreDevices(mojo::Array<serial::DeviceInfoPtr> devices) {
     devices_ = devices.Pass();
@@ -66,12 +62,48 @@ class SerialServiceTest : public testing::Test, public mojo::ErrorHandler {
     message_loop_.PostTask(FROM_HERE, run_loop_->QuitClosure());
   }
 
-  void OnGotInfo(serial::ConnectionInfoPtr options) { StopMessageLoop(); }
+  void OnGotInfo(serial::ConnectionInfoPtr options) {
+    connected_ = true;
+    StopMessageLoop();
+  }
+
+  scoped_refptr<SerialIoHandler> ReturnIoHandler() { return io_handler_; }
+
+  void RunConnectTest(const std::string& path, bool expecting_success) {
+    if (!io_handler_)
+      io_handler_ = new TestSerialIoHandler;
+    mojo::InterfacePtr<serial::SerialService> service;
+    mojo::BindToProxy(
+        new SerialServiceImpl(
+            new SerialConnectionFactory(
+                base::Bind(&SerialServiceTest::ReturnIoHandler,
+                           base::Unretained(this)),
+                base::MessageLoopProxy::current()),
+            scoped_ptr<SerialDeviceEnumerator>(new FakeSerialDeviceEnumerator)),
+        &service);
+    mojo::InterfacePtr<serial::Connection> connection;
+    mojo::InterfacePtr<serial::DataSink> sink;
+    mojo::InterfacePtr<serial::DataSource> source;
+    service->Connect(path,
+                     serial::ConnectionOptions::New(),
+                     mojo::Get(&connection),
+                     mojo::Get(&sink),
+                     mojo::Get(&source));
+    connection.set_error_handler(this);
+    expecting_error_ = !expecting_success;
+    connection->GetInfo(
+        base::Bind(&SerialServiceTest::OnGotInfo, base::Unretained(this)));
+    RunMessageLoop();
+    EXPECT_EQ(!expecting_success, connection.encountered_error());
+    EXPECT_EQ(expecting_success, connected_);
+    connection.reset();
+  }
 
   base::MessageLoop message_loop_;
   scoped_ptr<base::RunLoop> run_loop_;
   mojo::Array<serial::DeviceInfoPtr> devices_;
   scoped_refptr<TestSerialIoHandler> io_handler_;
+  bool connected_;
   bool expecting_error_;
   serial::ConnectionInfoPtr info_;
 
@@ -94,56 +126,16 @@ TEST_F(SerialServiceTest, GetDevices) {
 }
 
 TEST_F(SerialServiceTest, Connect) {
-  mojo::InterfacePtr<serial::SerialService> service;
-  mojo::BindToProxy(
-      new SerialServiceImpl(
-          new SerialConnectionFactory(base::Bind(&TestSerialIoHandler::Create),
-                                      base::MessageLoopProxy::current()),
-          scoped_ptr<SerialDeviceEnumerator>(new FakeSerialDeviceEnumerator)),
-      &service);
-  service.set_error_handler(this);
-  mojo::InterfacePtr<serial::Connection> connection;
-  service->Connect(
-      "device", serial::ConnectionOptions::New(), mojo::Get(&connection));
-  connection.set_error_handler(this);
-  connection->GetInfo(
-      base::Bind(&SerialServiceTest::OnGotInfo, base::Unretained(this)));
-  RunMessageLoop();
-  connection.reset();
+  RunConnectTest("device", true);
 }
 
 TEST_F(SerialServiceTest, ConnectInvalidPath) {
-  mojo::InterfacePtr<serial::SerialService> service;
-  mojo::BindToProxy(
-      new SerialServiceImpl(
-          new SerialConnectionFactory(base::Bind(&TestSerialIoHandler::Create),
-                                      base::MessageLoopProxy::current()),
-          scoped_ptr<SerialDeviceEnumerator>(new FakeSerialDeviceEnumerator)),
-      &service);
-  mojo::InterfacePtr<serial::Connection> connection;
-  service->Connect(
-      "invalid_path", serial::ConnectionOptions::New(), mojo::Get(&connection));
-  connection.set_error_handler(this);
-  expecting_error_ = true;
-  RunMessageLoop();
-  EXPECT_TRUE(connection.encountered_error());
+  RunConnectTest("invalid_path", false);
 }
 
 TEST_F(SerialServiceTest, ConnectOpenFailed) {
-  mojo::InterfacePtr<serial::SerialService> service;
-  mojo::BindToProxy(
-      new SerialServiceImpl(
-          new SerialConnectionFactory(base::Bind(&FailToOpenIoHandler::Create),
-                                      base::MessageLoopProxy::current()),
-          scoped_ptr<SerialDeviceEnumerator>(new FakeSerialDeviceEnumerator)),
-      &service);
-  mojo::InterfacePtr<serial::Connection> connection;
-  service->Connect(
-      "device", serial::ConnectionOptions::New(), mojo::Get(&connection));
-  expecting_error_ = true;
-  connection.set_error_handler(this);
-  RunMessageLoop();
-  EXPECT_TRUE(connection.encountered_error());
+  io_handler_ = new FailToOpenIoHandler;
+  RunConnectTest("device", false);
 }
 
 }  // namespace device

device/serial/test_serial_io_handler.cc

@@ -4,6 +4,8 @@
 
 #include "device/serial/test_serial_io_handler.h"
 
+#include <algorithm>
+
 #include "base/bind.h"
 #include "device/serial/serial.mojom.h"
 
@@ -44,24 +46,37 @@ bool TestSerialIoHandler::ConfigurePort(
 }
 
 void TestSerialIoHandler::ReadImpl() {
+  if (!pending_read_buffer())
+    return;
+  if (buffer_.empty())
+    return;
+
+  size_t num_bytes =
+      std::min(buffer_.size(), static_cast<size_t>(pending_read_buffer_len()));
+  memcpy(pending_read_buffer(), buffer_.c_str(), num_bytes);
+  buffer_ = buffer_.substr(num_bytes);
+  ReadCompleted(static_cast<uint32_t>(num_bytes), serial::RECEIVE_ERROR_NONE);
 }
 
 void TestSerialIoHandler::CancelReadImpl() {
-  QueueReadCompleted(0, read_cancel_reason());
+  ReadCompleted(0, read_cancel_reason());
 }
 
 void TestSerialIoHandler::WriteImpl() {
-  DCHECK(pending_read_buffer());
-  DCHECK_LE(pending_write_buffer_len(), pending_read_buffer_len());
-  memcpy(pending_read_buffer(),
-         pending_write_buffer(),
-         pending_write_buffer_len());
-  QueueReadCompleted(pending_write_buffer_len(), serial::RECEIVE_ERROR_NONE);
-  QueueWriteCompleted(pending_write_buffer_len(), serial::SEND_ERROR_NONE);
+  if (!send_callback_.is_null()) {
+    base::Closure callback = send_callback_;
+    send_callback_.Reset();
+    callback.Run();
+    return;
+  }
+  buffer_ += std::string(pending_write_buffer(), pending_write_buffer_len());
+  WriteCompleted(pending_write_buffer_len(), serial::SEND_ERROR_NONE);
+  if (pending_read_buffer())
+    ReadImpl();
 }
 
 void TestSerialIoHandler::CancelWriteImpl() {
-  QueueWriteCompleted(0, write_cancel_reason());
+  WriteCompleted(0, write_cancel_reason());
 }
 
 serial::DeviceControlSignalsPtr TestSerialIoHandler::GetControlSignals() const {

device/serial/test_serial_io_handler.h

@@ -40,6 +40,11 @@ class TestSerialIoHandler : public SerialIoHandler {
   bool dtr() { return dtr_; }
   bool rts() { return rts_; }
   int flushes() { return flushes_; }
+  // This callback will be called when this IoHandler processes its next write,
+  // instead of the normal behavior of echoing the data to reads.
+  void set_send_callback(const base::Closure& callback) {
+    send_callback_ = callback;
+  }
 
  protected:
   virtual ~TestSerialIoHandler();
@@ -51,6 +56,8 @@ class TestSerialIoHandler : public SerialIoHandler {
   bool dtr_;
   bool rts_;
   mutable int flushes_;
+  std::string buffer_;
+  base::Closure send_callback_;
 
   DISALLOW_COPY_AND_ASSIGN(TestSerialIoHandler);
 };

extensions/renderer/api/serial/serial_api_unittest.cc

@@ -308,6 +308,8 @@ class SerialApiTest : public ApiTestBase {
     ApiTestBase::SetUp();
     env()->RegisterModule("serial", IDR_SERIAL_CUSTOM_BINDINGS_JS);
     env()->RegisterModule("serial_service", IDR_SERIAL_SERVICE_JS);
+    env()->RegisterModule("device/serial/data_stream.mojom",
+                          IDR_DATA_STREAM_MOJOM_JS);
     env()->RegisterModule("device/serial/serial.mojom", IDR_SERIAL_MOJOM_JS);
     service_provider()->AddService<device::serial::SerialService>(base::Bind(
         &SerialApiTest::CreateSerialService, base::Unretained(this)));

extensions/renderer/resources/serial_service.js

@@ -122,7 +122,19 @@ define('serial_service', [
     options = options || {};
     var serviceOptions = getServiceOptions(options);
     var pipe = core.createMessagePipe();
-    service.connect(path, serviceOptions, pipe.handle0);
+    // Note: These two are created and closed because the service implementation
+    // requires that we provide valid message pipes for the data source and
+    // sink. Currently the client handles are immediately closed; the real
+    // implementation will come later.
+    var sendPipe = core.createMessagePipe();
+    var receivePipe = core.createMessagePipe();
+    service.connect(path,
+                    serviceOptions,
+                    pipe.handle0,
+                    sendPipe.handle0,
+                    receivePipe.handle0);
+    core.close(sendPipe.handle1);
+    core.close(receivePipe.handle1);
     var router = new routerModule.Router(pipe.handle1);
     var connection = new serialMojom.ConnectionProxy(router);
     return connection.getInfo().then(convertServiceInfo).then(