TySS-Dev/nearby
1
0
Fork 0
mirror of https://github.com/kidfromjupiter/nearby.git synced 2026-09-16 15:36:12 -04:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into release

Browse Source
This commit is contained in:
hai007
2021-02-04 13:53:51 -08:00
parent 0435b4b9ec 80b390ce45
commit f02498f17e
45 changed files with 1228 additions and 531 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cpp/core/internal/base_pcp_handler.cc
+21 -21
View File
@@ -260,7 +260,7 @@ void BasePcpHandler::OnEncryptionSuccessRunnable(
// Fail early, if there is no crypto context.
ProcessPreConnectionInitiationFailure(
endpoint_id, connection_info.channel.get(), {Status::kEndpointIoError},
connection_info.result.get());
connection_info.result.lock().get());
connection_info.result.reset();
return;
}
@@ -287,10 +287,10 @@ void BasePcpHandler::OnEncryptionSuccessRunnable(
connection_info.options, std::move(connection_info.channel),
connection_info.listener);
if (connection_info.result != nullptr) {
if (auto future_status = connection_info.result.lock()) {
NEARBY_LOG(INFO, "Connection established; Finalising future OK");
connection_info.result->Set({Status::kSuccess});
connection_info.result = nullptr;
future_status->Set({Status::kSuccess});
connection_info.result.reset();
}
}
@@ -321,7 +321,7 @@ void BasePcpHandler::OnEncryptionFailureRunnable(
ProcessPreConnectionInitiationFailure(endpoint_id, info.channel.get(),
{Status::kEndpointIoError},
info.result.get());
info.result.lock().get());
info.result.reset();
}
@@ -329,15 +329,15 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client,
const std::string& endpoint_id,
const ConnectionRequestInfo& info,
const ConnectionOptions& options) {
Future<Status> result;
RunOnPcpHandlerThread([this, client, &info, options, endpoint_id, &result]() {
auto result = std::make_shared<Future<Status>>();
RunOnPcpHandlerThread([this, client, &info, options, endpoint_id, result]() {
absl::Time start_time = SystemClock::ElapsedRealtime();
// If we already have a pending connection, then we shouldn't allow any more
// outgoing connections to this endpoint.
if (pending_connections_.count(endpoint_id)) {
NEARBY_LOG(INFO, "Connection already exists: id=%s", endpoint_id.c_str());
result.Set({Status::kAlreadyConnectedToEndpoint});
result->Set({Status::kAlreadyConnectedToEndpoint});
return;
}
@@ -347,7 +347,7 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client,
!CanSendOutgoingConnection(client)) {
NEARBY_LOG(INFO, "Outgoing connection not allowed: id=%s",
endpoint_id.c_str());
result.Set({Status::kOutOfOrderApiCall});
result->Set({Status::kOutOfOrderApiCall});
return;
}
@@ -355,7 +355,7 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client,
if (endpoint == nullptr) {
NEARBY_LOG(INFO, "Discovered endpoint not found: id=%s",
endpoint_id.c_str());
result.Set({Status::kEndpointUnknown});
result->Set({Status::kEndpointUnknown});
return;
}
@@ -377,8 +377,7 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client,
ConnectImplResult connect_impl_result;
for (auto connect_endpoint : discovered_endpoints) {
if (!MediumSupportedByClientOptions(connect_endpoint->medium,
client->GetDiscoveryOptions()))
if (!MediumSupportedByClientOptions(connect_endpoint->medium, options))
continue;
connect_impl_result = ConnectImpl(client, connect_endpoint);
if (connect_impl_result.status.Ok()) {
@@ -391,7 +390,7 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client,
NEARBY_LOG(INFO, "Endpoint channel not available: id=%s",
endpoint_id.c_str());
ProcessPreConnectionInitiationFailure(
endpoint_id, channel.get(), connect_impl_result.status, &result);
endpoint_id, channel.get(), connect_impl_result.status, result.get());
return;
}
@@ -403,12 +402,12 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client,
// endpoint about ourselves.
Exception write_exception = WriteConnectionRequestFrame(
channel.get(), client->GetLocalEndpointId(), info.endpoint_info, nonce,
GetSupportedConnectionMediumsByPriority(client->GetDiscoveryOptions()));
GetSupportedConnectionMediumsByPriority(options));
if (!write_exception.Ok()) {
NEARBY_LOG(INFO, "Failed to send connection request: id=%s",
endpoint_id.c_str());
ProcessPreConnectionInitiationFailure(
endpoint_id, channel.get(), {Status::kEndpointIoError}, &result);
endpoint_id, channel.get(), {Status::kEndpointIoError}, result.get());
return;
}
@@ -431,7 +430,7 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client,
.start_time = start_time,
.listener = info.listener,
.options = options,
.result = MakeSwapper(&result),
.result = result,
.channel = std::move(channel),
})
.first->second.channel.get();
@@ -447,7 +446,7 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client,
endpoint_id.c_str());
auto status =
WaitForResult(absl::StrCat("RequestConnection(", endpoint_id, ")"),
client->GetClientId(), &result);
client->GetClientId(), result.get());
NEARBY_LOG(INFO, "Wait is complete: id=%s; status=%d", endpoint_id.c_str(),
status.value);
return status;
@@ -1024,8 +1023,8 @@ void BasePcpHandler::ProcessTieBreakLoss(
BasePcpHandler::PendingConnectionInfo* info) {
ProcessPreConnectionInitiationFailure(endpoint_id, info->channel.get(),
{Status::kEndpointIoError},
info->result.get());
info->result = nullptr;
info->result.lock().get());
info->result.reset();
ProcessPreConnectionResultFailure(client, endpoint_id);
}
@@ -1269,9 +1268,10 @@ ExceptionOr<OfflineFrame> BasePcpHandler::ReadConnectionRequestFrame(
///////////////////// BasePcpHandler::PendingConnectionInfo ///////////////////
BasePcpHandler::PendingConnectionInfo::~PendingConnectionInfo() {
if (result != nullptr) {
auto future_status = result.lock();
if (future_status && !future_status->IsSet()) {
NEARBY_LOG(INFO, "Future was not set; destroying info");
result->Set({Status::kError});
future_status->Set({Status::kError});
}
if (channel != nullptr) {
cpp/core/internal/base_pcp_handler.h
+1 -31
View File
@@ -54,36 +54,6 @@ namespace location {
namespace nearby {
namespace connections {
// Define a class that supports move operation for pointers using std::swap.
// It replicates std::unique_ptr<> behavior, but it does not own the pointer,
// so it does not attempt destroy it.
// This approach was recommended during code review, as a better alternative to
// reuse of std::unique_ptr<> with custom no-op deleter, for the sake of
// readability.
template <typename T>
class Swapper {
public:
Swapper(T* pointer) : pointer_(pointer) {} // NOLINT.
Swapper(Swapper&& other) { *this = std::move(other); }
Swapper& operator=(Swapper&& other) {
std::swap(pointer_, other.pointer_);
return *this;
}
T* operator->() const { return pointer_; }
T& operator*() { return *pointer_; }
operator T*() { return pointer_; } // NOLINT.
T* get() const { return pointer_; }
void reset() { pointer_ = nullptr; }
private:
T* pointer_ = nullptr;
};
template <typename T>
Swapper<T> MakeSwapper(T* value) {
return Swapper<T>(value);
}
// Represents the WebRtc state that mediums are connectable or not.
enum class WebRtcState {
kUndefined = 0,
@@ -355,7 +325,7 @@ class BasePcpHandler : public PcpHandler,
// Only set for outgoing connections. If set, we must call
// result->Set() when connection is established, or rejected.
Swapper<Future<Status>> result = nullptr;
std::weak_ptr<Future<Status>> result;
// Only (possibly) vector for incoming connections.
std::vector<proto::connections::Medium> supported_mediums;
cpp/core/internal/base_pcp_handler_test.cc
+8 -5
View File
@@ -171,18 +171,21 @@ class MockPcpHandler : public BasePcpHandler {
class MockContext {
public:
explicit MockContext(std::atomic_int* destroyed = nullptr) {
destroyed_ = destroyed;
explicit MockContext(std::atomic_int* destroyed = nullptr)
: destroyed_{destroyed} {}
MockContext(MockContext&& other) { *this = std::move(other); }
MockContext& operator=(MockContext&& other) {
destroyed_ = other.destroyed_;
other.destroyed_ = nullptr;
return *this;
}
MockContext(MockContext&&) = default;
MockContext& operator=(MockContext&&) = default;
~MockContext() {
if (destroyed_) (*destroyed_)++;
}
private:
Swapper<std::atomic_int> destroyed_{nullptr};
std::atomic_int* destroyed_;
};
struct MockDiscoveredEndpoint : public MockPcpHandler::DiscoveredEndpoint {
cpp/core/internal/client_proxy.cc
+6
View File
@@ -19,6 +19,7 @@
#include <utility>
#include "platform/base/base64_utils.h"
#include "platform/base/feature_flags.h"
#include "platform/base/prng.h"
#include "platform/public/crypto.h"
#include "platform/public/logging.h"
@@ -493,6 +494,11 @@ bool ClientProxy::RemoteConnectionIsAccepted(std::string endpoint_id) const {
}
void ClientProxy::AddCancellationFlag(const std::string& endpoint_id) {
// Don't insert the CancellationFlag to the map if feature flag is disabled.
if (!FeatureFlags::GetInstance().GetFlags().enable_cancellation_flag) {
return;
}
auto item = cancellation_flags_.find(endpoint_id);
if (item != cancellation_flags_.end()) {
return;
cpp/core/internal/client_proxy_test.cc
+113 -71
View File
@@ -20,6 +20,8 @@
#include "core/options.h"
#include "core/strategy.h"
#include "platform/base/byte_array.h"
#include "platform/base/feature_flags.h"
#include "platform/base/medium_environment.h"
#include "gmock/gmock.h"
#include "gtest/gtest.h"
#include "absl/container/flat_hash_set.h"
@@ -30,10 +32,20 @@ namespace nearby {
namespace connections {
namespace {
using FeatureFlags = FeatureFlags::Flags;
using ::testing::MockFunction;
using ::testing::StrictMock;
class ClientProxyTest : public testing::Test {
constexpr FeatureFlags kTestCases[] = {
FeatureFlags{
.enable_cancellation_flag = true,
},
FeatureFlags{
.enable_cancellation_flag = false,
},
};
class ClientProxyTest : public ::testing::TestWithParam<FeatureFlags> {
protected:
struct MockDiscoveryListener {
StrictMock<MockFunction<void(const std::string& endpoint_id,
@@ -111,8 +123,6 @@ class ClientProxyTest : public testing::Test {
connection_options_,
discovery_connection_listener_);
EXPECT_TRUE(client->HasPendingConnectionToEndpoint(endpoint.id));
// Cancellation flag has been created and added into map.
EXPECT_FALSE(client->GetCancellationFlag(endpoint.id)->Cancelled());
}
void OnDiscoveryConnectionLocalAccepted(ClientProxy* client,
@@ -175,8 +185,6 @@ class ClientProxyTest : public testing::Test {
const Endpoint& endpoint) {
EXPECT_CALL(mock_discovery_connection_.disconnected_cb, Call).Times(1);
client->OnDisconnected(endpoint.id, true);
// The Cancelled is always true as the default flag being returned.
EXPECT_TRUE(client->GetCancellationFlag(endpoint.id)->Cancelled());
}
void OnPayload(ClientProxy* client, const Endpoint& endpoint) {
@@ -231,6 +239,106 @@ class ClientProxyTest : public testing::Test {
ConnectionOptions connection_options_;
};
TEST_P(ClientProxyTest, CanCancelEndpoint) {
FeatureFlags feature_flags = GetParam();
MediumEnvironment::Instance().SetFeatureFlags(feature_flags);
Endpoint advertising_endpoint =
StartAdvertising(&client1_, advertising_connection_listener_);
StartDiscovery(&client2_, discovery_listener_);
OnDiscoveryEndpointFound(&client2_, advertising_endpoint);
OnDiscoveryConnectionInitiated(&client2_, advertising_endpoint);
EXPECT_FALSE(
client2_.GetCancellationFlag(advertising_endpoint.id)->Cancelled());
client2_.CancelEndpoint(advertising_endpoint.id);
// If FeatureFlag is disabled, Cancelled is false as no-op.
if (!feature_flags.enable_cancellation_flag) {
EXPECT_FALSE(
client2_.GetCancellationFlag(advertising_endpoint.id)->Cancelled());
} else {
// The Cancelled is always true as the default flag being returned.
EXPECT_TRUE(
client2_.GetCancellationFlag(advertising_endpoint.id)->Cancelled());
}
}
TEST_P(ClientProxyTest, CanCancelAllEndpoints) {
FeatureFlags feature_flags = GetParam();
MediumEnvironment::Instance().SetFeatureFlags(feature_flags);
Endpoint advertising_endpoint =
StartAdvertising(&client1_, advertising_connection_listener_);
StartDiscovery(&client2_, discovery_listener_);
OnDiscoveryEndpointFound(&client2_, advertising_endpoint);
OnDiscoveryConnectionInitiated(&client2_, advertising_endpoint);
EXPECT_FALSE(
client2_.GetCancellationFlag(advertising_endpoint.id)->Cancelled());
client2_.CancelAllEndpoints();
// If FeatureFlag is disabled, Cancelled is false as no-op.
if (!feature_flags.enable_cancellation_flag) {
EXPECT_FALSE(
client2_.GetCancellationFlag(advertising_endpoint.id)->Cancelled());
} else {
// The Cancelled is always true as the default flag being returned.
EXPECT_TRUE(
client2_.GetCancellationFlag(advertising_endpoint.id)->Cancelled());
}
}
TEST_P(ClientProxyTest, CanCancelAllEndpointsWithDifferentEndpoint) {
FeatureFlags feature_flags = GetParam();
MediumEnvironment::Instance().SetFeatureFlags(feature_flags);
ConnectionListener advertising_connection_listener_2;
ConnectionListener advertising_connection_listener_3;
ClientProxy client3;
StartDiscovery(&client1_, discovery_listener_);
Endpoint advertising_endpoint_2 =
StartAdvertising(&client2_, advertising_connection_listener_2);
Endpoint advertising_endpoint_3 =
StartAdvertising(&client3, advertising_connection_listener_3);
OnDiscoveryEndpointFound(&client1_, advertising_endpoint_2);
OnDiscoveryConnectionInitiated(&client1_, advertising_endpoint_2);
OnDiscoveryEndpointFound(&client1_, advertising_endpoint_3);
OnDiscoveryConnectionInitiated(&client1_, advertising_endpoint_3);
// The CancellationFlag of endpoint_2 and endpoint_3 have been added. Default
// Cancelled is false.
EXPECT_FALSE(
client1_.GetCancellationFlag(advertising_endpoint_2.id)->Cancelled());
EXPECT_FALSE(
client1_.GetCancellationFlag(advertising_endpoint_3.id)->Cancelled());
client1_.CancelAllEndpoints();
if (!feature_flags.enable_cancellation_flag) {
// The CancellationFlag of endpoint_2 and endpoint_3 will not be removed
// since it is not added. The default flag returned as Cancelled being true,
// but Cancelled requested is false since the FeatureFlag is off.
EXPECT_FALSE(
client1_.GetCancellationFlag(advertising_endpoint_2.id)->Cancelled());
EXPECT_FALSE(
client1_.GetCancellationFlag(advertising_endpoint_3.id)->Cancelled());
} else {
// Expect the CancellationFlag of endpoint_2 and endpoint_3 has been
// removed. The Cancelled is always true as the default flag being returned.
EXPECT_TRUE(
client1_.GetCancellationFlag(advertising_endpoint_2.id)->Cancelled());
EXPECT_TRUE(
client1_.GetCancellationFlag(advertising_endpoint_3.id)->Cancelled());
}
}
INSTANTIATE_TEST_SUITE_P(ParametrisedClientProxyTest, ClientProxyTest,
::testing::ValuesIn(kTestCases));
TEST_F(ClientProxyTest, ConstructorDestructorWorks) { SUCCEED(); }
TEST_F(ClientProxyTest, ClientIdIsUnique) {
@@ -371,72 +479,6 @@ TEST_F(ClientProxyTest, OnPayloadProgressChangesState) {
OnPayloadProgress(&client2_, advertising_endpoint);
}
TEST_F(ClientProxyTest, CanCancelEndpoint) {
Endpoint advertising_endpoint =
StartAdvertising(&client1_, advertising_connection_listener_);
StartDiscovery(&client2_, discovery_listener_);
OnDiscoveryEndpointFound(&client2_, advertising_endpoint);
OnDiscoveryConnectionInitiated(&client2_, advertising_endpoint);
EXPECT_FALSE(
client2_.GetCancellationFlag(advertising_endpoint.id)->Cancelled());
client2_.CancelEndpoint(advertising_endpoint.id);
// The Cancelled is always true as the default flag being returned.
EXPECT_TRUE(
client2_.GetCancellationFlag(advertising_endpoint.id)->Cancelled());
}
TEST_F(ClientProxyTest, CanCancelAllEndpoints) {
Endpoint advertising_endpoint =
StartAdvertising(&client1_, advertising_connection_listener_);
StartDiscovery(&client2_, discovery_listener_);
OnDiscoveryEndpointFound(&client2_, advertising_endpoint);
OnDiscoveryConnectionInitiated(&client2_, advertising_endpoint);
EXPECT_FALSE(
client2_.GetCancellationFlag(advertising_endpoint.id)->Cancelled());
client2_.CancelAllEndpoints();
// The Cancelled is always true as the default flag being returned.
EXPECT_TRUE(
client2_.GetCancellationFlag(advertising_endpoint.id)->Cancelled());
}
TEST_F(ClientProxyTest, CanCancelAllEndpointsWithDifferentEndpoint) {
ConnectionListener advertising_connection_listener_2;
ConnectionListener advertising_connection_listener_3;
ClientProxy client3;
StartDiscovery(&client1_, discovery_listener_);
Endpoint advertising_endpoint_2 =
StartAdvertising(&client2_, advertising_connection_listener_2);
Endpoint advertising_endpoint_3 =
StartAdvertising(&client3, advertising_connection_listener_3);
OnDiscoveryEndpointFound(&client1_, advertising_endpoint_2);
OnDiscoveryConnectionInitiated(&client1_, advertising_endpoint_2);
OnDiscoveryEndpointFound(&client1_, advertising_endpoint_3);
OnDiscoveryConnectionInitiated(&client1_, advertising_endpoint_3);
// The CancellationFlag of endpoint_2 and endpoint_3 have been added. Default
// Cancelled is false.
EXPECT_FALSE(
client1_.GetCancellationFlag(advertising_endpoint_2.id)->Cancelled());
EXPECT_FALSE(
client1_.GetCancellationFlag(advertising_endpoint_3.id)->Cancelled());
client1_.CancelAllEndpoints();
// Expect the CancellationFlag of endpoint_2 and endpoint_3 has been removed.
// The Cancelled is always true as the default flag being returned.
EXPECT_TRUE(
client1_.GetCancellationFlag(advertising_endpoint_2.id)->Cancelled());
EXPECT_TRUE(
client1_.GetCancellationFlag(advertising_endpoint_3.id)->Cancelled());
}
} // namespace
} // namespace connections
} // namespace nearby
cpp/core/internal/mediums/ble.cc
+6 -1
View File
@@ -335,7 +335,12 @@ BleSocket Ble::Connect(BlePeripheral& peripheral, const std::string& service_id,
return socket;
}
socket = medium_.Connect(peripheral, service_id);
if (cancellation_flag->Cancelled()) {
NEARBY_LOGS(INFO) << "Can't create client BLE socket due to cancel.";
return socket;
}
socket = medium_.Connect(peripheral, service_id, cancellation_flag);
if (!socket.IsValid()) {
NEARBY_LOGS(INFO) << "Failed to Connect via BLE [service=" << service_id
<< "]";
cpp/core/internal/mediums/ble_test.cc
+129 -53
View File
@@ -29,12 +29,23 @@ namespace nearby {
namespace connections {
namespace {
using FeatureFlags = FeatureFlags::Flags;
constexpr FeatureFlags kTestCases[] = {
FeatureFlags{
.enable_cancellation_flag = true,
},
FeatureFlags{
.enable_cancellation_flag = false,
},
};
constexpr absl::Duration kWaitDuration = absl::Milliseconds(1000);
constexpr absl::string_view kServiceID{"com.google.location.nearby.apps.test"};
constexpr absl::string_view kAdvertisementString{"\x0a\x0b\x0c\x0d"};
constexpr absl::string_view kFastAdvertisementServiceUuid{"\xf3\xfe"};
class BleTest : public ::testing::Test {
class BleTest : public ::testing::TestWithParam<FeatureFlags> {
protected:
using DiscoveredPeripheralCallback = BleMedium::DiscoveredPeripheralCallback;
@@ -43,6 +54,123 @@ class BleTest : public ::testing::Test {
MediumEnvironment& env_{MediumEnvironment::Instance()};
};
TEST_P(BleTest, CanStartAcceptingConnectionsAndConnect) {
FeatureFlags feature_flags = GetParam();
env_.SetFeatureFlags(feature_flags);
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);
CountDownLatch found_latch(1);
CountDownLatch accept_latch(1);
ble_a.StartAdvertising(service_id, advertisement_bytes,
fast_advertisement_service_uuid);
ble_a.StartAcceptingConnections(
service_id,
{
.accepted_cb = [&accept_latch](
BleSocket socket,
const std::string&) { accept_latch.CountDown(); },
});
BlePeripheral discovered_peripheral;
ble_b.StartScanning(
service_id, fast_advertisement_service_uuid,
{
.peripheral_discovered_cb =
[&found_latch, &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);
found_latch.CountDown();
},
});
EXPECT_TRUE(found_latch.Await(kWaitDuration).result());
ASSERT_TRUE(discovered_peripheral.IsValid());
CancellationFlag flag;
BleSocket socket = ble_b.Connect(discovered_peripheral, service_id, &flag);
EXPECT_TRUE(accept_latch.Await(kWaitDuration).result());
EXPECT_TRUE(socket.IsValid());
ble_b.StopScanning(service_id);
ble_a.StopAdvertising(service_id);
env_.Stop();
}
TEST_P(BleTest, CanCancelConnect) {
FeatureFlags feature_flags = GetParam();
env_.SetFeatureFlags(feature_flags);
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);
CountDownLatch found_latch(1);
CountDownLatch accept_latch(1);
ble_a.StartAdvertising(service_id, advertisement_bytes,
fast_advertisement_service_uuid);
ble_a.StartAcceptingConnections(
service_id,
{
.accepted_cb = [&accept_latch](
BleSocket socket,
const std::string&) { accept_latch.CountDown(); },
});
BlePeripheral discovered_peripheral;
ble_b.StartScanning(
service_id, fast_advertisement_service_uuid,
{
.peripheral_discovered_cb =
[&found_latch, &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);
found_latch.CountDown();
},
});
EXPECT_TRUE(found_latch.Await(kWaitDuration).result());
ASSERT_TRUE(discovered_peripheral.IsValid());
CancellationFlag flag(true);
BleSocket socket = ble_b.Connect(discovered_peripheral, service_id, &flag);
// 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(socket.IsValid());
} else {
EXPECT_FALSE(accept_latch.Await(kWaitDuration).result());
EXPECT_FALSE(socket.IsValid());
}
ble_b.StopScanning(service_id);
ble_a.StopAdvertising(service_id);
env_.Stop();
}
INSTANTIATE_TEST_SUITE_P(ParametrisedBleTest, BleTest,
::testing::ValuesIn(kTestCases));
TEST_F(BleTest, CanConstructValidObject) {
env_.Start();
BluetoothRadio radio_a;
@@ -129,58 +257,6 @@ TEST_F(BleTest, CanStartDiscovery) {
env_.Stop();
}
TEST_F(BleTest, CanStartAcceptingConnectionsAndConnect) {
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);
CountDownLatch found_latch(1);
CountDownLatch accept_latch(1);
ble_a.StartAdvertising(service_id, advertisement_bytes,
fast_advertisement_service_uuid);
ble_a.StartAcceptingConnections(
service_id,
{
.accepted_cb = [&accept_latch](
BleSocket socket,
const std::string&) { accept_latch.CountDown(); },
});
BlePeripheral discovered_peripheral;
ble_b.StartScanning(
service_id, fast_advertisement_service_uuid,
{
.peripheral_discovered_cb =
[&found_latch, &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);
found_latch.CountDown();
},
});
EXPECT_TRUE(found_latch.Await(kWaitDuration).result());
ASSERT_TRUE(discovered_peripheral.IsValid());
CancellationFlag flag;
BleSocket socket = ble_b.Connect(discovered_peripheral, service_id, &flag);
EXPECT_TRUE(accept_latch.Await(kWaitDuration).result());
EXPECT_TRUE(socket.IsValid());
ble_b.StopScanning(service_id);
ble_a.StopAdvertising(service_id);
env_.Stop();
}
} // namespace
} // namespace connections
} // namespace nearby
cpp/core/internal/mediums/bluetooth_classic.cc
+11 -3
View File
@@ -350,7 +350,8 @@ BluetoothSocket BluetoothClassic::Connect(BluetoothDevice& bluetooth_device,
CancellationFlag* cancellation_flag) {
for (int attempts_count = 0; attempts_count < kConnectAttemptsLimit;
attempts_count++) {
auto wrapper_result = AttemptToConnect(bluetooth_device, service_name);
auto wrapper_result =
AttemptToConnect(bluetooth_device, service_name, cancellation_flag);
if (wrapper_result.IsValid()) {
return wrapper_result;
}
@@ -359,7 +360,8 @@ BluetoothSocket BluetoothClassic::Connect(BluetoothDevice& bluetooth_device,
}
BluetoothSocket BluetoothClassic::AttemptToConnect(
BluetoothDevice& bluetooth_device, const std::string& service_name) {
BluetoothDevice& bluetooth_device, const std::string& service_name,
CancellationFlag* cancellation_flag) {
MutexLock lock(&mutex_);
NEARBY_LOG(INFO, "BluetoothClassic::Connect: device=%p", &bluetooth_device);
// Socket to return. To allow for NRVO to work, it has to be a single object.
@@ -386,8 +388,14 @@ BluetoothSocket BluetoothClassic::AttemptToConnect(
return socket;
}
if (cancellation_flag->Cancelled()) {
NEARBY_LOGS(INFO) << "Can't create client BT socket due to cancel.";
return socket;
}
socket = medium_.ConnectToService(bluetooth_device,
GenerateUuidFromString(service_name));
GenerateUuidFromString(service_name),
cancellation_flag);
if (!socket.IsValid()) {
NEARBY_LOG(INFO, "Failed to Connect via BT [service=%s]",
service_name.c_str());
cpp/core/internal/mediums/bluetooth_classic.h
+2 -1
View File
@@ -170,7 +170,8 @@ class BluetoothClassic {
// Returns socket instance. On success, BluetoothSocket.IsValid() return true.
// Called by client.
BluetoothSocket AttemptToConnect(BluetoothDevice& bluetooth_device,
const std::string& service_name);
const std::string& service_name,
CancellationFlag* cancellation_flag);
mutable Mutex mutex_;
BluetoothRadio& radio_ ABSL_GUARDED_BY(mutex_);
cpp/core/internal/mediums/bluetooth_classic_test.cc
+131 -52
View File
@@ -31,9 +31,20 @@ namespace nearby {
namespace connections {
namespace {
using FeatureFlags = FeatureFlags::Flags;
constexpr FeatureFlags kTestCases[] = {
FeatureFlags{
.enable_cancellation_flag = true,
},
FeatureFlags{
.enable_cancellation_flag = false,
},
};
constexpr absl::Duration kWaitDuration = absl::Milliseconds(1000);
class BluetoothClassicTest : public ::testing::Test {
class BluetoothClassicTest : public ::testing::TestWithParam<FeatureFlags> {
protected:
using DiscoveryCallback = BluetoothClassicMedium::DiscoveryCallback;
@@ -72,6 +83,125 @@ class BluetoothClassicTest : public ::testing::Test {
std::unique_ptr<BluetoothClassic> bt_b_;
};
TEST_P(BluetoothClassicTest, CanConnect) {
FeatureFlags feature_flags = GetParam();
env_.SetFeatureFlags(feature_flags);
constexpr absl::string_view kDeviceName{"Simulated BT device #1"};
constexpr absl::string_view kServiceName{"service name"};
BluetoothRadio& radio_for_client = *radio_a_;
BluetoothRadio& radio_for_server = *radio_b_;
BluetoothClassic& bt_client = *bt_a_;
BluetoothClassic& bt_server = *bt_b_;
EXPECT_TRUE(radio_for_client.IsEnabled());
EXPECT_TRUE(radio_for_server.IsEnabled());
EXPECT_TRUE(bt_server.TurnOnDiscoverability(std::string(kDeviceName)));
EXPECT_EQ(radio_for_server.GetBluetoothAdapter().GetName(),
std::string(kDeviceName));
CountDownLatch latch(1);
BluetoothDevice discovered_device;
EXPECT_TRUE(bt_client.StartDiscovery({
.device_discovered_cb =
[&latch, &discovered_device](BluetoothDevice& device) {
discovered_device = device;
NEARBY_LOG(INFO, "Discovered device=%p [impl=%p]", &device,
&device.GetImpl());
latch.CountDown();
},
}));
EXPECT_TRUE(latch.Await(kWaitDuration).result());
EXPECT_TRUE(bt_server.TurnOffDiscoverability());
ASSERT_TRUE(discovered_device.IsValid());
BluetoothSocket socket_for_server;
CountDownLatch accept_latch(1);
EXPECT_TRUE(bt_server.StartAcceptingConnections(
std::string(kServiceName),
{
.accepted_cb =
[&socket_for_server, &accept_latch](BluetoothSocket socket) {
socket_for_server = std::move(socket);
accept_latch.CountDown();
},
}));
CancellationFlag flag;
BluetoothSocket socket_for_client =
bt_client.Connect(discovered_device, std::string(kServiceName), &flag);
EXPECT_TRUE(accept_latch.Await(kWaitDuration).result());
EXPECT_TRUE(bt_server.StopAcceptingConnections(std::string(kServiceName)));
EXPECT_TRUE(socket_for_server.IsValid());
EXPECT_TRUE(socket_for_client.IsValid());
EXPECT_TRUE(socket_for_server.GetRemoteDevice().IsValid());
EXPECT_TRUE(socket_for_client.GetRemoteDevice().IsValid());
}
TEST_P(BluetoothClassicTest, CanCancelConnect) {
FeatureFlags feature_flags = GetParam();
env_.SetFeatureFlags(feature_flags);
constexpr absl::string_view kDeviceName{"Simulated BT device #1"};
constexpr absl::string_view kServiceName{"service name"};
BluetoothRadio& radio_for_client = *radio_a_;
BluetoothRadio& radio_for_server = *radio_b_;
BluetoothClassic& bt_client = *bt_a_;
BluetoothClassic& bt_server = *bt_b_;
EXPECT_TRUE(radio_for_client.IsEnabled());
EXPECT_TRUE(radio_for_server.IsEnabled());
EXPECT_TRUE(bt_server.TurnOnDiscoverability(std::string(kDeviceName)));
EXPECT_EQ(radio_for_server.GetBluetoothAdapter().GetName(),
std::string(kDeviceName));
CountDownLatch latch(1);
BluetoothDevice discovered_device;
EXPECT_TRUE(bt_client.StartDiscovery({
.device_discovered_cb =
[&latch, &discovered_device](BluetoothDevice& device) {
discovered_device = device;
NEARBY_LOG(INFO, "Discovered device=%p [impl=%p]", &device,
&device.GetImpl());
latch.CountDown();
},
}));
EXPECT_TRUE(latch.Await(kWaitDuration).result());
EXPECT_TRUE(bt_server.TurnOffDiscoverability());
ASSERT_TRUE(discovered_device.IsValid());
BluetoothSocket socket_for_server;
CountDownLatch accept_latch(1);
EXPECT_TRUE(bt_server.StartAcceptingConnections(
std::string(kServiceName),
{
.accepted_cb =
[&socket_for_server, &accept_latch](BluetoothSocket socket) {
socket_for_server = std::move(socket);
accept_latch.CountDown();
},
}));
CancellationFlag flag(true);
BluetoothSocket socket_for_client =
bt_client.Connect(discovered_device, std::string(kServiceName), &flag);
// 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(bt_server.StopAcceptingConnections(std::string(kServiceName)));
EXPECT_TRUE(socket_for_server.IsValid());
EXPECT_TRUE(socket_for_client.IsValid());
EXPECT_TRUE(socket_for_server.GetRemoteDevice().IsValid());
EXPECT_TRUE(socket_for_client.GetRemoteDevice().IsValid());
} else {
EXPECT_FALSE(accept_latch.Await(kWaitDuration).result());
EXPECT_TRUE(bt_server.StopAcceptingConnections(std::string(kServiceName)));
EXPECT_FALSE(socket_for_server.IsValid());
EXPECT_FALSE(socket_for_client.IsValid());
}
}
INSTANTIATE_TEST_SUITE_P(ParametrisedBluetoothClassicTest, BluetoothClassicTest,
::testing::ValuesIn(kTestCases));
TEST_F(BluetoothClassicTest, CanConstructValidObject) {
EXPECT_TRUE(bt_a_->IsMediumValid());
EXPECT_TRUE(bt_a_->IsAdapterValid());
@@ -154,57 +284,6 @@ TEST_F(BluetoothClassicTest, CanStartAcceptingConnections) {
EXPECT_TRUE(bt_server.StopAcceptingConnections(std::string(kServiceName)));
}
TEST_F(BluetoothClassicTest, CanConnect) {
constexpr absl::string_view kDeviceName{"Simulated BT device #1"};
constexpr absl::string_view kServiceName{"service name"};
BluetoothRadio& radio_for_client = *radio_a_;
BluetoothRadio& radio_for_server = *radio_b_;
BluetoothClassic& bt_client = *bt_a_;
BluetoothClassic& bt_server = *bt_b_;
EXPECT_TRUE(radio_for_client.IsEnabled());
EXPECT_TRUE(radio_for_server.IsEnabled());
EXPECT_TRUE(bt_server.TurnOnDiscoverability(std::string(kDeviceName)));
EXPECT_EQ(radio_for_server.GetBluetoothAdapter().GetName(),
std::string(kDeviceName));
CountDownLatch latch(1);
BluetoothDevice discovered_device;
EXPECT_TRUE(bt_client.StartDiscovery({
.device_discovered_cb =
[&latch, &discovered_device](BluetoothDevice& device) {
discovered_device = device;
NEARBY_LOG(INFO, "Discovered device=%p [impl=%p]", &device,
&device.GetImpl());
latch.CountDown();
},
}));
EXPECT_TRUE(latch.Await(kWaitDuration).result());
EXPECT_TRUE(bt_server.TurnOffDiscoverability());
ASSERT_TRUE(discovered_device.IsValid());
BluetoothSocket socket_for_server;
CountDownLatch accept_latch(1);
EXPECT_TRUE(bt_server.StartAcceptingConnections(
std::string(kServiceName),
{
.accepted_cb =
[&socket_for_server, &accept_latch](BluetoothSocket socket) {
socket_for_server = std::move(socket);
accept_latch.CountDown();
},
}));
CancellationFlag flag;
BluetoothSocket socket_for_client =
bt_client.Connect(discovered_device, std::string(kServiceName), &flag);
EXPECT_TRUE(accept_latch.Await(kWaitDuration).result());
EXPECT_TRUE(bt_server.StopAcceptingConnections(std::string(kServiceName)));
EXPECT_TRUE(socket_for_server.IsValid());
EXPECT_TRUE(socket_for_client.IsValid());
EXPECT_TRUE(socket_for_server.GetRemoteDevice().IsValid());
EXPECT_TRUE(socket_for_client.GetRemoteDevice().IsValid());
}
} // namespace
} // namespace connections
} // namespace nearby
cpp/core/internal/mediums/webrtc.cc
+8 -3
View File
@@ -212,8 +212,8 @@ WebRtcSocketWrapper WebRtc::Connect(const std::string& service_id,
CancellationFlag* cancellation_flag) {
for (int attempts_count = 0; attempts_count < kConnectAttemptsLimit;
attempts_count++) {
auto wrapper_result =
AttemptToConnect(service_id, remote_peer_id, location_hint);
auto wrapper_result = AttemptToConnect(service_id, remote_peer_id,
location_hint, cancellation_flag);
if (wrapper_result.IsValid()) {
return wrapper_result;
}
@@ -223,7 +223,7 @@ WebRtcSocketWrapper WebRtc::Connect(const std::string& service_id,
WebRtcSocketWrapper WebRtc::AttemptToConnect(
const std::string& service_id, const PeerId& remote_peer_id,
const LocationHint& location_hint) {
const LocationHint& location_hint, CancellationFlag* cancellation_flag) {
ConnectionRequestInfo info = ConnectionRequestInfo();
info.self_peer_id = PeerId::FromRandom();
Future<WebRtcSocketWrapper> socket_future = info.socket_future;
@@ -238,6 +238,11 @@ WebRtcSocketWrapper WebRtc::AttemptToConnect(
return WebRtcSocketWrapper();
}
if (cancellation_flag->Cancelled()) {
NEARBY_LOGS(INFO) << "Cannot connect with WebRtc due to cancel.";
return WebRtcSocketWrapper();
}
// Create a new ConnectionFlow for this connection attempt.
std::unique_ptr<ConnectionFlow> connection_flow =
CreateConnectionFlow(service_id, remote_peer_id);
cpp/core/internal/mediums/webrtc.h
+2 -1
View File
@@ -143,7 +143,8 @@ class WebRtc {
// Runs on @MainThread.
WebRtcSocketWrapper AttemptToConnect(const std::string& service_id,
const PeerId& peer_id,
const LocationHint& location_hint)
const LocationHint& location_hint,
CancellationFlag* cancellation_flag)
ABSL_LOCKS_EXCLUDED(mutex_);
// Returns if the device is accepting connection with specific service id.
cpp/core/internal/mediums/webrtc_test.cc
+93 -37
View File
@@ -28,7 +28,18 @@ namespace mediums {
namespace {
class WebRtcTest : public ::testing::Test {
using FeatureFlags = FeatureFlags::Flags;
constexpr FeatureFlags kTestCases[] = {
FeatureFlags{
.enable_cancellation_flag = true,
},
FeatureFlags{
.enable_cancellation_flag = false,
},
};
class WebRtcTest : public ::testing::TestWithParam<FeatureFlags> {
protected:
WebRtcTest() {
MediumEnvironment::Instance().Stop();
@@ -36,6 +47,87 @@ class WebRtcTest : public ::testing::Test {
}
};
// Tests the flow when the two devices exchange SDP messages and connect to each
// other but the signaling channel is closed before sending the data.
TEST_P(WebRtcTest, ConnectBothDevices_ShutdownSignaling_SendData) {
FeatureFlags feature_flags = GetParam();
MediumEnvironment::Instance().SetFeatureFlags(feature_flags);
WebRtc receiver, sender;
WebRtcSocketWrapper receiver_socket, sender_socket;
const PeerId self_id("self_id");
const std::string service_id("NearbySharing");
LocationHint location_hint;
Future<bool> connected;
ByteArray message("message xyz");
receiver.StartAcceptingConnections(
service_id, self_id, location_hint,
{[&receiver_socket, connected](WebRtcSocketWrapper wrapper) mutable {
receiver_socket = wrapper;
connected.Set(receiver_socket.IsValid());
}});
CancellationFlag flag;
sender_socket = sender.Connect(service_id, self_id, location_hint, &flag);
EXPECT_TRUE(sender_socket.IsValid());
ExceptionOr<bool> devices_connected = connected.Get();
ASSERT_TRUE(devices_connected.ok());
EXPECT_TRUE(devices_connected.result());
// Only shuts down signaling channel.
receiver.StopAcceptingConnections(service_id);
sender_socket.GetOutputStream().Write(message);
ExceptionOr<ByteArray> received_msg =
receiver_socket.GetInputStream().Read(/*size=*/32);
ASSERT_TRUE(received_msg.ok());
EXPECT_EQ(message, received_msg.result());
}
TEST_P(WebRtcTest, CanCancelConnect) {
FeatureFlags feature_flags = GetParam();
MediumEnvironment::Instance().SetFeatureFlags(feature_flags);
WebRtc receiver, sender;
WebRtcSocketWrapper receiver_socket, sender_socket;
const PeerId self_id("self_id");
const std::string service_id("NearbySharing");
LocationHint location_hint;
Future<bool> connected;
ByteArray message("message");
receiver.StartAcceptingConnections(
service_id, self_id, location_hint,
{[&receiver_socket, connected](WebRtcSocketWrapper wrapper) mutable {
receiver_socket = wrapper;
connected.Set(receiver_socket.IsValid());
}});
CancellationFlag flag(true);
sender_socket = sender.Connect(service_id, self_id, location_hint, &flag);
// If FeatureFlag is disabled, Cancelled is false as no-op.
if (!feature_flags.enable_cancellation_flag) {
EXPECT_TRUE(sender_socket.IsValid());
ExceptionOr<bool> devices_connected = connected.Get();
ASSERT_TRUE(devices_connected.ok());
EXPECT_TRUE(devices_connected.result());
sender_socket.GetOutputStream().Write(message);
ExceptionOr<ByteArray> received_msg =
receiver_socket.GetInputStream().Read(/*size=*/32);
ASSERT_TRUE(received_msg.ok());
EXPECT_EQ(message, received_msg.result());
receiver_socket.Close();
} else {
EXPECT_FALSE(sender_socket.IsValid());
}
}
INSTANTIATE_TEST_SUITE_P(ParametrisedWebRtcTest, WebRtcTest,
::testing::ValuesIn(kTestCases));
// Basic test to check that device is accepting connections when initialized.
TEST_F(WebRtcTest, NotAcceptingConnections) {
WebRtc webrtc;
@@ -241,42 +333,6 @@ TEST_F(WebRtcTest, ConnectBothDevicesAndSendData) {
receiver_socket.Close();
}
// Tests the flow when the two devices exchange SDP messages and connect to each
// other but the signaling channel is closed before sending the data.
TEST_F(WebRtcTest, ConnectBothDevices_ShutdownSignaling_SendData) {
WebRtc receiver, sender;
WebRtcSocketWrapper receiver_socket, sender_socket;
const PeerId self_id("self_id");
const std::string service_id("NearbySharing");
LocationHint location_hint;
Future<bool> connected;
ByteArray message("message xyz");
receiver.StartAcceptingConnections(
service_id, self_id, location_hint,
{[&receiver_socket, connected](WebRtcSocketWrapper wrapper) mutable {
receiver_socket = wrapper;
connected.Set(receiver_socket.IsValid());
}});
CancellationFlag flag;
sender_socket = sender.Connect(service_id, self_id, location_hint, &flag);
EXPECT_TRUE(sender_socket.IsValid());
ExceptionOr<bool> devices_connected = connected.Get();
ASSERT_TRUE(devices_connected.ok());
EXPECT_TRUE(devices_connected.result());
// Only shuts down signaling channel.
receiver.StopAcceptingConnections(service_id);
sender_socket.GetOutputStream().Write(message);
ExceptionOr<ByteArray> received_msg =
receiver_socket.GetInputStream().Read(/*size=*/32);
ASSERT_TRUE(received_msg.ok());
EXPECT_EQ(message, received_msg.result());
}
TEST_F(WebRtcTest, Connect_NullPeerConnection) {
using MockAcceptedCallback =
testing::MockFunction<void(WebRtcSocketWrapper socket)>;
cpp/core/internal/mediums/wifi_lan.cc
+6 -1
View File
@@ -248,7 +248,12 @@ WifiLanSocket WifiLan::Connect(WifiLanService& wifi_lan_service,
return socket;
}
socket = medium_.Connect(wifi_lan_service, service_id);
if (cancellation_flag->Cancelled()) {
NEARBY_LOGS(INFO) << "Can't create client WifiLan socket due to cancel.";
return socket;
}
socket = medium_.Connect(wifi_lan_service, service_id, cancellation_flag);
if (!socket.IsValid()) {
NEARBY_LOG(INFO, "Failed to Connect via WifiLan [service_id=%s]",
service_id.c_str());
cpp/core/internal/mediums/wifi_lan_test.cc
+123 -50
View File
@@ -29,6 +29,17 @@ namespace nearby {
namespace connections {
namespace {
using FeatureFlags = FeatureFlags::Flags;
constexpr FeatureFlags kTestCases[] = {
FeatureFlags{
.enable_cancellation_flag = true,
},
FeatureFlags{
.enable_cancellation_flag = false,
},
};
constexpr absl::Duration kWaitDuration = absl::Milliseconds(1000);
constexpr absl::string_view kServiceID{"com.google.location.nearby.apps.test"};
constexpr absl::string_view kServiceInfoName{
@@ -36,7 +47,7 @@ constexpr absl::string_view kServiceInfoName{
constexpr absl::string_view kEndpointName{"Simulated endpoint name"};
constexpr absl::string_view kEndpointInfoKey{"n"};
class WifiLanTest : public ::testing::Test {
class WifiLanTest : public ::testing::TestWithParam<FeatureFlags> {
protected:
using DiscoveredServiceCallback = WifiLanMedium::DiscoveredServiceCallback;
@@ -45,6 +56,117 @@ class WifiLanTest : public ::testing::Test {
MediumEnvironment& env_{MediumEnvironment::Instance()};
};
TEST_P(WifiLanTest, CanStartAcceptingConnectionsAndConnect) {
FeatureFlags feature_flags = GetParam();
env_.SetFeatureFlags(feature_flags);
env_.Start();
WifiLan wifi_lan_a;
WifiLan wifi_lan_b;
std::string service_id(kServiceID);
std::string service_info_name{kServiceInfoName};
std::string endpoint_info_name{kEndpointName};
CountDownLatch found_latch(1);
CountDownLatch accept_latch(1);
NsdServiceInfo nsd_service_info;
nsd_service_info.SetServiceInfoName(service_info_name);
nsd_service_info.SetTxtRecord(std::string(kEndpointInfoKey),
endpoint_info_name);
wifi_lan_a.StartAdvertising(service_id, nsd_service_info);
wifi_lan_a.StartAcceptingConnections(
service_id,
{
.accepted_cb = [&accept_latch](
WifiLanSocket socket,
absl::string_view) { accept_latch.CountDown(); },
});
WifiLanService discovered_service;
wifi_lan_b.StartDiscovery(
service_id,
{
.service_discovered_cb =
[&found_latch, &discovered_service](
WifiLanService& service, absl::string_view service_id) {
discovered_service = service;
NEARBY_LOG(INFO, "Discovered service=%p [impl=%p]", &service,
&service.GetImpl());
found_latch.CountDown();
},
});
EXPECT_TRUE(found_latch.Await(kWaitDuration).result());
ASSERT_TRUE(discovered_service.IsValid());
CancellationFlag flag;
WifiLanSocket socket =
wifi_lan_b.Connect(discovered_service, service_id, &flag);
EXPECT_TRUE(accept_latch.Await(kWaitDuration).result());
EXPECT_TRUE(socket.IsValid());
wifi_lan_b.StopDiscovery(service_id);
wifi_lan_a.StopAcceptingConnections(service_id);
wifi_lan_a.StopAdvertising(service_id);
env_.Stop();
}
TEST_P(WifiLanTest, CanCancelConnect) {
FeatureFlags feature_flags = GetParam();
env_.SetFeatureFlags(feature_flags);
env_.Start();
WifiLan wifi_lan_a;
WifiLan wifi_lan_b;
std::string service_id(kServiceID);
std::string service_info_name{kServiceInfoName};
std::string endpoint_info_name{kEndpointName};
CountDownLatch found_latch(1);
CountDownLatch accept_latch(1);
NsdServiceInfo nsd_service_info;
nsd_service_info.SetServiceInfoName(service_info_name);
nsd_service_info.SetTxtRecord(std::string(kEndpointInfoKey),
endpoint_info_name);
wifi_lan_a.StartAdvertising(service_id, nsd_service_info);
wifi_lan_a.StartAcceptingConnections(
service_id,
{
.accepted_cb = [&accept_latch](
WifiLanSocket socket,
absl::string_view) { accept_latch.CountDown(); },
});
WifiLanService discovered_service;
wifi_lan_b.StartDiscovery(
service_id,
{
.service_discovered_cb =
[&found_latch, &discovered_service](
WifiLanService& service, absl::string_view service_id) {
discovered_service = service;
NEARBY_LOG(INFO, "Discovered service=%p [impl=%p]", &service,
&service.GetImpl());
found_latch.CountDown();
},
});
EXPECT_TRUE(found_latch.Await(kWaitDuration).result());
ASSERT_TRUE(discovered_service.IsValid());
CancellationFlag flag(true);
WifiLanSocket socket =
wifi_lan_b.Connect(discovered_service, service_id, &flag);
// 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(socket.IsValid());
} else {
EXPECT_FALSE(accept_latch.Await(kWaitDuration).result());
EXPECT_FALSE(socket.IsValid());
}
wifi_lan_b.StopDiscovery(service_id);
wifi_lan_a.StopAcceptingConnections(service_id);
wifi_lan_a.StopAdvertising(service_id);
env_.Stop();
}
INSTANTIATE_TEST_SUITE_P(ParametrisedWifiLanTest, WifiLanTest,
::testing::ValuesIn(kTestCases));
TEST_F(WifiLanTest, CanConstructValidObject) {
env_.Start();
WifiLan wifi_lan_a;
@@ -121,55 +243,6 @@ TEST_F(WifiLanTest, CanStartDiscovery) {
env_.Stop();
}
TEST_F(WifiLanTest, CanStartAcceptingConnectionsAndConnect) {
env_.Start();
WifiLan wifi_lan_a;
WifiLan wifi_lan_b;
std::string service_id(kServiceID);
std::string service_info_name{kServiceInfoName};
std::string endpoint_info_name{kEndpointName};
CountDownLatch found_latch(1);
CountDownLatch accept_latch(1);
NsdServiceInfo nsd_service_info;
nsd_service_info.SetServiceInfoName(service_info_name);
nsd_service_info.SetTxtRecord(std::string(kEndpointInfoKey),
endpoint_info_name);
wifi_lan_a.StartAdvertising(service_id, nsd_service_info);
wifi_lan_a.StartAcceptingConnections(
service_id,
{
.accepted_cb = [&accept_latch](
WifiLanSocket socket,
absl::string_view) { accept_latch.CountDown(); },
});
WifiLanService discovered_service;
wifi_lan_b.StartDiscovery(
service_id,
{
.service_discovered_cb =
[&found_latch, &discovered_service](
WifiLanService& service, absl::string_view service_id) {
discovered_service = service;
NEARBY_LOG(INFO, "Discovered service=%p [impl=%p]", &service,
&service.GetImpl());
found_latch.CountDown();
},
});
EXPECT_TRUE(found_latch.Await(kWaitDuration).result());
ASSERT_TRUE(discovered_service.IsValid());
CancellationFlag flag;
WifiLanSocket socket =
wifi_lan_b.Connect(discovered_service, service_id, &flag);
EXPECT_TRUE(accept_latch.Await(kWaitDuration).result());
EXPECT_TRUE(socket.IsValid());
wifi_lan_b.StopDiscovery(service_id);
wifi_lan_a.StopAcceptingConnections(service_id);
wifi_lan_a.StopAdvertising(service_id);
env_.Stop();
}
} // namespace
} // namespace connections
} // namespace nearby
cpp/core/internal/offline_simulation_user.cc
+1
View File
@@ -167,6 +167,7 @@ Status OfflineSimulationUser::RequestConnection(CountDownLatch* latch) {
.disconnected_cb =
absl::bind_front(&OfflineSimulationUser::OnEndpointDisconnect, this),
};
client_.AddCancellationFlag(discovered_.endpoint_id);
return ctrl_.RequestConnection(
&client_, discovered_.endpoint_id,
{
cpp/core/internal/p2p_cluster_pcp_handler_test.cc
+1
View File
@@ -227,6 +227,7 @@ TEST_P(P2pClusterPcpHandlerTest, CanConnect) {
EXPECT_TRUE(discover_latch.Await(absl::Milliseconds(1000)).result());
EXPECT_EQ(endpoint_name_a, std::string{discovered.endpoint_info});
client_b_.AddCancellationFlag(discovered.endpoint_id);
handler_b.RequestConnection(
&client_b_, discovered.endpoint_id,
{
cpp/core/internal/payload_manager.cc
+4 -2
View File
@@ -405,9 +405,11 @@ void PayloadManager::OnEndpointDisconnect(ClientProxy* client,
if (!pending_payload) continue;
auto endpoint_info = pending_payload->GetEndpoint(endpoint_id);
if (!endpoint_info) continue;
std::int64_t endpoint_offset = endpoint_info->offset;
// Stop tracking the endpoint for this payload.
pending_payload->RemoveEndpoints({endpoint_id});
// |endpoint_info| is longer valid after calling RemoveEndpoints.
endpoint_info = nullptr;
std::int64_t payload_total_size =
pending_payload->GetInternalPayload()->GetTotalSize();
@@ -420,7 +422,7 @@ void PayloadManager::OnEndpointDisconnect(ClientProxy* client,
// Create the payload transfer update.
PayloadProgressInfo update{payload_id,
PayloadProgressInfo::Status::kFailure,
payload_total_size, endpoint_info->offset};
payload_total_size, endpoint_offset};
// Send a client notification of a payload transfer failure.
client->OnPayloadProgress(endpoint_id, update);
cpp/core/internal/simulation_user.cc
+1
View File
@@ -148,6 +148,7 @@ void SimulationUser::RequestConnection(CountDownLatch* latch) {
.rejected_cb =
absl::bind_front(&SimulationUser::OnConnectionRejected, this),
};
client_.AddCancellationFlag(discovered_.endpoint_id);
EXPECT_TRUE(
mgr_.RequestConnection(&client_, discovered_.endpoint_id,
{