internal fix

PiperOrigin-RevId: 666482118
2026-09-15 07:06:11 -04:00 · 2024-08-22 14:08:02 -07:00
parent 52dac14a3c
commit 2c700115ef
7 changed files with 112 additions and 118 deletions
@@ -126,6 +126,7 @@ cc_test(
    deps = [
        ":mediums",
        ":utils",
+        "//connections:core_types",
        "//connections/implementation/flags:connections_flags",
        "//connections/implementation/mediums/ble_v2",
        "//internal/flags:nearby_flags",
@@ -137,7 +138,9 @@ cc_test(
        "//internal/platform/implementation:types",
        "//internal/platform/implementation/g3",  # build_cleaner: keep
        "@com_github_protobuf_matchers//protobuf-matchers",
+        "@com_google_absl//absl/base:core_headers",
        "@com_google_absl//absl/strings",
+        "@com_google_absl//absl/synchronization",
        "@com_google_absl//absl/time",
        "@com_google_googletest//:gtest_main",
    ],
@@ -14,14 +14,19 @@

 #include "connections/implementation/mediums/ble.h"

+#include <atomic>
 #include <string>

-#include "gmock/gmock.h"
-#include "protobuf-matchers/protocol-buffer-matchers.h"
 #include "gtest/gtest.h"
+#include "absl/strings/string_view.h"
+#include "absl/time/time.h"
 #include "connections/implementation/mediums/bluetooth_radio.h"
 #include "internal/platform/ble.h"
+#include "internal/platform/bluetooth_adapter.h"
+#include "internal/platform/byte_array.h"
+#include "internal/platform/cancellation_flag.h"
 #include "internal/platform/count_down_latch.h"
+#include "internal/platform/feature_flags.h"
 #include "internal/platform/logging.h"
 #include "internal/platform/medium_environment.h"

@@ -73,25 +78,27 @@ TEST_P(BleTest, CanStartAcceptingConnectionsAndConnect) {
  ble_a.StartAcceptingConnections(
      service_id,
      [&](BleSocket socket, const std::string&) { accept_latch.CountDown(); });
-  BlePeripheral discovered_peripheral;
+  std::atomic<BlePeripheral> atomic_discovered_peripheral;
  ble_b.StartScanning(
      service_id, fast_advertisement_service_uuid,
      {
          .peripheral_discovered_cb =
-              [&found_latch, &discovered_peripheral](
+              [&found_latch, &atomic_discovered_peripheral](
                  BlePeripheral& peripheral, const std::string& service_id,
                  const ByteArray& advertisement_bytes,
                  bool fast_advertisement) {
-                discovered_peripheral = peripheral;
-                NEARBY_LOG(
-                    INFO,
-                    "Discovered peripheral=%p [impl=%p], fast advertisement=%d",
-                    &peripheral, &peripheral.GetImpl(), fast_advertisement);
+                NEARBY_LOG(INFO,
+                           "Discovered peripheral=%p [impl=%p], fast "
+                           "advertisement=%d.",
+                           &peripheral, &peripheral.GetImpl(),
+                           fast_advertisement);
+                atomic_discovered_peripheral.store(peripheral);
                found_latch.CountDown();
              },
      });

  EXPECT_TRUE(found_latch.Await(kWaitDuration).result());
+  BlePeripheral discovered_peripheral = atomic_discovered_peripheral.load();
  ASSERT_TRUE(discovered_peripheral.IsValid());
  CancellationFlag flag;
  BleSocket socket = ble_b.Connect(discovered_peripheral, service_id, &flag);
@@ -123,25 +130,27 @@ TEST_P(BleTest, CanCancelConnect) {
  ble_a.StartAcceptingConnections(
      service_id,
      [&](BleSocket socket, const std::string&) { accept_latch.CountDown(); });
-  BlePeripheral discovered_peripheral;
+  std::atomic<BlePeripheral> atomic_discovered_peripheral;
  ble_b.StartScanning(
      service_id, fast_advertisement_service_uuid,
      {
          .peripheral_discovered_cb =
-              [&found_latch, &discovered_peripheral](
+              [&found_latch, &atomic_discovered_peripheral](
                  BlePeripheral& peripheral, const std::string& service_id,
                  const ByteArray& advertisement_bytes,
                  bool fast_advertisement) {
-                discovered_peripheral = peripheral;
-                NEARBY_LOG(
-                    INFO,
-                    "Discovered peripheral=%p [impl=%p], fast advertisement=%d",
-                    &peripheral, &peripheral.GetImpl(), fast_advertisement);
+                NEARBY_LOG(INFO,
+                           "Discovered peripheral=%p [impl=%p], fast "
+                           "advertisement = %d.",
+                           &peripheral, &peripheral.GetImpl(),
+                           fast_advertisement);
+                atomic_discovered_peripheral.store(peripheral);
                found_latch.CountDown();
              },
      });

  EXPECT_TRUE(found_latch.Await(kWaitDuration).result());
+  BlePeripheral discovered_peripheral = atomic_discovered_peripheral.load();
  ASSERT_TRUE(discovered_peripheral.IsValid());
  CancellationFlag flag(true);
  BleSocket socket = ble_b.Connect(discovered_peripheral, service_id, &flag);
@@ -247,48 +256,6 @@ TEST_F(BleTest, CanStartDiscovery) {
  env_.Stop();
 }

-TEST_F(BleTest, HandleDupeFindingsFromDiscovery) {
-  env_.Start();
-  BluetoothRadio radio_a;
-  BluetoothRadio radio_b;
-  Ble ble_a{radio_a};
-  Ble ble_b{radio_b};
-  radio_a.Enable();
-  radio_b.Enable();
-  std::string service_id(kServiceID);
-  ByteArray advertisement_bytes{std::string(kAdvertisementString)};
-  std::string fast_advertisement_service_uuid(kFastAdvertisementServiceUuid);
-  // expecting two discoveries from the same peripheral.
-  CountDownLatch discovery_latch(2);
-  // Expecting the peripheral lost will trigger lost_cb.
-  CountDownLatch lost_latch(1);
-
-  ble_b.StartAdvertising(service_id, advertisement_bytes,
-                         fast_advertisement_service_uuid);
-
-  EXPECT_TRUE(ble_a.StartScanning(
-      service_id, fast_advertisement_service_uuid,
-      DiscoveredPeripheralCallback{
-          .peripheral_discovered_cb =
-              [&discovery_latch](
-                  BlePeripheral& peripheral, const std::string& service_id,
-                  const ByteArray& advertisement_bytes,
-                  bool fast_advertisement) { discovery_latch.CountDown(); },
-          .peripheral_lost_cb =
-              [&lost_latch](BlePeripheral& peripheral,
-                            const std::string& service_id) {
-                lost_latch.CountDown();
-              },
-      }));
-  ble_b.StartAdvertising(service_id, advertisement_bytes,
-                         fast_advertisement_service_uuid);
-  EXPECT_TRUE(discovery_latch.Await(kWaitDuration).result());
-  ble_b.StopAdvertising(service_id);
-  EXPECT_TRUE(lost_latch.Await(kWaitDuration).result());
-  EXPECT_TRUE(ble_a.StopScanning(service_id));
-  env_.Stop();
-}
-
 TEST_F(BleTest, CanStartAndStopLegacyAdvertising) {
  env_.Start();
  BluetoothRadio radio_a;
@@ -1112,9 +1112,11 @@ bool BleV2::StartAsyncScanningLocked(absl::string_view service_id,
          .advertisement_found_cb =
              [this](api::ble_v2::BlePeripheral& peripheral,
                     BleAdvertisementData advertisement_data) {
-                RunOnBleThread([this, &peripheral, advertisement_data]() {
+                AssumeHeld(mutex_);
+                BleV2Peripheral proxy(medium_, peripheral);
+                RunOnBleThread([this, proxy = std::move(proxy),
+                                advertisement_data]() {
                  MutexLock lock(&mutex_);
-                  BleV2Peripheral proxy(medium_, peripheral);
                  discovered_peripheral_tracker_.ProcessFoundBleAdvertisement(
                      std::move(proxy), advertisement_data,
                      [this](BleV2Peripheral proxy, int num_slots, int psm,
@@ -14,14 +14,24 @@

 #include "connections/implementation/mediums/ble_v2.h"

+#include <cstdint>
 #include <string>
+#include <utility>

 #include "gtest/gtest.h"
+#include "absl/strings/string_view.h"
+#include "absl/time/time.h"
 #include "connections/implementation/flags/nearby_connections_feature_flags.h"
 #include "connections/implementation/mediums/ble_v2/discovered_peripheral_callback.h"
 #include "connections/implementation/mediums/bluetooth_radio.h"
+#include "connections/power_level.h"
 #include "internal/flags/nearby_flags.h"
+#include "internal/platform/ble_v2.h"
+#include "internal/platform/byte_array.h"
+#include "internal/platform/cancellation_flag.h"
 #include "internal/platform/count_down_latch.h"
+#include "internal/platform/feature_flags.h"
+#include "internal/platform/logging.h"
 #include "internal/platform/medium_environment.h"

 namespace nearby {
@@ -280,14 +280,15 @@ bool BluetoothClassic::StartDiscovery(const std::string& serviceId,
              }
            }};

+    AddDiscoveryCallback(serviceId, std::move(callback));
+
    if (!medium_->StartDiscovery(std::move(medium_callback))) {
      NEARBY_LOGS(INFO) << "Failed to start discovery of BT devices.";
+      RemoveDiscoveryCallback(serviceId);
      return false;
    }
  }

-  AddDiscoveryCallback(serviceId, std::move(callback));
-
  // Mark the fact that we're currently performing a Bluetooth scan.
  scan_info_.valid = true;

@@ -387,55 +388,53 @@ bool BluetoothClassic::StartAcceptingConnections(
  // Start the accept loop on a dedicated thread - this stays alive and
  // listening for new incoming connections until StopAcceptingConnections()
  // is invoked.
-  accept_loops_runner_.Execute(
-      "bt-accept",
-      [callback = std::move(callback), server_socket = std::move(owned_socket),
-       service_id, this]() mutable {
-        while (true) {
-          BluetoothSocket client_socket = server_socket.Accept();
-          if (!client_socket.IsValid()) {
-            NEARBY_LOGS(INFO) << "Failed to accept connection for "
-                              << service_id;
-            server_socket.Close();
-            break;
-          }
-          NEARBY_LOGS(INFO) << "Accepted connection for " << service_id;
-          bool callback_called = false;
-          {
-            MutexLock lock(&mutex_);
-            if (is_multiplex_enabled_) {
-              BluetoothSocket client_socket_bak = client_socket;
-              auto physical_socket_ptr =
-                  std::make_shared<BluetoothSocket>(client_socket_bak);
-              MultiplexSocket* multiplex_socket =
-                  MultiplexSocket::CreateIncomingSocket(
-                      physical_socket_ptr, service_id);
+  accept_loops_runner_.Execute("bt-accept", [callback = std::move(callback),
+                                             server_socket =
+                                                 std::move(owned_socket),
+                                             service_id, this]() mutable {
+    while (true) {
+      BluetoothSocket client_socket = server_socket.Accept();
+      if (!client_socket.IsValid()) {
+        NEARBY_LOGS(INFO) << "Failed to accept connection for " << service_id;
+        server_socket.Close();
+        break;
+      }
+      NEARBY_LOGS(INFO) << "Accepted connection for " << service_id;
+      bool callback_called = false;
+      {
+        MutexLock lock(&mutex_);
+        if (is_multiplex_enabled_) {
+          BluetoothSocket client_socket_bak = client_socket;
+          auto physical_socket_ptr =
+              std::make_shared<BluetoothSocket>(client_socket_bak);
+          MultiplexSocket* multiplex_socket =
+              MultiplexSocket::CreateIncomingSocket(physical_socket_ptr,
+                                                    service_id);

-              if (multiplex_socket != nullptr &&
-                  multiplex_socket->GetVirtualSocket(service_id)) {
-                multiplex_sockets_.emplace(
-                    client_socket.GetRemoteDevice().GetMacAddress(),
-                    multiplex_socket);
-                MultiplexSocket::StopListeningForIncomingConnection(
-                    service_id, Medium::BLUETOOTH);
-                NEARBY_LOGS(INFO) << "Multiplex virtaul socket created for "
-                                  << client_socket.GetRemoteDevice().GetName();
-                if (callback) {
-                  callback(
-                      service_id,
-                      *(down_cast<BluetoothSocket*>(
-                          multiplex_socket->GetVirtualSocket(service_id))));
-                  callback_called = true;
-                }
-              }
+          if (multiplex_socket != nullptr &&
+              multiplex_socket->GetVirtualSocket(service_id)) {
+            multiplex_sockets_.emplace(
+                client_socket.GetRemoteDevice().GetMacAddress(),
+                multiplex_socket);
+            MultiplexSocket::StopListeningForIncomingConnection(
+                service_id, Medium::BLUETOOTH);
+            NEARBY_LOGS(INFO) << "Multiplex virtaul socket created for "
+                              << client_socket.GetRemoteDevice().GetName();
+            if (callback) {
+              callback(service_id,
+                       *(down_cast<BluetoothSocket*>(
+                           multiplex_socket->GetVirtualSocket(service_id))));
+              callback_called = true;
            }
          }
-          if (callback && !callback_called) {
-                NEARBY_LOGS(INFO) << "Call back triggered for physical socket.";
-            callback(service_id, std::move(client_socket));
-          }
        }
-      });
+      }
+      if (callback && !callback_called) {
+        NEARBY_LOGS(INFO) << "Call back triggered for physical socket.";
+        callback(service_id, std::move(client_socket));
+      }
+    }
+  });

  return true;
 }
@@ -510,8 +509,7 @@ BluetoothSocket BluetoothClassic::Connect(BluetoothDevice& bluetooth_device,
          auto* virtual_socket =
              multiplex_socket->EstablishVirtualSocket(service_id);
          // Should not happen.
-          auto* bluetooth_socket =
-              down_cast<BluetoothSocket*>(virtual_socket);
+          auto* bluetooth_socket = down_cast<BluetoothSocket*>(virtual_socket);
          if (bluetooth_socket == nullptr) {
            NEARBY_LOGS(INFO)
                << "Failed to cast to BluetoothSocket for " << service_id
@@ -593,8 +591,7 @@ BluetoothSocket BluetoothClassic::AttemptToConnect(
  if (is_multiplex_enabled_) {
    // New MultiplexSocket but default disabled, should be enabled after
    // negotiated
-    auto physical_socket_ptr =
-        std::make_shared<BluetoothSocket>(socket);
+    auto physical_socket_ptr = std::make_shared<BluetoothSocket>(socket);
    MultiplexSocket* multiplex_socket = MultiplexSocket::CreateOutgoingSocket(
        std::move(physical_socket_ptr), service_id);

@@ -88,7 +88,7 @@ class BluetoothClassicTest : public ::testing::TestWithParam<FeatureFlags> {
 protected:
  using DiscoveryCallback = BluetoothClassicMedium::DiscoveryCallback;

-  BluetoothClassicTest() {
+  void SetUp() override {
    env_.Start();
    radio_a_ = std::make_unique<BluetoothRadio>();
    radio_b_ = std::make_unique<BluetoothRadio>();
@@ -109,7 +109,7 @@ class BluetoothClassicTest : public ::testing::TestWithParam<FeatureFlags> {
    env_.Sync();
  }

-  ~BluetoothClassicTest() override {
+  void TearDown() override {
    env_.Sync(false);
    radio_a_->Disable();
    radio_b_->Disable();
@@ -317,9 +317,9 @@ TEST_P(BluetoothClassicTest, CanCancelBeforeConnect) {
      {
          .device_discovered_cb =
              [&latch, &discovered_device](BluetoothDevice& device) {
-                discovered_device = device;
                NEARBY_LOG(INFO, "Discovered device=%p [impl=%p]", &device,
                           &device.GetImpl());
+                discovered_device = device;
                latch.CountDown();
              },
      }));
@@ -352,7 +352,7 @@ TEST_P(BluetoothClassicTest, CanCancelBeforeConnect) {
    EXPECT_FALSE(socket_for_client.IsValid());

    // Expect an invalid socket from stopping during the first attempt to
-    // connect, because `Connect` returned immediatley when it checked for
+    // connect, because `Connect` returned immediately when it checked for
    // cancellation.
    EXPECT_EQ(1, bt_client.connect_attempts_count(std::string(kServiceId1)));
  }
@@ -474,19 +474,23 @@ TEST_P(BluetoothClassicTest, CanCancelDuringConnect_MultipleEndpoints) {
  // Simulate the flag being cancelled during connection attempt to a different
  // endpoint.
  medium_a_->CancelDuringConnectToService();
+  EXPECT_TRUE(accept_latch.Await(kWaitDuration).result());
+
+  CountDownLatch accept_latch2(1);
  EXPECT_TRUE(bt_server.StartAcceptingConnections(
      std::string(kServiceId2),
      [&](const std::string& service_id, BluetoothSocket socket) {
        socket_for_server2 = std::move(socket);
-        accept_latch.CountDown();
+        accept_latch2.CountDown();
      }));

+  CancellationFlag flag2;
  BluetoothSocket socket_for_client2 =
-      bt_client.Connect(discovered_device, std::string(kServiceId2), &flag);
+      bt_client.Connect(discovered_device, std::string(kServiceId2), &flag2);

  // If FeatureFlag is disabled, Cancelled is false as no-op.
  if (!feature_flags.enable_cancellation_flag) {
-    EXPECT_TRUE(accept_latch.Await(kWaitDuration).result());
+    EXPECT_TRUE(accept_latch2.Await(kWaitDuration).result());
    EXPECT_TRUE(bt_server.StopAcceptingConnections(std::string(kServiceId1)));
    EXPECT_TRUE(bt_server.StopAcceptingConnections(std::string(kServiceId2)));
    EXPECT_TRUE(socket_for_server1.IsValid());
@@ -648,6 +652,7 @@ TEST_F(BluetoothClassicTest, CanStartAcceptingConnections) {
  // This is best effort, because no callbacks are invoked in this scenario.
  SystemClock::Sleep(kWaitDuration);
  EXPECT_TRUE(bt_server.StopAcceptingConnections(std::string(kServiceId1)));
+  EXPECT_TRUE(bt_client.StopDiscovery(std::string(kServiceId1)));
 }

 TEST_F(BluetoothClassicTest, CheckDiscoveryingStatus) {
@@ -217,6 +217,16 @@ std::unique_ptr<api::BluetoothSocket> BluetoothClassicMedium::ConnectToService(
        << service_uuid;
    return {};
  }
+
+  if (cancellation_flag->Cancelled()) {
+    NEARBY_LOGS(ERROR)
+        << "G3 Bluetooth Connect: Has been cancelled after connected: "
+           "service_uuid="
+        << service_uuid;
+    socket->Close();
+    return {};
+  }
+
  NEARBY_LOGS(INFO) << "G3 ConnectToService: connected: socket="
                    << socket.get();
  return socket;