[Bluetooth] Add mojom::Socket and associated mojom::Adapter method.

Implement mojom::Socket, a Mojo interface for interacting with
device::BluetoothSocket. This implementation mirrors the
implementation of tcp_socket.mojom: it uses a pair of Mojo
DataPipes to stream sent and received data, instead of mirroring
device::BluetoothSocket's Send() and Receive() methods.

This CL only adds mojom::Adapter::ConnectToServiceInsecurely(), to
allow clients to initiate a connection. A subsequent CL will add
mojom::Adapter::CreateRfcommService() to allow listening for an
incoming connection.

This implementation has been manually verified by running a Nearby Share
(go/nearby-chrome-bt) prototype on a Pixelbook, communicating with a
Pixel 3.

Please see design doc go/nearby-chrome-bt for more details.

Bug: 1043300
Change-Id: Ice8110ef9bce905ab46cc38fee3110f8d0775644
Fixed: b:157750558, b:157484150
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2292871
Commit-Queue: Ryan Hansberry <hansberry@chromium.org>
Reviewed-by: Daniel Cheng <dcheng@chromium.org>
Reviewed-by: James Vecore <vecore@google.com>
Reviewed-by: Alex Chau <alexchau@chromium.org>
Reviewed-by: Reilly Grant <reillyg@chromium.org>
Cr-Commit-Position: refs/heads/master@{#795318}

chrome/services/sharing/nearby/test_support/fake_adapter.cc

@@ -27,4 +27,9 @@ void FakeAdapter::SetClient(
 void FakeAdapter::StartDiscoverySession(
     StartDiscoverySessionCallback callback) {}
 
+void FakeAdapter::ConnectToServiceInsecurely(
+    const std::string& address,
+    const device::BluetoothUUID& service_uuid,
+    ConnectToServiceInsecurelyCallback callback) {}
+
 }  // namespace bluetooth

chrome/services/sharing/nearby/test_support/fake_adapter.h

@@ -24,6 +24,10 @@ class FakeAdapter : public mojom::Adapter {
   void GetInfo(GetInfoCallback callback) override;
   void SetClient(::mojo::PendingRemote<mojom::AdapterClient> client) override;
   void StartDiscoverySession(StartDiscoverySessionCallback callback) override;
+  void ConnectToServiceInsecurely(
+      const std::string& address,
+      const device::BluetoothUUID& service_uuid,
+      ConnectToServiceInsecurelyCallback callback) override;
 
   mojo::Receiver<mojom::Adapter> adapter{this};
   bool present = true;

chrome/test/data/webui/bluetooth_internals_browsertest.js

@@ -130,6 +130,10 @@ BluetoothInternalsTest.prototype = {
         return {session: null};
       }
 
+      async connectToServiceInsecurely(address, service_uuid) {
+        return {result: null};
+      }
+
       setTestConnectResult(connectResult) {
         this.connectResult_ = connectResult;
       }

device/BUILD.gn

@@ -41,6 +41,7 @@ test("device_unittests") {
     "bluetooth/bluetooth_task_manager_win_unittest.cc",
     "bluetooth/device_unittest.cc",
     "bluetooth/public/cpp/bluetooth_uuid_unittest.cc",
+    "bluetooth/socket_unittest.cc",
     "bluetooth/string_util_icu_unittest.cc",
     "bluetooth/test/bluetooth_gatt_server_test.cc",
     "bluetooth/test/bluetooth_gatt_server_test.h",

device/bluetooth/BUILD.gn

@@ -40,12 +40,15 @@ source_set("deprecated_experimental_mojo") {
     "device.h",
     "discovery_session.cc",
     "discovery_session.h",
+    "socket.cc",
+    "socket.h",
   ]
 
   deps = [
     ":bluetooth",
     "//device/bluetooth/public/mojom:deprecated_experimental_interfaces",
     "//mojo/public/cpp/bindings",
+    "//net",
   ]
 
   visibility = [

device/bluetooth/adapter.cc

@@ -2,6 +2,8 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
+#include "device/bluetooth/adapter.h"
+
 #include <string>
 #include <utility>
 #include <vector>
@@ -10,10 +12,12 @@
 #include "base/callback_helpers.h"
 #include "base/memory/ptr_util.h"
 #include "build/build_config.h"
-#include "device/bluetooth/adapter.h"
+#include "device/bluetooth/bluetooth_socket.h"
 #include "device/bluetooth/device.h"
 #include "device/bluetooth/discovery_session.h"
+#include "device/bluetooth/public/cpp/bluetooth_uuid.h"
 #include "device/bluetooth/public/mojom/connect_result_type_converter.h"
+#include "device/bluetooth/socket.h"
 #include "mojo/public/cpp/bindings/pending_remote.h"
 #include "mojo/public/cpp/bindings/self_owned_receiver.h"
 
@@ -35,7 +39,7 @@ void Adapter::ConnectToDevice(const std::string& address,
 
   if (!device) {
     std::move(callback).Run(mojom::ConnectResult::DEVICE_NO_LONGER_IN_RANGE,
-                            /* device */ mojo::NullRemote());
+                            /*device=*/mojo::NullRemote());
     return;
   }
 
@@ -87,6 +91,19 @@ void Adapter::StartDiscoverySession(StartDiscoverySessionCallback callback) {
                      weak_ptr_factory_.GetWeakPtr(), copyable_callback));
 }
 
+void Adapter::ConnectToServiceInsecurely(
+    const std::string& address,
+    const device::BluetoothUUID& service_uuid,
+    ConnectToServiceInsecurelyCallback callback) {
+  auto copyable_callback = base::AdaptCallbackForRepeating(std::move(callback));
+  adapter_->GetDevice(address)->ConnectToServiceInsecurely(
+      service_uuid,
+      base::BindOnce(&Adapter::OnConnectToService,
+                     weak_ptr_factory_.GetWeakPtr(), copyable_callback),
+      base::BindOnce(&Adapter::OnConnectToServiceError,
+                     weak_ptr_factory_.GetWeakPtr(), copyable_callback));
+}
+
 void Adapter::AdapterPresentChanged(device::BluetoothAdapter* adapter,
                                     bool present) {
   if (client_)
@@ -148,7 +165,7 @@ void Adapter::OnConnectError(
     ConnectToDeviceCallback callback,
     device::BluetoothDevice::ConnectErrorCode error_code) {
   std::move(callback).Run(mojo::ConvertTo<mojom::ConnectResult>(error_code),
-                          /* device */ mojo::NullRemote());
+                          /*device=*/mojo::NullRemote());
 }
 
 void Adapter::OnStartDiscoverySession(
@@ -162,7 +179,57 @@ void Adapter::OnStartDiscoverySession(
 }
 
 void Adapter::OnDiscoverySessionError(StartDiscoverySessionCallback callback) {
-  std::move(callback).Run(mojo::NullRemote() /* session */);
+  std::move(callback).Run(/*session=*/mojo::NullRemote());
+}
+
+void Adapter::OnConnectToService(
+    ConnectToServiceInsecurelyCallback callback,
+    scoped_refptr<device::BluetoothSocket> socket) {
+  mojo::ScopedDataPipeProducerHandle receive_pipe_producer_handle;
+  mojo::ScopedDataPipeConsumerHandle receive_pipe_consumer_handle;
+  MojoResult result =
+      mojo::CreateDataPipe(/*options=*/nullptr, &receive_pipe_producer_handle,
+                           &receive_pipe_consumer_handle);
+  if (result != MOJO_RESULT_OK) {
+    socket->Close();
+    OnConnectToServiceError(std::move(callback),
+                            "Failed to create receiving DataPipe.");
+    return;
+  }
+
+  mojo::ScopedDataPipeProducerHandle send_pipe_producer_handle;
+  mojo::ScopedDataPipeConsumerHandle send_pipe_consumer_handle;
+  result = mojo::CreateDataPipe(/*options=*/nullptr, &send_pipe_producer_handle,
+                                &send_pipe_consumer_handle);
+  if (result != MOJO_RESULT_OK) {
+    socket->Close();
+    OnConnectToServiceError(std::move(callback),
+                            "Failed to create sending DataPipe.");
+    return;
+  }
+
+  mojo::PendingRemote<mojom::Socket> pending_socket;
+
+  mojo::MakeSelfOwnedReceiver(
+      std::make_unique<Socket>(std::move(socket),
+                               std::move(receive_pipe_producer_handle),
+                               std::move(send_pipe_consumer_handle)),
+      pending_socket.InitWithNewPipeAndPassReceiver());
+
+  mojom::ConnectToServiceResultPtr connect_to_service_result =
+      mojom::ConnectToServiceResult::New();
+  connect_to_service_result->socket = std::move(pending_socket);
+  connect_to_service_result->receive_stream =
+      std::move(receive_pipe_consumer_handle);
+  connect_to_service_result->send_stream = std::move(send_pipe_producer_handle);
+  std::move(callback).Run(std::move(connect_to_service_result));
+}
+
+void Adapter::OnConnectToServiceError(
+    ConnectToServiceInsecurelyCallback callback,
+    const std::string& message) {
+  DLOG(ERROR) << "Failed to connect to service: '" << message << "'";
+  std::move(callback).Run(/*result=*/nullptr);
 }
 
 }  // namespace bluetooth

device/bluetooth/adapter.h

@@ -12,6 +12,7 @@
 #include "base/memory/ref_counted.h"
 #include "device/bluetooth/bluetooth_adapter.h"
 #include "device/bluetooth/bluetooth_gatt_connection.h"
+#include "device/bluetooth/public/cpp/bluetooth_uuid.h"
 #include "device/bluetooth/public/mojom/adapter.mojom.h"
 #include "device/bluetooth/public/mojom/device.mojom-forward.h"
 #include "mojo/public/cpp/bindings/pending_remote.h"
@@ -36,6 +37,10 @@ class Adapter : public mojom::Adapter,
   void GetInfo(GetInfoCallback callback) override;
   void SetClient(mojo::PendingRemote<mojom::AdapterClient> client) override;
   void StartDiscoverySession(StartDiscoverySessionCallback callback) override;
+  void ConnectToServiceInsecurely(
+      const std::string& address,
+      const device::BluetoothUUID& service_uuid,
+      ConnectToServiceInsecurelyCallback callback) override;
 
   // device::BluetoothAdapter::Observer overrides:
   void AdapterPresentChanged(device::BluetoothAdapter* adapter,
@@ -67,6 +72,12 @@ class Adapter : public mojom::Adapter,
 
   void OnDiscoverySessionError(StartDiscoverySessionCallback callback);
 
+  void OnConnectToService(ConnectToServiceInsecurelyCallback callback,
+                          scoped_refptr<device::BluetoothSocket> socket);
+
+  void OnConnectToServiceError(ConnectToServiceInsecurelyCallback callback,
+                               const std::string& message);
+
   // The current Bluetooth adapter.
   scoped_refptr<device::BluetoothAdapter> adapter_;
 

device/bluetooth/public/mojom/adapter.mojom

@@ -5,6 +5,7 @@
 module bluetooth.mojom;
 
 import "device/bluetooth/public/mojom/device.mojom";
+import "device/bluetooth/public/mojom/uuid.mojom";
 
 // Possible errors sent as a response by Adapter.ConnectToDevice on a Device
 // connection request.
@@ -21,7 +22,15 @@ enum ConnectResult {
   DEVICE_NO_LONGER_IN_RANGE,
 };
 
+// The results of a successful connection via ConnectToServiceInsecurely().
+struct ConnectToServiceResult {
+  pending_remote<Socket> socket;
+  handle<data_pipe_consumer> receive_stream;
+  handle<data_pipe_producer> send_stream;
+};
+
 struct AdapterInfo {
+  // TODO(crbug.com/1112886): Use fixed-size array, not string, for address.
   string address;
   string name;
   string system_name;
@@ -52,10 +61,25 @@ interface DiscoverySession {
   Stop() => (bool success);
 };
 
+// Represents an open connection to a remote device. Releasing it will destroy
+// the underlying connection, but callers should prefer to let a call to
+// Disconnect() to finish first.
+interface Socket {
+  // Use to gracefully close the underlying connection before destroying. The
+  // reply callback can be used to synchronize a reconnection attempt;
+  // attempting to reconnect to the same service on this device may fail with a
+  // busy error until the reply callback is invoked.
+  [Sync]
+  Disconnect() => ();
+};
+
+// Handles requests to either query Bluetooth adapter capabilities or state, or
+// find or connect to remote devices. Backed by //device/bluetooth.
 interface Adapter {
   // Creates a GATT connection to the device with |address| and returns a
   // Device if the connection was succesful. The GATT connection is tied to the
   // the lifetime of the Device message pipe.
+  // TODO(crbug.com/1112886): Use fixed-size array, not string, for address.
   ConnectToDevice(string address) =>
       (ConnectResult result, pending_remote<Device>? device);
 
@@ -73,6 +97,16 @@ interface Adapter {
   // Requests the adapter to start a new discovery session. Returns null if
   // session not created successfully.
   StartDiscoverySession() => (pending_remote<DiscoverySession>? session);
+
+  // Attempts to initiate an insecure outgoing L2CAP or RFCOMM connection to the
+  // advertised service on this device matching |service_uuid|. This method is
+  // marked as "Insecurely" because the outgoing connection will not request
+  // bonding (pairing) to the remote device. All return values will be null if
+  // the connection attempt fails.
+  // TODO(crbug.com/1112886): Use fixed-size array, not string, for address.
+  [Sync]
+  ConnectToServiceInsecurely(string address, UUID service_uuid)
+      => (ConnectToServiceResult? result);
 };
 
 interface AdapterClient {

device/bluetooth/socket.cc

+// Copyright 2020 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 "device/bluetooth/socket.h"
+
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "base/bind.h"
+#include "base/callback_helpers.h"
+#include "base/memory/ptr_util.h"
+#include "device/bluetooth/bluetooth_socket.h"
+#include "mojo/public/cpp/bindings/pending_receiver.h"
+#include "mojo/public/cpp/bindings/receiver.h"
+#include "net/base/io_buffer.h"
+
+namespace bluetooth {
+
+Socket::Socket(scoped_refptr<device::BluetoothSocket> bluetooth_socket,
+               mojo::ScopedDataPipeProducerHandle receive_stream,
+               mojo::ScopedDataPipeConsumerHandle send_stream)
+    : bluetooth_socket_(std::move(bluetooth_socket)),
+      receive_stream_(std::move(receive_stream)),
+      send_stream_(std::move(send_stream)),
+      receive_stream_watcher_(FROM_HERE,
+                              mojo::SimpleWatcher::ArmingPolicy::MANUAL),
+      send_stream_watcher_(FROM_HERE,
+                           mojo::SimpleWatcher::ArmingPolicy::MANUAL) {
+  receive_stream_watcher_.Watch(
+      receive_stream_.get(),
+      MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+      base::BindRepeating(&Socket::OnReceiveStreamWritable,
+                          base::Unretained(this)));
+  send_stream_watcher_.Watch(
+      send_stream_.get(),
+      MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+      base::BindRepeating(&Socket::OnSendStreamReadable,
+                          base::Unretained(this)));
+  ReceiveMore();
+  SendMore();
+}
+
+Socket::~Socket() {
+  ShutdownReceive();
+  ShutdownSend();
+  bluetooth_socket_->Close();
+}
+
+void Socket::Disconnect(DisconnectCallback callback) {
+  bluetooth_socket_->Disconnect(std::move(callback));
+}
+
+void Socket::OnReceiveStreamWritable(MojoResult result) {
+  DCHECK(receive_stream_.is_valid());
+  if (result == MOJO_RESULT_OK) {
+    ReceiveMore();
+    return;
+  }
+  ShutdownReceive();
+}
+
+void Socket::ShutdownReceive() {
+  receive_stream_watcher_.Cancel();
+  receive_stream_.reset();
+}
+
+void Socket::ReceiveMore() {
+  DCHECK(receive_stream_.is_valid());
+
+  // The destination to which we will write incoming bytes from
+  // |bluetooth_socket_|. The allocated buffer and its max available size
+  // (assigned to |pending_write_buffer_max_size|) will be fetched by calling
+  // BeginWriteData() below. This already-allocated buffer is a buffer shared
+  // between the 2 sides of |receive_stream_|.
+  void* pending_write_buffer = nullptr;
+
+  // Passing 0 as the initial value allows |pending_write_buffer_max_size| to be
+  // assigned the buffer's max size.
+  uint32_t pending_write_buffer_max_size = 0;
+
+  MojoResult result = receive_stream_->BeginWriteData(
+      &pending_write_buffer, &pending_write_buffer_max_size,
+      MOJO_WRITE_DATA_FLAG_NONE);
+  if (result == MOJO_RESULT_SHOULD_WAIT) {
+    receive_stream_watcher_.ArmOrNotify();
+    return;
+  } else if (result != MOJO_RESULT_OK) {
+    ShutdownReceive();
+    return;
+  }
+
+  bluetooth_socket_->Receive(
+      pending_write_buffer_max_size,
+      base::BindOnce(&Socket::OnBluetoothSocketReceive,
+                     weak_ptr_factory_.GetWeakPtr(), pending_write_buffer),
+      base::BindOnce(&Socket::OnBluetoothSocketReceiveError,
+                     weak_ptr_factory_.GetWeakPtr()));
+}
+
+void Socket::OnBluetoothSocketReceive(void* pending_write_buffer,
+                                      int num_bytes_received,
+                                      scoped_refptr<net::IOBuffer> io_buffer) {
+  DCHECK_GT(num_bytes_received, 0);
+  DCHECK(io_buffer->data());
+
+  if (!receive_stream_.is_valid())
+    return;
+
+  memcpy(pending_write_buffer, io_buffer->data(), num_bytes_received);
+  receive_stream_->EndWriteData(static_cast<uint32_t>(num_bytes_received));
+
+  ReceiveMore();
+}
+
+void Socket::OnBluetoothSocketReceiveError(
+    device::BluetoothSocket::ErrorReason error_reason,
+    const std::string& error_message) {
+  DLOG(ERROR) << "Failed to receive data for reason '" << error_reason << "': '"
+              << error_message << "'";
+  if (receive_stream_.is_valid()) {
+    receive_stream_->EndWriteData(0);
+    ShutdownReceive();
+  }
+}
+
+void Socket::OnSendStreamReadable(MojoResult result) {
+  DCHECK(send_stream_.is_valid());
+  if (result == MOJO_RESULT_OK)
+    SendMore();
+  else
+    ShutdownSend();
+}
+
+void Socket::ShutdownSend() {
+  send_stream_watcher_.Cancel();
+  send_stream_.reset();
+}
+
+void Socket::SendMore() {
+  DCHECK(send_stream_.is_valid());
+
+  // The source from which we will write outgoing bytes to
+  // |bluetooth_socket_|. The allocated buffer and the number of bytes already
+  // written by the other side of |send_stream_| (assigned to
+  // |pending_read_buffer_size|) will be fetched by calling BeginReadData()
+  // below. This already-allocated buffer is a buffer shared between the 2 sides
+  // of |send_stream_|.
+  const void* pending_read_buffer = nullptr;
+
+  // Passing 0 as the initial value allows |pending_read_buffer_size| to be
+  // assigned the number of bytes that the other side of |send_stream_| has
+  // already written.
+  uint32_t pending_read_buffer_size = 0;
+
+  MojoResult result = send_stream_->BeginReadData(&pending_read_buffer,
+                                                  &pending_read_buffer_size,
+                                                  MOJO_WRITE_DATA_FLAG_NONE);
+  if (result == MOJO_RESULT_SHOULD_WAIT) {
+    send_stream_watcher_.ArmOrNotify();
+    return;
+  } else if (result != MOJO_RESULT_OK) {
+    ShutdownSend();
+    return;
+  }
+
+  bluetooth_socket_->Send(base::MakeRefCounted<net::WrappedIOBuffer>(
+                              static_cast<const char*>(pending_read_buffer)),
+                          pending_read_buffer_size,
+                          base::BindOnce(&Socket::OnBluetoothSocketSend,
+                                         weak_ptr_factory_.GetWeakPtr()),
+                          base::BindOnce(&Socket::OnBluetoothSocketSendError,
+                                         weak_ptr_factory_.GetWeakPtr()));
+}
+
+void Socket::OnBluetoothSocketSend(int num_bytes_sent) {
+  DCHECK_GE(num_bytes_sent, 0);
+
+  if (!send_stream_.is_valid())
+    return;
+
+  send_stream_->EndReadData(static_cast<uint32_t>(num_bytes_sent));
+  SendMore();
+}
+
+void Socket::OnBluetoothSocketSendError(const std::string& error_message) {
+  DLOG(ERROR) << "Failed to send data: '" << error_message << "'";
+  if (send_stream_.is_valid()) {
+    send_stream_->EndReadData(0);
+    ShutdownSend();
+  }
+}
+
+}  // namespace bluetooth

device/bluetooth/socket.h

+// Copyright 2020 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.
+
+#ifndef DEVICE_BLUETOOTH_SOCKET_H_
+#define DEVICE_BLUETOOTH_SOCKET_H_
+
+#include <memory>
+#include <string>
+
+#include "base/memory/ref_counted.h"
+#include "base/memory/weak_ptr.h"
+#include "device/bluetooth/bluetooth_socket.h"
+#include "device/bluetooth/public/mojom/adapter.mojom.h"
+#include "mojo/public/cpp/system/data_pipe.h"
+#include "mojo/public/cpp/system/simple_watcher.h"
+
+namespace net {
+class IOBuffer;
+}  // namespace net
+
+namespace bluetooth {
+
+// Implementation of Mojo Socket in
+// device/bluetooth/public/mojom/adapter.mojom.
+// It handles requests to interact with a Socket.
+// Uses the platform abstraction of //device/bluetooth.
+// An instance of this class is constructed by Adapter and strongly bound to its
+// MessagePipe. When the instance is destroyed, the underlying BluetoothSocket
+// is destroyed.
+class Socket : public mojom::Socket {
+ public:
+  Socket(scoped_refptr<device::BluetoothSocket> bluetooth_socket,
+         mojo::ScopedDataPipeProducerHandle receive_stream,
+         mojo::ScopedDataPipeConsumerHandle send_stream);
+  ~Socket() override;
+  Socket(const Socket&) = delete;
+  Socket& operator=(const Socket&) = delete;
+
+  // mojom::Socket:
+  void Disconnect(DisconnectCallback callback) override;
+
+ private:
+  // "Receiving" in this context means receiving data from |bluetooth_socket_|
+  // via BluetoothSocket::Receive() and *writing* it to |receive_stream_|.
+  void OnReceiveStreamWritable(MojoResult result);
+  void ShutdownReceive();
+  void ReceiveMore();
+  void OnBluetoothSocketReceive(void* pending_write_buffer,
+                                int num_bytes_received,
+                                scoped_refptr<net::IOBuffer> io_buffer);
+  void OnBluetoothSocketReceiveError(
+      device::BluetoothSocket::ErrorReason error_reason,
+      const std::string& error_message);
+
+  // "Sending" in this context means *reading* data from |send_stream_| and
+  // sending it over the |bluetooth_socket_| via BluetoothSocket::Send().
+  void OnSendStreamReadable(MojoResult result);
+  void ShutdownSend();
+  void SendMore();
+  void OnBluetoothSocketSend(int num_bytes_sent);
+  void OnBluetoothSocketSendError(const std::string& error_message);
+
+  scoped_refptr<device::BluetoothSocket> bluetooth_socket_;
+
+  mojo::ScopedDataPipeProducerHandle receive_stream_;
+  mojo::ScopedDataPipeConsumerHandle send_stream_;
+
+  mojo::SimpleWatcher receive_stream_watcher_;
+  mojo::SimpleWatcher send_stream_watcher_;
+
+  base::WeakPtrFactory<Socket> weak_ptr_factory_{this};
+};
+
+}  // namespace bluetooth
+
+#endif  // DEVICE_BLUETOOTH_SOCKET_H_

device/bluetooth/socket_unittest.cc

+// Copyright 2020 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 "device/bluetooth/socket.h"
+
+#include <tuple>
+#include <vector>
+
+#include "base/callback_helpers.h"
+#include "base/location.h"
+#include "base/optional.h"
+#include "base/run_loop.h"
+#include "base/test/gmock_callback_support.h"
+#include "base/test/task_environment.h"
+#include "device/bluetooth/bluetooth_socket.h"
+#include "mojo/public/cpp/system/simple_watcher.h"
+#include "net/base/io_buffer.h"
+#include "testing/gmock/include/gmock/gmock.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace bluetooth {
+
+namespace {
+
+class FakeBluetoothSocket : public device::BluetoothSocket {
+ public:
+  using ReceiveArgs = std::
+      tuple<int, ReceiveCompletionCallback, ReceiveErrorCompletionCallback>;
+  using SendArgs = std::tuple<scoped_refptr<net::IOBuffer>,
+                              int,
+                              SendCompletionCallback,
+                              ErrorCompletionCallback>;
+
+  FakeBluetoothSocket() = default;
+  FakeBluetoothSocket(const FakeBluetoothSocket&) = delete;
+  FakeBluetoothSocket& operator=(const FakeBluetoothSocket&) = delete;
+
+  // device::BluetoothSocket:
+  void Close() override { called_close_ = true; }
+  void Disconnect(base::OnceClosure success_callback) override {
+    called_disconnect_ = true;
+    std::move(success_callback).Run();
+  }
+  void Receive(int buffer_size,
+               ReceiveCompletionCallback success_callback,
+               ReceiveErrorCompletionCallback error_callback) override {
+    EXPECT_FALSE(receive_args_);
+    receive_args_ = std::make_unique<ReceiveArgs>(
+        buffer_size, std::move(success_callback), std::move(error_callback));
+  }
+  void Send(scoped_refptr<net::IOBuffer> buffer,
+            int buffer_size,
+            SendCompletionCallback success_callback,
+            ErrorCompletionCallback error_callback) override {
+    EXPECT_FALSE(send_args_);
+    send_args_ = std::make_unique<SendArgs>(std::move(buffer), buffer_size,
+                                            std::move(success_callback),
+                                            std::move(error_callback));
+  }
+  void Accept(AcceptCompletionCallback success_callback,
+              ErrorCompletionCallback error_callback) override {
+    // TODO(hansberry): To be implemented in a subsequent CL.
+    NOTIMPLEMENTED();
+  }
+
+  bool called_close() { return called_close_; }
+  bool called_disconnect() { return called_disconnect_; }
+
+  bool HasReceiveArgs() { return receive_args_.get(); }
+
+  bool HasSendArgs() { return send_args_.get(); }
+
+  std::unique_ptr<ReceiveArgs> TakeReceiveArgs() {
+    return std::move(receive_args_);
+  }
+
+  std::unique_ptr<SendArgs> TakeSendArgs() { return std::move(send_args_); }
+
+ protected:
+  ~FakeBluetoothSocket() override = default;
+
+ private:
+  bool called_close_ = false;
+  bool called_disconnect_ = false;
+  std::unique_ptr<ReceiveArgs> receive_args_;
+  std::unique_ptr<SendArgs> send_args_;
+};
+
+}  // namespace
+
+class SocketTest : public testing::Test {
+ public:
+  SocketTest() = default;
+  ~SocketTest() override = default;
+  SocketTest(const SocketTest&) = delete;
+  SocketTest& operator=(const SocketTest&) = delete;
+
+  void SetUp() override {
+    mojo::ScopedDataPipeProducerHandle receive_pipe_producer_handle;
+    mojo::ScopedDataPipeConsumerHandle receive_pipe_consumer_handle;
+    ASSERT_EQ(
+        MOJO_RESULT_OK,
+        mojo::CreateDataPipe(/*options=*/nullptr, &receive_pipe_producer_handle,
+                             &receive_pipe_consumer_handle));
+
+    mojo::ScopedDataPipeProducerHandle send_pipe_producer_handle;
+    mojo::ScopedDataPipeConsumerHandle send_pipe_consumer_handle;
+    ASSERT_EQ(MOJO_RESULT_OK, mojo::CreateDataPipe(/*options=*/nullptr,
+                                                   &send_pipe_producer_handle,
+                                                   &send_pipe_consumer_handle));
+
+    receive_stream_ = std::move(receive_pipe_consumer_handle);
+    send_stream_ = std::move(send_pipe_producer_handle);
+
+    fake_bluetooth_socket_ = base::MakeRefCounted<FakeBluetoothSocket>();
+    socket_ = std::make_unique<Socket>(fake_bluetooth_socket_,
+                                       std::move(receive_pipe_producer_handle),
+                                       std::move(send_pipe_consumer_handle));
+  }
+
+  void VerifyReceiveAndRead(const std::string& message, bool success) {
+    EXPECT_FALSE(receive_stream_->QuerySignalsState().never_readable());
+
+    // Socket should only have 1 outstanding invocation of
+    // BluetoothSocket::Receive().
+    EXPECT_TRUE(fake_bluetooth_socket_->HasReceiveArgs());
+    auto receive_args = fake_bluetooth_socket_->TakeReceiveArgs();
+    EXPECT_FALSE(fake_bluetooth_socket_->HasReceiveArgs());
+
+    int max_buffer_size = std::get<0>(*receive_args);
+    EXPECT_GT(max_buffer_size, 0);
+
+    // Attempting to read from |receive_stream_| before the BluetoothSocket
+    // provides received data should signal a MOJO_RESULT_SHOULD_WAIT result.
+    uint32_t buffer_size = static_cast<uint32_t>(max_buffer_size);
+    EXPECT_EQ(MOJO_RESULT_SHOULD_WAIT,
+              receive_stream_->ReadData(nullptr, &buffer_size,
+                                        MOJO_READ_DATA_FLAG_NONE));
+
+    if (success) {
+      // Emulate a successful response from the remote device on the other side
+      // of the BluetoothSocket. |receive_stream_| should then be ready to be
+      // read from.
+      auto success_callback = std::move(std::get<1>(*receive_args));
+      std::move(success_callback)
+          .Run(
+              /*num_bytes_received=*/message.size(),
+              /*io_buffer=*/base::MakeRefCounted<net::StringIOBuffer>(message));
+
+      std::vector<char> buffer(max_buffer_size);
+      EXPECT_EQ(MOJO_RESULT_OK,
+                receive_stream_->ReadData(buffer.data(), &buffer_size,
+                                          MOJO_READ_DATA_FLAG_NONE));
+      std::string received_string(buffer.data(), buffer_size);
+      EXPECT_EQ(message, received_string);
+    } else {
+      // Emulate an error in the stack. We should not be able to read from
+      // |receive_stream_|.
+      auto error_callback = std::move(std::get<2>(*receive_args));
+      std::move(error_callback)
+          .Run(device::BluetoothSocket::ErrorReason::kSystemError, "Error");
+    }
+
+    // Socket should only invoke BluetoothSocket::Receive() if it received
+    // a success callback from the previous invocation.
+    EXPECT_EQ(success, fake_bluetooth_socket_->HasReceiveArgs());
+
+    // |receive_stream_| should only remain readable if Socket received a
+    // success callback.
+    EXPECT_EQ(success, !receive_stream_->QuerySignalsState().never_readable());
+  }
+
+  void WriteAndVerifySend(const std::string& message, bool success) {
+    EXPECT_FALSE(send_stream_->QuerySignalsState().never_writable());
+
+    // Verify that Socket has not attempted to invoke BluetoothSocket::Send(),
+    // because no bytes have been written over |send_stream_| yet.
+    EXPECT_FALSE(fake_bluetooth_socket_->HasSendArgs());
+
+    uint32_t message_size = message.size();
+    EXPECT_EQ(MOJO_RESULT_OK,
+              send_stream_->WriteData(message.data(), &message_size,
+                                      MOJO_WRITE_DATA_FLAG_NONE));
+    EXPECT_EQ(message.size(), message_size);
+
+    // Allow Socket to be notified that it can now read |send_stream_|.
+    base::RunLoop().RunUntilIdle();
+
+    // Socket should have now attempted to send our |message| to the remote
+    // device on the other side of the BluetoothSocket.
+    EXPECT_TRUE(fake_bluetooth_socket_->HasSendArgs());
+    auto send_args = fake_bluetooth_socket_->TakeSendArgs();
+
+    int buffer_size = std::get<1>(*send_args);
+    EXPECT_EQ(message_size, static_cast<uint32_t>(buffer_size));
+
+    char* buffer = std::get<0>(*send_args)->data();
+    std::string sent_string(buffer, buffer_size);
+    EXPECT_EQ(message, sent_string);
+
+    if (success) {
+      auto success_callback = std::move(std::get<2>(*send_args));
+      std::move(success_callback).Run(/*num_bytes_sent=*/message.size());
+    } else {
+      auto error_callback = std::move(std::get<3>(*send_args));
+      std::move(error_callback).Run(/*error_message=*/"Error");
+    }
+
+    // Never expect an outstanding invocation of BluetoothSocket::Send().
+    EXPECT_FALSE(fake_bluetooth_socket_->HasSendArgs());
+
+    // |send_stream_| should only remain writeable if Socket received a success
+    // callback.
+    EXPECT_EQ(success, !send_stream_->QuerySignalsState().never_writable());
+  }
+
+ protected:
+  scoped_refptr<FakeBluetoothSocket> fake_bluetooth_socket_;
+  mojo::ScopedDataPipeConsumerHandle receive_stream_;
+  mojo::ScopedDataPipeProducerHandle send_stream_;
+  std::unique_ptr<Socket> socket_;
+
+ private:
+  base::test::SingleThreadTaskEnvironment task_environment;
+};
+
+TEST_F(SocketTest, TestOnDestroyCallsClose) {
+  // When destroyed, |socket_| is expected to tear down its BluetoothSocket.
+  socket_.reset();
+  EXPECT_TRUE(fake_bluetooth_socket_->called_close());
+}
+
+TEST_F(SocketTest, TestDisconnect) {
+  socket_->Disconnect(base::DoNothing());
+  EXPECT_TRUE(fake_bluetooth_socket_->called_disconnect());
+}
+
+TEST_F(SocketTest, TestReceive_Success) {
+  VerifyReceiveAndRead("received_message", /*success=*/true);
+}
+
+TEST_F(SocketTest, TestReceive_Error) {
+  VerifyReceiveAndRead("received_message", /*success=*/false);
+}
+
+TEST_F(SocketTest, TestSend_Success) {
+  WriteAndVerifySend("sent_message", /*success=*/true);
+}
+
+TEST_F(SocketTest, TestSend_Error) {
+  WriteAndVerifySend("sent_message", /*success=*/false);
+}
+
+TEST_F(SocketTest, TestReceiveAndSendMultiple) {
+  VerifyReceiveAndRead("message_1", /*success=*/true);
+  VerifyReceiveAndRead("message_2", /*success=*/true);
+  WriteAndVerifySend("message_3", /*success=*/true);
+  WriteAndVerifySend("message_4", /*success=*/true);
+  VerifyReceiveAndRead("message_5", /*success=*/true);
+  WriteAndVerifySend("message_6", /*success=*/true);
+}
+
+}  // namespace bluetooth