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

Revert "Revert incoming connections listener changes for internal breakages."

Browse Source 
PiperOrigin-RevId: 542381558
This commit is contained in:
Anay Wadhera
2023-06-21 15:36:44 -07:00
committed by Copybara-Service
parent 5a2c972707
commit 5da4b33824
39 changed files with 1002 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
connections/core.cc
+4 -3
View File
@@ -341,9 +341,10 @@ void Core::StopDiscoveryV3(ResultCallback result_cb) {
void Core::StartListeningForIncomingConnectionsV3(
const v3::ConnectionListeningOptions& options, absl::string_view service_id,
v3::ConnectionListener listener_cb, ResultCallback result_cb) {
result_cb.result_cb(router_->StartListeningForIncomingConnectionsV3(
&client_, service_id, std::move(listener_cb), options));
v3::ConnectionListener listener_cb, v3::ListeningResultListener result_cb) {
router_->StartListeningForIncomingConnectionsV3(
&client_, service_id, std::move(listener_cb), options,
std::move(result_cb));
}
void Core::StopListeningForIncomingConnectionsV3() {
connections/core.h
+2 -1
View File
@@ -30,6 +30,7 @@
#include "connections/payload.h"
#include "connections/v3/connection_listening_options.h"
#include "connections/v3/listeners.h"
#include "connections/v3/listening_result.h"
#include "internal/analytics/event_logger.h"
#include "internal/interop/device.h"
#include "internal/interop/device_provider.h"
@@ -348,7 +349,7 @@ class Core {
void StartListeningForIncomingConnectionsV3(
const v3::ConnectionListeningOptions& options,
absl::string_view service_id, v3::ConnectionListener listener_cb,
ResultCallback result_cb);
v3::ListeningResultListener result_cb);
// Stops listening for incoming connections. Should be called after
// calling StartListeningForIncomingConnections.
connections/implementation/BUILD
+1
View File
@@ -123,6 +123,7 @@ cc_library(
"//internal/platform:base",
"//internal/platform:cancellation_flag",
"//internal/platform:comm",
"//internal/platform:connection_info",
"//internal/platform:error_code_recorder",
"//internal/platform:logging",
"//internal/platform:types",
connections/implementation/analytics/analytics_recorder.cc
+20
View File
@@ -212,6 +212,26 @@ void AnalyticsRecorder::OnStopDiscovery() {
RecordDiscoveryPhaseDurationLocked();
}
void AnalyticsRecorder::OnStartedIncomingConnectionListening(
connections::Strategy strategy) {
MutexLock lock(&mutex_);
if (!CanRecordAnalyticsLocked("OnStartedIncomingConnectionListening")) {
return;
}
UpdateStrategySessionLocked(strategy, ADVERTISER);
if (started_advertising_phase_time_ == absl::Now()) {
started_advertising_phase_time_ = SystemClock::ElapsedRealtime();
}
}
void AnalyticsRecorder::OnStoppedIncomingConnectionListening() {
MutexLock lock(&mutex_);
if (!CanRecordAnalyticsLocked("OnStoppedIncomingConnectionListening")) {
return;
}
RecordAdvertisingPhaseDurationLocked();
}
void AnalyticsRecorder::OnEndpointFound(Medium medium) {
MutexLock lock(&mutex_);
if (!CanRecordAnalyticsLocked("OnEndpointFound")) {
connections/implementation/analytics/analytics_recorder.h
+5
View File
@@ -57,6 +57,11 @@ class AnalyticsRecorder {
ABSL_LOCKS_EXCLUDED(mutex_);
void OnStopAdvertising() ABSL_LOCKS_EXCLUDED(mutex_);
// Connection listening
void OnStartedIncomingConnectionListening(connections::Strategy strategy)
ABSL_LOCKS_EXCLUDED(mutex_);
void OnStoppedIncomingConnectionListening() ABSL_LOCKS_EXCLUDED(mutex_);
// Discovery phase
void OnStartDiscovery(
connections::Strategy strategy,
connections/implementation/analytics/analytics_recorder_test.cc
+55
View File
@@ -25,6 +25,7 @@
#include "gmock/gmock.h"
#include "protobuf-matchers/protocol-buffer-matchers.h"
#include "gtest/gtest.h"
#include "absl/time/clock.h"
#include "absl/time/time.h"
#include "internal/analytics/event_logger.h"
#include "internal/platform/count_down_latch.h"
@@ -926,6 +927,60 @@ TEST(AnalyticsRecorderTest, UpgradeAttemptWorks) {
Partially(EqualsProto(strategy_session_proto)));
}
TEST(AnalyticsRecorderTest, StartListeningForIncomingConnectionsWorks) {
std::string endpoint_id = "endpoint_id";
std::string endpoint_id_1 = "endpoint_id_1";
std::string endpoint_id_2 = "endpoint_id_2";
std::string connection_token = "connection_token";
CountDownLatch client_session_done_latch(1);
FakeEventLogger event_logger(client_session_done_latch);
AnalyticsRecorder analytics_recorder(&event_logger);
analytics_recorder.OnStartedIncomingConnectionListening(
connections::Strategy::kP2pStar);
analytics_recorder.OnBandwidthUpgradeStarted(endpoint_id, BLE, WIFI_LAN,
INCOMING, connection_token);
analytics_recorder.OnBandwidthUpgradeStarted(
endpoint_id_1, BLUETOOTH, WIFI_LAN, INCOMING, connection_token);
// Error to upgrade.
analytics_recorder.OnBandwidthUpgradeError(endpoint_id, WIFI_LAN_MEDIUM_ERROR,
WIFI_LAN_SOCKET_CREATION);
// Success to upgrade.
analytics_recorder.OnBandwidthUpgradeSuccess(endpoint_id_1);
analytics_recorder.LogSession();
// ASSERT_TRUE(client_session_done_latch.Await(kDefaultTimeout).result());
ConnectionsLog::ClientSession strategy_session_proto =
ParseTextProtoOrDie(R"pb(
strategy_session <
strategy: P2P_STAR
role: ADVERTISER
upgrade_attempt <
direction: INCOMING
from_medium: BLE
to_medium: WIFI_LAN
upgrade_result: WIFI_LAN_MEDIUM_ERROR
error_stage: WIFI_LAN_SOCKET_CREATION
connection_token: "connection_token"
>
upgrade_attempt <
direction: INCOMING
from_medium: BLUETOOTH
to_medium: WIFI_LAN
upgrade_result: UPGRADE_RESULT_SUCCESS
error_stage: UPGRADE_SUCCESS
connection_token: "connection_token"
>
>)pb");
EXPECT_THAT(event_logger.GetLoggedClientSession(),
Partially(EqualsProto(strategy_session_proto)));
}
TEST(AnalyticsRecorderTest, SetErrorCodeFieldsCorrectly) {
CountDownLatch client_session_done_latch(1);
FakeEventLogger event_logger(client_session_done_latch);
connections/implementation/base_pcp_handler.cc
+75 -3
View File
@@ -33,11 +33,20 @@
#include "connections/implementation/offline_frames.h"
#include "connections/implementation/proto/offline_wire_formats.pb.h"
#include "connections/medium_selector.h"
#include "connections/status.h"
#include "connections/v3/connections_device.h"
#include "connections/v3/listeners.h"
#include "internal/flags/nearby_flags.h"
#include "internal/platform/base64_utils.h"
#include "internal/platform/bluetooth_connection_info.h"
#include "internal/platform/bluetooth_utils.h"
#include "internal/platform/cancelable_alarm.h"
#include "internal/platform/connection_info.h"
#include "internal/platform/count_down_latch.h"
#include "internal/platform/future.h"
#include "internal/platform/logging.h"
#include "internal/platform/wifi_lan_connection_info.h"
#include "proto/connections_enums.pb.h"
namespace nearby {
namespace connections {
@@ -88,6 +97,67 @@ void BasePcpHandler::DisconnectFromEndpointManager() {
this);
}
std::pair<Status, std::vector<ConnectionInfoVariant>>
BasePcpHandler::StartListeningForIncomingConnections(
ClientProxy* client, absl::string_view service_id,
v3::ConnectionListeningOptions options,
v3::ConnectionListener connection_listener) {
Future<std::pair<Status, std::vector<ConnectionInfoVariant>>> response;
RunOnPcpHandlerThread(
"start-listening-for-incoming-conn",
[this, client, service_id, options, &response,
connection_listener = std::move(
connection_listener)]() RUN_ON_PCP_HANDLER_THREAD() mutable {
StartOperationResult result = StartListeningForIncomingConnectionsImpl(
client, service_id, client->GetLocalEndpointId(), options);
if (!result.status.Ok()) {
response.Set({result.status, {}});
return;
}
client->StartedListeningForIncomingConnections(
service_id, GetStrategy(), std::move(connection_listener), options);
response.Set(
{result.status, GetConnectionInfoFromResult(service_id, result)});
});
return response.Get().GetResult();
}
std::vector<ConnectionInfoVariant> BasePcpHandler::GetConnectionInfoFromResult(
absl::string_view service_id, StartOperationResult result) {
std::vector<ConnectionInfoVariant> connection_infos;
for (const auto& medium : result.mediums) {
if (medium == location::nearby::proto::connections::BLUETOOTH) {
BluetoothConnectionInfo info(
mediums_->GetBluetoothClassic().GetMacAddress(), "", {});
connection_infos.push_back(info);
} else if (medium == location::nearby::proto::connections::BLE) {
// TODO(b/284311319): Add relevant information.
BleConnectionInfo info("", "", "", {});
connection_infos.push_back(info);
} else if (medium == location::nearby::proto::connections::WIFI_LAN) {
std::pair<std::string, int> ip_port_pair =
mediums_->GetWifiLan().GetCredentials(std::string(service_id));
WifiLanConnectionInfo info(
ip_port_pair.first,
absl::StrCat(absl::Hex(ip_port_pair.second, absl::kZeroPad16)), "",
{});
connection_infos.push_back(info);
}
}
return connection_infos;
}
void BasePcpHandler::StopListeningForIncomingConnections(ClientProxy* client) {
CountDownLatch latch(1);
RunOnPcpHandlerThread("stop-listening-for-incoming-conn",
[this, client, &latch]() RUN_ON_PCP_HANDLER_THREAD() {
StopListeningForIncomingConnectionsImpl(client);
client->StoppedListeningForIncomingConnections();
latch.CountDown();
});
WaitForLatch("StopListeningForIncomingConnections", &latch);
}
Status BasePcpHandler::StartAdvertising(
ClientProxy* client, const std::string& service_id,
const AdvertisingOptions& advertising_options,
@@ -1262,12 +1332,13 @@ Exception BasePcpHandler::OnIncomingConnection(
// Fixes an NPE in ClientProxy.OnConnectionAccepted. The crash happened when
// the client stopped advertising and we nulled out state, followed by an
// incoming connection where we attempted to check that state.
if (!client->IsAdvertising()) {
if (!client->IsAdvertising() &&
!client->IsListeningForIncomingConnections()) {
NEARBY_LOGS(WARNING) << "Ignoring incoming connection on medium "
<< location::nearby::proto::connections::Medium_Name(
channel->GetMedium())
<< " because client=" << client->GetClientId()
<< " is no longer advertising.";
<< " is no longer waiting for incoming connections.";
return {Exception::kIo};
}
@@ -1385,7 +1456,8 @@ Exception BasePcpHandler::OnIncomingConnection(
pendingConnectionInfo.nonce = connection_request.nonce();
pendingConnectionInfo.is_incoming = true;
pendingConnectionInfo.start_time = start_time;
pendingConnectionInfo.listener = advertising_listener_;
pendingConnectionInfo.listener =
client->GetAdvertisingOrIncomingConnectionListener();
pendingConnectionInfo.connection_options = connection_options;
pendingConnectionInfo.supported_mediums =
parser::ConnectionRequestMediumsToMediums(connection_request);
connections/implementation/base_pcp_handler.h
+22
View File
@@ -36,9 +36,11 @@
#include "connections/listeners.h"
#include "connections/medium_selector.h"
#include "connections/status.h"
#include "connections/v3/listeners.h"
#include "internal/platform/atomic_boolean.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/cancelable_alarm.h"
#include "internal/platform/connection_info.h"
#include "internal/platform/count_down_latch.h"
#include "internal/platform/future.h"
#include "internal/platform/prng.h"
@@ -77,6 +79,14 @@ class BasePcpHandler : public PcpHandler,
BasePcpHandler(BasePcpHandler&&) = delete;
BasePcpHandler& operator=(BasePcpHandler&&) = delete;
std::pair<Status, std::vector<ConnectionInfoVariant>>
StartListeningForIncomingConnections(
ClientProxy* client, absl::string_view service_id,
v3::ConnectionListeningOptions options,
v3::ConnectionListener connection_listener) override;
void StopListeningForIncomingConnections(ClientProxy* client) override;
// Starts advertising. Once successfully started, changes ClientProxy's state.
// Notifies ConnectionListener (info.listener) in case of any event.
// See
@@ -274,6 +284,14 @@ class BasePcpHandler : public PcpHandler,
virtual Status StopDiscoveryImpl(ClientProxy* client)
RUN_ON_PCP_HANDLER_THREAD() = 0;
virtual StartOperationResult StartListeningForIncomingConnectionsImpl(
ClientProxy* client_proxy, absl::string_view service_id,
absl::string_view local_endpoint_id,
v3::ConnectionListeningOptions options) RUN_ON_PCP_HANDLER_THREAD() = 0;
virtual void StopListeningForIncomingConnectionsImpl(ClientProxy* client)
RUN_ON_PCP_HANDLER_THREAD() = 0;
virtual Status InjectEndpointImpl(ClientProxy* client,
const std::string& service_id,
const OutOfBandConnectionMetadata& metadata)
@@ -309,6 +327,10 @@ class BasePcpHandler : public PcpHandler,
absl::string_view endpoint_id,
location::nearby::proto::connections::Medium medium);
// Returns a vector of ConnectionInfos generated from a StartOperationResult.
std::vector<ConnectionInfoVariant> GetConnectionInfoFromResult(
absl::string_view service_id, StartOperationResult result);
mediums::WebrtcPeerId CreatePeerIdFromAdvertisement(
const string& service_id, const string& endpoint_id,
const ByteArray& endpoint_info);
connections/implementation/base_pcp_handler_test.cc
+133
View File
@@ -31,6 +31,9 @@
#include "connections/implementation/proto/offline_wire_formats.pb.h"
#include "connections/listeners.h"
#include "connections/params.h"
#include "connections/status.h"
#include "connections/strategy.h"
#include "connections/v3/connection_listening_options.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/exception.h"
#include "internal/platform/medium_environment.h"
@@ -153,6 +156,13 @@ class MockPcpHandler : public BasePcpHandler {
const DiscoveryOptions& discovery_options),
(override));
MOCK_METHOD(Status, StopDiscoveryImpl, (ClientProxy * client), (override));
MOCK_METHOD(StartOperationResult, StartListeningForIncomingConnectionsImpl,
(ClientProxy * client_proxy, absl::string_view service_id,
absl::string_view local_endpoint_id,
v3::ConnectionListeningOptions options),
(override));
MOCK_METHOD(void, StopListeningForIncomingConnectionsImpl,
(ClientProxy * client_proxy), (override));
MOCK_METHOD(Status, InjectEndpointImpl,
(ClientProxy * client, const std::string& service_id,
const OutOfBandConnectionMetadata& metadata),
@@ -220,6 +230,12 @@ class MockPcpHandler : public BasePcpHandler {
GetMediumsFromSelector(BooleanMediumSelector allowed) {
return allowed.GetMediums(true);
}
std::vector<ConnectionInfoVariant> GetConnectionInfoFromResult(
absl::string_view service_id,
BasePcpHandler::StartOperationResult result) {
return BasePcpHandler::GetConnectionInfoFromResult(service_id, result);
}
};
class MockContext {
@@ -1181,6 +1197,123 @@ TEST_F(BasePcpHandlerTest, TestEndpointFoundStopsAlarm) {
env_.Stop();
}
TEST_P(BasePcpHandlerTest, TestGetConnectionInfosFromMediums) {
env_.Start();
std::string service_id{"service"};
Mediums mediums;
EndpointChannelManager endpoint_channel_manager;
EndpointManager endpoint_manager(&endpoint_channel_manager);
BwuManager bwu_manager(mediums, endpoint_manager, endpoint_channel_manager,
{}, {});
MockPcpHandler pcp_handler(&mediums, &endpoint_manager,
&endpoint_channel_manager, &bwu_manager);
BooleanMediumSelector selector = GetParam();
// Flip on a medium we should not get info for.
selector.web_rtc = true;
std::vector<ConnectionInfoVariant> infos =
pcp_handler.GetConnectionInfoFromResult(
service_id, {.mediums = selector.GetMediums(true)});
// Make sure we don't count webrtc.
EXPECT_EQ(infos.size(), selector.Count(true) - 1);
env_.Stop();
}
TEST_F(BasePcpHandlerTest, TestCanStartListeningForIncomingConnections) {
env_.Start();
ClientProxy client;
Mediums mediums;
EndpointChannelManager endpoint_channel_manager;
EndpointManager endpoint_manager(&endpoint_channel_manager);
BwuManager bwu_manager(mediums, endpoint_manager, endpoint_channel_manager,
{}, {});
MockPcpHandler pcp_handler(&mediums, &endpoint_manager,
&endpoint_channel_manager, &bwu_manager);
EXPECT_CALL(pcp_handler, StartListeningForIncomingConnectionsImpl)
.Times(1)
.WillOnce(Return(
MockPcpHandler::StartOperationResult{.status = {Status::kSuccess}}));
v3::ConnectionListeningOptions options = {.strategy = Strategy::kP2pCluster,
.enable_ble_listening = true,
.enable_bluetooth_listening = true,
.enable_wlan_listening = true};
pcp_handler.StartListeningForIncomingConnections(&client, "service_id",
options, {});
EXPECT_TRUE(client.IsListeningForIncomingConnections());
}
TEST_F(BasePcpHandlerTest, TestStartListeningForIncomingConnectionsBadStatus) {
env_.Start();
ClientProxy client;
Mediums m;
EndpointChannelManager ecm;
EndpointManager em(&ecm);
BwuManager bwu(m, em, ecm, {}, {});
MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
EXPECT_CALL(pcp_handler, StartListeningForIncomingConnectionsImpl)
.Times(1)
.WillOnce(Return(MockPcpHandler::StartOperationResult{
.status = {Status::kAlreadyListening}}));
v3::ConnectionListeningOptions options = {.strategy = Strategy::kP2pCluster,
.enable_ble_listening = true,
.enable_bluetooth_listening = true,
.enable_wlan_listening = true};
pcp_handler.StartListeningForIncomingConnections(&client, "service_id",
options, {});
EXPECT_FALSE(client.IsListeningForIncomingConnections());
}
TEST_F(BasePcpHandlerTest, TestCanStopListeningForIncomingConnections) {
env_.Start();
std::string service_id{"service"};
std::string endpoint_id{"ABCD"};
ClientProxy client;
Mediums m;
EndpointChannelManager ecm;
EndpointManager em(&ecm);
BwuManager bwu(m, em, ecm, {}, {});
MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
EXPECT_CALL(pcp_handler, StartListeningForIncomingConnectionsImpl)
.Times(1)
.WillOnce(Return(
MockPcpHandler::StartOperationResult{.status = {Status::kSuccess}}));
EXPECT_CALL(pcp_handler, StopListeningForIncomingConnectionsImpl).Times(1);
v3::ConnectionListeningOptions options = {.strategy = Strategy::kP2pCluster,
.enable_ble_listening = true,
.enable_bluetooth_listening = true,
.enable_wlan_listening = true};
pcp_handler.StartListeningForIncomingConnections(&client, service_id, options,
{});
pcp_handler.StopListeningForIncomingConnections(&client);
EXPECT_FALSE(client.IsListeningForIncomingConnections());
}
TEST_F(BasePcpHandlerTest,
TestWifiLanStopListeningForIncomingConnectionsSuccessWhenStopped) {
env_.Start();
std::string service_id{"service"};
std::string endpoint_id{"ABCD"};
ClientProxy client;
Mediums m;
EndpointChannelManager ecm;
EndpointManager em(&ecm);
BwuManager bwu(m, em, ecm, {}, {});
MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
EXPECT_CALL(pcp_handler, StartListeningForIncomingConnectionsImpl)
.Times(1)
.WillOnce(Return(
MockPcpHandler::StartOperationResult{.status = {Status::kSuccess}}));
EXPECT_CALL(pcp_handler, StopListeningForIncomingConnectionsImpl).Times(1);
v3::ConnectionListeningOptions options = {.strategy = Strategy::kP2pCluster,
.enable_ble_listening = true,
.enable_bluetooth_listening = true,
.enable_wlan_listening = true};
pcp_handler.StartListeningForIncomingConnections(&client, service_id, options,
{});
m.GetWifiLan().StopAcceptingConnections(service_id);
pcp_handler.StopListeningForIncomingConnections(&client);
EXPECT_FALSE(client.IsListeningForIncomingConnections());
}
} // namespace
} // namespace connections
} // namespace nearby
connections/implementation/client_proxy.cc
+87
View File
@@ -28,6 +28,8 @@
#include "absl/container/flat_hash_set.h"
#include "absl/strings/escaping.h"
#include "absl/strings/str_format.h"
#include "connections/v3/bandwidth_info.h"
#include "connections/v3/connection_listening_options.h"
#include "connections/v3/connections_device_provider.h"
#include "internal/analytics/event_logger.h"
#include "internal/platform/error_code_recorder.h"
@@ -182,6 +184,87 @@ std::string ClientProxy::GetAdvertisingServiceId() const {
return advertising_info_.service_id;
}
void ClientProxy::StartedListeningForIncomingConnections(
absl::string_view service_id, Strategy strategy,
v3::ConnectionListener listener,
const v3::ConnectionListeningOptions& options) {
MutexLock lock(&mutex_);
listening_options_ = options;
listening_info_ = ListeningInfo{
.service_id = std::string(service_id),
.listener = std::move(listener),
};
analytics_recorder_->OnStartedIncomingConnectionListening(strategy);
}
void ClientProxy::StoppedListeningForIncomingConnections() {
MutexLock lock(&mutex_);
listening_info_.Clear();
analytics_recorder_->OnStoppedIncomingConnectionListening();
}
bool ClientProxy::IsListeningForIncomingConnections() const {
MutexLock lock(&mutex_);
return !listening_info_.IsEmpty();
}
std::string ClientProxy::GetListeningForIncomingConnectionsServiceId() const {
MutexLock lock(&mutex_);
if (IsListeningForIncomingConnections()) {
return listening_info_.service_id;
}
return "";
}
ConnectionListener ClientProxy::GetAdvertisingOrIncomingConnectionListener() {
if (IsListeningForIncomingConnections()) {
ConnectionListener listener = {
.initiated_cb =
[this](const std::string& endpoint_id,
const ConnectionResponseInfo& info) {
auto remote_device = v3::ConnectionsDevice(
endpoint_id, info.remote_endpoint_info.AsStringView(), {});
this->listening_info_.listener.initiated_cb(
remote_device,
v3::InitialConnectionInfo{
.authentication_digits = info.authentication_token,
.raw_authentication_token =
info.raw_authentication_token.string_data(),
.is_incoming_connection = info.is_incoming_connection,
});
},
.accepted_cb =
[this](const std::string& endpoint_id) {
auto remote_device = v3::ConnectionsDevice(endpoint_id, "", {});
this->listening_info_.listener.result_cb(
remote_device,
v3::ConnectionResult{.status = Status{
.value = Status::kSuccess,
}});
},
.rejected_cb =
[this](const std::string& endpoint_id, Status status) {
auto remote_device = v3::ConnectionsDevice(endpoint_id, "", {});
this->listening_info_.listener.result_cb(
remote_device, v3::ConnectionResult{.status = status});
},
.disconnected_cb =
[this](const std::string& endpoint_id) {
auto remote_device = v3::ConnectionsDevice(endpoint_id, "", {});
this->listening_info_.listener.disconnected_cb(remote_device);
},
.bandwidth_changed_cb =
[this](const std::string& endpoint_id, Medium medium) {
auto remote_device = v3::ConnectionsDevice(endpoint_id, "", {});
this->listening_info_.listener.bandwidth_changed_cb(
remote_device, v3::BandwidthInfo{.medium = medium});
},
};
return listener;
}
return advertising_info_.listener;
}
void ClientProxy::StartedDiscovery(
const std::string& service_id, Strategy strategy,
const DiscoveryListener& listener,
@@ -797,6 +880,10 @@ DiscoveryOptions ClientProxy::GetDiscoveryOptions() const {
return discovery_options_;
}
v3::ConnectionListeningOptions ClientProxy::GetListeningOptions() const {
return listening_options_;
}
void ClientProxy::EnterHighVisibilityMode() {
MutexLock lock(&mutex_);
NEARBY_LOGS(INFO) << "ClientProxy [EnterHighVisibilityMode]: client="
connections/implementation/client_proxy.h
+27
View File
@@ -30,6 +30,8 @@
#include "connections/listeners.h"
#include "connections/status.h"
#include "connections/strategy.h"
#include "connections/v3/connection_listening_options.h"
#include "connections/v3/listeners.h"
#include "internal/analytics/event_logger.h"
#include "internal/interop/device.h"
#include "internal/interop/device_provider.h"
@@ -91,6 +93,17 @@ class ClientProxy final {
bool IsAdvertising() const;
std::string GetAdvertisingServiceId() const;
// Marks this client as listening for incoming connections.
void StartedListeningForIncomingConnections(
absl::string_view service_id, Strategy strategy,
v3::ConnectionListener listener,
const v3::ConnectionListeningOptions& options);
void StoppedListeningForIncomingConnections();
bool IsListeningForIncomingConnections() const;
std::string GetListeningForIncomingConnectionsServiceId() const;
ConnectionListener GetAdvertisingOrIncomingConnectionListener();
// Marks this client as discovering with the given callback.
void StartedDiscovery(
const std::string& service_id, Strategy strategy,
@@ -200,6 +213,7 @@ class ClientProxy final {
void CancelAllEndpoints();
AdvertisingOptions GetAdvertisingOptions() const;
DiscoveryOptions GetDiscoveryOptions() const;
v3::ConnectionListeningOptions GetListeningOptions() const;
// The endpoint id will be stable for 30 seconds after high visibility mode
// (high power and Bluetooth Classic) advertisement stops.
@@ -269,6 +283,13 @@ class ClientProxy final {
bool IsEmpty() const { return service_id.empty(); }
};
struct ListeningInfo {
std::string service_id;
v3::ConnectionListener listener;
void Clear() { service_id.clear(); }
bool IsEmpty() const { return service_id.empty(); }
};
// `RemoveAllEndpoints` is expected to only be called during destruction of
// ClientProxy via `ClientProxy::Reset`, which makes destroying
// CancellationFlags safe here since we are destroying ClientProxy. Do not
@@ -327,6 +348,9 @@ class ClientProxy final {
// If not empty, we are currently discovering for the given service_id.
DiscoveryInfo discovery_info_;
// If not empty, we are currently listening for the given service_id.
ListeningInfo listening_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
@@ -340,6 +364,9 @@ class ClientProxy final {
// discovery (eg: connection speed, etc.)
DiscoveryOptions discovery_options_;
// The active ClientProxy's listening constraints.
v3::ConnectionListeningOptions listening_options_;
// Maps endpoint_id to endpoint connection state.
absl::flat_hash_map<std::string, ConnectionPair> connections_;
connections/implementation/client_proxy_test.cc
+40
View File
@@ -29,10 +29,12 @@
#include "absl/types/span.h"
#include "connections/listeners.h"
#include "connections/strategy.h"
#include "connections/v3/bandwidth_info.h"
#include "connections/v3/connections_device_provider.h"
#include "internal/analytics/event_logger.h"
#include "internal/interop/device_provider.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/count_down_latch.h"
#include "internal/platform/feature_flags.h"
#include "internal/platform/medium_environment.h"
@@ -1023,6 +1025,44 @@ TEST_F(ClientProxyTest, TestGetSetLocalEndpointInfo) {
EXPECT_EQ(client1_.GetLocalEndpointInfo(), "endpoint_info");
}
TEST_F(ClientProxyTest, TestGetIncomingConnectionListener) {
CountDownLatch result_latch(2);
CountDownLatch bwu_latch(1);
CountDownLatch disconnect_latch(1);
CountDownLatch init_latch(1);
client1_.StartedListeningForIncomingConnections(
service_id_, Strategy::kP2pCluster,
{
.initiated_cb =
[&init_latch](const NearbyDevice&,
const v3::InitialConnectionInfo&) {
init_latch.CountDown();
},
.result_cb = [&result_latch](
const NearbyDevice&,
v3::ConnectionResult) { result_latch.CountDown(); },
.disconnected_cb =
[&disconnect_latch](const NearbyDevice&) {
disconnect_latch.CountDown();
},
.bandwidth_changed_cb =
[&bwu_latch](const NearbyDevice&, v3::BandwidthInfo) {
bwu_latch.CountDown();
},
},
{});
auto listener = client1_.GetAdvertisingOrIncomingConnectionListener();
listener.accepted_cb("endpoint-id");
listener.initiated_cb("endpoint-id", {.is_incoming_connection = false});
listener.disconnected_cb("endpoint-id");
listener.rejected_cb("endpoint-id", {Status::Value::kConnectionRejected});
listener.bandwidth_changed_cb("endpoint-id", Medium::WIFI_LAN);
EXPECT_TRUE(result_latch.Await().Ok());
EXPECT_TRUE(init_latch.Await().Ok());
EXPECT_TRUE(bwu_latch.Await().Ok());
EXPECT_TRUE(disconnect_latch.Await().Ok());
}
} // namespace
} // namespace connections
} // namespace nearby
connections/implementation/mock_service_controller.h
+5 -1
View File
@@ -15,6 +15,9 @@
#ifndef CORE_INTERNAL_MOCK_SERVICE_CONTROLLER_H_
#define CORE_INTERNAL_MOCK_SERVICE_CONTROLLER_H_
#include <utility>
#include <vector>
#include "gmock/gmock.h"
#include "connections/implementation/service_controller.h"
#include "connections/v3/connection_listening_options.h"
@@ -53,7 +56,8 @@ class MockServiceController : public ServiceController {
const OutOfBandConnectionMetadata& metadata),
(override));
MOCK_METHOD(Status, StartListeningForIncomingConnections,
MOCK_METHOD((std::pair<Status, std::vector<ConnectionInfoVariant>>),
StartListeningForIncomingConnections,
(ClientProxy * client, absl::string_view service_id,
v3::ConnectionListener listener,
const v3::ConnectionListeningOptions& options),
connections/implementation/offline_service_controller.cc
+14
View File
@@ -69,6 +69,20 @@ void OfflineServiceController::StopDiscovery(ClientProxy* client) {
pcp_manager_.StopDiscovery(client);
}
std::pair<Status, std::vector<ConnectionInfoVariant>>
OfflineServiceController::StartListeningForIncomingConnections(
ClientProxy* client, absl::string_view service_id,
v3::ConnectionListener listener,
const v3::ConnectionListeningOptions& options) {
return pcp_manager_.StartListeningForIncomingConnections(
client, service_id, std::move(listener), options);
}
void OfflineServiceController::StopListeningForIncomingConnections(
ClientProxy* client) {
pcp_manager_.StopListeningForIncomingConnections(client);
}
void OfflineServiceController::InjectEndpoint(
ClientProxy* client, const std::string& service_id,
const OutOfBandConnectionMetadata& metadata) {
connections/implementation/offline_service_controller.h
+4 -8
View File
@@ -55,17 +55,13 @@ class OfflineServiceController : public ServiceController {
void InjectEndpoint(ClientProxy* client, const std::string& service_id,
const OutOfBandConnectionMetadata& metadata) override;
Status StartListeningForIncomingConnections(
std::pair<Status, std::vector<ConnectionInfoVariant>>
StartListeningForIncomingConnections(
ClientProxy* client, absl::string_view service_id,
v3::ConnectionListener listener,
const v3::ConnectionListeningOptions& options) override {
// TODO(b/283823898): Implement.
return Status{.value = Status::kError};
}
const v3::ConnectionListeningOptions& options) override;
void StopListeningForIncomingConnections(ClientProxy* client) override {
// TODO(b/283823898): Implement.
}
void StopListeningForIncomingConnections(ClientProxy* client) override;
Status RequestConnection(
ClientProxy* client, const std::string& endpoint_id,
connections/implementation/p2p_cluster_pcp_handler.cc
+122
View File
@@ -31,8 +31,11 @@
#include "connections/implementation/flags/nearby_connections_feature_flags.h"
#include "connections/implementation/mediums/utils.h"
#include "connections/implementation/wifi_lan_endpoint_channel.h"
#include "connections/medium_selector.h"
#include "connections/power_level.h"
#include "connections/status.h"
#include "internal/flags/nearby_flags.h"
#include "internal/platform/logging.h"
#include "internal/platform/nsd_service_info.h"
#include "internal/platform/types.h"
#include "proto/connections_enums.pb.h"
@@ -1140,6 +1143,125 @@ BasePcpHandler::ConnectImplResult P2pClusterPcpHandler::ConnectImpl(
};
}
BasePcpHandler::StartOperationResult
P2pClusterPcpHandler::StartListeningForIncomingConnectionsImpl(
ClientProxy* client_proxy, absl::string_view service_id,
absl::string_view local_endpoint_id,
v3::ConnectionListeningOptions options) {
std::vector<Medium> started_mediums;
if (options.enable_bluetooth_listening &&
!bluetooth_medium_.IsAcceptingConnections(std::string(service_id))) {
if (!bluetooth_medium_.StartAcceptingConnections(
std::string(service_id),
{.accepted_cb = absl::bind_front(
&P2pClusterPcpHandler::BluetoothConnectionAcceptedHandler,
this, client_proxy, local_endpoint_id)})) {
NEARBY_LOGS(WARNING)
<< "Failed to start listening for incoming connections on Bluetooth";
} else {
started_mediums.push_back(
location::nearby::proto::connections::BLUETOOTH);
}
}
// ble
if (NearbyFlags::GetInstance().GetBoolFlag(
config_package_nearby::nearby_connections_feature::kEnableBleV2)) {
// ble_v2
if (options.enable_ble_listening &&
!ble_v2_medium_.IsAcceptingConnections(std::string(service_id))) {
if (!ble_v2_medium_.StartAcceptingConnections(
std::string(service_id),
{.accepted_cb = absl::bind_front(
&P2pClusterPcpHandler::BleV2ConnectionAcceptedHandler, this,
client_proxy, local_endpoint_id)})) {
NEARBY_LOGS(WARNING)
<< "Failed to start listening for incoming connections on ble_v2";
} else {
started_mediums.push_back(location::nearby::proto::connections::BLE);
}
}
} else {
// ble v1
if (options.enable_ble_listening &&
!ble_medium_.IsAcceptingConnections(std::string(service_id))) {
if (!ble_medium_.StartAcceptingConnections(
std::string(service_id),
{.accepted_cb = absl::bind_front(
&P2pClusterPcpHandler::BleConnectionAcceptedHandler, this,
client_proxy, local_endpoint_id)})) {
NEARBY_LOGS(WARNING)
<< "Failed to start listening for incoming connections on ble";
} else {
started_mediums.push_back(location::nearby::proto::connections::BLE);
}
}
}
if (options.enable_wlan_listening &&
!wifi_lan_medium_.IsAcceptingConnections(std::string(service_id))) {
if (!wifi_lan_medium_.StartAcceptingConnections(
std::string(service_id),
{.accepted_cb = absl::bind_front(
&P2pClusterPcpHandler::WifiLanConnectionAcceptedHandler, this,
client_proxy, local_endpoint_id, "")})) {
NEARBY_LOGS(WARNING)
<< "Failed to start listening for incoming connections on wifi_lan";
} else {
started_mediums.push_back(location::nearby::proto::connections::WIFI_LAN);
}
}
if (started_mediums.empty()) {
NEARBY_LOGS(WARNING) << absl::StrFormat(
"Failed StartListeningForIncomingConnectionsImpl() for client %d for "
"service_id %s",
client_proxy->GetClientId(), service_id);
return StartOperationResult{
.status = {Status::kError},
};
}
return BasePcpHandler::StartOperationResult{
.status = {Status::kSuccess}, .mediums = std::move(started_mediums)};
}
void P2pClusterPcpHandler::StopListeningForIncomingConnectionsImpl(
ClientProxy* client) {
if (wifi_lan_medium_.IsAcceptingConnections(
client->GetListeningForIncomingConnectionsServiceId())) {
if (!wifi_lan_medium_.StopAcceptingConnections(
client->GetListeningForIncomingConnectionsServiceId())) {
NEARBY_LOGS(WARNING)
<< "Unable to stop wifi lan from accepting connections.";
}
}
if (bluetooth_medium_.IsAcceptingConnections(
client->GetListeningForIncomingConnectionsServiceId())) {
if (!bluetooth_medium_.StopAcceptingConnections(
client->GetListeningForIncomingConnectionsServiceId())) {
NEARBY_LOGS(WARNING)
<< "Unable to stop bluetooth medium from accepting connections.";
}
}
if (NearbyFlags::GetInstance().GetBoolFlag(
config_package_nearby::nearby_connections_feature::kEnableBleV2)) {
if (ble_v2_medium_.IsAcceptingConnections(
client->GetListeningForIncomingConnectionsServiceId())) {
if (!ble_v2_medium_.StopAcceptingConnections(
client->GetListeningForIncomingConnectionsServiceId())) {
NEARBY_LOGS(WARNING)
<< "Unable to stop ble_v2 medium from accepting connections.";
}
}
} else {
if (ble_medium_.IsAcceptingConnections(
client->GetListeningForIncomingConnectionsServiceId())) {
if (!ble_medium_.StopAcceptingConnections(
client->GetListeningForIncomingConnectionsServiceId())) {
NEARBY_LOGS(WARNING)
<< "Unable to stop ble medium from accepting connections.";
}
}
}
}
void P2pClusterPcpHandler::BluetoothConnectionAcceptedHandler(
ClientProxy* client, absl::string_view local_endpoint_info,
const std::string& service_id, BluetoothSocket socket) {
connections/implementation/p2p_cluster_pcp_handler.h
+9
View File
@@ -92,6 +92,15 @@ class P2pClusterPcpHandler : public BasePcpHandler {
ClientProxy* client,
BasePcpHandler::DiscoveredEndpoint* endpoint) override;
// @PCPHandlerThread
BasePcpHandler::StartOperationResult StartListeningForIncomingConnectionsImpl(
ClientProxy* client_proxy, absl::string_view service_id,
absl::string_view local_endpoint_id,
v3::ConnectionListeningOptions options) override;
// @PCPHandlerThread
void StopListeningForIncomingConnectionsImpl(ClientProxy* client) override;
private:
// Holds the state required to re-create a BleEndpoint we see on a
// BlePeripheral, so BlePeripheralLostHandler can call
connections/implementation/p2p_cluster_pcp_handler_test.cc
+106
View File
@@ -24,6 +24,7 @@
#include "connections/implementation/bwu_manager.h"
#include "connections/implementation/flags/nearby_connections_feature_flags.h"
#include "connections/implementation/injected_bluetooth_device_store.h"
#include "connections/v3/connection_listening_options.h"
#include "internal/flags/nearby_flags.h"
#include "internal/platform/count_down_latch.h"
#include "internal/platform/logging.h"
@@ -312,6 +313,111 @@ TEST_P(P2pClusterPcpHandlerTest, CanConnect) {
env_.Stop();
}
TEST_P(P2pClusterPcpHandlerTest, CanStartListeningForIncomingConnections) {
env_.Start();
std::string endpoint_name_a{"endpoint_name"};
Mediums mediums_a;
BluetoothRadio& radio_a = mediums_a.GetBluetoothRadio();
radio_a.GetBluetoothAdapter().SetName("BT Device A");
EndpointChannelManager ecm_a;
EndpointManager em_a(&ecm_a);
BwuManager bwu_a(mediums_a, em_a, ecm_a, {},
{.allow_upgrade_to = {.bluetooth = true}});
InjectedBluetoothDeviceStore ibds_a;
P2pClusterPcpHandler handler_a(&mediums_a, &em_a, &ecm_a, &bwu_a, ibds_a);
v3::ConnectionListeningOptions v3_options{
.strategy = Strategy::kP2pCluster,
.enable_ble_listening = true,
.enable_bluetooth_listening = true,
.enable_wlan_listening = true,
};
// make sure mediums are not accepting before calling handler.
ASSERT_FALSE(
mediums_a.GetBluetoothClassic().IsAcceptingConnections(service_id_));
ASSERT_FALSE(mediums_a.GetWifiLan().IsAcceptingConnections(service_id_));
if (std::get<1>(GetParam())) {
ASSERT_FALSE(mediums_a.GetBleV2().IsAcceptingConnections(service_id_));
} else {
ASSERT_FALSE(mediums_a.GetBle().IsAcceptingConnections(service_id_));
}
// call handler.
auto result = handler_a.StartListeningForIncomingConnections(
&client_a_, service_id_, v3_options, {});
// now check to make sure we are in fact accepting connections.
EXPECT_TRUE(
mediums_a.GetBluetoothClassic().IsAcceptingConnections(service_id_));
EXPECT_TRUE(mediums_a.GetWifiLan().IsAcceptingConnections(service_id_));
if (std::get<1>(GetParam())) {
EXPECT_TRUE(mediums_a.GetBleV2().IsAcceptingConnections(service_id_));
} else {
EXPECT_TRUE(mediums_a.GetBle().IsAcceptingConnections(service_id_));
}
EXPECT_EQ(result.second.size(), 3);
ASSERT_TRUE(client_a_.IsListeningForIncomingConnections());
EXPECT_EQ(client_a_.GetListeningForIncomingConnectionsServiceId(),
service_id_);
EXPECT_EQ(client_a_.GetListeningOptions().enable_ble_listening,
v3_options.enable_ble_listening);
EXPECT_EQ(client_a_.GetListeningOptions().enable_bluetooth_listening,
v3_options.enable_bluetooth_listening);
EXPECT_EQ(client_a_.GetListeningOptions().enable_wlan_listening,
v3_options.enable_wlan_listening);
EXPECT_EQ(client_a_.GetListeningOptions().strategy, v3_options.strategy);
env_.Stop();
}
TEST_P(P2pClusterPcpHandlerTest, CanStopListeningForIncomingConnections) {
env_.Start();
std::string endpoint_name_a{"endpoint_name"};
Mediums mediums_a;
BluetoothRadio& radio_a = mediums_a.GetBluetoothRadio();
radio_a.GetBluetoothAdapter().SetName("BT Device A");
EndpointChannelManager ecm_a;
EndpointManager em_a(&ecm_a);
BwuManager bwu_a(mediums_a, em_a, ecm_a, {},
{.allow_upgrade_to = {.bluetooth = true}});
InjectedBluetoothDeviceStore ibds_a;
P2pClusterPcpHandler handler_a(&mediums_a, &em_a, &ecm_a, &bwu_a, ibds_a);
v3::ConnectionListeningOptions v3_options{
.strategy = Strategy::kP2pCluster,
.enable_ble_listening = true,
.enable_bluetooth_listening = true,
.enable_wlan_listening = true,
};
// make sure mediums are not accepting before calling handler.
ASSERT_FALSE(
mediums_a.GetBluetoothClassic().IsAcceptingConnections(service_id_));
ASSERT_FALSE(mediums_a.GetWifiLan().IsAcceptingConnections(service_id_));
if (std::get<1>(GetParam())) {
ASSERT_FALSE(mediums_a.GetBleV2().IsAcceptingConnections(service_id_));
} else {
ASSERT_FALSE(mediums_a.GetBle().IsAcceptingConnections(service_id_));
}
// call handler.
auto result = handler_a.StartListeningForIncomingConnections(
&client_a_, service_id_, v3_options, {});
// now check to make sure we are in fact accepting connections.
ASSERT_TRUE(
mediums_a.GetBluetoothClassic().IsAcceptingConnections(service_id_));
ASSERT_TRUE(mediums_a.GetWifiLan().IsAcceptingConnections(service_id_));
if (std::get<1>(GetParam())) {
ASSERT_TRUE(mediums_a.GetBleV2().IsAcceptingConnections(service_id_));
} else {
ASSERT_TRUE(mediums_a.GetBle().IsAcceptingConnections(service_id_));
}
// stop.
handler_a.StopListeningForIncomingConnections(&client_a_);
EXPECT_FALSE(
mediums_a.GetBluetoothClassic().IsAcceptingConnections(service_id_));
EXPECT_FALSE(mediums_a.GetWifiLan().IsAcceptingConnections(service_id_));
if (std::get<1>(GetParam())) {
EXPECT_FALSE(mediums_a.GetBleV2().IsAcceptingConnections(service_id_));
} else {
EXPECT_FALSE(mediums_a.GetBle().IsAcceptingConnections(service_id_));
}
env_.Stop();
}
INSTANTIATE_TEST_SUITE_P(ParametrisedPcpHandlerTest, P2pClusterPcpHandlerTest,
::testing::Combine(::testing::ValuesIn(kTestCases),
::testing::Bool()));
connections/implementation/pcp_handler.h
+9
View File
@@ -15,6 +15,7 @@
#ifndef CORE_INTERNAL_PCP_HANDLER_H_
#define CORE_INTERNAL_PCP_HANDLER_H_
#include <utility>
#include <vector>
#include "connections/implementation/client_proxy.h"
@@ -87,6 +88,14 @@ class PcpHandler {
// otherwise do nothing.
virtual void StopDiscovery(ClientProxy* client) = 0;
virtual std::pair<Status, std::vector<ConnectionInfoVariant>>
StartListeningForIncomingConnections(
ClientProxy* client, absl::string_view service_id,
v3::ConnectionListeningOptions options,
v3::ConnectionListener connection_listener) = 0;
virtual void StopListeningForIncomingConnections(ClientProxy* client) = 0;
// If Discovery is active with is_out_of_band_connection == true, invoke the
// callback with the provided endpoint info.
virtual void InjectEndpoint(ClientProxy* client,
connections/implementation/pcp_manager.cc
+20
View File
@@ -14,6 +14,8 @@
#include "connections/implementation/pcp_manager.h"
#include <vector>
#include "connections/implementation/p2p_cluster_pcp_handler.h"
#include "connections/implementation/p2p_point_to_point_pcp_handler.h"
#include "connections/implementation/p2p_star_pcp_handler.h"
@@ -87,6 +89,24 @@ void PcpManager::StopDiscovery(ClientProxy* client) {
}
}
std::pair<Status, std::vector<ConnectionInfoVariant>>
PcpManager::StartListeningForIncomingConnections(
ClientProxy* client, absl::string_view service_id,
v3::ConnectionListener listener,
const v3::ConnectionListeningOptions& options) {
if (!SetCurrentPcpHandler(options.strategy)) {
return {{Status::kError}, {}};
}
return {current_->StartListeningForIncomingConnections(
client, service_id, options, std::move(listener))};
}
void PcpManager::StopListeningForIncomingConnections(ClientProxy* client) {
if (current_) {
current_->StopListeningForIncomingConnections(client);
}
}
void PcpManager::InjectEndpoint(ClientProxy* client,
const std::string& service_id,
const OutOfBandConnectionMetadata& metadata) {
connections/implementation/pcp_manager.h
+8
View File
@@ -57,6 +57,14 @@ class PcpManager {
DiscoveryListener listener);
void StopDiscovery(ClientProxy* client);
std::pair<Status, std::vector<ConnectionInfoVariant>>
StartListeningForIncomingConnections(
ClientProxy* client, absl::string_view service_id,
v3::ConnectionListener listener,
const v3::ConnectionListeningOptions& options);
void StopListeningForIncomingConnections(ClientProxy* client);
void InjectEndpoint(ClientProxy* client, const std::string& service_id,
const OutOfBandConnectionMetadata& metadata);
connections/implementation/pcp_manager_test.cc
+37
View File
@@ -23,6 +23,8 @@
#include "absl/time/time.h"
#include "connections/implementation/endpoint_channel_manager.h"
#include "connections/implementation/simulation_user.h"
#include "connections/medium_selector.h"
#include "connections/v3/connection_listening_options.h"
#include "internal/platform/count_down_latch.h"
#include "internal/platform/medium_environment.h"
@@ -165,6 +167,41 @@ TEST_P(PcpManagerTest, CanReject) {
env_.Stop();
}
TEST_P(PcpManagerTest, CanStartListeningForIncomingConnections) {
env_.Start();
BooleanMediumSelector selector = GetParam();
SimulationUser user_a(kDeviceA, selector);
CountDownLatch start_latch(1);
v3::ConnectionListeningOptions options;
options.enable_ble_listening = selector.ble;
options.enable_bluetooth_listening = selector.bluetooth;
options.enable_wlan_listening = selector.wifi_lan;
options.listening_mediums = selector.GetMediums(true);
options.upgrade_mediums = selector.GetMediums(true);
options.strategy = Strategy::kP2pCluster;
user_a.StartListeningForIncomingConnections(&start_latch, "service", options,
{Status::kSuccess});
user_a.Stop();
env_.Stop();
}
TEST_P(PcpManagerTest, StartListeningForIncomingConnectionsFailsNoStrategy) {
env_.Start();
BooleanMediumSelector selector = GetParam();
SimulationUser user_a(kDeviceA, selector);
CountDownLatch start_latch(1);
v3::ConnectionListeningOptions options;
options.enable_ble_listening = selector.ble;
options.enable_bluetooth_listening = selector.bluetooth;
options.enable_wlan_listening = selector.wifi_lan;
options.listening_mediums = selector.GetMediums(true);
options.upgrade_mediums = selector.GetMediums(true);
user_a.StartListeningForIncomingConnections(&start_latch, "service", options,
{Status::kError});
user_a.Stop();
env_.Stop();
}
INSTANTIATE_TEST_SUITE_P(ParametrisedPcpManagerTest, PcpManagerTest,
::testing::ValuesIn(kTestCases));
connections/implementation/service_controller.h
+3 -1
View File
@@ -17,6 +17,7 @@
#include <cstdint>
#include <string>
#include <utility>
#include <vector>
#include "connections/advertising_options.h"
@@ -79,7 +80,8 @@ class ServiceController {
const std::string& service_id,
const OutOfBandConnectionMetadata& metadata) = 0;
virtual Status StartListeningForIncomingConnections(
virtual std::pair<Status, std::vector<ConnectionInfoVariant>>
StartListeningForIncomingConnections(
ClientProxy* client, absl::string_view service_id,
v3::ConnectionListener listener,
const v3::ConnectionListeningOptions& options) = 0;
connections/implementation/service_controller_router.cc
+32 -5
View File
@@ -28,6 +28,7 @@
#include "connections/v3/bandwidth_info.h"
#include "connections/v3/connection_result.h"
#include "connections/v3/connections_device.h"
#include "connections/v3/listening_result.h"
#include "internal/platform/logging.h"
// TODO(b/285657711): Add tests for uncovered logic, even if trivial.
@@ -345,17 +346,43 @@ void ServiceControllerRouter::DisconnectFromEndpoint(
});
}
Status ServiceControllerRouter::StartListeningForIncomingConnectionsV3(
void ServiceControllerRouter::StartListeningForIncomingConnectionsV3(
ClientProxy* client, absl::string_view service_id,
v3::ConnectionListener listener,
const v3::ConnectionListeningOptions& options) {
return GetServiceController()->StartListeningForIncomingConnections(
client, service_id, std::move(listener), options);
const v3::ConnectionListeningOptions& options,
v3::ListeningResultListener callback) {
RouteToServiceController(
"scr-start-listening-for-incoming-connections",
[this, client, callback = std::move(callback), service_id,
listener = std::move(listener), options]() mutable {
if (client->IsListeningForIncomingConnections()) {
callback({{Status::kAlreadyListening},
{
.endpoint_id = client->GetLocalEndpointId(),
.connection_info = {},
}});
return;
}
auto pair =
GetServiceController()->StartListeningForIncomingConnections(
client, service_id, std::move(listener), options);
v3::ListeningResult result = {
.endpoint_id = client->GetLocalEndpointId(),
.connection_info = pair.second,
};
callback(std::make_pair(pair.first, result));
});
}
void ServiceControllerRouter::StopListeningForIncomingConnectionsV3(
ClientProxy* client) {
GetServiceController()->StopListeningForIncomingConnections(client);
RouteToServiceController(
"scr-stop-listening-for-incoming-connections", [this, client]() {
if (!client->IsListeningForIncomingConnections()) {
return;
}
GetServiceController()->StopListeningForIncomingConnections(client);
});
}
void ServiceControllerRouter::RequestConnectionV3(
connections/implementation/service_controller_router.h
+5 -2
View File
@@ -24,9 +24,11 @@
#include "absl/types/span.h"
#include "connections/implementation/client_proxy.h"
#include "connections/implementation/service_controller.h"
#include "connections/listeners.h"
#include "connections/params.h"
#include "connections/v3/connection_listening_options.h"
#include "connections/v3/listeners.h"
#include "connections/v3/listening_result.h"
#include "connections/v3/params.h"
#include "internal/interop/device.h"
#include "internal/platform/runnable.h"
@@ -116,10 +118,11 @@ class ServiceControllerRouter {
const ResultCallback& callback);
////////////////////////////// V3 ////////////////////////////////////////////
virtual Status StartListeningForIncomingConnectionsV3(
virtual void StartListeningForIncomingConnectionsV3(
ClientProxy* client, absl::string_view service_id,
v3::ConnectionListener listener,
const v3::ConnectionListeningOptions& options);
const v3::ConnectionListeningOptions& options,
v3::ListeningResultListener callback);
virtual void StopListeningForIncomingConnectionsV3(ClientProxy* client);
connections/implementation/service_controller_router_test.cc
+70
View File
@@ -30,8 +30,10 @@
#include "connections/listeners.h"
#include "connections/params.h"
#include "connections/v3/bandwidth_info.h"
#include "connections/v3/connection_listening_options.h"
#include "connections/v3/connection_result.h"
#include "connections/v3/connections_device.h"
#include "connections/v3/listening_result.h"
#include "connections/v3/params.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/condition_variable.h"
@@ -423,6 +425,34 @@ class ServiceControllerRouterTest : public testing::Test {
EXPECT_FALSE(client->IsConnectedToEndpoint(kRemoteDevice.GetEndpointId()));
}
void StartListeningForIncomingConnectionsV3(
ClientProxy* client, absl::string_view service_id,
v3::ConnectionListener listener,
const v3::ConnectionListeningOptions& options,
v3::ListeningResultListener result_listener, bool expecting_call = true) {
if (expecting_call) {
EXPECT_CALL(*mock_, StartListeningForIncomingConnections)
.Times(1)
.WillOnce([](ClientProxy* client, absl::string_view service_id,
v3::ConnectionListener,
const v3::ConnectionListeningOptions& options) {
client->StartedListeningForIncomingConnections(
service_id, options.strategy, {}, options);
return std::pair<Status, std::vector<ConnectionInfoVariant>>{
Status{Status::kSuccess}, {}};
});
}
{
MutexLock lock(&mutex_);
complete_ = false;
router_.StartListeningForIncomingConnectionsV3(
client, service_id, std::move(listener), options,
std::move(result_listener));
while (!complete_) cond_.Wait();
EXPECT_TRUE(client->IsListeningForIncomingConnections());
}
}
protected:
const ResultCallback kCallback{
.result_cb =
@@ -845,6 +875,46 @@ TEST_F(ServiceControllerRouterTest, CancelPayloadV3Called) {
CancelPayloadV3(&client_, kRemoteDevice, kPayloadId, kCallback);
}
TEST_F(ServiceControllerRouterTest,
StartListeningForIncomingConnectionsCalledV3) {
StartListeningForIncomingConnectionsV3(
&client_, kServiceId, {}, {},
[this](std::pair<Status, v3::ListeningResult> result) {
EXPECT_TRUE(result.first.Ok());
{
MutexLock lock(&mutex_);
complete_ = true;
}
cond_.Notify();
});
}
TEST_F(ServiceControllerRouterTest,
StartListeningForIncomingConnectionsCalledV3TwiceFails) {
StartListeningForIncomingConnectionsV3(
&client_, kServiceId, {}, {},
[this](std::pair<Status, v3::ListeningResult> result) {
EXPECT_TRUE(result.first.Ok());
{
MutexLock lock(&mutex_);
complete_ = true;
}
cond_.Notify();
});
StartListeningForIncomingConnectionsV3(
&client_, kServiceId, {}, {},
[this](std::pair<Status, v3::ListeningResult> result) {
EXPECT_EQ(result.first.value, Status::kAlreadyListening);
{
MutexLock lock(&mutex_);
complete_ = true;
}
cond_.Notify();
},
/*expecting_call=*/false);
}
} // namespace
} // namespace connections
} // namespace nearby
connections/implementation/simulation_user.cc
+10
View File
@@ -174,5 +174,15 @@ void SimulationUser::RejectConnection(CountDownLatch* latch) {
EXPECT_TRUE(mgr_.RejectConnection(&client_, discovered_.endpoint_id).Ok());
}
void SimulationUser::StartListeningForIncomingConnections(
CountDownLatch* latch, absl::string_view service_id,
const v3::ConnectionListeningOptions& options, Status expected_status) {
auto result = mgr_.StartListeningForIncomingConnections(
&client_, service_id, /*listener=*/{}, options);
latch->CountDown();
NEARBY_LOGS(INFO) << "status: " << result.first.ToString();
EXPECT_EQ(expected_status, result.first);
}
} // namespace connections
} // namespace nearby
connections/implementation/simulation_user.h
+6
View File
@@ -110,6 +110,12 @@ class SimulationUser {
// callback.
void RejectConnection(CountDownLatch* latch);
// Calls PcpManager::StartListeningForIncomingConnections.
// If latch is provided, latch->CountDown() will be called on call completion.
void StartListeningForIncomingConnections(
CountDownLatch* latch, absl::string_view service_id,
const v3::ConnectionListeningOptions& options, Status expected_status);
// Unlike acceptance, rejection does not have to be mutual, in order to work.
// This method will allow to synchronize on the remote rejection, without
// performing a local rejection.
connections/status.cc
+4
View File
@@ -33,6 +33,8 @@ std::string Status::ToString() const {
return "kAlreadyAdvertising";
case Status::kAlreadyDiscovering:
return "kAlreadyDiscovering";
case Status::kAlreadyListening:
return "kAlreadyListening";
case Status::kEndpointIoError:
return "kEndpointIoError";
case Status::kEndpointUnknown:
@@ -51,6 +53,8 @@ std::string Status::ToString() const {
return "kWifiLanError";
case Status::kPayloadUnknown:
return "kPayloadUnknown";
default:
return "Unknown";
}
}
connections/status.h
+8
View File
@@ -22,6 +22,7 @@ namespace connections {
// Protocol operation result: kSuccess, if operation was successful;
// descriptive error code otherwise.
// LINT.IfChange
struct Status {
// Status is a struct, so it is possible to pass some context about failure,
// by adding extra fields to it when necessary, and not change any of the
@@ -33,6 +34,7 @@ struct Status {
kAlreadyHaveActiveStrategy,
kAlreadyAdvertising,
kAlreadyDiscovering,
kAlreadyListening,
kEndpointIoError,
kEndpointUnknown,
kConnectionRejected,
@@ -42,6 +44,7 @@ struct Status {
kBleError,
kWifiLanError,
kPayloadUnknown,
kNextValue,
};
Value value{kError};
bool Ok() const { return value == kSuccess; }
@@ -49,6 +52,11 @@ struct Status {
// Converts the status to a logging-friendly string.
std::string ToString() const;
};
// LINT.ThenChange(
// //depot/google3/location/nearby/cpp/sharing/implementation/nearby_connections_manager.cc:24
// //depot/google3/location/nearby/cpp/sharing/implementation/nearby_connections_types.h:46
// //depot/google3/location/nearby/cpp/sharing/implementation/nearby_connections_types_test.cc
// )
inline bool operator==(const Status& a, const Status& b) {
return a.value == b.value;
connections/status_test.cc
+1
View File
@@ -70,6 +70,7 @@ std::vector<StatusToStringData> GetTestData() {
"kAlreadyHaveActiveStrategy"},
{Status{.value = Status::kAlreadyAdvertising}, "kAlreadyAdvertising"},
{Status{.value = Status::kAlreadyDiscovering}, "kAlreadyDiscovering"},
{Status{.value = Status::kAlreadyListening}, "kAlreadyListening"},
{Status{.value = Status::kEndpointIoError}, "kEndpointIoError"},
{Status{.value = Status::kEndpointUnknown}, "kEndpointUnknown"},
{Status{.value = Status::kConnectionRejected}, "kConnectionRejected"},
connections/swift/NearbyCoreAdapter/Sources/GNCCoreAdapter.mm
+4
View File
@@ -67,6 +67,8 @@ GNCStatus GNCStatusFromCppStatus(Status status) {
return GNCStatusAlreadyAdvertising;
case Status::kAlreadyDiscovering:
return GNCStatusAlreadyDiscovering;
case Status::kAlreadyListening:
return GNCStatusAlreadyListening;
case Status::kEndpointIoError:
return GNCStatusEndpointIoError;
case Status::kEndpointUnknown:
@@ -85,6 +87,8 @@ GNCStatus GNCStatusFromCppStatus(Status status) {
return GNCStatusWifiLanError;
case Status::kPayloadUnknown:
return GNCStatusPayloadUnknown;
case Status::kNextValue:
return GNCStatusUnknown;
}
}
connections/swift/NearbyCoreAdapter/Sources/GNCError.mm
+4
View File
@@ -41,6 +41,8 @@ NSError *NSErrorFromCppStatus(Status status) {
return [NSError errorWithDomain:GNCErrorDomain code:GNCErrorAlreadyAdvertising userInfo:nil];
case Status::kAlreadyDiscovering:
return [NSError errorWithDomain:GNCErrorDomain code:GNCErrorAlreadyDiscovering userInfo:nil];
case Status::kAlreadyListening:
return [NSError errorWithDomain:GNCErrorDomain code:GNCErrorAlreadyListening userInfo:nil];
case Status::kEndpointIoError:
return [NSError errorWithDomain:GNCErrorDomain code:GNCErrorEndpointIoError userInfo:nil];
case Status::kEndpointUnknown:
@@ -63,6 +65,8 @@ NSError *NSErrorFromCppStatus(Status status) {
return [NSError errorWithDomain:GNCErrorDomain code:GNCErrorWifiLanError userInfo:nil];
case Status::kPayloadUnknown:
return [NSError errorWithDomain:GNCErrorDomain code:GNCErrorPayloadUnknown userInfo:nil];
case Status::kNextValue:
return [NSError errorWithDomain:GNCErrorDomain code:GNCErrorUnknown userInfo:nil];
}
}
connections/swift/NearbyCoreAdapter/Sources/Public/NearbyCoreAdapter/GNCConnectionDelegate.h
+2
View File
@@ -24,6 +24,7 @@ typedef NS_CLOSED_ENUM(NSInteger, GNCStatus) {
GNCStatusAlreadyHaveActiveStrategy,
GNCStatusAlreadyAdvertising,
GNCStatusAlreadyDiscovering,
GNCStatusAlreadyListening,
GNCStatusEndpointIoError,
GNCStatusEndpointUnknown,
GNCStatusConnectionRejected,
@@ -33,6 +34,7 @@ typedef NS_CLOSED_ENUM(NSInteger, GNCStatus) {
GNCStatusBleError,
GNCStatusWifiLanError,
GNCStatusPayloadUnknown,
GNCStatusUnknown,
};
/**
connections/swift/NearbyCoreAdapter/Sources/Public/NearbyCoreAdapter/GNCError.h
+1
View File
@@ -28,6 +28,7 @@ typedef NS_ERROR_ENUM(GNCErrorDomain, GNCErrorCode){
GNCErrorAlreadyHaveActiveStrategy,
GNCErrorAlreadyAdvertising,
GNCErrorAlreadyDiscovering,
GNCErrorAlreadyListening,
GNCErrorEndpointIoError,
GNCErrorEndpointUnknown,
GNCErrorConnectionRejected,
connections/v3/BUILD
+1
View File
@@ -7,6 +7,7 @@ cc_library(
"connections_device.h",
"connections_device_provider.h",
"listeners.h",
"listening_result.h",
"params.h",
],
visibility = [
connections/v3/connection_listening_options.h
+1 -1
View File
@@ -35,7 +35,7 @@ struct ConnectionListeningOptions {
bool auto_upgrade_bandwidth = true;
bool enforce_topology_constraints = true;
std::vector<location::nearby::proto::connections::Medium> upgrade_mediums;
std::vector<::location::nearby::proto::connections::Medium> listening_mediums;
std::vector<location::nearby::proto::connections::Medium> listening_mediums;
nearby::NearbyDevice::Type listening_endpoint_type =
NearbyDevice::Type::kConnectionsDevice;
};
connections/v3/listening_result.h
+44
View File
@@ -0,0 +1,44 @@
// Copyright 2023 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#ifndef THIRD_PARTY_NEARBY_CONNECTIONS_V3_LISTENING_RESULT_H_
#define THIRD_PARTY_NEARBY_CONNECTIONS_V3_LISTENING_RESULT_H_
#include <string>
#include <utility>
#include <vector>
#include "absl/functional/any_invocable.h"
#include "connections/status.h"
#include "internal/platform/connection_info.h"
namespace nearby {
namespace connections {
namespace v3 {
// Returned from StartListeningForIncomingConnections(). Can be passed to
// requestConnection().
struct ListeningResult {
std::string endpoint_id;
std::vector<ConnectionInfoVariant> connection_info;
};
using ListeningResultListener = absl::AnyInvocable<void(
std::pair<Status, v3::ListeningResult> result) const>;
} // namespace v3
} // namespace connections
} // namespace nearby
#endif // THIRD_PARTY_NEARBY_CONNECTIONS_V3_LISTENING_RESULT_H_
internal/platform/BUILD
+1
View File
@@ -153,6 +153,7 @@ cc_library(
"wifi_lan_connection_info.h",
],
visibility = [
"//connections/implementation:__pkg__",
"//connections/v3:__pkg__",
"//internal/interop:__pkg__",
"//presence:__subpackages__",