Add Message Stream

Define MessageStream class for exchanging messages betweek the seeker and the
provider over rfcomm.

PiperOrigin-RevId: 516986541

fastpair/common/constant.h

@@ -23,6 +23,8 @@ namespace fastpair {
 // Bluetooth Uuid
 constexpr char kServiceId[] = "Fast Pair";
 
+constexpr char kRfcommUuid[] = "df21fe2c-2515-4fdb-8886-f12c4d67927c";
+
 // Key pair
 constexpr int kSharedSecretKeyByteSize = 16;
 constexpr int kPublicKeyByteSize = 64;

fastpair/message_stream/BUILD

@@ -0,0 +1,139 @@
+# Copyright 2023 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+licenses(["notice"])
+
+cc_library(
+    name = "message_stream",
+    srcs = [
+        "medium.cc",
+        "message_stream.cc",
+    ],
+    hdrs = [
+        "medium.h",
+        "message.h",
+        "message_stream.h",
+    ],
+    visibility = [
+        "//:__subpackages__",
+        "//fastpair:__subpackages__",
+    ],
+    deps = [
+        "//fastpair/common",
+        "//internal/platform:base",
+        "//internal/platform:comm",
+        "//internal/platform:types",
+        "@com_google_absl//absl/base:core_headers",
+        "@com_google_absl//absl/log:check",
+        "@com_google_absl//absl/status",
+        "@com_google_absl//absl/strings",
+        "@com_google_absl//absl/time",
+    ],
+)
+
+cc_library(
+    name = "fake_provider",
+    testonly = True,
+    hdrs = [
+        "fake_provider.h",
+    ],
+    visibility = [
+        "//:__subpackages__",
+        "//fastpair:__subpackages__",
+    ],
+    deps = [
+        ":message_stream",
+        "//fastpair/common",
+        "//internal/platform:comm",
+        "//internal/platform:logging",
+        "//internal/platform:test_util",
+        "//internal/platform:types",
+        "@com_google_absl//absl/status",
+        "@com_google_absl//absl/strings",
+        "@com_google_absl//absl/time",
+        "@com_google_googletest//:gtest_for_library_testonly",
+    ],
+)
+
+cc_library(
+    name = "fake_medium_observer",
+    testonly = True,
+    hdrs = [
+        "fake_medium_observer.h",
+    ],
+    visibility = [
+        "//:__subpackages__",
+        "//fastpair:__subpackages__",
+    ],
+    deps = [
+        ":message_stream",
+        "//fastpair/common",
+        "//internal/platform:comm",
+        "//internal/platform:logging",
+        "//internal/platform:test_util",
+        "//internal/platform:types",
+        "@com_google_absl//absl/status",
+        "@com_google_absl//absl/strings",
+        "@com_google_absl//absl/time",
+        "@com_google_googletest//:gtest_for_library_testonly",
+    ],
+)
+
+cc_test(
+    name = "medium_test",
+    size = "small",
+    srcs = [
+        "medium_test.cc",
+    ],
+    deps = [
+        ":fake_medium_observer",
+        ":fake_provider",
+        ":message_stream",
+        "//fastpair/common",
+        "//internal/platform:comm",
+        "//internal/platform:logging",
+        "//internal/platform:test_util",
+        "//internal/platform:types",
+        "//internal/platform/implementation/g3",  # build_cleaner: keep
+        "@com_github_protobuf_matchers//protobuf-matchers",
+        "@com_google_absl//absl/status",
+        "@com_google_absl//absl/strings",
+        "@com_google_absl//absl/time",
+        "@com_google_googletest//:gtest_main",
+    ],
+)
+
+cc_test(
+    name = "message_stream_test",
+    size = "small",
+    srcs = [
+        "message_stream_test.cc",
+    ],
+    deps = [
+        ":fake_medium_observer",
+        ":fake_provider",
+        ":message_stream",
+        "//fastpair/common",
+        "//internal/platform:comm",
+        "//internal/platform:logging",
+        "//internal/platform:test_util",
+        "//internal/platform:types",
+        "//internal/platform/implementation/g3",  # build_cleaner: keep
+        "@com_github_protobuf_matchers//protobuf-matchers",
+        "@com_google_absl//absl/status",
+        "@com_google_absl//absl/strings",
+        "@com_google_absl//absl/time",
+        "@com_google_googletest//:gtest_main",
+    ],
+)

fastpair/message_stream/fake_medium_observer.h

@@ -0,0 +1,90 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef THIRD_PARTY_NEARBY_FASTPAIR_MESSAGE_STREAM_FAKE_MEDIUM_OBSERVER_H_
+#define THIRD_PARTY_NEARBY_FASTPAIR_MESSAGE_STREAM_FAKE_MEDIUM_OBSERVER_H_
+
+#include <deque>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+#include "absl/status/status.h"
+#include "absl/strings/escaping.h"
+#include "absl/time/clock.h"
+#include "absl/time/time.h"
+#include "fastpair/common/constant.h"
+#include "fastpair/common/fast_pair_device.h"
+#include "fastpair/message_stream/medium.h"
+#include "fastpair/message_stream/message.h"
+#include "internal/platform/bluetooth_classic.h"
+#include "internal/platform/count_down_latch.h"
+#include "internal/platform/logging.h"
+#include "internal/platform/medium_environment.h"
+#include "internal/platform/single_thread_executor.h"
+
+namespace nearby {
+namespace fastpair {
+
+class FakeMediumObserver : public Medium::Observer {
+ public:
+  void OnConnectionResult(absl::Status result) override {
+    NEARBY_LOGS(INFO) << "OnConnectionResult " << result;
+    connection_result_.Set(result);
+  }
+
+  void OnDisconnected(absl::Status status) override {
+    NEARBY_LOGS(INFO) << "OnDisconnected " << status;
+    disconnected_reason_.Set(status);
+  }
+
+  void OnReceived(Message message) override {
+    NEARBY_LOGS(INFO) << "OnReceived " << message;
+    MutexLock lock(&messages_mutex_);
+    messages_.push_back(std::move(message));
+  }
+
+  std::vector<Message> GetMessages() {
+    MutexLock lock(&messages_mutex_);
+    return messages_;
+  }
+
+  absl::Status WaitForMessages(int message_count, absl::Duration timeout) {
+    constexpr absl::Duration kStep = absl::Milliseconds(100);
+    while (timeout > absl::ZeroDuration()) {
+      std::vector<Message> messages = GetMessages();
+      if (messages.size() >= message_count) {
+        return absl::OkStatus();
+      }
+      timeout -= kStep;
+      absl::SleepFor(kStep);
+    }
+    return absl::DeadlineExceededError("timeout waiting for messages");
+  }
+
+  Future<absl::Status> connection_result_;
+  Future<absl::Status> disconnected_reason_;
+
+ private:
+  std::vector<Message> messages_;
+  Mutex messages_mutex_;
+};
+
+}  // namespace fastpair
+}  // namespace nearby
+
+#endif  // THIRD_PARTY_NEARBY_FASTPAIR_MESSAGE_STREAM_FAKE_MEDIUM_OBSERVER_H_

fastpair/message_stream/fake_provider.h

@@ -0,0 +1,128 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef THIRD_PARTY_NEARBY_FASTPAIR_MESSAGE_STREAM_FAKE_PROVIDER_H_
+#define THIRD_PARTY_NEARBY_FASTPAIR_MESSAGE_STREAM_FAKE_PROVIDER_H_
+
+#include <deque>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+#include "absl/status/status.h"
+#include "absl/strings/escaping.h"
+#include "absl/time/clock.h"
+#include "absl/time/time.h"
+#include "fastpair/common/constant.h"
+#include "fastpair/common/fast_pair_device.h"
+#include "fastpair/message_stream/message.h"
+#include "internal/platform/bluetooth_classic.h"
+#include "internal/platform/count_down_latch.h"
+#include "internal/platform/logging.h"
+#include "internal/platform/medium_environment.h"
+#include "internal/platform/single_thread_executor.h"
+
+namespace nearby {
+namespace fastpair {
+
+// Fake BT device with the Provider role. Tailored to testing message stream.
+class FakeProvider {
+ public:
+  ~FakeProvider() { Shutdown(); }
+
+  void Shutdown() { provider_thread_.Shutdown(); }
+
+  void DiscoverProvider(BluetoothClassicMedium& seeker_medium) {
+    CountDownLatch found_latch(1);
+    seeker_medium.StartDiscovery(BluetoothClassicMedium::DiscoveryCallback{
+        .device_discovered_cb =
+            [&](BluetoothDevice& device) {
+              NEARBY_LOG(INFO, "Device discovered: %s",
+                         device.GetName().c_str());
+              found_latch.CountDown();
+            },
+    });
+    provider_adapter_.SetScanMode(
+        BluetoothAdapter::ScanMode::kConnectableDiscoverable);
+    ASSERT_TRUE(found_latch.Await().Ok());
+  }
+
+  void EnableProviderRfcomm() {
+    std::string service_name{"service"};
+    std::string uuid(kRfcommUuid);
+    provider_server_socket_ =
+        provider_medium_.ListenForService(service_name, uuid);
+    provider_thread_.Execute(
+        [this]() { provider_socket_ = provider_server_socket_.Accept(); });
+  }
+
+  Future<std::string> ReadProviderBytes(size_t num_bytes) {
+    Future<std::string> result;
+    provider_thread_.Execute([this, result, num_bytes]() mutable {
+      if (!provider_socket_.IsValid()) {
+        result.SetException({Exception::kIo});
+        return;
+      }
+      ExceptionOr<ByteArray> bytes =
+          provider_socket_.GetInputStream().Read(num_bytes);
+      if (bytes.ok()) {
+        result.Set(std::string(bytes.GetResult().AsStringView()));
+      } else {
+        result.SetException(bytes.GetException());
+      }
+    });
+    return result;
+  }
+
+  void WriteProviderBytes(std::string bytes) {
+    CountDownLatch latch(1);
+    provider_thread_.Execute([&, data = ByteArray(bytes)]() {
+      if (provider_socket_.IsValid()) {
+        provider_socket_.GetOutputStream().Write(data);
+      }
+      latch.CountDown();
+    });
+    latch.Await();
+  }
+
+  void DisableProviderRfcomm() {
+    if (provider_server_socket_.IsValid()) {
+      provider_server_socket_.Close();
+    }
+    provider_thread_.Execute([this]() {
+      if (provider_socket_.IsValid()) {
+        provider_socket_.Close();
+      }
+    });
+  }
+
+  std::string GetMacAddress() const {
+    return provider_adapter_.GetMacAddress();
+  }
+
+ private:
+  BluetoothAdapter provider_adapter_;
+  BluetoothClassicMedium provider_medium_{provider_adapter_};
+  BluetoothServerSocket provider_server_socket_;
+  BluetoothSocket provider_socket_;
+  SingleThreadExecutor provider_thread_;
+};
+
+}  // namespace fastpair
+}  // namespace nearby
+
+#endif  // THIRD_PARTY_NEARBY_FASTPAIR_MESSAGE_STREAM_FAKE_PROVIDER_H_

fastpair/message_stream/medium.cc

@@ -0,0 +1,174 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "fastpair/message_stream/medium.h"
+
+#include <string>
+#include <utility>
+
+#include "absl/log/check.h"
+#include "absl/status/status.h"
+#include "absl/strings/escaping.h"
+#include "absl/strings/string_view.h"
+#include "fastpair/common/constant.h"
+#include "fastpair/message_stream/message.h"
+#include "internal/platform/bluetooth_classic.h"
+#include "internal/platform/future.h"
+
+namespace nearby {
+namespace fastpair {
+
+namespace {
+constexpr int kHeaderSize = 4;
+}
+
+absl::Status Medium::OpenRfcomm() {
+  if (!bt_classic_medium_.has_value()) {
+    return absl::FailedPreconditionError("BT classic unsupported");
+  }
+  if (!device_.public_address().has_value()) {
+    return absl::FailedPreconditionError(
+        "Connect open RFCOMM without public BT address");
+  }
+  BluetoothClassicMedium* classic_medium = bt_classic_medium_.value();
+  executor_.Execute("open-rfcomm", [this, classic_medium]() {
+    if (cancellation_flag_.Cancelled()) return;
+    BluetoothDevice device =
+        classic_medium->GetRemoteDevice(device_.public_address().value());
+    if (!device.IsValid()) {
+      observer_.OnConnectionResult(absl::UnavailableError(absl::StrFormat(
+          "Remote BT device %s not found", device_.public_address().value())));
+      return;
+    }
+    SetSocket(classic_medium->ConnectToService(device, kRfcommUuid,
+                                               &cancellation_flag_));
+    BluetoothSocket socket = GetSocket();
+    absl::Status status = socket.IsValid()
+                              ? absl::OkStatus()
+                              : absl::UnavailableError(absl::StrFormat(
+                                    "Failed to open RFCOMM with %s",
+                                    device_.public_address().value()));
+    observer_.OnConnectionResult(status);
+    if (status.ok()) {
+      RunLoop(std::move(socket));
+    }
+  });
+  return absl::OkStatus();
+}
+
+// Opens L2CAP connection with the remote party.
+// Returns an error if a connection attempt could not be made.
+// Otherwise, `OnConnected()` will be called if the connection was successful,
+// or `OnDisconnected()` if connection failed.
+absl::Status Medium::OpenL2cap(absl::string_view ble_address) {
+  return absl::UnimplementedError("L2CAP unimplemented");
+}
+
+absl::Status Medium::Disconnect() {
+  NEARBY_LOGS(INFO) << "Disconnect";
+  cancellation_flag_.Cancel();
+  CloseSocket();
+  return absl::OkStatus();
+}
+
+// Returns OK if the message was queued for delivery. It does not mean the
+// message was delivered to the remote party.
+absl::Status Medium::Send(Message message, bool compute_and_append_mac) {
+  BluetoothSocket socket = GetSocket();
+  if (cancellation_flag_.Cancelled() || !socket.IsValid()) {
+    return absl::FailedPreconditionError("Not connected");
+  }
+  ByteArray byte_array = Serialize(std::move(message), compute_and_append_mac);
+  if (socket.GetOutputStream().Write(byte_array).Ok()) {
+    return absl::OkStatus();
+  } else {
+    cancellation_flag_.Cancel();
+    return absl::DataLossError("Failed to send data to remote");
+  }
+}
+
+void Medium::RunLoop(BluetoothSocket socket) {
+  NEARBY_LOGS(INFO) << "Run loop";
+  InputStream& input = socket.GetInputStream();
+  while (!cancellation_flag_.Cancelled()) {
+    ExceptionOr<ByteArray> header = input.Read(kHeaderSize);
+    if (!header.ok() || header.GetResult().size() != kHeaderSize) {
+      break;
+    }
+    absl::string_view data = header.GetResult().AsStringView();
+    MessageGroup group = static_cast<MessageGroup>(data[0]);
+    MessageCode code = static_cast<MessageCode>(data[1]);
+    int length = static_cast<unsigned int>(data[2]) * 256 +
+                 static_cast<unsigned int>(data[3]);
+    ExceptionOr<ByteArray> payload;
+    if (length > 0) {
+      payload = input.Read(length);
+    } else {
+      payload = ExceptionOr<ByteArray>(ByteArray(""));
+    }
+    if (!payload.ok() || payload.GetResult().size() != length) {
+      break;
+    }
+    observer_.OnReceived(Message{.message_group = group,
+                                 .message_code = code,
+                                 .payload = std::string(payload.GetResult())});
+  }
+  socket.Close();
+  if (!cancellation_flag_.Cancelled()) {
+    observer_.OnDisconnected(absl::DataLossError("Failed to read from remote"));
+  }
+  NEARBY_LOGS(INFO) << "Run loop done";
+}
+
+ByteArray Medium::Serialize(Message message, bool compute_and_append_mac) {
+  DCHECK_EQ(compute_and_append_mac, false);
+  uint16_t payload_size = message.payload.size();
+  int message_size = kHeaderSize + payload_size;
+  ByteArray byte_array = ByteArray(message_size);
+  uint8_t* data = reinterpret_cast<uint8_t*>(byte_array.data());
+  data[0] = static_cast<uint8_t>(message.message_group);
+  data[1] = static_cast<uint8_t>(message.message_code);
+  data[2] = payload_size >> 8;
+  data[3] = payload_size & 0xFF;
+  memcpy(&data[kHeaderSize], message.payload.data(), payload_size);
+  return byte_array;
+}
+
+void Medium::SetSocket(BluetoothSocket socket) {
+  MutexLock lock(&mutex_);
+  if (cancellation_flag_.Cancelled()) {
+    NEARBY_LOGS(INFO) << "Medium already closed. Closing socket";
+    socket.Close();
+  } else {
+    socket_ = std::move(socket);
+  }
+}
+
+BluetoothSocket Medium::GetSocket() {
+  MutexLock lock(&mutex_);
+  // Returns a stable copy of the socket. Both `socket_` and the returned copy
+  // refer to the same platform socket.
+  return socket_;
+}
+
+void Medium::CloseSocket() {
+  MutexLock lock(&mutex_);
+  if (socket_.IsValid()) {
+    socket_.Close();
+    socket_ = BluetoothSocket();
+  }
+}
+
+}  // namespace fastpair
+}  // namespace nearby

fastpair/message_stream/medium.h

@@ -0,0 +1,103 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef THIRD_PARTY_NEARBY_FASTPAIR_MESSAGE_STREAM_MEDIUM_H_
+#define THIRD_PARTY_NEARBY_FASTPAIR_MESSAGE_STREAM_MEDIUM_H_
+
+#include <functional>
+#include <memory>
+#include <optional>
+#include <string>
+#include <utility>
+
+#include "absl/base/thread_annotations.h"
+#include "absl/strings/string_view.h"
+#include "fastpair/common/fast_pair_device.h"
+#include "fastpair/message_stream/message.h"
+#include "internal/platform/bluetooth_classic.h"
+#include "internal/platform/future.h"
+#include "internal/platform/logging.h"
+#include "internal/platform/mutex.h"
+#include "internal/platform/single_thread_executor.h"
+
+namespace nearby {
+namespace fastpair {
+
+class Medium {
+ public:
+  class Observer {
+   public:
+    virtual ~Observer() = default;
+    virtual void OnConnectionResult(absl::Status result) = 0;
+
+    virtual void OnDisconnected(absl::Status reason) = 0;
+
+    virtual void OnReceived(Message message) = 0;
+  };
+  Medium(const FastPairDevice& device,
+         std::optional<BluetoothClassicMedium*> bt_classic, Observer& observer)
+      : device_(device), bt_classic_medium_(bt_classic), observer_(observer) {}
+  Medium(Medium&& other) = default;
+
+  ~Medium() {
+    NEARBY_LOGS(INFO) << "Destructing FP Medium";
+    cancellation_flag_.Cancel();
+    if (socket_.IsValid()) {
+      socket_.Close();
+    }
+    executor_.Shutdown();
+    NEARBY_LOGS(INFO) << "Destructed FP Medium";
+  }
+
+  // Opens RFCOMM connection with the remote party.
+  // Returns an error if a connection attempt could not be made.
+  // Otherwise, `OnConnectionResult()` will be called with the connection
+  // result. When the connection is successful, Medium starts processing
+  // messages coming in from the remote party, and calls `OnReceived()` for each
+  // complete message.
+  absl::Status OpenRfcomm();
+
+  // Opens L2CAP connection with the remote party.
+  // Returns an error if a connection attempt could not be made.
+  // Otherwise, `OnConnectionResult()` will be called with the connection
+  // result. When the connection is successful, Medium starts processing
+  // messages coming in from the remote party, and calls `OnReceived()` for each
+  // complete message.
+  absl::Status OpenL2cap(absl::string_view ble_address);
+
+  absl::Status Disconnect();
+
+  // Returns OK if the message was queued for delivery. It does not mean the
+  // message was delivered to the remote party.
+  absl::Status Send(Message message, bool compute_and_append_mac = false);
+
+ private:
+  void RunLoop(BluetoothSocket socket);
+  ByteArray Serialize(Message message, bool compute_and_append_mac);
+  void SetSocket(BluetoothSocket socket);
+  void CloseSocket();
+  BluetoothSocket GetSocket();
+  const FastPairDevice& device_;
+  std::optional<BluetoothClassicMedium*> bt_classic_medium_;
+  Observer& observer_;
+  BluetoothSocket socket_ ABSL_GUARDED_BY(mutex_);
+  Mutex mutex_;
+  CancellationFlag cancellation_flag_;
+  SingleThreadExecutor executor_;
+};
+
+}  // namespace fastpair
+}  // namespace nearby
+
+#endif  // THIRD_PARTY_NEARBY_FASTPAIR_MESSAGE_STREAM_MEDIUM_H_

fastpair/message_stream/medium_test.cc

@@ -0,0 +1,190 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "fastpair/message_stream/medium.h"
+
+#include <deque>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "gmock/gmock.h"
+#include "protobuf-matchers/protocol-buffer-matchers.h"
+#include "gtest/gtest.h"
+#include "absl/status/status.h"
+#include "absl/strings/escaping.h"
+#include "absl/time/clock.h"
+#include "absl/time/time.h"
+#include "fastpair/common/constant.h"
+#include "fastpair/common/fast_pair_device.h"
+#include "fastpair/message_stream/fake_medium_observer.h"
+#include "fastpair/message_stream/fake_provider.h"
+#include "fastpair/message_stream/message.h"
+#include "internal/platform/bluetooth_classic.h"
+#include "internal/platform/count_down_latch.h"
+#include "internal/platform/logging.h"
+#include "internal/platform/medium_environment.h"
+#include "internal/platform/single_thread_executor.h"
+
+namespace nearby {
+namespace fastpair {
+
+namespace {
+
+using ::testing::status::StatusIs;
+
+class MediumEnvironmentStarter {
+ public:
+  MediumEnvironmentStarter() { MediumEnvironment::Instance().Start(); }
+  ~MediumEnvironmentStarter() { MediumEnvironment::Instance().Stop(); }
+};
+
+class MediumTest : public testing::Test {
+ protected:
+  void SetUp() override { MediumEnvironment::Instance().Start(); }
+  void TearDown() override {
+    provider_.Shutdown();
+    MediumEnvironment::Instance().Stop();
+  }
+  // The medium environment must be initialized (started) before adding
+  // adapters.
+  MediumEnvironmentStarter env_;
+  BluetoothAdapter seeker_adapter_;
+  BluetoothClassicMedium seeker_medium_{seeker_adapter_};
+  FakeProvider provider_;
+  FakeMediumObserver observer_;
+};
+
+TEST_F(MediumTest, ConnectWithNonExistingDeviceFails) {
+  FastPairDevice fp_device("model id", "ble address",
+                           Protocol::kFastPairRetroactivePairing);
+  fp_device.set_public_address("11:22:33:44:55:66");
+  Medium medium =
+      Medium(fp_device, std::optional<BluetoothClassicMedium*>(&seeker_medium_),
+             observer_);
+
+  ASSERT_OK(medium.OpenRfcomm());
+  ASSERT_TRUE(observer_.connection_result_.Get().ok());
+  EXPECT_THAT(observer_.connection_result_.Get().GetResult(),
+              StatusIs(absl::StatusCode::kUnavailable));
+}
+
+TEST_F(MediumTest, Connect) {
+  FastPairDevice fp_device("model id", "ble address",
+                           Protocol::kFastPairRetroactivePairing);
+  fp_device.set_public_address(provider_.GetMacAddress());
+  provider_.DiscoverProvider(seeker_medium_);
+  provider_.EnableProviderRfcomm();
+  Medium medium =
+      Medium(fp_device, std::optional<BluetoothClassicMedium*>(&seeker_medium_),
+             observer_);
+
+  ASSERT_OK(medium.OpenRfcomm());
+  ASSERT_TRUE(observer_.connection_result_.Get().ok());
+  EXPECT_OK(observer_.connection_result_.Get().GetResult());
+}
+
+TEST_F(MediumTest, ProviderDisconnectsCallsOnDisconnectCallback) {
+  FastPairDevice fp_device("model id", "ble address",
+                           Protocol::kFastPairRetroactivePairing);
+  fp_device.set_public_address(provider_.GetMacAddress());
+  provider_.DiscoverProvider(seeker_medium_);
+  provider_.EnableProviderRfcomm();
+  Medium medium =
+      Medium(fp_device, std::optional<BluetoothClassicMedium*>(&seeker_medium_),
+             observer_);
+  ASSERT_OK(medium.OpenRfcomm());
+  ASSERT_TRUE(observer_.connection_result_.Get().ok());
+
+  provider_.DisableProviderRfcomm();
+
+  ASSERT_TRUE(observer_.disconnected_reason_.Get().ok());
+  EXPECT_THAT(observer_.disconnected_reason_.Get().GetResult(),
+              StatusIs(absl::StatusCode::kDataLoss));
+}
+
+TEST_F(MediumTest, DisconnectSendFails) {
+  FastPairDevice fp_device("model id", "ble address",
+                           Protocol::kFastPairRetroactivePairing);
+  fp_device.set_public_address(provider_.GetMacAddress());
+  provider_.DiscoverProvider(seeker_medium_);
+  provider_.EnableProviderRfcomm();
+  Medium medium =
+      Medium(fp_device, std::optional<BluetoothClassicMedium*>(&seeker_medium_),
+             observer_);
+  ASSERT_OK(medium.OpenRfcomm());
+  ASSERT_TRUE(observer_.connection_result_.Get().ok());
+
+  ASSERT_OK(medium.Disconnect());
+
+  // Medium is disconnected, Send should fail
+  EXPECT_THAT(medium.Send(Message{}, false),
+              StatusIs(absl::StatusCode::kFailedPrecondition));
+}
+
+TEST_F(MediumTest, SendMessage) {
+  Message message = {.message_group = MessageGroup::kDeviceInformationEvent,
+                     .message_code = MessageCode::kSessionNonce,
+                     .payload = absl::HexStringToBytes("ABCDEF")};
+  // See the format definition in
+  // https://developers.google.com/nearby/fast-pair/specifications/extensions/messagestreamhttps://developers.google.com/nearby/fast-pair/specifications/extensions/messagestream
+  std::string expected_result = absl::HexStringToBytes("030A0003ABCDEF");
+  FastPairDevice fp_device("model id", "ble address",
+                           Protocol::kFastPairRetroactivePairing);
+  fp_device.set_public_address(provider_.GetMacAddress());
+  provider_.DiscoverProvider(seeker_medium_);
+  provider_.EnableProviderRfcomm();
+  Medium medium =
+      Medium(fp_device, std::optional<BluetoothClassicMedium*>(&seeker_medium_),
+             observer_);
+  ASSERT_OK(medium.OpenRfcomm());
+  ASSERT_TRUE(observer_.connection_result_.Get().ok());
+
+  EXPECT_OK(medium.Send(message, false));
+
+  Future<std::string> result =
+      provider_.ReadProviderBytes(expected_result.size());
+  ASSERT_TRUE(result.Get().ok());
+  EXPECT_EQ(result.Get().GetResult(), expected_result);
+}
+
+TEST_F(MediumTest, ReceiveMessage) {
+  Message expected_message = {
+      .message_group = MessageGroup::kDeviceInformationEvent,
+      .message_code = MessageCode::kSessionNonce,
+      .payload = absl::HexStringToBytes("ABCDEF")};
+  std::string input = absl::HexStringToBytes("030A0003ABCDEF");
+  FastPairDevice fp_device("model id", "ble address",
+                           Protocol::kFastPairRetroactivePairing);
+  fp_device.set_public_address(provider_.GetMacAddress());
+  provider_.DiscoverProvider(seeker_medium_);
+  provider_.EnableProviderRfcomm();
+  Medium medium =
+      Medium(fp_device, std::optional<BluetoothClassicMedium*>(&seeker_medium_),
+             observer_);
+  ASSERT_OK(medium.OpenRfcomm());
+  ASSERT_TRUE(observer_.connection_result_.Get().ok());
+
+  provider_.WriteProviderBytes(input);
+
+  ASSERT_OK(observer_.WaitForMessages(1, absl::Seconds(10)));
+  std::vector<Message> messages = observer_.GetMessages();
+  ASSERT_EQ(messages.size(), 1);
+  EXPECT_EQ(messages[0], expected_message);
+}
+
+}  // namespace
+}  // namespace fastpair
+}  // namespace nearby

fastpair/message_stream/message.h

@@ -0,0 +1,103 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef THIRD_PARTY_NEARBY_FASTPAIR_MESSAGE_STREAM_MESSAGE_H_
+#define THIRD_PARTY_NEARBY_FASTPAIR_MESSAGE_STREAM_MESSAGE_H_
+
+#include <stdint.h>
+
+#include <ostream>
+#include <string>
+
+#include "absl/strings/escaping.h"
+#include "internal/platform/byte_array.h"
+
+namespace nearby {
+namespace fastpair {
+
+enum class MessageGroup {
+  kBluetooth = 1,
+  kCompanionAppEvent = 2,
+  kDeviceInformationEvent = 3,
+  kDeviceActionEvent = 4,
+  kSass = 7,
+  kAcknowledgement = 255
+};
+
+// Note, message code values are not unique, because every message group has
+// their own list of message codes.
+enum class MessageCode {
+  // Message codes for kBluetooth message group
+  kEnableSilenceMode = 1,
+  kDisableSilenceMode = 2,
+  // Message codes for kCompanionAppEvent message group
+  kLogBufferFull = 1,
+  // Message codes for kDeviceInformationEvent message group
+  kModelId = 1,
+  kBleAddressUpdated = 2,
+  kBatteryUpdated = 3,
+  kRemainingBatteryTime = 4,
+  kActiveComponentRequest = 5,
+  kActiveComponentResponse = 6,
+  kCapabilites = 7,
+  kPlatformType = 8,
+  kSessionNonce = 0x0A,
+  // Message codes for kDeviceActionEvent message group
+  kRing = 1,
+  // Message codes for kSass message group
+  kSassGetCapability = 0x10,
+  kSassNotifyCapability = 0x11,
+  kSassSetMultipointState = 0x12,
+  kSassSetSwitchingPreference = 0x20,
+  kSassGetSwitchingPreference = 0x21,
+  kSassNotifySwitchingPreference = 0x22,
+  kSassSwitchActiveSourceCode = 0x30,
+  kSassSwitchBackAudioSource = 0x31,
+  kSassNotifyMultipointSwitchEvent = 0x32,
+  kSassGetConnectionStatus = 0x33,
+  kSassNotifyConnectionStatus = 0x34,
+  kSassNotifySassInitiatedConnection = 0x40,
+  kSassInUseAccountKey = 0x41,
+  kSassSendCustomData = 0x42,
+  kSassSetDropConnectionTarget = 0x43,
+  // Message codes for kAcknowledgement message group
+  kAck = 1,
+  kNack = 2
+};
+
+struct Message {
+  MessageGroup message_group;
+  MessageCode message_code;
+  std::string payload;
+};
+
+inline std::ostream& operator<<(std::ostream& os, const Message& message) {
+  os << "Message{" << static_cast<int>(message.message_group) << ", "
+     << static_cast<int>(message.message_code);
+  if (!message.payload.empty()) {
+    os << ", '" << absl::BytesToHexString(message.payload) << "'";
+  }
+  os << "}";
+
+  return os;
+}
+
+inline bool operator==(const Message& a, const Message& b) {
+  return a.message_group == b.message_group &&
+         a.message_code == b.message_code && a.payload == b.payload;
+}
+}  // namespace fastpair
+}  // namespace nearby
+
+#endif  // THIRD_PARTY_NEARBY_FASTPAIR_MESSAGE_STREAM_MESSAGE_H_

fastpair/message_stream/message_stream.cc

@@ -0,0 +1,277 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "fastpair/message_stream/message_stream.h"
+
+#include <algorithm>
+#include <memory>
+#include <optional>
+#include <string>
+#include <utility>
+
+#include "absl/time/time.h"
+#include "fastpair/message_stream/message.h"
+
+namespace nearby {
+namespace fastpair {
+
+namespace {
+constexpr uint8_t kCompanionInstalledBit = 0x02;
+constexpr uint8_t kSupportsSilenceBit = 0x01;
+// The default active components response.
+constexpr uint8_t kDefaultComponents = 0;
+constexpr int kModelIdSize = 3;
+
+int GetModelIdFromString(absl::string_view s) {
+  int model_id = static_cast<uint8_t>(s[0]);
+  model_id <<= 8;
+  model_id |= static_cast<uint8_t>(s[1]);
+  model_id <<= 8;
+  model_id |= static_cast<uint8_t>(s[2]);
+  return model_id;
+}
+
+}  // namespace
+
+MessageStream::MessageStream(const FastPairDevice& device,
+                             std::optional<BluetoothClassicMedium*> bt_classic,
+                             Observer& observer)
+    : observer_(observer), medium_(device, bt_classic, *this) {}
+
+absl::Status MessageStream::OpenRfcomm() { return medium_.OpenRfcomm(); }
+
+absl::Status MessageStream::OpenL2cap(absl::string_view ble_address) {
+  return medium_.OpenL2cap(ble_address);
+}
+
+absl::Status MessageStream::Disconnect() { return medium_.Disconnect(); }
+
+Future<uint8_t> MessageStream::GetActiveComponents() {
+  Future<uint8_t> future;
+  absl::Status status = medium_.Send(
+      Message{.message_group = MessageGroup::kDeviceInformationEvent,
+              .message_code = MessageCode::kActiveComponentRequest});
+  if (status.ok()) {
+    MutexLock lock(&mutex_);
+    if (get_active_components_request_) {
+      get_active_components_request_->SetException({Exception::kInterrupted});
+    }
+    get_active_components_request_ = std::make_unique<Future<uint8_t>>(future);
+  } else {
+    future.SetException({Exception::kIo});
+  }
+  return future;
+}
+
+absl::Status MessageStream::SendCapabilities(bool companion_app_installed,
+                                             bool supports_silence_mode) {
+  uint8_t capabilites = 0;
+  if (companion_app_installed) {
+    capabilites |= kCompanionInstalledBit;
+  }
+  if (supports_silence_mode) {
+    capabilites |= kSupportsSilenceBit;
+  }
+  return medium_.Send(
+      Message{.message_group = MessageGroup::kDeviceInformationEvent,
+              .message_code = MessageCode::kCapabilites,
+              .payload = {capabilites}});
+}
+
+// Asks the Provider to ring.
+// Returns true if the Provider replies with an ACK.
+Future<bool> MessageStream::Ring(uint8_t components, absl::Duration duration) {
+  constexpr MessageGroup kGroup = MessageGroup::kDeviceActionEvent;
+  constexpr MessageCode kCode = MessageCode::kRing;
+  Future<bool> future;
+  std::string payload = {components};
+  uint8_t minutes = absl::ToInt64Minutes(duration);
+  if (minutes != 0) {
+    payload.append({minutes});
+  }
+  {
+    MutexLock lock(&mutex_);
+    waiting_for_ack_.push_back(MessageWithAck{
+        .message_group = kGroup, .message_code = kCode, .future = future});
+  }
+  absl::Status status = medium_.Send(Message{
+      .message_group = kGroup, .message_code = kCode, .payload = payload});
+  if (!status.ok()) {
+    FinishCall(kGroup, kCode, false);
+  }
+  return future;
+}
+
+void MessageStream::FinishCall(MessageGroup group, MessageCode code,
+                               bool result) {
+  NEARBY_LOGS(INFO) << "Finish call " << static_cast<int>(group) << ", "
+                    << static_cast<int>(code) << " with result: " << result;
+  MutexLock lock(&mutex_);
+  auto it = std::find_if(waiting_for_ack_.begin(), waiting_for_ack_.end(),
+                         [&](const MessageWithAck& item) {
+                           return item.message_group == group &&
+                                  item.message_code == code;
+                         });
+  if (it != waiting_for_ack_.end()) {
+    NEARBY_LOGS(INFO) << "Finishing call with result: " << result;
+    it->future.Set(result);
+    waiting_for_ack_.erase(it);
+  }
+}
+
+// Medium::Observer
+void MessageStream::OnConnectionResult(absl::Status result) {
+  observer_.OnConnectionResult(result);
+}
+void MessageStream::OnDisconnected(absl::Status status) {
+  observer_.OnDisconnected(status);
+}
+
+void MessageStream::OnReceived(Message message) {
+  bool handled = false;
+  NEARBY_LOGS(INFO) << "Received: " << message;
+  switch (message.message_group) {
+    case MessageGroup::kAcknowledgement:
+      handled = HandleAcknowledgement(message);
+      break;
+    case MessageGroup::kBluetooth:
+      handled = HandleBluetooth(message);
+      break;
+    case MessageGroup::kCompanionAppEvent:
+      handled = HandleCompanionAppEvent(message);
+      break;
+    case MessageGroup::kDeviceInformationEvent:
+      handled = HandleDeviceInformationEvent(message);
+      break;
+    case MessageGroup::kDeviceActionEvent:
+      handled = HandleDeviceActionEvent(message);
+      break;
+    case MessageGroup::kSass:
+      break;
+    default:
+      break;
+  }
+  if (!handled) {
+    NEARBY_LOGS(INFO) << "Unrecognized " << message;
+  }
+}
+
+bool MessageStream::HandleDeviceInformationEvent(const Message& message) {
+  switch (message.message_code) {
+    case MessageCode::kModelId: {
+      if (message.payload.size() != kModelIdSize) {
+        NEARBY_LOGS(WARNING) << "Model id event size should be " << kModelIdSize
+                             << " but is " << message.payload.size();
+        break;
+      }
+      int model_id = GetModelIdFromString(message.payload);
+      observer_.OnModelId(model_id);
+      return true;
+    }
+    case MessageCode::kActiveComponentResponse: {
+      uint8_t components = message.payload.size() == 1
+                               ? message.payload.data()[0]
+                               : kDefaultComponents;
+      MutexLock lock(&mutex_);
+      if (get_active_components_request_) {
+        get_active_components_request_->Set(components);
+        get_active_components_request_.reset();
+      }
+      return true;
+    }
+    default:
+      break;
+  }
+  return false;
+}
+
+bool MessageStream::HandleAcknowledgement(const Message& message) {
+  bool result;
+  if (message.message_code == MessageCode::kAck) {
+    result = true;
+  } else if (message.message_code == MessageCode::kNack) {
+    result = false;
+  } else {
+    return false;
+  }
+  // The payload of ACK/NACK message contains the group/code of the original
+  // message.
+  if (message.payload.size() < 2) {
+    NEARBY_LOGS(INFO) << "ACK/NACK too short: " << message;
+    return false;
+  }
+  MessageGroup group = static_cast<MessageGroup>(message.payload[0]);
+  MessageCode code = static_cast<MessageCode>(message.payload[1]);
+  FinishCall(group, code, result);
+  return true;
+}
+
+bool MessageStream::HandleBluetooth(const Message& message) {
+  switch (message.message_code) {
+    case MessageCode::kEnableSilenceMode:
+      observer_.OnEnableSilenceMode(true);
+      return true;
+    case MessageCode::kDisableSilenceMode:
+      observer_.OnEnableSilenceMode(false);
+      return true;
+    default:
+      break;
+  }
+  return false;
+}
+
+bool MessageStream::HandleCompanionAppEvent(const Message& message) {
+  switch (message.message_code) {
+    case MessageCode::kLogBufferFull:
+      observer_.OnLogBufferFull();
+      return true;
+    default:
+      break;
+  }
+  return false;
+}
+
+bool MessageStream::HandleDeviceActionEvent(const Message& message) {
+  switch (message.message_code) {
+    case MessageCode::kRing: {
+      uint8_t components = 0;
+      absl::Duration duration = absl::ZeroDuration();
+      if (!message.payload.empty()) {
+        components = static_cast<uint8_t>(message.payload[0]);
+      }
+      if (message.payload.size() >= 2) {
+        duration = absl::Minutes(static_cast<uint8_t>(message.payload[1]));
+      }
+      bool result = observer_.OnRing(components, duration);
+      SendAcknowledgement(message, result);
+      return true;
+    }
+    default:
+      break;
+  }
+  return false;
+}
+
+void MessageStream::SendAcknowledgement(const Message& message, bool ack) {
+  absl::Status status = medium_.Send(
+      Message{.message_group = MessageGroup::kAcknowledgement,
+              .message_code = ack ? MessageCode::kAck : MessageCode::kNack,
+              .payload = {static_cast<uint8_t>(message.message_group),
+                          static_cast<uint8_t>(message.message_code)}});
+  if (!status.ok()) {
+    NEARBY_LOGS(WARNING) << "Failed to send ACK/NACK " << status;
+  }
+}
+}  // namespace fastpair
+}  // namespace nearby

fastpair/message_stream/message_stream.h

@@ -0,0 +1,104 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef THIRD_PARTY_NEARBY_FASTPAIR_MESSAGE_STREAM_MESSAGE_STREAM_H_
+#define THIRD_PARTY_NEARBY_FASTPAIR_MESSAGE_STREAM_MESSAGE_STREAM_H_
+
+#include <memory>
+#include <utility>
+#include <vector>
+
+#include "absl/base/thread_annotations.h"
+#include "fastpair/message_stream/medium.h"
+#include "fastpair/message_stream/message.h"
+
+namespace nearby {
+namespace fastpair {
+
+class MessageStream : public Medium::Observer {
+ public:
+  class Observer {
+   public:
+    virtual ~Observer() = default;
+    virtual void OnConnectionResult(absl::Status result) = 0;
+
+    virtual void OnDisconnected(absl::Status status) = 0;
+
+    // `void' callbacks don't send acknowledgements to the Provider.
+    // 'bool' callbacks send ACK or NACK depending on the return value.
+    virtual void OnEnableSilenceMode(bool enable) = 0;
+
+    virtual void OnLogBufferFull() = 0;
+
+    virtual void OnModelId(int model_id) = 0;
+
+    virtual bool OnRing(uint8_t components, absl::Duration duration) = 0;
+  };
+
+  MessageStream(const FastPairDevice& device,
+                std::optional<BluetoothClassicMedium*> bt_classic,
+                Observer& observer);
+  MessageStream(MessageStream&& other) = default;
+  absl::Status OpenRfcomm();
+
+  absl::Status OpenL2cap(absl::string_view ble_address);
+
+  absl::Status Disconnect();
+
+  // Asks the Provider for active components.
+  // Returns the active components bitmap.
+  Future<uint8_t> GetActiveComponents();
+
+  absl::Status SendCapabilities(bool companion_app_installed,
+                                bool supports_silence_mode);
+
+  // Asks the Provider to ring.
+  // Returns true if the Provider replies with an ACK.
+  Future<bool> Ring(uint8_t components, absl::Duration duration);
+
+  // Medium::Observer
+  void OnConnectionResult(absl::Status result) override;
+
+  void OnDisconnected(absl::Status status) override;
+
+  void OnReceived(Message message) override;
+
+ private:
+  // Notifies the caller that message {group, code} was handled by the provider
+  // with the `result`.
+  void FinishCall(MessageGroup group, MessageCode code, bool result);
+  bool HandleBluetooth(const Message& message);
+  bool HandleCompanionAppEvent(const Message& message);
+  bool HandleDeviceActionEvent(const Message& message);
+  bool HandleAcknowledgement(const Message& message);
+  bool HandleDeviceInformationEvent(const Message& message);
+  void SendAcknowledgement(const Message& message, bool ack);
+
+  struct MessageWithAck {
+    MessageGroup message_group;
+    MessageCode message_code;
+    Future<bool> future;
+  };
+  // A list of messages waiting for ACK/NACK.
+  std::vector<MessageWithAck> waiting_for_ack_ ABSL_GUARDED_BY(mutex_);
+  std::unique_ptr<Future<uint8_t>> get_active_components_request_
+      ABSL_GUARDED_BY(mutex_);
+  Mutex mutex_;
+  Observer& observer_;
+  Medium medium_;
+};
+
+}  // namespace fastpair
+}  // namespace nearby
+#endif  // THIRD_PARTY_NEARBY_FASTPAIR_MESSAGE_STREAM_MESSAGE_STREAM_H_

fastpair/message_stream/message_stream_test.cc

@@ -0,0 +1,313 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "fastpair/message_stream/message_stream.h"
+
+#include <memory>
+#include <string>
+#include <utility>
+
+#include "gmock/gmock.h"
+#include "protobuf-matchers/protocol-buffer-matchers.h"
+#include "gtest/gtest.h"
+#include "absl/status/status.h"
+#include "absl/strings/escaping.h"
+#include "absl/time/clock.h"
+#include "absl/time/time.h"
+#include "fastpair/common/constant.h"
+#include "fastpair/common/fast_pair_device.h"
+#include "fastpair/message_stream/fake_medium_observer.h"
+#include "fastpair/message_stream/fake_provider.h"
+#include "fastpair/message_stream/message.h"
+#include "internal/platform/bluetooth_classic.h"
+#include "internal/platform/count_down_latch.h"
+#include "internal/platform/logging.h"
+#include "internal/platform/medium_environment.h"
+#include "internal/platform/single_thread_executor.h"
+
+namespace nearby {
+namespace fastpair {
+
+namespace {
+
+using ::testing::status::StatusIs;
+
+class MediumEnvironmentStarter {
+ public:
+  MediumEnvironmentStarter() { MediumEnvironment::Instance().Start(); }
+  ~MediumEnvironmentStarter() { MediumEnvironment::Instance().Stop(); }
+};
+
+class MessageStreamTest : public testing::Test {
+ protected:
+  void SetUp() override {
+    MediumEnvironment::Instance().Start();
+
+    fp_device_.set_public_address(provider_.GetMacAddress());
+    provider_.DiscoverProvider(seeker_medium_);
+    provider_.EnableProviderRfcomm();
+  }
+  void TearDown() override {
+    provider_.Shutdown();
+    MediumEnvironment::Instance().Stop();
+  }
+
+  MessageStream OpenMessageStream() {
+    MessageStream message_stream = MessageStream(
+        fp_device_, std::optional<BluetoothClassicMedium*>(&seeker_medium_),
+        observer_);
+    CHECK(message_stream.OpenRfcomm().ok());
+    CHECK(observer_.connection_result_.Get().ok());
+    CHECK(observer_.connection_result_.Get().GetResult().ok());
+    return message_stream;
+  }
+
+  void VerifySentMessage(absl::string_view bytes) {
+    Future<std::string> result = provider_.ReadProviderBytes(bytes.size());
+    ASSERT_TRUE(result.Get().ok());
+    ASSERT_EQ(result.Get().GetResult(), bytes);
+  }
+  // The medium environment must be initialized (started) before adding
+  // adapters.
+  MediumEnvironmentStarter env_;
+  BluetoothAdapter seeker_adapter_;
+  BluetoothClassicMedium seeker_medium_{seeker_adapter_};
+  FakeProvider provider_;
+  FastPairDevice fp_device_{"model id", "ble address",
+                            Protocol::kFastPairRetroactivePairing};
+
+  class FakeObserver : public MessageStream::Observer {
+   public:
+    void OnConnectionResult(absl::Status result) override {
+      NEARBY_LOGS(INFO) << "OnConnectionResult " << result;
+      connection_result_.Set(result);
+    }
+
+    void OnDisconnected(absl::Status status) override {
+      NEARBY_LOGS(INFO) << "OnDisconnected " << status;
+      disconnected_reason_.Set(status);
+    }
+
+    void OnEnableSilenceMode(bool enable) override {
+      silence_mode_.Set(enable);
+    }
+
+    void OnLogBufferFull() override { log_buffer_full_.Set(true); }
+
+    void OnModelId(int model_id) override { model_id_.Set(model_id); }
+
+    bool OnRing(uint8_t components, absl::Duration duration) override {
+      on_ring_event_.Set({components, duration});
+      // This allows us to test returning ACK/NACK to the seeker.
+      return components != 0xAB;
+    }
+    Future<absl::Status> connection_result_;
+    Future<absl::Status> disconnected_reason_;
+    Future<int> model_id_;
+    Future<bool> silence_mode_;
+    Future<bool> log_buffer_full_;
+    struct OnRingData {
+      uint8_t components;
+      absl::Duration duration;
+    };
+    Future<OnRingData> on_ring_event_;
+  };
+
+  FakeObserver observer_;
+};
+
+TEST_F(MessageStreamTest, ConnectRfcomm) {
+  MessageStream message_stream = MessageStream(
+      fp_device_, std::optional<BluetoothClassicMedium*>(&seeker_medium_),
+      observer_);
+
+  ASSERT_OK(message_stream.OpenRfcomm());
+
+  ASSERT_TRUE(observer_.connection_result_.Get().ok());
+  EXPECT_OK(observer_.connection_result_.Get().GetResult());
+}
+
+TEST_F(MessageStreamTest, Disconnect) {
+  MessageStream message_stream = OpenMessageStream();
+
+  ASSERT_OK(message_stream.Disconnect());
+
+  ASSERT_THAT(message_stream.SendCapabilities(/*companion_app_installed=*/true,
+                                              /*supports_silence_mode=*/false),
+              StatusIs(absl::StatusCode::kFailedPrecondition));
+}
+
+TEST_F(MessageStreamTest, ProviderDisconnectCallsOnDisconnectCallback) {
+  MessageStream message_stream = OpenMessageStream();
+
+  provider_.DisableProviderRfcomm();
+
+  ASSERT_TRUE(observer_.disconnected_reason_.Get().ok());
+  EXPECT_THAT(observer_.disconnected_reason_.Get().GetResult(),
+              StatusIs(absl::StatusCode::kDataLoss));
+}
+
+TEST_F(MessageStreamTest, SendCapabilites) {
+  MessageStream message_stream = OpenMessageStream();
+
+  ASSERT_OK(message_stream.SendCapabilities(/*companion_app_installed=*/true,
+                                            /*supports_silence_mode=*/false));
+  VerifySentMessage(absl::HexStringToBytes("0307000102"));
+
+  ASSERT_OK(message_stream.SendCapabilities(/*companion_app_installed=*/false,
+                                            /*supports_silence_mode=*/true));
+  VerifySentMessage(absl::HexStringToBytes("0307000101"));
+
+  ASSERT_OK(message_stream.SendCapabilities(/*companion_app_installed=*/true,
+                                            /*supports_silence_mode=*/true));
+  VerifySentMessage(absl::HexStringToBytes("0307000103"));
+
+  ASSERT_OK(message_stream.SendCapabilities(/*companion_app_installed=*/false,
+                                            /*supports_silence_mode=*/false));
+  VerifySentMessage(absl::HexStringToBytes("0307000100"));
+}
+
+TEST_F(MessageStreamTest, RingAcked) {
+  constexpr uint8_t kComponents = 0x50;
+  MessageStream message_stream = OpenMessageStream();
+
+  Future<bool> result = message_stream.Ring(kComponents, absl::Minutes(10));
+
+  VerifySentMessage(absl::HexStringToBytes("04010002500A"));
+  provider_.WriteProviderBytes(absl::HexStringToBytes("FF0100020401"));
+  ASSERT_TRUE(result.Get().ok());
+  ASSERT_TRUE(result.Get().GetResult());
+}
+
+TEST_F(MessageStreamTest, RingNacked) {
+  constexpr uint8_t kComponents = 0x50;
+  MessageStream message_stream = OpenMessageStream();
+
+  Future<bool> result = message_stream.Ring(kComponents, absl::Minutes(10));
+
+  VerifySentMessage(absl::HexStringToBytes("04010002500A"));
+  provider_.WriteProviderBytes(absl::HexStringToBytes("FF0200020401"));
+  ASSERT_TRUE(result.Get().ok());
+  ASSERT_FALSE(result.Get().GetResult());
+}
+
+TEST_F(MessageStreamTest, GetActiveComponents) {
+  MessageStream message_stream = OpenMessageStream();
+
+  Future<uint8_t> result = message_stream.GetActiveComponents();
+
+  VerifySentMessage(absl::HexStringToBytes("0305"));
+  provider_.WriteProviderBytes(absl::HexStringToBytes("03060001AB"));
+  ASSERT_TRUE(result.Get().ok());
+  ASSERT_EQ(result.Get().GetResult(), 0xAB);
+}
+
+TEST_F(MessageStreamTest, GetActiveComponentsEmptyResponse) {
+  MessageStream message_stream = OpenMessageStream();
+
+  Future<uint8_t> result = message_stream.GetActiveComponents();
+
+  VerifySentMessage(absl::HexStringToBytes("0305"));
+  provider_.WriteProviderBytes(absl::HexStringToBytes("03060000"));
+  ASSERT_TRUE(result.Get().ok());
+  ASSERT_EQ(result.Get().GetResult(), 0);
+}
+
+TEST_F(MessageStreamTest, ReceiveModelId) {
+  MessageStream message_stream = OpenMessageStream();
+
+  provider_.WriteProviderBytes(absl::HexStringToBytes("03010003ABCDEF"));
+
+  ASSERT_TRUE(observer_.model_id_.Get().ok());
+  ASSERT_EQ(observer_.model_id_.Get().GetResult(), 0xABCDEF);
+}
+
+TEST_F(MessageStreamTest, ReceiveEnableSilenceMode) {
+  MessageStream message_stream = OpenMessageStream();
+
+  provider_.WriteProviderBytes(absl::HexStringToBytes("01010000"));
+
+  ASSERT_TRUE(observer_.silence_mode_.Get().ok());
+  ASSERT_TRUE(observer_.silence_mode_.Get().GetResult());
+}
+
+TEST_F(MessageStreamTest, ReceiveDisableSilenceMode) {
+  MessageStream message_stream = OpenMessageStream();
+
+  provider_.WriteProviderBytes(absl::HexStringToBytes("01020000"));
+
+  ASSERT_TRUE(observer_.silence_mode_.Get().ok());
+  ASSERT_FALSE(observer_.silence_mode_.Get().GetResult());
+}
+
+TEST_F(MessageStreamTest, ReceiveLogBufferFull) {
+  MessageStream message_stream = OpenMessageStream();
+
+  provider_.WriteProviderBytes(absl::HexStringToBytes("02010000"));
+
+  ASSERT_TRUE(observer_.log_buffer_full_.Get().ok());
+  ASSERT_TRUE(observer_.log_buffer_full_.Get().GetResult());
+}
+
+TEST_F(MessageStreamTest, ReceiveOnRingEmpty) {
+  MessageStream message_stream = OpenMessageStream();
+
+  provider_.WriteProviderBytes(absl::HexStringToBytes("04010000"));
+
+  ASSERT_TRUE(observer_.on_ring_event_.Get().ok());
+  ASSERT_EQ(observer_.on_ring_event_.Get().GetResult().components, 0);
+  ASSERT_EQ(observer_.on_ring_event_.Get().GetResult().duration,
+            absl::ZeroDuration());
+  VerifySentMessage(absl::HexStringToBytes("FF0100020401"));
+}
+
+TEST_F(MessageStreamTest, ReceiveOnRingWithComponents) {
+  MessageStream message_stream = OpenMessageStream();
+
+  provider_.WriteProviderBytes(absl::HexStringToBytes("04010001CD"));
+
+  ASSERT_TRUE(observer_.on_ring_event_.Get().ok());
+  ASSERT_EQ(observer_.on_ring_event_.Get().GetResult().components, 0xCD);
+  ASSERT_EQ(observer_.on_ring_event_.Get().GetResult().duration,
+            absl::ZeroDuration());
+  VerifySentMessage(absl::HexStringToBytes("FF0100020401"));
+}
+
+TEST_F(MessageStreamTest, ReceiveOnRingWithTimeout) {
+  MessageStream message_stream = OpenMessageStream();
+
+  provider_.WriteProviderBytes(absl::HexStringToBytes("04010002CDFE"));
+
+  ASSERT_TRUE(observer_.on_ring_event_.Get().ok());
+  ASSERT_EQ(observer_.on_ring_event_.Get().GetResult().components, 0xCD);
+  ASSERT_EQ(observer_.on_ring_event_.Get().GetResult().duration,
+            absl::Minutes(0xFE));
+  VerifySentMessage(absl::HexStringToBytes("FF0100020401"));
+}
+
+TEST_F(MessageStreamTest, OnRingFailSendsNack) {
+  MessageStream message_stream = OpenMessageStream();
+
+  provider_.WriteProviderBytes(absl::HexStringToBytes("04010002ABFE"));
+
+  ASSERT_TRUE(observer_.on_ring_event_.Get().ok());
+  ASSERT_EQ(observer_.on_ring_event_.Get().GetResult().components, 0xAB);
+  ASSERT_EQ(observer_.on_ring_event_.Get().GetResult().duration,
+            absl::Minutes(0xFE));
+  VerifySentMessage(absl::HexStringToBytes("FF0200020401"));
+}
+
+}  // namespace
+}  // namespace fastpair
+}  // namespace nearby

internal/platform/medium_environment.cc

@@ -25,6 +25,7 @@
 
 #include "absl/container/flat_hash_set.h"
 #include "absl/status/status.h"
+#include "absl/strings/escaping.h"
 #include "internal/platform/count_down_latch.h"
 #include "internal/platform/feature_flags.h"
 #include "internal/platform/implementation/ble_v2.h"
@@ -199,9 +200,13 @@ api::BluetoothDevice* MediumEnvironment::FindBluetoothDevice(
   api::BluetoothDevice* device = nullptr;
   CountDownLatch latch(1);
   RunOnMediumEnvironmentThread([this, &device, &latch, &mac_address]() {
+    NEARBY_LOGS(INFO) << " Looking for: "
+                      << absl::BytesToHexString(mac_address);
     for (auto& item : bluetooth_mediums_) {
       auto* adapter = item.second.adapter;
       if (!adapter) continue;
+      NEARBY_LOGS(INFO) << " Adapter: "
+                        << absl::BytesToHexString(adapter->GetMacAddress());
       if (adapter->GetMacAddress() == mac_address) {
         device = bluetooth_adapters_[adapter];
         break;