diff --git a/connections/implementation/mediums/BUILD b/connections/implementation/mediums/BUILD index 9e425460..82a6fd34 100644 --- a/connections/implementation/mediums/BUILD +++ b/connections/implementation/mediums/BUILD @@ -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", ], diff --git a/connections/implementation/mediums/ble_test.cc b/connections/implementation/mediums/ble_test.cc index c054e911..a3856b63 100644 --- a/connections/implementation/mediums/ble_test.cc +++ b/connections/implementation/mediums/ble_test.cc @@ -14,14 +14,19 @@ #include "connections/implementation/mediums/ble.h" +#include #include -#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 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 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; diff --git a/connections/implementation/mediums/ble_v2.cc b/connections/implementation/mediums/ble_v2.cc index 9021fdc2..24472de5 100644 --- a/connections/implementation/mediums/ble_v2.cc +++ b/connections/implementation/mediums/ble_v2.cc @@ -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, diff --git a/connections/implementation/mediums/ble_v2_test.cc b/connections/implementation/mediums/ble_v2_test.cc index 618a1e74..b0832f53 100644 --- a/connections/implementation/mediums/ble_v2_test.cc +++ b/connections/implementation/mediums/ble_v2_test.cc @@ -14,14 +14,24 @@ #include "connections/implementation/mediums/ble_v2.h" +#include #include +#include #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 { diff --git a/connections/implementation/mediums/bluetooth_classic.cc b/connections/implementation/mediums/bluetooth_classic.cc index 388fad75..6eb98ec4 100644 --- a/connections/implementation/mediums/bluetooth_classic.cc +++ b/connections/implementation/mediums/bluetooth_classic.cc @@ -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(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(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( - 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( + 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(virtual_socket); + auto* bluetooth_socket = down_cast(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(socket); + auto physical_socket_ptr = std::make_shared(socket); MultiplexSocket* multiplex_socket = MultiplexSocket::CreateOutgoingSocket( std::move(physical_socket_ptr), service_id); diff --git a/connections/implementation/mediums/bluetooth_classic_test.cc b/connections/implementation/mediums/bluetooth_classic_test.cc index 94ee7272..268865dd 100644 --- a/connections/implementation/mediums/bluetooth_classic_test.cc +++ b/connections/implementation/mediums/bluetooth_classic_test.cc @@ -88,7 +88,7 @@ class BluetoothClassicTest : public ::testing::TestWithParam { protected: using DiscoveryCallback = BluetoothClassicMedium::DiscoveryCallback; - BluetoothClassicTest() { + void SetUp() override { env_.Start(); radio_a_ = std::make_unique(); radio_b_ = std::make_unique(); @@ -109,7 +109,7 @@ class BluetoothClassicTest : public ::testing::TestWithParam { 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) { diff --git a/internal/platform/implementation/g3/bluetooth_classic.cc b/internal/platform/implementation/g3/bluetooth_classic.cc index a4d7b121..e8377304 100644 --- a/internal/platform/implementation/g3/bluetooth_classic.cc +++ b/internal/platform/implementation/g3/bluetooth_classic.cc @@ -217,6 +217,16 @@ std::unique_ptr 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;