Parse events from message stream.

Adds parsing for standard events, everything except audio switch.

PiperOrigin-RevId: 518425430

fastpair/message_stream/BUILD

@@ -34,6 +34,7 @@ cc_library(
         "//internal/platform:base",
         "//internal/platform:comm",
         "//internal/platform:types",
+        "//internal/platform/implementation:types",
         "@com_google_absl//absl/base:core_headers",
         "@com_google_absl//absl/log:check",
         "@com_google_absl//absl/status",
@@ -129,6 +130,7 @@ cc_test(
         "//internal/platform:logging",
         "//internal/platform:test_util",
         "//internal/platform:types",
+        "//internal/platform/implementation:types",
         "//internal/platform/implementation/g3",  # build_cleaner: keep
         "@com_github_protobuf_matchers//protobuf-matchers",
         "@com_google_absl//absl/status",

fastpair/message_stream/message_stream.cc

@@ -19,6 +19,7 @@
 #include <optional>
 #include <string>
 #include <utility>
+#include <vector>
 
 #include "absl/time/time.h"
 #include "fastpair/message_stream/message.h"
@@ -32,6 +33,13 @@ constexpr uint8_t kSupportsSilenceBit = 0x01;
 // The default active components response.
 constexpr uint8_t kDefaultComponents = 0;
 constexpr int kModelIdSize = 3;
+constexpr int kBleAddressSize = 6;
+// We don't accept more than `kMaxBatteryLevels` battery values from the
+// provider.
+constexpr int kMaxBatteryLevels = 3;
+constexpr int kUnknownBatteryLevel = 0x7F;
+constexpr absl::Duration kActiveComponentsTimeout = absl::Seconds(1);
+constexpr absl::Duration kRingAckTimeout = absl::Seconds(1);
 
 int GetModelIdFromString(absl::string_view s) {
   int model_id = static_cast<uint8_t>(s[0]);
@@ -42,6 +50,17 @@ int GetModelIdFromString(absl::string_view s) {
   return model_id;
 }
 
+MessageStream::BatteryInfo ConvertBatteryInfo(uint8_t battery_value) {
+  // The highest bit represents `is_charging`, the rest the battery level (in
+  // %).
+  bool is_charging = battery_value & 0x80;
+  int level = battery_value & 0x7F;
+  MessageStream::BatteryInfo battery_info{.is_charging = is_charging};
+  if (level != kUnknownBatteryLevel) {
+    battery_info.level = level;
+  }
+  return battery_info;
+}
 }  // namespace
 
 MessageStream::MessageStream(const FastPairDevice& device,
@@ -58,7 +77,7 @@ absl::Status MessageStream::OpenL2cap(absl::string_view ble_address) {
 absl::Status MessageStream::Disconnect() { return medium_.Disconnect(); }
 
 Future<uint8_t> MessageStream::GetActiveComponents() {
-  Future<uint8_t> future;
+  Future<uint8_t> future(kActiveComponentsTimeout);
   absl::Status status = medium_.Send(
       Message{.message_group = MessageGroup::kDeviceInformationEvent,
               .message_code = MessageCode::kActiveComponentRequest});
@@ -89,12 +108,20 @@ absl::Status MessageStream::SendCapabilities(bool companion_app_installed,
               .payload = {capabilites}});
 }
 
+absl::Status MessageStream::SendPlatformType(
+    api::DeviceInfo::OsType platform_type, uint8_t custom_code) {
+  return medium_.Send(
+      Message{.message_group = MessageGroup::kDeviceInformationEvent,
+              .message_code = MessageCode::kPlatformType,
+              .payload = {static_cast<uint8_t>(platform_type), custom_code}});
+}
+
 // 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;
+  Future<bool> future(kRingAckTimeout);
   std::string payload = {components};
   uint8_t minutes = absl::ToInt64Minutes(duration);
   if (minutes != 0) {
@@ -179,6 +206,46 @@ bool MessageStream::HandleDeviceInformationEvent(const Message& message) {
       observer_.OnModelId(model_id);
       return true;
     }
+    case MessageCode::kBleAddressUpdated: {
+      if (message.payload.size() != kBleAddressSize) {
+        NEARBY_LOGS(WARNING)
+            << "BLE address updated event size should be " << kBleAddressSize
+            << " but is " << message.payload.size();
+        break;
+      }
+      observer_.OnBleAddressUpdated(message.payload);
+      return true;
+    }
+    case MessageCode::kBatteryUpdated: {
+      if (message.payload.size() > kMaxBatteryLevels) {
+        NEARBY_LOGS(WARNING)
+            << "Battery level event size should be <= " << kMaxBatteryLevels
+            << " but is " << message.payload.size();
+        break;
+      }
+      std::vector<BatteryInfo> battery_levels(message.payload.size());
+      for (size_t i = 0; i < message.payload.size(); i++) {
+        battery_levels[i] = ConvertBatteryInfo(message.payload[i]);
+      }
+      observer_.OnBatteryUpdated(std::move(battery_levels));
+      return true;
+    }
+    case MessageCode::kRemainingBatteryTime: {
+      int battery_time = 0;
+      if (message.payload.size() == 1) {
+        battery_time = static_cast<uint8_t>(message.payload[0]);
+      } else if (message.payload.size() == 2) {
+        battery_time = 256 * static_cast<uint8_t>(message.payload[0]) +
+                       static_cast<uint8_t>(message.payload[1]);
+      } else {
+        NEARBY_LOGS(WARNING)
+            << "Remaining battery event size should be 1 or 2 bytes but is "
+            << message.payload.size();
+        break;
+      }
+      observer_.OnRemainingBatteryTime(absl::Minutes(battery_time));
+      return true;
+    }
     case MessageCode::kActiveComponentResponse: {
       uint8_t components = message.payload.size() == 1
                                ? message.payload.data()[0]

fastpair/message_stream/message_stream.h

@@ -16,18 +16,24 @@
 #define THIRD_PARTY_NEARBY_FASTPAIR_MESSAGE_STREAM_MESSAGE_STREAM_H_
 
 #include <memory>
+#include <optional>
 #include <utility>
 #include <vector>
 
 #include "absl/base/thread_annotations.h"
 #include "fastpair/message_stream/medium.h"
 #include "fastpair/message_stream/message.h"
+#include "internal/platform/implementation/device_info.h"
 
 namespace nearby {
 namespace fastpair {
 
 class MessageStream : public Medium::Observer {
  public:
+  struct BatteryInfo {
+    bool is_charging;
+    std::optional<int> level;
+  };
   class Observer {
    public:
     virtual ~Observer() = default;
@@ -43,6 +49,15 @@ class MessageStream : public Medium::Observer {
 
     virtual void OnModelId(int model_id) = 0;
 
+    virtual void OnBleAddressUpdated(absl::string_view address) = 0;
+
+    virtual void OnBatteryUpdated(std::vector<BatteryInfo> battery_levels) = 0;
+
+    virtual void OnRemainingBatteryTime(absl::Duration duration) = 0;
+
+    // Note, there are no callbacks for Active Components Request and Active
+    // Components Response. Use `GetActiveComponents()` instead.
+
     virtual bool OnRing(uint8_t components, absl::Duration duration) = 0;
   };
 
@@ -63,6 +78,9 @@ class MessageStream : public Medium::Observer {
   absl::Status SendCapabilities(bool companion_app_installed,
                                 bool supports_silence_mode);
 
+  absl::Status SendPlatformType(api::DeviceInfo::OsType platform_type,
+                                uint8_t custom_code);
+
   // Asks the Provider to ring.
   // Returns true if the Provider replies with an ACK.
   Future<bool> Ring(uint8_t components, absl::Duration duration);

fastpair/message_stream/message_stream_test.cc

@@ -17,6 +17,7 @@
 #include <memory>
 #include <string>
 #include <utility>
+#include <vector>
 
 #include "gmock/gmock.h"
 #include "protobuf-matchers/protocol-buffer-matchers.h"
@@ -32,6 +33,7 @@
 #include "fastpair/message_stream/message.h"
 #include "internal/platform/bluetooth_classic.h"
 #include "internal/platform/count_down_latch.h"
+#include "internal/platform/implementation/device_info.h"
 #include "internal/platform/logging.h"
 #include "internal/platform/medium_environment.h"
 #include "internal/platform/single_thread_executor.h"
@@ -107,6 +109,19 @@ class MessageStreamTest : public testing::Test {
 
     void OnModelId(int model_id) override { model_id_.Set(model_id); }
 
+    void OnBleAddressUpdated(absl::string_view address) override {
+      ble_address_updated_.Set(std::string(address));
+    }
+
+    void OnBatteryUpdated(
+        std::vector<MessageStream::BatteryInfo> battery_levels) override {
+      battery_levels_.Set(battery_levels);
+    }
+
+    void OnRemainingBatteryTime(absl::Duration duration) override {
+      remaining_battery_time_.Set(duration);
+    }
+
     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.
@@ -115,6 +130,9 @@ class MessageStreamTest : public testing::Test {
     Future<absl::Status> connection_result_;
     Future<absl::Status> disconnected_reason_;
     Future<int> model_id_;
+    Future<std::string> ble_address_updated_;
+    Future<std::vector<MessageStream::BatteryInfo>> battery_levels_;
+    Future<absl::Duration> remaining_battery_time_;
     Future<bool> silence_mode_;
     Future<bool> log_buffer_full_;
     struct OnRingData {
@@ -202,6 +220,17 @@ TEST_F(MessageStreamTest, RingNacked) {
   ASSERT_FALSE(result.Get().GetResult());
 }
 
+TEST_F(MessageStreamTest, RingNoResponse) {
+  constexpr uint8_t kComponents = 0x50;
+  MessageStream message_stream = OpenMessageStream();
+
+  Future<bool> result = message_stream.Ring(kComponents, absl::Minutes(10));
+
+  VerifySentMessage(absl::HexStringToBytes("04010002500A"));
+  ASSERT_FALSE(result.Get().ok());
+  EXPECT_EQ(result.Get().exception(), Exception::kTimeout);
+}
+
 TEST_F(MessageStreamTest, GetActiveComponents) {
   MessageStream message_stream = OpenMessageStream();
 
@@ -224,6 +253,16 @@ TEST_F(MessageStreamTest, GetActiveComponentsEmptyResponse) {
   ASSERT_EQ(result.Get().GetResult(), 0);
 }
 
+TEST_F(MessageStreamTest, GetActiveComponentsNoResponse) {
+  MessageStream message_stream = OpenMessageStream();
+
+  Future<uint8_t> result = message_stream.GetActiveComponents();
+
+  VerifySentMessage(absl::HexStringToBytes("0305"));
+  ASSERT_FALSE(result.Get().ok());
+  EXPECT_EQ(result.Get().exception(), Exception::kTimeout);
+}
+
 TEST_F(MessageStreamTest, ReceiveModelId) {
   MessageStream message_stream = OpenMessageStream();
 
@@ -308,6 +347,72 @@ TEST_F(MessageStreamTest, OnRingFailSendsNack) {
   VerifySentMessage(absl::HexStringToBytes("FF0200020401"));
 }
 
+TEST_F(MessageStreamTest, SendPlatformType) {
+  MessageStream message_stream = OpenMessageStream();
+
+  ASSERT_OK(
+      message_stream.SendPlatformType(api::DeviceInfo::OsType::kAndroid, 0x1C));
+  VerifySentMessage(absl::HexStringToBytes("03080002011C"));
+}
+
+TEST_F(MessageStreamTest, ReceiveBleAddressUpdated) {
+  MessageStream message_stream = OpenMessageStream();
+
+  provider_.WriteProviderBytes(absl::HexStringToBytes("03020006AABBCCDDEEFF"));
+
+  ASSERT_TRUE(observer_.ble_address_updated_.Get().ok());
+  EXPECT_EQ(observer_.ble_address_updated_.Get().GetResult(),
+            absl::HexStringToBytes("AABBCCDDEEFF"));
+}
+
+TEST_F(MessageStreamTest, ReceiveBatteryUpdated) {
+  MessageStream message_stream = OpenMessageStream();
+
+  // The values copied from the specification.
+  provider_.WriteProviderBytes(absl::HexStringToBytes("0303000357417F"));
+
+  ASSERT_TRUE(observer_.battery_levels_.Get().ok());
+  ASSERT_EQ(observer_.battery_levels_.Get().result().size(), 3);
+  EXPECT_FALSE(observer_.battery_levels_.Get().result()[0].is_charging);
+  EXPECT_EQ(observer_.battery_levels_.Get().result()[0].level, 87);
+  EXPECT_FALSE(observer_.battery_levels_.Get().result()[1].is_charging);
+  EXPECT_EQ(observer_.battery_levels_.Get().result()[1].level, 65);
+  EXPECT_FALSE(observer_.battery_levels_.Get().result()[2].is_charging);
+  EXPECT_FALSE(observer_.battery_levels_.Get().result()[2].level.has_value());
+}
+
+TEST_F(MessageStreamTest, ReceiveBatteryUpdatedOneBattery) {
+  MessageStream message_stream = OpenMessageStream();
+
+  provider_.WriteProviderBytes(absl::HexStringToBytes("030300019A"));
+
+  ASSERT_TRUE(observer_.battery_levels_.Get().ok());
+  ASSERT_EQ(observer_.battery_levels_.Get().result().size(), 1);
+  EXPECT_TRUE(observer_.battery_levels_.Get().result()[0].is_charging);
+  EXPECT_EQ(observer_.battery_levels_.Get().result()[0].level, 0x1A);
+}
+
+TEST_F(MessageStreamTest, ReceiveRemainingBatteryTime) {
+  MessageStream message_stream = OpenMessageStream();
+
+  // The values copied from the specification
+  provider_.WriteProviderBytes(absl::HexStringToBytes("03040001F0"));
+
+  ASSERT_TRUE(observer_.remaining_battery_time_.Get().ok());
+  EXPECT_EQ(observer_.remaining_battery_time_.Get().GetResult(),
+            absl::Minutes(240));
+}
+
+TEST_F(MessageStreamTest, ReceiveRemainingBatteryTimeHighValue) {
+  MessageStream message_stream = OpenMessageStream();
+
+  provider_.WriteProviderBytes(absl::HexStringToBytes("03040002ABCD"));
+
+  ASSERT_TRUE(observer_.remaining_battery_time_.Get().ok());
+  EXPECT_EQ(observer_.remaining_battery_time_.Get().GetResult(),
+            absl::Minutes(0xABCD));
+}
+
 }  // namespace
 }  // namespace fastpair
 }  // namespace nearby