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

Merge branch 'google3' to roll forward up to cl/353292511.

Browse Source
This commit is contained in:
hai007
2021-01-22 12:15:03 -08:00
parent f657fea81c 335b3a8d45
commit 39d141ad73
36 changed files with 592 additions and 155 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cpp/core/internal/BUILD
+2
View File
@@ -84,6 +84,7 @@ cc_library(
],
visibility = [
"//core:__pkg__",
"//core/internal/fuzzers:__pkg__",
],
deps = [
":message_lite",
@@ -94,6 +95,7 @@ cc_library(
"//proto/connections:offline_wire_formats_portable_proto",
"//platform/api:comm",
"//platform/base",
"//platform/base:cancellation_flag",
"//platform/base:util",
"//platform/public:comm",
"//platform/public:logging",
cpp/core/internal/base_pcp_handler.cc
+61 -52
View File
@@ -95,14 +95,12 @@ Status BasePcpHandler::StartAdvertising(ClientProxy* client,
// Now that we've succeeded, mark the client as advertising.
// Save the advertising options for local reference in later process like
// upgrading bandwidth.
// TODO(hais): saving advertising_options_ in clientProxy instead of here
// as java implementation does.
std::vector<proto::connections::Medium> supported_mediums =
advertising_options.GetMediums();
advertising_options_ = advertising_options;
advertising_listener_ = info.listener;
client->StartedAdvertising(service_id, GetStrategy(), info.listener,
absl::MakeSpan(result.mediums));
absl::MakeSpan(result.mediums),
advertising_options);
response.Set({Status::kSuccess});
});
return WaitForResult(
@@ -115,7 +113,6 @@ void BasePcpHandler::StopAdvertising(ClientProxy* client) {
RunOnPcpHandlerThread([this, client, &latch]() {
StopAdvertisingImpl(client);
client->StoppedAdvertising();
// advertising_options_ is purposefully not cleared here.
latch.CountDown();
});
WaitForLatch("StopAdvertising", &latch);
@@ -142,23 +139,22 @@ Status BasePcpHandler::StartDiscovery(ClientProxy* client,
NEARBY_LOG(INFO, "StartDiscovery with supported mediums: %s",
GetStringValueOfSupportedMediums(options).c_str());
RunOnPcpHandlerThread(
[this, client, service_id, discovery_options, &listener, &response]() {
// Ask the implementation to attempt to start discovery.
auto result = StartDiscoveryImpl(client, service_id, discovery_options);
if (!result.status.Ok()) {
response.Set(result.status);
return;
}
RunOnPcpHandlerThread([this, client, service_id, discovery_options, &listener,
&response]() {
// Ask the implementation to attempt to start discovery.
auto result = StartDiscoveryImpl(client, service_id, discovery_options);
if (!result.status.Ok()) {
response.Set(result.status);
return;
}
// Now that we've succeeded, mark the client as discovering and clear
// out any old endpoints we had discovered.
discovery_options_ = discovery_options;
discovered_endpoints_.clear();
client->StartedDiscovery(service_id, GetStrategy(), listener,
absl::MakeSpan(result.mediums));
response.Set({Status::kSuccess});
});
// Now that we've succeeded, mark the client as discovering and clear
// out any old endpoints we had discovered.
discovered_endpoints_.clear();
client->StartedDiscovery(service_id, GetStrategy(), listener,
absl::MakeSpan(result.mediums), discovery_options);
response.Set({Status::kSuccess});
});
return WaitForResult(absl::StrCat("StartDiscovery(", service_id, ")"),
client->GetClientId(), &response);
}
@@ -168,7 +164,6 @@ void BasePcpHandler::StopDiscovery(ClientProxy* client) {
RunOnPcpHandlerThread([this, client, &latch]() {
StopDiscoveryImpl(client);
client->StoppedDiscovery();
// discovery_options_ is purposefully not cleared here.
latch.CountDown();
});
@@ -346,7 +341,7 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client,
// If our child class says we can't send any more outgoing connections,
// listen to them.
if (ShouldEnforceTopologyConstraints() &&
if (ShouldEnforceTopologyConstraints(client->GetAdvertisingOptions()) &&
!CanSendOutgoingConnection(client)) {
NEARBY_LOG(INFO, "Outgoing connection not allowed: id=%s",
endpoint_id.c_str());
@@ -365,13 +360,14 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client,
auto remote_bluetooth_mac_address =
BluetoothUtils::ToString(options.remote_bluetooth_mac_address);
if (!remote_bluetooth_mac_address.empty()) {
if (AppendRemoteBluetoothMacAddressEndpoint(endpoint_id,
remote_bluetooth_mac_address))
if (AppendRemoteBluetoothMacAddressEndpoint(
endpoint_id, remote_bluetooth_mac_address,
client->GetDiscoveryOptions()))
NEARBY_LOGS(INFO) << "Appended remote Bluetooth MAC Address endpoint "
<< "[" << remote_bluetooth_mac_address << "]";
}
if (AppendWebRTCEndpoint(endpoint_id))
if (AppendWebRTCEndpoint(endpoint_id, client->GetDiscoveryOptions()))
NEARBY_LOGS(INFO) << "Appended Web RTC endpoint.";
auto discovered_endpoints = GetDiscoveredEndpoints(endpoint_id);
@@ -380,7 +376,7 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client,
for (auto connect_endpoint : discovered_endpoints) {
if (!MediumSupportedByClientOptions(connect_endpoint->medium,
discovery_options_))
client->GetDiscoveryOptions()))
continue;
connect_impl_result = ConnectImpl(client, connect_endpoint);
if (connect_impl_result.status.Ok()) {
@@ -405,7 +401,7 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client,
// endpoint about ourselves.
Exception write_exception = WriteConnectionRequestFrame(
channel.get(), client->GetLocalEndpointId(), info.endpoint_info, nonce,
GetSupportedConnectionMediumsByPriority(discovery_options_));
GetSupportedConnectionMediumsByPriority(client->GetDiscoveryOptions()));
if (!write_exception.Ok()) {
NEARBY_LOG(INFO, "Failed to send connection request: id=%s",
endpoint_id.c_str());
@@ -467,7 +463,7 @@ bool BasePcpHandler::MediumSupportedByClientOptions(
}
// Get ordered supported connection medium based on local advertising/discovery
// option. local_option is either advertising_options_ or discovery_options_.
// option.
std::vector<proto::connections::Medium>
BasePcpHandler::GetSupportedConnectionMediumsByPriority(
const ConnectionOptions& local_option) {
@@ -501,6 +497,19 @@ BasePcpHandler::GetDiscoveredEndpoints(const std::string& endpoint_id) {
[this](DiscoveredEndpoint* a, DiscoveredEndpoint* b) -> bool {
return IsPreferred(*a, *b);
});
return result;
}
std::vector<BasePcpHandler::DiscoveredEndpoint*>
BasePcpHandler::GetDiscoveredEndpoints(
const proto::connections::Medium medium) {
std::vector<BasePcpHandler::DiscoveredEndpoint*> result;
for (const auto& item : discovered_endpoints_) {
if (item.second->medium == medium) {
result.push_back(item.second.get());
}
}
return result;
}
@@ -569,22 +578,24 @@ void BasePcpHandler::ProcessPreConnectionResultFailure(
client->OnConnectionRejected(endpoint_id, {Status::kError});
}
bool BasePcpHandler::ShouldEnforceTopologyConstraints() const {
bool BasePcpHandler::ShouldEnforceTopologyConstraints(
const ConnectionOptions& local_advertising_options) const {
// Topology constraints only matter for the advertiser.
// For discoverers, we'll always enforce them.
if (advertising_options_.strategy.IsNone()) {
if (local_advertising_options.strategy.IsNone()) {
return true;
}
return advertising_options_.enforce_topology_constraints;
return local_advertising_options.enforce_topology_constraints;
}
bool BasePcpHandler::AutoUpgradeBandwidth() const {
if (advertising_options_.strategy.IsNone()) {
bool BasePcpHandler::AutoUpgradeBandwidth(
const ConnectionOptions& local_advertising_options) const {
if (local_advertising_options.strategy.IsNone()) {
return true;
}
return advertising_options_.auto_upgrade_bandwidth;
return local_advertising_options.auto_upgrade_bandwidth;
}
Status BasePcpHandler::AcceptConnection(
@@ -765,14 +776,6 @@ BluetoothDevice BasePcpHandler::GetRemoteBluetoothDevice(
remote_bluetooth_mac_address);
}
ConnectionOptions BasePcpHandler::GetConnectionOptions() const {
return advertising_options_;
}
ConnectionOptions BasePcpHandler::GetDiscoveryOptions() const {
return discovery_options_;
}
void BasePcpHandler::OnEndpointFound(
ClientProxy* client, std::shared_ptr<DiscoveredEndpoint> endpoint) {
// Check if we've seen this endpoint ID before.
@@ -933,7 +936,7 @@ Exception BasePcpHandler::OnIncomingConnection(
// If our child class says we can't accept any more incoming connections,
// listen to them.
if (ShouldEnforceTopologyConstraints() &&
if (ShouldEnforceTopologyConstraints(client->GetAdvertisingOptions()) &&
!CanReceiveIncomingConnection(client)) {
return {Exception::kIo};
}
@@ -1029,15 +1032,18 @@ void BasePcpHandler::InitiateBandwidthUpgrade(
// sense to dynamically select the proper medium for upgrading.
// TODO(hais): when we add more mediums like Wifi Hotspot, we need to prevent
// upgrading interfering with active connections.
Medium bwu_medium = ChooseBestUpgradeMedium(their_supported_mediums);
Medium bwu_medium = ChooseBestUpgradeMedium(their_supported_mediums,
client->GetAdvertisingOptions());
if (AutoUpgradeBandwidth() && bwu_medium != Medium::UNKNOWN_MEDIUM) {
if (AutoUpgradeBandwidth(client->GetAdvertisingOptions()) &&
bwu_medium != Medium::UNKNOWN_MEDIUM) {
bwu_manager_->InitiateBwuForEndpoint(client, endpoint_id, bwu_medium);
}
}
proto::connections::Medium BasePcpHandler::ChooseBestUpgradeMedium(
const std::vector<proto::connections::Medium>& their_supported_mediums) {
const std::vector<proto::connections::Medium>& their_supported_mediums,
const ConnectionOptions& local_advertising_options) {
// If the remote side did not report their supported mediums, choose an
// appropriate default.
std::vector<proto::connections::Medium> their_mediums =
@@ -1048,7 +1054,7 @@ proto::connections::Medium BasePcpHandler::ChooseBestUpgradeMedium(
// Otherwise, pick the best medium we support.
std::vector<proto::connections::Medium> my_mediums =
GetSupportedConnectionMediumsByPriority(advertising_options_);
GetSupportedConnectionMediumsByPriority(local_advertising_options);
for (const auto& my_medium : my_mediums) {
for (const auto& their_medium : their_mediums) {
if (my_medium == their_medium) {
@@ -1062,8 +1068,9 @@ proto::connections::Medium BasePcpHandler::ChooseBestUpgradeMedium(
bool BasePcpHandler::AppendRemoteBluetoothMacAddressEndpoint(
const std::string& endpoint_id,
const std::string& remote_bluetooth_mac_address) {
if (!discovery_options_.allowed.bluetooth) {
const std::string& remote_bluetooth_mac_address,
const ConnectionOptions& local_discovery_options) {
if (!local_discovery_options.allowed.bluetooth) {
return false;
}
@@ -1103,8 +1110,10 @@ bool BasePcpHandler::AppendRemoteBluetoothMacAddressEndpoint(
return true;
}
bool BasePcpHandler::AppendWebRTCEndpoint(const std::string& endpoint_id) {
if (!discovery_options_.allowed.web_rtc) {
bool BasePcpHandler::AppendWebRTCEndpoint(
const std::string& endpoint_id,
const ConnectionOptions& local_discovery_options) {
if (!local_discovery_options.allowed.web_rtc) {
return false;
}
cpp/core/internal/base_pcp_handler.h
+14 -17
View File
@@ -311,6 +311,10 @@ class BasePcpHandler : public PcpHandler,
std::vector<BasePcpHandler::DiscoveredEndpoint*> GetDiscoveredEndpoints(
const std::string& endpoint_id);
// Returns a vector of discovered endpoints that share a given Medium.
std::vector<BasePcpHandler::DiscoveredEndpoint*> GetDiscoveredEndpoints(
const proto::connections::Medium medium);
mediums::PeerId CreatePeerIdFromAdvertisement(const string& service_id,
const string& endpoint_id,
const ByteArray& endpoint_info);
@@ -407,11 +411,13 @@ class BasePcpHandler : public PcpHandler,
const BasePcpHandler::DiscoveredEndpoint& old_endpoint);
// Returns true, if connection party should respect the specified topology.
bool ShouldEnforceTopologyConstraints() const;
bool ShouldEnforceTopologyConstraints(
const ConnectionOptions& local_advertising_options) const;
// Returns true, if connection party should attempt to upgrade itself to
// use a higher bandwidth medium, if it is available.
bool AutoUpgradeBandwidth() const;
bool AutoUpgradeBandwidth(
const ConnectionOptions& local_advertising_options) const;
// Returns true if the incoming connection should be killed. This only
// happens when an incoming connection arrives while we have an outgoing
@@ -438,18 +444,21 @@ class BasePcpHandler : public PcpHandler,
// Returns the optimal medium supported by both devices.
proto::connections::Medium ChooseBestUpgradeMedium(
const std::vector<proto::connections::Medium>& supported_mediums);
const std::vector<proto::connections::Medium>& supported_mediums,
const ConnectionOptions& local_advertising_options);
// Returns true if the bluetooth endpoint based on remote bluetooth mac
// address is created and appended into discovered_endpoints_ with key
// endpoint_id.
bool AppendRemoteBluetoothMacAddressEndpoint(
const std::string& endpoint_id,
const std::string& remote_bluetooth_mac_address);
const std::string& remote_bluetooth_mac_address,
const ConnectionOptions& local_discovery_options);
// Returns true if the webrtc endpoint is created and appended into
// discovered_endpoints_ with key endpoint_id.
bool AppendWebRTCEndpoint(const std::string& endpoint_id);
bool AppendWebRTCEndpoint(const std::string& endpoint_id,
const ConnectionOptions& local_discovery_options);
void ProcessPreConnectionInitiationFailure(const std::string& endpoint_id,
EndpointChannel* channel,
@@ -506,22 +515,10 @@ class BasePcpHandler : public PcpHandler,
// doesn't happen.
absl::flat_hash_map<std::string, CancelableAlarm> pending_alarms_;
// The active ClientProxy's advertising constraints. Empty()
// returns true if the client hasn't started advertising false otherwise.
// Note: this is not cleared when the client stops advertising because it
// might still be useful downstream of advertising (eg: establishing
// connections, performing bandwidth upgrades, etc.)
ConnectionOptions advertising_options_;
// The active ClientProxy's connection lifecycle listener. Non-null while
// advertising.
ConnectionListener advertising_listener_;
// The active ClientProxy's discovery constraints. Null if the client
// hasn't started discovering. Note: this is not cleared when the client
// stops discovering because it might still be useful downstream of
// discovery (eg: connection speed, etc.)
ConnectionOptions discovery_options_;
AtomicBoolean stop_{false};
Pcp pcp_;
Strategy strategy_{PcpToStrategy(pcp_)};
cpp/core/internal/base_pcp_handler_test.cc
+28 -23
View File
@@ -130,7 +130,8 @@ class MockPcpHandler : public BasePcpHandler {
MOCK_METHOD(Status, StopDiscoveryImpl, (ClientProxy * client), (override));
MOCK_METHOD(Status, InjectEndpointImpl,
(ClientProxy * client, const std::string& service_id,
const OutOfBandConnectionMetadata& metadata), (override));
const OutOfBandConnectionMetadata& metadata),
(override));
MOCK_METHOD(ConnectImplResult, ConnectImpl,
(ClientProxy * client, DiscoveredEndpoint* endpoint), (override));
MOCK_METHOD(proto::connections::Medium, GetDefaultUpgradeMedium, (),
@@ -138,7 +139,9 @@ class MockPcpHandler : public BasePcpHandler {
std::vector<proto::connections::Medium> GetConnectionMediumsByPriority()
override {
return GetDiscoveryMediums();
return std::vector<proto::connections::Medium>{
proto::connections::WIFI_LAN, proto::connections::WEB_RTC,
proto::connections::BLUETOOTH, proto::connections::BLE};
}
// Mock adapters for protected non-virtual methods of a base class.
@@ -154,9 +157,9 @@ class MockPcpHandler : public BasePcpHandler {
return BasePcpHandler::GetDiscoveredEndpoints(endpoint_id);
}
std::vector<proto::connections::Medium> GetDiscoveryMediums() {
auto allowed =
BasePcpHandler::GetDiscoveryOptions().CompatibleOptions().allowed;
std::vector<proto::connections::Medium> GetDiscoveryMediums(
ClientProxy* client) {
auto allowed = client->GetDiscoveryOptions().CompatibleOptions().allowed;
return GetMediumsFromSelector(allowed);
}
@@ -322,7 +325,7 @@ class BasePcpHandlerTest
EXPECT_CALL(mock_connection_listener_.initiated_cb, Call).Times(1);
// Simulate successful discovery.
auto encryption_runner = std::make_unique<EncryptionRunner>();
auto allowed_mediums = pcp_handler->GetDiscoveryMediums();
auto allowed_mediums = pcp_handler->GetDiscoveryMediums(client);
EXPECT_CALL(*pcp_handler, ConnectImpl)
.WillOnce(Invoke([&channel_a, connect_medium](
@@ -457,7 +460,7 @@ TEST_P(BasePcpHandlerTest, RequestConnectionChangesState) {
BwuManager bwu(m, em, ecm, {}, {});
MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
StartDiscovery(&client, &pcp_handler);
auto mediums = pcp_handler.GetDiscoveryMediums();
auto mediums = pcp_handler.GetDiscoveryMediums(&client);
auto connect_medium = mediums[mediums.size() - 1];
auto channel_pair = SetupConnection(pipe_a_, pipe_b_, connect_medium);
auto& channel_a = channel_pair.first;
@@ -482,7 +485,7 @@ TEST_P(BasePcpHandlerTest, AcceptConnectionChangesState) {
BwuManager bwu(m, em, ecm, {}, {});
MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
StartDiscovery(&client, &pcp_handler);
auto mediums = pcp_handler.GetDiscoveryMediums();
auto mediums = pcp_handler.GetDiscoveryMediums(&client);
auto connect_medium = mediums[mediums.size() - 1];
auto channel_pair = SetupConnection(pipe_a_, pipe_b_, connect_medium);
auto& channel_a = channel_pair.first;
@@ -511,7 +514,7 @@ TEST_P(BasePcpHandlerTest, RejectConnectionChangesState) {
BwuManager bwu(m, em, ecm, {}, {});
MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
StartDiscovery(&client, &pcp_handler);
auto mediums = pcp_handler.GetDiscoveryMediums();
auto mediums = pcp_handler.GetDiscoveryMediums(&client);
auto connect_medium = mediums[mediums.size() - 1];
auto channel_pair = SetupConnection(pipe_a_, pipe_b_, connect_medium);
auto& channel_b = channel_pair.second;
@@ -536,7 +539,7 @@ TEST_P(BasePcpHandlerTest, OnIncomingFrameChangesState) {
BwuManager bwu(m, em, ecm, {}, {});
MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
StartDiscovery(&client, &pcp_handler);
auto mediums = pcp_handler.GetDiscoveryMediums();
auto mediums = pcp_handler.GetDiscoveryMediums(&client);
auto connect_medium = mediums[mediums.size() - 1];
auto channel_pair = SetupConnection(pipe_a_, pipe_b_, connect_medium);
auto& channel_a = channel_pair.first;
@@ -574,7 +577,7 @@ TEST_P(BasePcpHandlerTest, DestructorIsCalledOnProtocolEndpoint) {
BwuManager bwu(m, em, ecm, {}, {});
MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
StartDiscovery(&client, &pcp_handler);
auto mediums = pcp_handler.GetDiscoveryMediums();
auto mediums = pcp_handler.GetDiscoveryMediums(&client);
auto connect_medium = mediums[mediums.size() - 1];
auto channel_pair = SetupConnection(pipe_a_, pipe_b_, connect_medium);
auto& channel_a = channel_pair.first;
@@ -616,7 +619,7 @@ TEST_P(BasePcpHandlerTest, MultipleMediumsProduceSingleEndpointLostEvent) {
BwuManager bwu(m, em, ecm, {}, {});
MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
StartDiscovery(&client, &pcp_handler);
auto mediums = pcp_handler.GetDiscoveryMediums();
auto mediums = pcp_handler.GetDiscoveryMediums(&client);
auto connect_medium = mediums[mediums.size() - 1];
auto channel_pair = SetupConnection(pipe_a_, pipe_b_, connect_medium);
auto& channel_a = channel_pair.first;
@@ -626,7 +629,7 @@ TEST_P(BasePcpHandlerTest, MultipleMediumsProduceSingleEndpointLostEvent) {
EXPECT_CALL(mock_discovery_listener_.endpoint_lost_cb, Call).Times(1);
RequestConnection(endpoint_id, std::move(channel_a), channel_b.get(),
&client, &pcp_handler, connect_medium, &destroyed_flag);
auto allowed_mediums = pcp_handler.GetDiscoveryMediums();
auto allowed_mediums = pcp_handler.GetDiscoveryMediums(&client);
mediums_count = allowed_mediums.size();
NEARBY_LOG(INFO, "Attempting to accept connection: id=%s",
endpoint_id.c_str());
@@ -657,9 +660,11 @@ TEST_F(BasePcpHandlerTest, InjectEndpoint) {
EndpointManager em(&ecm);
BwuManager bwu(m, em, ecm, {}, {});
MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
BooleanMediumSelector allowed{ .bluetooth = true, };
BooleanMediumSelector allowed{
.bluetooth = true,
};
ConnectionOptions options{
.allowed = allowed,
.allowed = allowed,
.is_out_of_band_connection = true,
};
EXPECT_CALL(mock_discovery_listener_.endpoint_found_cb, Call);
@@ -675,9 +680,8 @@ TEST_F(BasePcpHandlerTest, InjectEndpoint) {
EXPECT_CALL(pcp_handler, InjectEndpointImpl(&client, service_id, _))
.WillOnce(Invoke([&pcp_handler, &endpoint_id](
ClientProxy* client,
const std::string& service_id,
const OutOfBandConnectionMetadata& metadata) {
ClientProxy* client, const std::string& service_id,
const OutOfBandConnectionMetadata& metadata) {
pcp_handler.OnEndpointFound(
client,
std::make_shared<MockDiscoveredEndpoint>(MockDiscoveredEndpoint{
@@ -692,11 +696,12 @@ TEST_F(BasePcpHandlerTest, InjectEndpoint) {
}));
return Status{Status::kSuccess};
}));
pcp_handler.InjectEndpoint(&client, service_id,
OutOfBandConnectionMetadata{
.medium = Medium::BLUETOOTH,
.remote_bluetooth_mac_address = ByteArray(kFakeMacAddress),
});
pcp_handler.InjectEndpoint(
&client, service_id,
OutOfBandConnectionMetadata{
.medium = Medium::BLUETOOTH,
.remote_bluetooth_mac_address = ByteArray(kFakeMacAddress),
});
bwu.Shutdown();
}
cpp/core/internal/bluetooth_bwu_handler.cc
+2 -1
View File
@@ -110,7 +110,8 @@ BluetoothBwuHandler::CreateUpgradedEndpointChannel(
return nullptr;
}
BluetoothSocket socket = bluetooth_medium_.Connect(device, service_name);
BluetoothSocket socket = bluetooth_medium_.Connect(
device, service_name, client->GetCancellationFlag(endpoint_id));
if (!socket.IsValid()) {
return nullptr;
}
cpp/core/internal/bwu_manager.cc
-8
View File
@@ -32,7 +32,6 @@ namespace location {
namespace nearby {
namespace connections {
using ::location::nearby::proto::connections::ConnectionAttemptResult;
using ::location::nearby::proto::connections::DisconnectionReason;
// Required for C++ 14 support in Chrome
@@ -391,13 +390,6 @@ void BwuManager::ProcessBwuPathAvailableEvent(
auto channel = ProcessBwuPathAvailableEventInternal(client, endpoint_id,
upgrade_path_info);
ConnectionAttemptResult connectionAttemptResult;
if (channel != nullptr) {
connectionAttemptResult = ConnectionAttemptResult::RESULT_SUCCESS;
} else {
connectionAttemptResult = ConnectionAttemptResult::RESULT_ERROR;
}
if (channel == nullptr) {
RunUpgradeFailedProtocol(client, endpoint_id, upgrade_path_info);
return;
cpp/core/internal/client_proxy.cc
+60 -2
View File
@@ -68,9 +68,11 @@ void ClientProxy::Reset() {
void ClientProxy::StartedAdvertising(
const std::string& service_id, Strategy strategy,
const ConnectionListener& listener,
absl::Span<proto::connections::Medium> mediums) {
absl::Span<proto::connections::Medium> mediums,
const ConnectionOptions& advertising_options) {
MutexLock lock(&mutex_);
advertising_info_ = {service_id, listener};
advertising_options_ = advertising_options;
}
void ClientProxy::StoppedAdvertising() {
@@ -79,6 +81,7 @@ void ClientProxy::StoppedAdvertising() {
if (IsAdvertising()) {
advertising_info_.Clear();
}
// advertising_options_ is purposefully not cleared here.
ResetLocalEndpointIdIfNeeded();
}
@@ -103,9 +106,11 @@ std::string ClientProxy::GetServiceId() const {
void ClientProxy::StartedDiscovery(
const std::string& service_id, Strategy strategy,
const DiscoveryListener& listener,
absl::Span<proto::connections::Medium> mediums) {
absl::Span<proto::connections::Medium> mediums,
const ConnectionOptions& discovery_options) {
MutexLock lock(&mutex_);
discovery_info_ = DiscoveryInfo{service_id, listener};
discovery_options_ = discovery_options;
}
void ClientProxy::StoppedDiscovery() {
@@ -115,6 +120,7 @@ void ClientProxy::StoppedDiscovery() {
discovered_endpoint_ids_.clear();
discovery_info_.Clear();
}
// discovery_options_ is purposefully not cleared here.
ResetLocalEndpointIdIfNeeded();
}
@@ -202,6 +208,11 @@ void ClientProxy::OnConnectionInitiated(const std::string& endpoint_id,
// Note: we allow devices to connect to an advertiser even after it stops
// advertising, so no need to check IsAdvertising() here.
item.connection_listener.initiated_cb(endpoint_id, info);
if (info.is_incoming_connection) {
// Add CancellationFlag for advertisers once encryption succeeds.
AddCancellationFlag(endpoint_id);
}
}
void ClientProxy::OnConnectionAccepted(const std::string& endpoint_id) {
@@ -262,6 +273,8 @@ void ClientProxy::OnDisconnected(const std::string& endpoint_id, bool notify) {
connections_.erase(endpoint_id);
ResetLocalEndpointIdIfNeeded();
}
CancelEndpoint(endpoint_id);
}
bool ClientProxy::ConnectionStatusMatches(const std::string& endpoint_id,
@@ -457,6 +470,42 @@ bool ClientProxy::RemoteConnectionIsAccepted(std::string endpoint_id) const {
endpoint_id, ClientProxy::Connection::kRemoteEndpointAccepted);
}
void ClientProxy::AddCancellationFlag(const std::string& endpoint_id) {
auto item = cancellation_flags_.find(endpoint_id);
if (item != cancellation_flags_.end()) {
return;
}
cancellation_flags_.emplace(endpoint_id,
std::make_unique<CancellationFlag>());
}
CancellationFlag* ClientProxy::GetCancellationFlag(
const std::string& endpoint_id) {
const auto item = cancellation_flags_.find(endpoint_id);
if (item == cancellation_flags_.end()) {
return default_cancellation_flag_.get();
}
return item->second.get();
}
void ClientProxy::CancelEndpoint(const std::string& endpoint_id) {
const auto item = cancellation_flags_.find(endpoint_id);
if (item == cancellation_flags_.end()) return;
item->second->Cancel();
cancellation_flags_.erase(item);
}
void ClientProxy::CancelAllEndpoints() {
for (const auto& item : cancellation_flags_) {
CancellationFlag* cancellation_flag = item.second.get();
if (cancellation_flag->Cancelled()) {
continue;
}
cancellation_flag->Cancel();
}
cancellation_flags_.clear();
}
void ClientProxy::OnPayload(const std::string& endpoint_id, Payload payload) {
MutexLock lock(&mutex_);
@@ -507,6 +556,7 @@ void ClientProxy::RemoveAllEndpoints() {
// endpoint, in the case when this is called from stopAllEndpoints(). For now,
// just remove without notifying.
connections_.clear();
cancellation_flags_.clear();
local_endpoint_id_.clear();
}
@@ -535,6 +585,14 @@ void ClientProxy::AppendConnectionStatus(const std::string& endpoint_id,
}
}
ConnectionOptions ClientProxy::GetAdvertisingOptions() const {
return advertising_options_;
}
ConnectionOptions ClientProxy::GetDiscoveryOptions() const {
return discovery_options_;
}
} // namespace connections
} // namespace nearby
} // namespace location
cpp/core/internal/client_proxy.h
+39 -4
View File
@@ -24,6 +24,7 @@
#include "core/status.h"
#include "core/strategy.h"
#include "platform/base/byte_array.h"
#include "platform/base/cancellation_flag.h"
#include "platform/base/prng.h"
#include "platform/public/mutex.h"
#include "proto/connections_enums.pb.h"
@@ -59,7 +60,8 @@ class ClientProxy final {
void StartedAdvertising(
const std::string& service_id, Strategy strategy,
const ConnectionListener& connection_lifecycle_listener,
absl::Span<proto::connections::Medium> mediums);
absl::Span<proto::connections::Medium> mediums,
const ConnectionOptions& advertising_options = ConnectionOptions{});
// Marks this client as not advertising.
void StoppedAdvertising();
bool IsAdvertising() const;
@@ -70,9 +72,11 @@ class ClientProxy final {
std::string GetServiceId() const;
// Marks this client as discovering with the given callback.
void StartedDiscovery(const std::string& service_id, Strategy strategy,
const DiscoveryListener& discovery_listener,
absl::Span<proto::connections::Medium> mediums);
void StartedDiscovery(
const std::string& service_id, Strategy strategy,
const DiscoveryListener& discovery_listener,
absl::Span<proto::connections::Medium> mediums,
const ConnectionOptions& discovery_options = ConnectionOptions{});
// Marks this client as not discovering at all.
void StoppedDiscovery();
bool IsDiscoveringServiceId(const std::string& service_id) const;
@@ -153,6 +157,17 @@ class ClientProxy final {
bool LocalConnectionIsAccepted(std::string endpoint_id) const;
bool RemoteConnectionIsAccepted(std::string endpoint_id) const;
// Adds a CancellationFlag for endpoint id.
void AddCancellationFlag(const std::string& endpoint_id);
// Returns the CancellationFlag for endpoint id,
CancellationFlag* GetCancellationFlag(const std::string& endpoint_id);
// Sets the CancellationFlag to true for endpoint id.
void CancelEndpoint(const std::string& endpoint_id);
// Cancels all CancellationFlags.
void CancelAllEndpoints();
ConnectionOptions GetAdvertisingOptions() const;
ConnectionOptions GetDiscoveryOptions() const;
private:
struct Connection {
// Status: may be either:
@@ -221,6 +236,19 @@ class ClientProxy final {
// If not empty, we are currently discovering for the given service_id.
DiscoveryInfo discovery_info_;
// The active ClientProxy's advertising constraints. Empty()
// returns true if the client hasn't started advertising false otherwise.
// Note: this is not cleared when the client stops advertising because it
// might still be useful downstream of advertising (eg: establishing
// connections, performing bandwidth upgrades, etc.)
ConnectionOptions advertising_options_;
// The active ClientProxy's discovery constraints. Null if the client
// hasn't started discovering. Note: this is not cleared when the client
// stops discovering because it might still be useful downstream of
// discovery (eg: connection speed, etc.)
ConnectionOptions discovery_options_;
// Maps endpoint_id to endpoint connection state.
absl::flat_hash_map<std::string, Connection> connections_;
@@ -230,6 +258,13 @@ class ClientProxy final {
// happen because some mediums (like Bluetooth) repeatedly give us the same
// endpoints after each scan.
absl::flat_hash_set<std::string> discovered_endpoint_ids_;
// Maps endpoint_id to CancellationFlag.
absl::flat_hash_map<std::string, std::unique_ptr<CancellationFlag>>
cancellation_flags_;
// A default cancellation flag with isCancelled set be true.
std::unique_ptr<CancellationFlag> default_cancellation_flag_ =
std::make_unique<CancellationFlag>(true);
};
// Operator overloads when comparing Ptr<ClientProxy>.
cpp/core/internal/client_proxy_test.cc
+72 -4
View File
@@ -93,8 +93,7 @@ class ClientProxyTest : public testing::Test {
void OnDiscoveryEndpointFound(ClientProxy* client, const Endpoint& endpoint) {
EXPECT_CALL(mock_discovery_.endpoint_found_cb, Call).Times(1);
client->OnEndpointFound(service_id_, endpoint.id, endpoint.info,
medium_);
client->OnEndpointFound(service_id_, endpoint.id, endpoint.info, medium_);
}
void OnDiscoveryEndpointLost(ClientProxy* client, const Endpoint& endpoint) {
@@ -112,6 +111,8 @@ 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,
@@ -174,6 +175,8 @@ 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) {
@@ -235,8 +238,7 @@ TEST_F(ClientProxyTest, ClientIdIsUnique) {
}
TEST_F(ClientProxyTest, GeneratedEndpointIdIsUnique) {
EXPECT_NE(client1_.GetLocalEndpointId(),
client2_.GetLocalEndpointId());
EXPECT_NE(client1_.GetLocalEndpointId(), client2_.GetLocalEndpointId());
}
TEST_F(ClientProxyTest, ResetClearsState) {
@@ -369,6 +371,72 @@ 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/endpoint_manager.cc
-2
View File
@@ -310,8 +310,6 @@ EndpointManager::FrameProcessor* EndpointManager::GetFrameProcessor(
latch.CountDown();
});
latch.Await();
NEARBY_LOG(INFO, "GetFrameProcessor: type=%d; processor=%p", frame_type,
processor);
return processor;
}
cpp/core/internal/fuzzers/BUILD
+12
View File
@@ -0,0 +1,12 @@
load("//security/fuzzing/blaze:cc_fuzz_target.bzl", "cc_fuzz_target")
cc_fuzz_target(
name = "offline_frames_fuzzer",
srcs = ["offline_frames_fuzzer.cc"],
componentid = 148515,
deps = [
"//core/internal",
"//platform/base",
"//security/fuzzing/blaze:default_init_google_for_cc_fuzz_target",
],
)
cpp/core/internal/fuzzers/offline_frames_fuzzer.cc
+11
View File
@@ -0,0 +1,11 @@
#include "core/internal/offline_frames.h"
#include "platform/base/byte_array.h"
extern "C" int LLVMFuzzerTestOneInput(const uint8_t* data, size_t size) {
location::nearby::ByteArray byte_array;
byte_array.SetData(reinterpret_cast<const char*>(data), size);
location::nearby::connections::parser::FromBytes(byte_array);
return 0;
}
cpp/core/internal/mediums/BUILD
+1
View File
@@ -45,6 +45,7 @@ cc_library(
"//core/internal/mediums/webrtc",
"//proto/connections:offline_wire_formats_portable_proto",
"//platform/base",
"//platform/base:cancellation_flag",
"//platform/public:comm",
"//platform/public:logging",
"//platform/public:types",
cpp/core/internal/mediums/ble.cc
+3 -2
View File
@@ -304,8 +304,9 @@ bool Ble::IsAcceptingConnectionsLocked(const std::string& service_id) {
return accepting_connections_info_.Existed(service_id);
}
BleSocket Ble::Connect(BlePeripheral& peripheral,
const std::string& service_id) {
// TODO(b/169303284): Handles Cancellation and registration.
BleSocket Ble::Connect(BlePeripheral& peripheral, const std::string& service_id,
CancellationFlag* cancellation_flag) {
MutexLock lock(&mutex_);
NEARBY_LOGS(INFO) << "BLE::Connect: service=" << &peripheral;
// Socket to return. To allow for NRVO to work, it has to be a single object.
cpp/core/internal/mediums/ble.h
+3 -1
View File
@@ -21,6 +21,7 @@
#include "core/internal/mediums/bluetooth_radio.h"
#include "core/listeners.h"
#include "platform/base/byte_array.h"
#include "platform/base/cancellation_flag.h"
#include "platform/public/ble.h"
#include "platform/public/multi_thread_executor.h"
#include "platform/public/mutex.h"
@@ -98,7 +99,8 @@ class Ble {
// service_id. Blocks until connection is established, or server-side is
// terminated. Returns socket instance. On success, BleSocket.IsValid() return
// true.
BleSocket Connect(BlePeripheral& peripheral, const std::string& service_id)
BleSocket Connect(BlePeripheral& peripheral, const std::string& service_id,
CancellationFlag* cancellation_flag)
ABSL_LOCKS_EXCLUDED(mutex_);
private:
cpp/core/internal/mediums/ble_test.cc
+2 -4
View File
@@ -172,10 +172,8 @@ TEST_F(BleTest, CanStartAcceptingConnectionsAndConnect) {
EXPECT_TRUE(found_latch.Await(kWaitDuration).result());
ASSERT_TRUE(discovered_peripheral.IsValid());
BleSocket socket =
ble_b.Connect(discovered_peripheral, service_id);
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);
cpp/core/internal/mediums/bluetooth_classic.cc
+3 -1
View File
@@ -344,8 +344,10 @@ bool BluetoothClassic::StopAcceptingConnections(
return true;
}
// TODO(b/169303284): Handles Cancellation and registration.
BluetoothSocket BluetoothClassic::Connect(BluetoothDevice& bluetooth_device,
const std::string& service_name) {
const std::string& service_name,
CancellationFlag* cancellation_flag) {
for (int attempts_count = 0; attempts_count < kConnectAttemptsLimit;
attempts_count++) {
auto wrapper_result = AttemptToConnect(bluetooth_device, service_name);
cpp/core/internal/mediums/bluetooth_classic.h
+3 -1
View File
@@ -21,6 +21,7 @@
#include "core/internal/mediums/bluetooth_radio.h"
#include "core/listeners.h"
#include "platform/base/byte_array.h"
#include "platform/base/cancellation_flag.h"
#include "platform/public/bluetooth_adapter.h"
#include "platform/public/bluetooth_classic.h"
#include "platform/public/multi_thread_executor.h"
@@ -111,7 +112,8 @@ class BluetoothClassic {
// Returns socket instance. On success, BluetoothSocket.IsValid() return true.
// Called by client.
BluetoothSocket Connect(BluetoothDevice& bluetooth_device,
const std::string& service_name)
const std::string& service_name,
CancellationFlag* cancellation_flag)
ABSL_LOCKS_EXCLUDED(mutex_);
std::string GetMacAddress() const ABSL_LOCKS_EXCLUDED(mutex_);
cpp/core/internal/mediums/bluetooth_classic_test.cc
+2 -1
View File
@@ -194,8 +194,9 @@ TEST_F(BluetoothClassicTest, CanConnect) {
accept_latch.CountDown();
},
}));
CancellationFlag flag;
BluetoothSocket socket_for_client =
bt_client.Connect(discovered_device, std::string(kServiceName));
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());
cpp/core/internal/mediums/webrtc.cc
+2 -1
View File
@@ -208,7 +208,8 @@ void WebRtc::StopAcceptingConnections(const std::string& service_id) {
WebRtcSocketWrapper WebRtc::Connect(const std::string& service_id,
const PeerId& remote_peer_id,
const LocationHint& location_hint) {
const LocationHint& location_hint,
CancellationFlag* cancellation_flag) {
for (int attempts_count = 0; attempts_count < kConnectAttemptsLimit;
attempts_count++) {
auto wrapper_result =
cpp/core/internal/mediums/webrtc.h
+3 -1
View File
@@ -27,6 +27,7 @@
#include "proto/connections/offline_wire_formats.pb.h"
#include "proto/connections/offline_wire_formats.pb.h"
#include "platform/base/byte_array.h"
#include "platform/base/cancellation_flag.h"
#include "platform/base/listeners.h"
#include "platform/base/runnable.h"
#include "platform/public/atomic_boolean.h"
@@ -92,7 +93,8 @@ class WebRtc {
// Runs on @MainThread.
WebRtcSocketWrapper Connect(const std::string& service_id,
const PeerId& peer_id,
const LocationHint& location_hint)
const LocationHint& location_hint,
CancellationFlag* cancellation_flag)
ABSL_LOCKS_EXCLUDED(mutex_);
private:
cpp/core/internal/mediums/webrtc_test.cc
+17 -10
View File
@@ -75,8 +75,9 @@ TEST_F(WebRtcTest, Connect_DataChannelTimeOut) {
LocationHint location_hint;
ASSERT_TRUE(webrtc.IsAvailable());
CancellationFlag flag;
WebRtcSocketWrapper wrapper_1 =
webrtc.Connect(service_id, peer_id, location_hint);
webrtc.Connect(service_id, peer_id, location_hint, &flag);
EXPECT_FALSE(wrapper_1.IsValid());
EXPECT_TRUE(webrtc.StartAcceptingConnections(
@@ -99,8 +100,9 @@ TEST_F(WebRtcTest, StartAcceptingConnection_ThenConnect) {
ASSERT_TRUE(webrtc.StartAcceptingConnections(
service_id, self_id, location_hint,
{mock_accepted_callback_.AsStdFunction()}));
WebRtcSocketWrapper wrapper =
webrtc.Connect(service_id, PeerId("random_peer_id"), location_hint);
CancellationFlag flag;
WebRtcSocketWrapper wrapper = webrtc.Connect(
service_id, PeerId("random_peer_id"), location_hint, &flag);
EXPECT_TRUE(webrtc.IsAcceptingConnections(service_id));
EXPECT_FALSE(wrapper.IsValid());
EXPECT_FALSE(webrtc.StartAcceptingConnections(
@@ -150,7 +152,8 @@ TEST_F(WebRtcTest, ConnectTwice) {
device_c.StartAcceptingConnections(service_id, other_id, location_hint,
{[](WebRtcSocketWrapper wrapper) {}});
sender_socket = sender.Connect(service_id, self_id, location_hint);
CancellationFlag flag;
sender_socket = sender.Connect(service_id, self_id, location_hint, &flag);
EXPECT_TRUE(sender_socket.IsValid());
ExceptionOr<bool> devices_connected = connected.Get();
@@ -158,7 +161,7 @@ TEST_F(WebRtcTest, ConnectTwice) {
EXPECT_TRUE(devices_connected.result());
WebRtcSocketWrapper socket =
sender.Connect(service_id, other_id, location_hint);
sender.Connect(service_id, other_id, location_hint, &flag);
EXPECT_TRUE(socket.IsValid());
socket.Close();
@@ -192,7 +195,8 @@ TEST_F(WebRtcTest, ConnectBothDevicesAndAbort) {
connected.Set(receiver_socket.IsValid());
}});
sender_socket = sender.Connect(service_id, self_id, location_hint);
CancellationFlag flag;
sender_socket = sender.Connect(service_id, self_id, location_hint, &flag);
EXPECT_TRUE(sender_socket.IsValid());
ExceptionOr<bool> devices_connected = connected.Get();
@@ -220,7 +224,8 @@ TEST_F(WebRtcTest, ConnectBothDevicesAndSendData) {
connected.Set(receiver_socket.IsValid());
}});
sender_socket = sender.Connect(service_id, self_id, location_hint);
CancellationFlag flag;
sender_socket = sender.Connect(service_id, self_id, location_hint, &flag);
EXPECT_TRUE(sender_socket.IsValid());
ExceptionOr<bool> devices_connected = connected.Get();
@@ -254,7 +259,8 @@ TEST_F(WebRtcTest, ConnectBothDevices_ShutdownSignaling_SendData) {
connected.Set(receiver_socket.IsValid());
}});
sender_socket = sender.Connect(service_id, self_id, location_hint);
CancellationFlag flag;
sender_socket = sender.Connect(service_id, self_id, location_hint, &flag);
EXPECT_TRUE(sender_socket.IsValid());
ExceptionOr<bool> devices_connected = connected.Get();
@@ -285,8 +291,9 @@ TEST_F(WebRtcTest, Connect_NullPeerConnection) {
LocationHint location_hint;
ASSERT_TRUE(webrtc.IsAvailable());
WebRtcSocketWrapper wrapper =
webrtc.Connect(service_id, PeerId("random_peer_id"), location_hint);
CancellationFlag flag;
WebRtcSocketWrapper wrapper = webrtc.Connect(
service_id, PeerId("random_peer_id"), location_hint, &flag);
EXPECT_FALSE(wrapper.IsValid());
}
cpp/core/internal/mediums/wifi_lan.cc
+3 -1
View File
@@ -222,8 +222,10 @@ bool WifiLan::IsAcceptingConnectionsLocked(const std::string& service_id) {
return accepting_connections_info_.Existed(service_id);
}
// TODO(b/169303284): Handles Cancellation and registration.
WifiLanSocket WifiLan::Connect(WifiLanService& wifi_lan_service,
const std::string& service_id) {
const std::string& service_id,
CancellationFlag* cancellation_flag) {
MutexLock lock(&mutex_);
NEARBY_LOGS(INFO) << "WifiLan::Connect: wifi_lan_service="
<< &wifi_lan_service << ", service_info_name="
cpp/core/internal/mediums/wifi_lan.h
+3 -1
View File
@@ -19,6 +19,7 @@
#include <string>
#include "platform/base/byte_array.h"
#include "platform/base/cancellation_flag.h"
#include "platform/public/multi_thread_executor.h"
#include "platform/public/mutex.h"
#include "platform/public/wifi_lan.h"
@@ -81,7 +82,8 @@ class WifiLan {
// Blocks until connection is established, or server-side is terminated.
// Returns socket instance. On success, WifiLanSocket.IsValid() return true.
WifiLanSocket Connect(WifiLanService& wifi_lan_service,
const std::string& service_id)
const std::string& service_id,
CancellationFlag* cancellation_flag)
ABSL_LOCKS_EXCLUDED(mutex_);
WifiLanService GetRemoteWifiLanService(const std::string& ip_address,
cpp/core/internal/mediums/wifi_lan_test.cc
+2 -3
View File
@@ -159,10 +159,9 @@ TEST_F(WifiLanTest, CanStartAcceptingConnectionsAndConnect) {
EXPECT_TRUE(found_latch.Await(kWaitDuration).result());
ASSERT_TRUE(discovered_service.IsValid());
CancellationFlag flag;
WifiLanSocket socket =
wifi_lan_b.Connect(discovered_service, service_id);
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);
cpp/core/internal/p2p_cluster_pcp_handler.cc
+90 -8
View File
@@ -241,6 +241,83 @@ void P2pClusterPcpHandler::BluetoothDeviceDiscoveredHandler(
});
}
void P2pClusterPcpHandler::BluetoothNameChangedHandler(
ClientProxy* client, const std::string& service_id,
BluetoothDevice device) {
RunOnPcpHandlerThread([this, client, service_id, device]() {
// Make sure we are still discovering before proceeding.
if (!client->IsDiscovering()) {
NEARBY_LOG(INFO,
"BT discovery handler (CHANGED) [client=%p, service=%s]: not "
"in discovery mode",
client, service_id.c_str());
return;
}
// Parse the Bluetooth device name.
const std::string device_name_string = device.GetName();
BluetoothDeviceName device_name(device_name_string);
NEARBY_LOG(INFO,
"BT discovery handler (CHANGED) [client=%p, service=%s]: "
"processing new name %s",
client, service_id.c_str(), device_name_string.c_str());
// By this point, the BluetoothDevice passed to us has a different name than
// what we may have discovered before. We need to iterate over the found
// BluetoothEndpoints and compare their addresses to see the devices are the
// same. We are not guaranteed to discover a match, since the old name may
// not have been formatted for Nearby Connections.
for (auto endpoint :
GetDiscoveredEndpoints(proto::connections::Medium::BLUETOOTH)) {
BluetoothEndpoint* bluetoothEndpoint =
static_cast<BluetoothEndpoint*>(endpoint);
NEARBY_LOG(INFO,
"BT discovery handler (CHANGED) [client=%p, service=%s]: "
"comparing MAC addresses with existing endpoint %s. They have "
"MAC address %s and the new endpoint has MAC address %s.",
client, service_id.c_str(),
bluetoothEndpoint->bluetooth_device.GetName().c_str(),
bluetoothEndpoint->bluetooth_device.GetMacAddress().c_str(),
device.GetMacAddress().c_str());
if (bluetoothEndpoint->bluetooth_device.GetMacAddress() ==
device.GetMacAddress()) {
// Report the BluetoothEndpoint as lost to the client.
NEARBY_LOG(
INFO,
"BT discovery handler (LOST) [client=%p, service=%s]: report "
"to client",
client, service_id.c_str());
OnEndpointLost(client, *endpoint);
break;
}
}
// Make sure the Bluetooth device name points to a valid
// endpoint we're discovering.
if (!IsRecognizedBluetoothEndpoint(device_name_string, service_id,
device_name)) {
NEARBY_LOG(INFO,
"BT discovery handler (CHANGED) [client=%p, service=%s]: The "
"new name is not recognized. Ignoring.",
client, service_id.c_str());
return;
}
// Report the discovered endpoint to the client.
NEARBY_LOGS(INFO)
<< "Invoking BasePcpHandler::OnEndpointFound() for BT service="
<< service_id << "; id=" << device_name.GetEndpointId() << "; name="
<< absl::BytesToHexString(device_name.GetEndpointInfo().data());
OnEndpointFound(
client, std::make_shared<BluetoothEndpoint>(BluetoothEndpoint{
{device_name.GetEndpointId(), device_name.GetEndpointInfo(),
service_id, proto::connections::Medium::BLUETOOTH,
device_name.GetWebRtcState()},
device,
}));
});
}
void P2pClusterPcpHandler::BluetoothDeviceLostHandler(
ClientProxy* client, const std::string& service_id,
BluetoothDevice& device) {
@@ -264,7 +341,7 @@ void P2pClusterPcpHandler::BluetoothDeviceLostHandler(
device_name))
return;
// Report the discovered endpoint to the client.
// Report the BluetoothEndpoint as lost to the client.
NEARBY_LOG(INFO,
"BT discovery handler (LOST) [client=%p, service=%s]: report "
"to client",
@@ -590,7 +667,7 @@ BasePcpHandler::StartOperationResult P2pClusterPcpHandler::StartDiscoveryImpl(
&P2pClusterPcpHandler::BluetoothDeviceDiscoveredHandler, this,
client, service_id),
.device_name_changed_cb = absl::bind_front(
&P2pClusterPcpHandler::BluetoothDeviceDiscoveredHandler, this,
&P2pClusterPcpHandler::BluetoothNameChangedHandler, this,
client, service_id),
.device_lost_cb = absl::bind_front(
&P2pClusterPcpHandler::BluetoothDeviceLostHandler, this, client,
@@ -828,8 +905,9 @@ BasePcpHandler::ConnectImplResult P2pClusterPcpHandler::BluetoothConnectImpl(
ClientProxy* client, BluetoothEndpoint* endpoint) {
BluetoothDevice& device = endpoint->bluetooth_device;
BluetoothSocket bluetooth_socket =
bluetooth_medium_.Connect(device, endpoint->service_id);
BluetoothSocket bluetooth_socket = bluetooth_medium_.Connect(
device, endpoint->service_id,
client->GetCancellationFlag(endpoint->endpoint_id));
if (!bluetooth_socket.IsValid()) {
return BasePcpHandler::ConnectImplResult{
.status = {Status::kBluetoothError},
@@ -1009,7 +1087,9 @@ BasePcpHandler::ConnectImplResult P2pClusterPcpHandler::BleConnectImpl(
ClientProxy* client, BleEndpoint* endpoint) {
BlePeripheral& peripheral = endpoint->ble_peripheral;
BleSocket ble_socket = ble_medium_.Connect(peripheral, endpoint->service_id);
BleSocket ble_socket =
ble_medium_.Connect(peripheral, endpoint->service_id,
client->GetCancellationFlag(endpoint->endpoint_id));
if (!ble_socket.IsValid()) {
return BasePcpHandler::ConnectImplResult{
.status = {Status::kBleError},
@@ -1139,8 +1219,9 @@ BasePcpHandler::ConnectImplResult P2pClusterPcpHandler::WifiLanConnectImpl(
ClientProxy* client, WifiLanEndpoint* endpoint) {
WifiLanService& wifi_lan_service = endpoint->wifi_lan_service;
WifiLanSocket wifi_lan_socket =
wifi_lan_medium_.Connect(wifi_lan_service, endpoint->service_id);
WifiLanSocket wifi_lan_socket = wifi_lan_medium_.Connect(
wifi_lan_service, endpoint->service_id,
client->GetCancellationFlag(endpoint->endpoint_id));
if (!wifi_lan_socket.IsValid()) {
return BasePcpHandler::ConnectImplResult{
.status = {Status::kWifiLanError},
@@ -1204,7 +1285,8 @@ BasePcpHandler::ConnectImplResult P2pClusterPcpHandler::WebRtcConnectImpl(
std::string empty_country_code;
mediums::WebRtcSocketWrapper socket_wrapper = webrtc_medium_.Connect(
webrtc_endpoint->service_id, webrtc_endpoint->peer_id,
Utils::BuildLocationHint(empty_country_code));
Utils::BuildLocationHint(empty_country_code),
client->GetCancellationFlag(webrtc_endpoint->endpoint_id));
if (!socket_wrapper.IsValid()) {
return BasePcpHandler::ConnectImplResult{.status = {Status::kError}};
}
cpp/core/internal/p2p_cluster_pcp_handler.h
+3
View File
@@ -130,6 +130,9 @@ class P2pClusterPcpHandler : public BasePcpHandler {
void BluetoothDeviceDiscoveredHandler(ClientProxy* client,
const std::string& service_id,
BluetoothDevice device);
void BluetoothNameChangedHandler(ClientProxy* client,
const std::string& service_id,
BluetoothDevice device);
void BluetoothDeviceLostHandler(ClientProxy* client,
const std::string& service_id,
BluetoothDevice& device);
cpp/core/internal/service_controller_router.cc
+24 -2
View File
@@ -160,6 +160,10 @@ void ServiceControllerRouter::RequestConnection(
ClientProxy* client, absl::string_view endpoint_id,
const ConnectionRequestInfo& info, const ConnectionOptions& options,
const ResultCallback& callback) {
// Cancellations can be fired from clients anytime, need to add the
// CancellationListener as soon as possible.
client->AddCancellationFlag(std::string(endpoint_id));
RouteToServiceController([this, client,
endpoint_id = std::string(endpoint_id), info,
options, callback]() {
@@ -174,8 +178,12 @@ void ServiceControllerRouter::RequestConnection(
return;
}
callback.result_cb(service_controller_->RequestConnection(
client, endpoint_id, info, options));
Status status = service_controller_->RequestConnection(client, endpoint_id,
info, options);
if (!status.Ok()) {
client->CancelEndpoint(endpoint_id);
}
callback.result_cb(status);
});
}
@@ -213,6 +221,8 @@ void ServiceControllerRouter::AcceptConnection(ClientProxy* client,
void ServiceControllerRouter::RejectConnection(ClientProxy* client,
absl::string_view endpoint_id,
const ResultCallback& callback) {
client->CancelEndpoint(std::string(endpoint_id));
RouteToServiceController(
[this, client, endpoint_id = std::string(endpoint_id), callback]() {
if (!ClientHasAcquiredServiceController(client)) {
@@ -310,6 +320,10 @@ void ServiceControllerRouter::CancelPayload(ClientProxy* client,
void ServiceControllerRouter::DisconnectFromEndpoint(
ClientProxy* client, absl::string_view endpoint_id,
const ResultCallback& callback) {
// Client can emit the cancellation at anytime, we need to execute the request
// without further posting it.
client->CancelEndpoint(std::string(endpoint_id));
RouteToServiceController(
[this, client, endpoint_id = std::string(endpoint_id), callback]() {
if (ClientHasAcquiredServiceController(client)) {
@@ -326,6 +340,10 @@ void ServiceControllerRouter::DisconnectFromEndpoint(
void ServiceControllerRouter::StopAllEndpoints(ClientProxy* client,
const ResultCallback& callback) {
// Client can emit the cancellation at anytime, we need to execute the request
// without further posting it.
client->CancelAllEndpoints();
RouteToServiceController([this, client, callback]() {
if (ClientHasAcquiredServiceController(client)) {
DoneWithStrategySessionForClient(client);
@@ -336,6 +354,10 @@ void ServiceControllerRouter::StopAllEndpoints(ClientProxy* client,
void ServiceControllerRouter::ClientDisconnecting(
ClientProxy* client, const ResultCallback& callback) {
// Client can emit the cancellation at anytime, we need to execute the request
// without further posting it.
client->CancelAllEndpoints();
RouteToServiceController([this, client, callback]() {
if (ClientHasAcquiredServiceController(client)) {
DoneWithStrategySessionForClient(client);
cpp/core/internal/webrtc_bwu_handler.cc
+2 -1
View File
@@ -123,7 +123,8 @@ WebrtcBwuHandler::CreateUpgradedEndpointChannel(
peer_id.GetId().c_str(), location_hint.DebugString().c_str());
mediums::WebRtcSocketWrapper socket =
webrtc_.Connect(service_id, peer_id, location_hint);
webrtc_.Connect(service_id, peer_id, location_hint,
client->GetCancellationFlag(endpoint_id));
if (!socket.IsValid()) {
NEARBY_LOG(ERROR,
"WebRtcBwuHandler failed to connect to remote peer (%s) on "
cpp/core/internal/wifi_lan_bwu_handler.cc
+2 -2
View File
@@ -117,8 +117,8 @@ WifiLanBwuHandler::CreateUpgradedEndpointChannel(
if (!wifi_lan_service.IsValid()) {
return nullptr;
}
WifiLanSocket socket =
wifi_lan_medium_.Connect(wifi_lan_service, service_id);
WifiLanSocket socket = wifi_lan_medium_.Connect(
wifi_lan_service, service_id, client->GetCancellationFlag(endpoint_id));
if (!socket.IsValid()) {
return nullptr;
}
cpp/platform/base/BUILD
+33
View File
@@ -94,6 +94,28 @@ cc_library(
],
)
cc_library(
name = "cancellation_flag",
srcs = [
"cancellation_flag.cc",
],
hdrs = [
"cancellation_flag.h",
],
visibility = [
"//core/internal:__subpackages__",
"//platform/api:__subpackages__",
"//platform/impl:__subpackages__",
"//platform/public:__pkg__",
],
deps = [
":base",
":util",
"//absl/container:flat_hash_set",
"//absl/synchronization",
],
)
cc_library(
name = "test_util",
testonly = True,
@@ -143,6 +165,17 @@ cc_test(
],
)
cc_test(
name = "cancellation_flag_test",
srcs = [
"cancellation_flag_test.cc",
],
deps = [
":cancellation_flag",
"//testing/base/public:gunit_main",
],
)
cc_with_non_compile_test(
name = "exception_test",
srcs = [
cpp/platform/base/cancellation_flag.cc
+32
View File
@@ -0,0 +1,32 @@
#include "platform/base/cancellation_flag.h"
namespace location {
namespace nearby {
CancellationFlag::CancellationFlag() {
mutex_ = std::make_unique<absl::Mutex>();
}
CancellationFlag::CancellationFlag(bool cancelled) {
mutex_ = std::make_unique<absl::Mutex>();
cancelled_ = cancelled;
}
void CancellationFlag::Cancel() {
absl::MutexLock lock(mutex_.get());
if (cancelled_) {
// Someone already cancelled. Return immediately.
return;
}
cancelled_ = true;
}
bool CancellationFlag::Cancelled() const {
absl::MutexLock lock(mutex_.get());
return cancelled_;
}
} // namespace nearby
} // namespace location
cpp/platform/base/cancellation_flag.h
+37
View File
@@ -0,0 +1,37 @@
#ifndef PLATFORM_BASE_CANCELLATION_FLAG_H_
#define PLATFORM_BASE_CANCELLATION_FLAG_H_
#include <memory>
#include "absl/synchronization/mutex.h"
namespace location {
namespace nearby {
// A cancellation flag to mark an operation has been cancelled and should be
// cleaned up as soon as possible.
class CancellationFlag {
public:
CancellationFlag();
explicit CancellationFlag(bool cancelled);
CancellationFlag(const CancellationFlag &) = delete;
CancellationFlag &operator=(const CancellationFlag &) = delete;
CancellationFlag(CancellationFlag &&) = default;
CancellationFlag &operator=(CancellationFlag &&) = default;
virtual ~CancellationFlag() = default;
// Set the flag as cancelled.
void Cancel() ABSL_LOCKS_EXCLUDED(mutex_);
// Returns true if the flag has been set to cancelled.
bool Cancelled() const ABSL_LOCKS_EXCLUDED(mutex_);
private:
std::unique_ptr<absl::Mutex> mutex_;
bool cancelled_ ABSL_GUARDED_BY(mutex_) = false;
};
} // namespace nearby
} // namespace location
#endif // PLATFORM_BASE_CANCELLATION_FLAG_H_
cpp/platform/base/cancellation_flag_test.cc
+20
View File
@@ -0,0 +1,20 @@
#include "platform/base/cancellation_flag.h"
#include "gtest/gtest.h"
namespace location {
namespace nearby {
TEST(CancellationFlagTest, InitialValueIsFalse) {
CancellationFlag flag;
EXPECT_FALSE(flag.Cancelled());
}
TEST(CancellationFlagTest, CanCancel) {
CancellationFlag flag;
flag.Cancel();
EXPECT_TRUE(flag.Cancelled());
}
} // namespace nearby
} // namespace location
cpp/platform/impl/g3/webrtc.h
+1 -1
View File
@@ -41,7 +41,7 @@ class WebRtcSignalingMessenger : public api::WebRtcSignalingMessenger {
void StopReceivingMessages() override;
private:
absl::string_view self_id_;
std::string self_id_;
connections::LocationHint location_hint_;
};
cpp/platform/public/BUILD
-1
View File
@@ -52,7 +52,6 @@ cc_library(
"//platform/base:logging",
"//platform/base:util",
"//absl/base:core_headers",
"//absl/container:flat_hash_map",
"//absl/time",
],
)