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

Internal change

Browse Source 
PiperOrigin-RevId: 379422825
This commit is contained in:
Edwin Wu
2021-06-14 23:20:15 -07:00
committed by Copybara-Service
parent b924c9200d
commit 5ed0903d6e
4 changed files with 698 additions and 532 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cpp/core/internal/base_pcp_handler.cc
+206 -147
View File
@@ -53,22 +53,22 @@ BasePcpHandler::BasePcpHandler(Mediums* mediums,
bwu_manager_(bwu_manager) {}
BasePcpHandler::~BasePcpHandler() {
NEARBY_LOGS(INFO) << "BasePcpHandler: going down; strategy="
<< strategy_.GetName();
NEARBY_LOGS(INFO) << "Initiating shutdown of BasePcpHandler("
<< strategy_.GetName() << ")";
DisconnectFromEndpointManager();
// Stop all the ongoing Runnables (as gracefully as possible).
NEARBY_LOGS(INFO) << "BasePcpHandler: bringing down executors; strategy="
<< strategy_.GetName();
NEARBY_LOGS(INFO) << "BasePcpHandler(" << strategy_.GetName()
<< ") is bringing down executors.";
serial_executor_.Shutdown();
alarm_executor_.Shutdown();
NEARBY_LOGS(INFO) << "BasePcpHandler: is down; strategy="
<< strategy_.GetName();
NEARBY_LOGS(INFO) << "BasePcpHandler(" << strategy_.GetName()
<< ") has shut down.";
}
void BasePcpHandler::DisconnectFromEndpointManager() {
if (stop_.Set(true)) return;
NEARBY_LOGS(INFO) << "BasePcpHandler: Unregister from EPM; strategy="
<< strategy_.GetName();
NEARBY_LOGS(INFO) << "BasePcpHandler(" << strategy_.GetName()
<< ") unregister from EPM.";
// Unregister ourselves from EPM message dispatcher.
endpoint_manager_->UnregisterFrameProcessor(V1Frame::CONNECTION_RESPONSE,
this);
@@ -79,8 +79,8 @@ Status BasePcpHandler::StartAdvertising(ClientProxy* client,
const ConnectionOptions& options,
const ConnectionRequestInfo& info) {
Future<Status> response;
NEARBY_LOG(INFO, "StartAdvertising with supported mediums: %s",
GetStringValueOfSupportedMediums(options).c_str());
NEARBY_LOGS(INFO) << "StartAdvertising with supported mediums: "
<< GetStringValueOfSupportedMediums(options);
ConnectionOptions advertising_options = options.CompatibleOptions();
RunOnPcpHandlerThread(
"start-advertising",
@@ -118,7 +118,8 @@ Status BasePcpHandler::StartAdvertising(ClientProxy* client,
}
void BasePcpHandler::StopAdvertising(ClientProxy* client) {
NEARBY_LOGS(INFO) << "StopAdvertising id=" << client->GetLocalEndpointId();
NEARBY_LOGS(INFO) << "StopAdvertising local_endpoint_id="
<< client->GetLocalEndpointId();
CountDownLatch latch(1);
RunOnPcpHandlerThread("stop-advertising",
[this, client, &latch]() RUN_ON_PCP_HANDLER_THREAD() {
@@ -133,10 +134,18 @@ std::string BasePcpHandler::GetStringValueOfSupportedMediums(
const ConnectionOptions& options) const {
std::ostringstream result;
result << "{ ";
if (options.allowed.bluetooth) result << "bluetooth ";
if (options.allowed.ble) result << "ble ";
if (options.allowed.web_rtc) result << "webrtc ";
if (options.allowed.wifi_lan) result << "wifilan ";
if (options.allowed.bluetooth) {
result << proto::connections::Medium_Name(Medium::BLUETOOTH) << " ";
}
if (options.allowed.ble) {
result << proto::connections::Medium_Name(Medium::BLE) << " ";
}
if (options.allowed.web_rtc) {
result << proto::connections::Medium_Name(Medium::WEB_RTC) << " ";
}
if (options.allowed.wifi_lan) {
result << proto::connections::Medium_Name(Medium::WIFI_LAN) << " ";
}
result << "}";
return result.str();
}
@@ -194,8 +203,8 @@ Status BasePcpHandler::StartDiscovery(ClientProxy* client,
Future<Status> response;
ConnectionOptions discovery_options = options.CompatibleOptions();
NEARBY_LOG(INFO, "StartDiscovery with supported mediums: %s",
GetStringValueOfSupportedMediums(options).c_str());
NEARBY_LOGS(INFO) << "StartDiscovery with supported mediums:"
<< GetStringValueOfSupportedMediums(options);
RunOnPcpHandlerThread(
"start-discovery", [this, client, service_id, discovery_options,
&listener, &response]() RUN_ON_PCP_HANDLER_THREAD() {
@@ -249,7 +258,7 @@ void BasePcpHandler::WaitForLatch(const std::string& method_name,
Exception await_exception = latch->Await();
if (!await_exception.Ok()) {
if (await_exception.Raised(Exception::kTimeout)) {
NEARBY_LOG(INFO, "Blocked in %s", method_name.c_str());
NEARBY_LOGS(INFO) << "Blocked in " << method_name;
}
}
}
@@ -258,18 +267,18 @@ Status BasePcpHandler::WaitForResult(const std::string& method_name,
std::int64_t client_id,
Future<Status>* future) {
if (!future) {
NEARBY_LOG(INFO, "No future to wait for; return with error");
NEARBY_LOGS(INFO) << "No future to wait for; return with error";
return {Status::kError};
}
NEARBY_LOG(INFO, "Waiting for future to complete: %s", method_name.c_str());
NEARBY_LOGS(INFO) << "Waiting for future to complete: " << method_name;
ExceptionOr<Status> result = future->Get();
if (!result.ok()) {
NEARBY_LOG(INFO, "Future:[%s] completed with exception: %d",
method_name.c_str(), result.exception());
NEARBY_LOGS(INFO) << "Future:[" << method_name
<< "] completed with exception:" << result.exception();
return {Status::kError};
}
NEARBY_LOG(INFO, "Future:[%s] completed with status: %d", method_name.c_str(),
result.result().value);
NEARBY_LOGS(INFO) << "Future:[" << method_name
<< "] completed with status:" << result.result().value;
return result.result();
}
@@ -299,8 +308,10 @@ EncryptionRunner::ResultListener BasePcpHandler::GetResultListener() {
RunOnPcpHandlerThread(
"encryption-failure",
[this, endpoint_id, channel]() RUN_ON_PCP_HANDLER_THREAD() {
NEARBY_LOG(ERROR, "Encryption failed for %s on medium %d",
endpoint_id.c_str(), channel->GetMedium());
NEARBY_LOGS(ERROR)
<< "Encryption failed for endpoint_id=" << endpoint_id
<< " on medium="
<< proto::connections::Medium_Name(channel->GetMedium());
OnEncryptionFailureRunnable(endpoint_id, channel);
});
},
@@ -314,9 +325,9 @@ void BasePcpHandler::OnEncryptionSuccessRunnable(
// busy running UKEY2.
auto it = pending_connections_.find(endpoint_id);
if (it == pending_connections_.end()) {
NEARBY_LOG(INFO,
"Connection not found on UKEY negotination complete; id=%s",
endpoint_id.c_str());
NEARBY_LOGS(INFO)
<< "Connection not found on UKEY negotination complete; endpoint_id="
<< endpoint_id;
return;
}
@@ -331,8 +342,9 @@ void BasePcpHandler::OnEncryptionSuccessRunnable(
}
connection_info.SetCryptoContext(std::move(ukey2));
NEARBY_LOG(INFO, "Register encrypted connection; wait for response; id=%s",
endpoint_id.c_str());
NEARBY_LOGS(INFO)
<< "Register encrypted connection; wait for response; endpoint_id="
<< endpoint_id;
// Set ourselves up so that we receive all acceptance/rejection messages
endpoint_manager_->RegisterFrameProcessor(V1Frame::CONNECTION_RESPONSE, this);
@@ -373,7 +385,7 @@ void BasePcpHandler::OnEncryptionSuccessRunnable(
std::move(connection_info.channel), connection_info.listener);
if (auto future_status = connection_info.result.lock()) {
NEARBY_LOG(INFO, "Connection established; Finalising future OK");
NEARBY_LOGS(INFO) << "Connection established; Finalising future OK.";
future_status->Set({Status::kSuccess});
connection_info.result.reset();
}
@@ -383,9 +395,9 @@ void BasePcpHandler::OnEncryptionFailureRunnable(
const std::string& endpoint_id, EndpointChannel* endpoint_channel) {
auto it = pending_connections_.find(endpoint_id);
if (it == pending_connections_.end()) {
NEARBY_LOG(INFO,
"Connection not found on UKEY negotination complete; id=%s",
endpoint_id.c_str());
NEARBY_LOGS(INFO)
<< "Connection not found on UKEY negotination complete; endpoint_id="
<< endpoint_id;
return;
}
@@ -398,9 +410,9 @@ void BasePcpHandler::OnEncryptionFailureRunnable(
// the map had already updated with the winning EndpointChannel, we closed
// it too by accident.
if (*endpoint_channel != *info.channel) {
NEARBY_LOG(
INFO, "Not destroying channel [mismatch]: passed=%s; expected=%s",
endpoint_channel->GetName().c_str(), info.channel->GetName().c_str());
NEARBY_LOGS(INFO) << "Not destroying channel [mismatch]: passed="
<< endpoint_channel->GetName()
<< "; expected=" << info.channel->GetName();
return;
}
@@ -422,8 +434,11 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client,
// If we already have a pending connection, then we shouldn't allow any
// more outgoing connections to this endpoint.
if (pending_connections_.count(endpoint_id)) {
NEARBY_LOG(INFO, "Connection already exists: id=%s",
endpoint_id.c_str());
NEARBY_LOGS(INFO)
<< "In requestConnection(), connection requested with "
"endpoint(id="
<< endpoint_id
<< "), but we already have a pending connection with them.";
result->Set({Status::kAlreadyConnectedToEndpoint});
return;
}
@@ -432,16 +447,18 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client,
// listen to them.
if (ShouldEnforceTopologyConstraints(client->GetAdvertisingOptions()) &&
!CanSendOutgoingConnection(client)) {
NEARBY_LOG(INFO, "Outgoing connection not allowed: id=%s",
endpoint_id.c_str());
NEARBY_LOGS(INFO)
<< "In requestConnection(), client=" << client->GetClientId()
<< " attempted a connection with endpoint(id=" << endpoint_id
<< "), but outgoing connections are disallowed";
result->Set({Status::kOutOfOrderApiCall});
return;
}
DiscoveredEndpoint* endpoint = GetDiscoveredEndpoint(endpoint_id);
if (endpoint == nullptr) {
NEARBY_LOG(INFO, "Discovered endpoint not found: id=%s",
endpoint_id.c_str());
NEARBY_LOGS(INFO)
<< "Discovered endpoint not found: endpoint_id=" << endpoint_id;
result->Set({Status::kEndpointUnknown});
return;
}
@@ -453,8 +470,8 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client,
endpoint_id, remote_bluetooth_mac_address,
client->GetDiscoveryOptions()))
NEARBY_LOGS(INFO)
<< "Appended remote Bluetooth MAC Address endpoint "
<< "[" << remote_bluetooth_mac_address << "]";
<< "Appended remote Bluetooth MAC Address endpoint ["
<< remote_bluetooth_mac_address << "]";
}
if (AppendWebRTCEndpoint(endpoint_id, client->GetDiscoveryOptions()))
@@ -476,16 +493,18 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client,
}
if (channel == nullptr) {
NEARBY_LOG(INFO, "Endpoint channel not available: id=%s",
endpoint_id.c_str());
NEARBY_LOGS(INFO)
<< "Endpoint channel not available: endpoint_id=" << endpoint_id;
ProcessPreConnectionInitiationFailure(endpoint_id, channel.get(),
connect_impl_result.status,
result.get());
return;
}
NEARBY_LOG(INFO, "Sending connection request: id=%s",
endpoint_id.c_str());
NEARBY_LOGS(INFO)
<< "In requestConnection(), wrote ConnectionRequestFrame "
"to endpoint_id="
<< endpoint_id;
// Generate the nonce to use for this connection.
std::int32_t nonce = prng_.NextInt32();
@@ -497,16 +516,16 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client,
options.keep_alive_interval_millis,
options.keep_alive_timeout_millis);
if (!write_exception.Ok()) {
NEARBY_LOG(INFO, "Failed to send connection request: id=%s",
endpoint_id.c_str());
NEARBY_LOGS(INFO) << "Failed to send connection request: endpoint_id="
<< endpoint_id;
ProcessPreConnectionInitiationFailure(endpoint_id, channel.get(),
{Status::kEndpointIoError},
result.get());
return;
}
NEARBY_LOG(INFO, "adding connection to pending set: id=%s",
endpoint_id.c_str());
NEARBY_LOGS(INFO) << "Adding connection to pending set: endpoint_id="
<< endpoint_id;
// We've successfully connected to the device, and are now about to jump
// on to the EncryptionRunner thread to start running our encryption
@@ -531,20 +550,20 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client,
.emplace(endpoint_id, std::move(pendingConnectionInfo))
.first->second.channel.get();
NEARBY_LOG(INFO, "Initiating secure connection: id=%s",
endpoint_id.c_str());
NEARBY_LOGS(INFO) << "Initiating secure connection: endpoint_id="
<< endpoint_id;
// Next, we'll set up encryption. When it's done, our future will return
// and RequestConnection() will finish.
encryption_runner_.StartClient(client, endpoint_id, endpoint_channel,
GetResultListener());
});
NEARBY_LOG(INFO, "Waiting for connection to complete: id=%s",
endpoint_id.c_str());
NEARBY_LOGS(INFO) << "Waiting for connection to complete: endpoint_id="
<< endpoint_id;
auto status =
WaitForResult(absl::StrCat("RequestConnection(", endpoint_id, ")"),
client->GetClientId(), result.get());
NEARBY_LOG(INFO, "Wait is complete: id=%s; status=%d", endpoint_id.c_str(),
status.value);
NEARBY_LOGS(INFO) << "Wait is complete: endpoint_id=" << endpoint_id
<< "; status=" << status.value;
return status;
}
@@ -663,7 +682,7 @@ void BasePcpHandler::ProcessPreConnectionInitiationFailure(
}
if (result != nullptr) {
NEARBY_LOG(INFO, "Connection failed; aborting future");
NEARBY_LOGS(INFO) << "Connection failed; aborting future";
result->Set(status);
}
@@ -707,10 +726,12 @@ Status BasePcpHandler::AcceptConnection(
RunOnPcpHandlerThread(
"accept-connection", [this, client, endpoint_id, payload_listener,
&response]() RUN_ON_PCP_HANDLER_THREAD() {
NEARBY_LOG(INFO, "AcceptConnection: id=%s", endpoint_id.c_str());
NEARBY_LOGS(INFO) << "AcceptConnection: endpoint_id=" << endpoint_id;
if (!pending_connections_.count(endpoint_id)) {
NEARBY_LOG(INFO, "AcceptConnection: no pending connection for id=%s",
endpoint_id.c_str());
NEARBY_LOGS(INFO)
<< "AcceptConnection: no pending connection for endpoint_id="
<< endpoint_id;
response.Set({Status::kEndpointUnknown});
return;
}
@@ -724,10 +745,9 @@ Status BasePcpHandler::AcceptConnection(
std::shared_ptr<EndpointChannel> channel =
channel_manager_->GetChannelForEndpoint(endpoint_id);
if (channel == nullptr) {
NEARBY_LOG(
ERROR,
"Channel destroyed before Accept; bring down connection: id=%s",
endpoint_id.c_str());
NEARBY_LOGS(ERROR) << "Channel destroyed before Accept; bring down "
"connection: endpoint_id="
<< endpoint_id;
ProcessPreConnectionResultFailure(client, endpoint_id);
response.Set({Status::kEndpointUnknown});
return;
@@ -736,15 +756,16 @@ Status BasePcpHandler::AcceptConnection(
Exception write_exception =
channel->Write(parser::ForConnectionResponse(Status::kSuccess));
if (!write_exception.Ok()) {
NEARBY_LOG(INFO, "AcceptConnection: failed to send response: id=%s",
endpoint_id.c_str());
NEARBY_LOGS(INFO)
<< "AcceptConnection: failed to send response: endpoint_id="
<< endpoint_id;
ProcessPreConnectionResultFailure(client, endpoint_id);
response.Set({Status::kEndpointIoError});
return;
}
NEARBY_LOG(INFO, "AcceptConnection: accepting locally: id=%s",
endpoint_id.c_str());
NEARBY_LOGS(INFO) << "AcceptConnection: accepting locally: endpoint_id="
<< endpoint_id;
connection_info.LocalEndpointAcceptedConnection(endpoint_id,
payload_listener);
EvaluateConnectionResult(client, endpoint_id,
@@ -764,8 +785,9 @@ Status BasePcpHandler::RejectConnection(ClientProxy* client,
[this, client, endpoint_id, &response]() RUN_ON_PCP_HANDLER_THREAD() {
NEARBY_LOG(INFO, "RejectConnection: id=%s", endpoint_id.c_str());
if (!pending_connections_.count(endpoint_id)) {
NEARBY_LOG(INFO, "RejectConnection: no pending connection for id=%s",
endpoint_id.c_str());
NEARBY_LOGS(INFO)
<< "RejectConnection: no pending connection for endpoint_id="
<< endpoint_id;
response.Set({Status::kEndpointUnknown});
return;
}
@@ -779,10 +801,10 @@ Status BasePcpHandler::RejectConnection(ClientProxy* client,
std::shared_ptr<EndpointChannel> channel =
channel_manager_->GetChannelForEndpoint(endpoint_id);
if (channel == nullptr) {
NEARBY_LOG(
ERROR,
"Channel destroyed before Reject; bring down connection: id=%s",
endpoint_id.c_str());
NEARBY_LOGS(ERROR)
<< "Channel destroyed before Reject; bring down connection: "
"endpoint_id="
<< endpoint_id;
ProcessPreConnectionResultFailure(client, endpoint_id);
response.Set({Status::kEndpointUnknown});
return;
@@ -791,15 +813,16 @@ Status BasePcpHandler::RejectConnection(ClientProxy* client,
Exception write_exception = channel->Write(
parser::ForConnectionResponse(Status::kConnectionRejected));
if (!write_exception.Ok()) {
NEARBY_LOG(INFO, "RejectConnection: failed to send response: id=%s",
endpoint_id.c_str());
NEARBY_LOGS(INFO)
<< "RejectConnection: failed to send response: endpoint_id="
<< endpoint_id;
ProcessPreConnectionResultFailure(client, endpoint_id);
response.Set({Status::kEndpointIoError});
return;
}
NEARBY_LOG(INFO, "RejectConnection: rejecting locally: id=%s",
endpoint_id.c_str());
NEARBY_LOGS(INFO) << "RejectConnection: rejecting locally: endpoint_id="
<< endpoint_id;
connection_info.LocalEndpointRejectedConnection(endpoint_id);
EvaluateConnectionResult(client, endpoint_id,
false /* can_close_immediately */);
@@ -818,11 +841,13 @@ void BasePcpHandler::OnIncomingFrame(OfflineFrame& frame,
RunOnPcpHandlerThread(
"incoming-frame",
[this, client, endpoint_id, frame, &latch]() RUN_ON_PCP_HANDLER_THREAD() {
NEARBY_LOG(INFO, "OnConnectionResponse: id=%s", endpoint_id.c_str());
NEARBY_LOGS(INFO) << "OnConnectionResponse: endpoint_id="
<< endpoint_id;
if (client->HasRemoteEndpointResponded(endpoint_id)) {
NEARBY_LOG(INFO, "OnConnectionResponse: already handled; id=%s",
endpoint_id.c_str());
NEARBY_LOGS(INFO)
<< "OnConnectionResponse: already handled; endpoint_id="
<< endpoint_id;
return;
}
@@ -840,13 +865,14 @@ void BasePcpHandler::OnIncomingFrame(OfflineFrame& frame,
accepted = connection_response.status() == Status::kSuccess;
}
if (accepted) {
NEARBY_LOG(INFO, "OnConnectionResponse: remote accepted; id=%s",
endpoint_id.c_str());
NEARBY_LOGS(INFO)
<< "OnConnectionResponse: remote accepted; endpoint_id="
<< endpoint_id;
client->RemoteEndpointAcceptedConnection(endpoint_id);
} else {
NEARBY_LOG(INFO,
"OnConnectionResponse: remote rejected; id=%s; status=%d",
endpoint_id.c_str(), connection_response.status());
NEARBY_LOGS(INFO)
<< "OnConnectionResponse: remote rejected; endpoint_id="
<< endpoint_id << "; status=" << connection_response.status();
client->RemoteEndpointRejectedConnection(endpoint_id);
}
@@ -890,7 +916,7 @@ void BasePcpHandler::OnEndpointFound(
ClientProxy* client, std::shared_ptr<DiscoveredEndpoint> endpoint) {
// Check if we've seen this endpoint ID before.
std::string& endpoint_id = endpoint->endpoint_id;
NEARBY_LOG(INFO, "OnEndpointFound: id='%s' [enter]", endpoint_id.c_str());
NEARBY_LOGS(INFO) << "OnEndpointFound: id=" << endpoint_id << " [enter]";
auto range = discovered_endpoints_.equal_range(endpoint->endpoint_id);
@@ -920,14 +946,14 @@ void BasePcpHandler::OnEndpointFound(
// Range is empty: this is the first endpoint we discovered so far.
// Report this endpoint_id to client.
if (range.first == range.second) {
NEARBY_LOG(INFO, "Adding new endpoint: id=%s", endpoint_id.c_str());
NEARBY_LOGS(INFO) << "Adding new endpoint: endpoint_id=" << endpoint_id;
// And, as it's the first time, report it to the client.
client->OnEndpointFound(
owned_endpoint->service_id, owned_endpoint->endpoint_id,
owned_endpoint->endpoint_info, owned_endpoint->medium);
} else {
NEARBY_LOGS(INFO) << "Adding new medium for endpoint: id=" << endpoint_id
<< "; medium=" << owned_endpoint->medium;
NEARBY_LOGS(INFO) << "Adding new medium for endpoint: endpoint_id="
<< endpoint_id << "; medium=" << owned_endpoint->medium;
}
}
@@ -936,8 +962,8 @@ void BasePcpHandler::OnEndpointLost(
// Look up the DiscoveredEndpoint we have in our cache.
const auto* discovered_endpoint = GetDiscoveredEndpoint(endpoint.endpoint_id);
if (discovered_endpoint == nullptr) {
NEARBY_LOG(INFO, "No previous endpoint (nothing to lose): id=%s",
endpoint.endpoint_id.c_str());
NEARBY_LOGS(INFO) << "No previous endpoint (nothing to lose): endpoint_id="
<< endpoint.endpoint_id;
return;
}
@@ -946,10 +972,11 @@ void BasePcpHandler::OnEndpointLost(
// device changed their info. We reported onFound for the new info and are
// just now figuring out that we lost the old info.
if (discovered_endpoint->endpoint_info != endpoint.endpoint_info) {
NEARBY_LOG(INFO, "Previous endpoint name mismatch; passed=%s; expected=%s",
absl::BytesToHexString(endpoint.endpoint_info.data()).c_str(),
absl::BytesToHexString(discovered_endpoint->endpoint_info.data())
.c_str());
NEARBY_LOGS(INFO) << "Previous endpoint name mismatch; passed="
<< absl::BytesToHexString(endpoint.endpoint_info.data())
<< "; expected="
<< absl::BytesToHexString(
discovered_endpoint->endpoint_info.data());
return;
}
@@ -981,10 +1008,11 @@ bool BasePcpHandler::IsPreferred(
for (const auto& medium : mediums) {
absl::StrAppend(&medium_string, medium, "; ");
}
NEARBY_LOG(FATAL,
"Failed to determine preferred medium; bailing out; mediums=%s; "
"new=%d; old=%d",
medium_string.c_str(), new_endpoint.medium, old_endpoint.medium);
NEARBY_LOGS(FATAL) << "Failed to find either " << new_endpoint.medium
<< " or " << old_endpoint.medium
<< " in the list of locally supported mediums despite "
"expecting to find both, when deciding which medium "
<< medium_string << " is preferred.";
return false;
}
@@ -998,10 +1026,11 @@ Exception BasePcpHandler::OnIncomingConnection(
// the client stopped advertising and we nulled out state, followed by an
// incoming connection where we attempted to check that state.
if (!client->IsAdvertising()) {
NEARBY_LOG(WARNING,
"Ignoring incoming connection because client 0x%" PRIX64
" is no longer advertising.",
client->GetClientId());
NEARBY_LOGS(WARNING) << "Ignoring incoming connection on medium "
<< proto::connections::Medium_Name(
channel->GetMedium())
<< " because client=" << client->GetClientId()
<< " is no longer advertising.";
return {Exception::kIo};
}
@@ -1011,12 +1040,10 @@ Exception BasePcpHandler::OnIncomingConnection(
if (!wrapped_frame.ok()) {
if (wrapped_frame.exception()) {
NEARBY_LOG(
ERROR,
"Failed to parse incoming connection request; client_id=0x%" PRIX64
"; device=%s",
client->GetClientId(),
absl::BytesToHexString(remote_endpoint_info.data()).c_str());
NEARBY_LOGS(ERROR)
<< "Failed to parse incoming connection request; client="
<< client->GetClientId()
<< "; device=" << absl::BytesToHexString(remote_endpoint_info.data());
ProcessPreConnectionInitiationFailure("", channel.get(), {Status::kError},
nullptr);
return {Exception::kSuccess};
@@ -1027,13 +1054,17 @@ Exception BasePcpHandler::OnIncomingConnection(
OfflineFrame& frame = wrapped_frame.result();
const ConnectionRequestFrame& connection_request =
frame.v1().connection_request();
NEARBY_LOG(INFO,
"Incoming connection request; client_id=0x%" PRIX64
"; device=%s; id=%s",
client->GetClientId(),
absl::BytesToHexString(remote_endpoint_info.data()).c_str(),
connection_request.endpoint_id().c_str());
NEARBY_LOGS(INFO) << "In onIncomingConnection("
<< proto::connections::Medium_Name(channel->GetMedium())
<< ") for client=" << client->GetClientId()
<< ", read ConnectionRequestFrame from endpoint(id="
<< connection_request.endpoint_id() << ")";
if (client->IsConnectedToEndpoint(connection_request.endpoint_id())) {
NEARBY_LOGS(ERROR) << "Incoming connection on medium "
<< proto::connections::Medium_Name(channel->GetMedium())
<< " was denied because we're "
"already connected to endpoint(id="
<< connection_request.endpoint_id() << ").";
return {Exception::kIo};
}
@@ -1048,6 +1079,7 @@ Exception BasePcpHandler::OnIncomingConnection(
// listen to them.
if (ShouldEnforceTopologyConstraints(client->GetAdvertisingOptions()) &&
!CanReceiveIncomingConnection(client)) {
NEARBY_LOGS(ERROR) << "Incoming connections are currently disallowed.";
return {Exception::kIo};
}
@@ -1074,13 +1106,13 @@ Exception BasePcpHandler::OnIncomingConnection(
if (options.keep_alive_interval_millis == 0 ||
options.keep_alive_timeout_millis == 0 ||
options.keep_alive_interval_millis >= options.keep_alive_timeout_millis) {
NEARBY_LOG(WARNING,
"Incoming connection has wrong keep-alive frame interval=%d, "
"timeout=%d values; correct them as default.",
options.keep_alive_interval_millis,
options.keep_alive_timeout_millis);
options.keep_alive_interval_millis =
FeatureFlags::GetInstance().GetFlags().keep_alive_interval_millis;
NEARBY_LOGS(WARNING)
<< "Incoming connection has wrong keep-alive frame interval="
<< options.keep_alive_interval_millis
<< ", timeout=" << options.keep_alive_timeout_millis
<< " values; correct them as default.",
options.keep_alive_interval_millis =
FeatureFlags::GetInstance().GetFlags().keep_alive_interval_millis;
options.keep_alive_timeout_millis =
FeatureFlags::GetInstance().GetFlags().keep_alive_timeout_millis;
}
@@ -1122,21 +1154,38 @@ bool BasePcpHandler::BreakTie(ClientProxy* client,
if (it != pending_connections_.end()) {
BasePcpHandler::PendingConnectionInfo& info = it->second;
NEARBY_LOG(INFO, "BreakTie: id=%s", endpoint_id.c_str());
NEARBY_LOGS(INFO)
<< "In onIncomingConnection("
<< proto::connections::Medium_Name(endpoint_channel->GetMedium())
<< ") for client=" << client->GetClientId()
<< ", found a collision with endpoint " << endpoint_id
<< ". We've already sent a connection request to them with nonce "
<< info.nonce
<< ", but they're also trying to connect to us with nonce "
<< incoming_nonce;
// Break the lowest connection. In the (extremely) rare case of a tie, break
// both.
if (info.nonce > incoming_nonce) {
// Our connection won! Clean up their connection.
endpoint_channel->Close();
NEARBY_LOG(INFO, "BreakTie: We won; id=%s", endpoint_id.c_str());
NEARBY_LOGS(INFO) << "In onIncomingConnection("
<< proto::connections::Medium_Name(
endpoint_channel->GetMedium())
<< ") for client=" << client->GetClientId()
<< ", cleaned up the collision with endpoint "
<< endpoint_id << " by closing their channel.";
return true;
} else if (info.nonce < incoming_nonce) {
// Aw, we lost. Clean up our connection, and then we'll let their
// connection continue on.
ProcessTieBreakLoss(client, endpoint_id, &info);
NEARBY_LOG(INFO, "BreakTie: We lost; id=%s", endpoint_id.c_str());
NEARBY_LOGS(INFO)
<< "In onIncomingConnection("
<< proto::connections::Medium_Name(endpoint_channel->GetMedium())
<< ") for client=" << client->GetClientId()
<< ", cleaned up the collision with endpoint " << endpoint_id
<< " by closing our channel and notifying our client of the failure.";
} else {
// Oh. Huh. We both lost. Well, that's awkward. We'll clean up both and
// just force the devices to retry.
@@ -1144,7 +1193,13 @@ bool BasePcpHandler::BreakTie(ClientProxy* client,
ProcessTieBreakLoss(client, endpoint_id, &info);
NEARBY_LOG(INFO, "BreakTie: Both lost; id=%s", endpoint_id.c_str());
NEARBY_LOGS(INFO)
<< "In onIncomingConnection("
<< proto::connections::Medium_Name(endpoint_channel->GetMedium())
<< ") for client=" << client->GetClientId()
<< ", cleaned up the collision with endpoint " << endpoint_id
<< " by closing both channels. Our nonces were identical, so we "
"couldn't decide which channel to use.";
return true;
}
}
@@ -1177,9 +1232,9 @@ bool BasePcpHandler::AppendRemoteBluetoothMacAddressEndpoint(
for (auto item = it.first; item != it.second; item++) {
if (item->second->medium == proto::connections::Medium::BLUETOOTH) {
NEARBY_LOGS(INFO)
<< "Cannot append remote Bluetooth MAC Address endpoint, because the "
"endpoint has already been found over Bluetooth "
<< "[" << remote_bluetooth_mac_address << "]";
<< "Cannot append remote Bluetooth MAC Address endpoint, because "
"the endpoint has already been found over Bluetooth ["
<< remote_bluetooth_mac_address << "]";
return false;
}
}
@@ -1187,10 +1242,10 @@ bool BasePcpHandler::AppendRemoteBluetoothMacAddressEndpoint(
auto remote_bluetooth_device =
GetRemoteBluetoothDevice(remote_bluetooth_mac_address);
if (!remote_bluetooth_device.IsValid()) {
NEARBY_LOGS(INFO) << "Cannot append remote Bluetooth MAC Address endpoint, "
"because a valid "
"Bluetooth device could not be derived "
<< "[" << remote_bluetooth_mac_address << "]";
NEARBY_LOGS(INFO)
<< "Cannot append remote Bluetooth MAC Address endpoint, because a "
"valid Bluetooth device could not be derived ["
<< remote_bluetooth_mac_address << "]";
return false;
}
@@ -1243,11 +1298,13 @@ void BasePcpHandler::EvaluateConnectionResult(ClientProxy* client,
if (!client->IsConnectionAccepted(endpoint_id) &&
!client->IsConnectionRejected(endpoint_id)) {
if (!client->HasLocalEndpointResponded(endpoint_id)) {
NEARBY_LOG(INFO, "ConnectionResult: local client did not respond; id=%s",
endpoint_id.c_str());
NEARBY_LOGS(INFO)
<< "ConnectionResult: local client did not respond; endpoint_id="
<< endpoint_id;
} else if (!client->HasRemoteEndpointResponded(endpoint_id)) {
NEARBY_LOG(INFO, "ConnectionResult: remote client did not respond; id=%s",
endpoint_id.c_str());
NEARBY_LOGS(INFO)
<< "ConnectionResult: remote client did not respond; endpoint_id="
<< endpoint_id;
}
return;
}
@@ -1256,8 +1313,8 @@ void BasePcpHandler::EvaluateConnectionResult(ClientProxy* client,
// no longer pending.
auto it = pending_connections_.find(endpoint_id);
if (it == pending_connections_.end()) {
NEARBY_LOG(INFO, "No pending connection to evaluate; id=%s",
endpoint_id.c_str());
NEARBY_LOGS(INFO) << "No pending connection to evaluate; endpoint_id="
<< endpoint_id;
return;
}
@@ -1267,7 +1324,8 @@ void BasePcpHandler::EvaluateConnectionResult(ClientProxy* client,
Status response_code;
if (is_connection_accepted) {
NEARBY_LOG(INFO, "Pending connection accepted; id=%s", endpoint_id.c_str());
NEARBY_LOGS(INFO) << "Pending connection accepted; endpoint_id="
<< endpoint_id;
response_code = {Status::kSuccess};
// Both sides have accepted, so we can now start talking over encrypted
@@ -1284,7 +1342,8 @@ void BasePcpHandler::EvaluateConnectionResult(ClientProxy* client,
channel_manager_->EncryptChannelForEndpoint(endpoint_id,
std::move(context));
} else {
NEARBY_LOG(INFO, "Pending connection rejected; id=%s", endpoint_id.c_str());
NEARBY_LOGS(INFO) << "Pending connection rejected; endpoint_id="
<< endpoint_id;
response_code = {Status::kConnectionRejected};
}
cpp/core/internal/p2p_cluster_pcp_handler.cc
+489 -384
View File
File diff suppressed because it is too large Load Diff
cpp/core/internal/p2p_cluster_pcp_handler_test.cc
+1 -1
View File
@@ -54,7 +54,7 @@ class P2pClusterPcpHandlerTest
NEARBY_LOG(INFO, "SetUp: BT enabled");
}
if (options_.allowed.wifi_lan) {
NEARBY_LOG(INFO, "SetUp: Wifi LAN enabled");
NEARBY_LOG(INFO, "SetUp: WifiLan enabled");
}
if (options_.allowed.web_rtc) {
NEARBY_LOG(INFO, "SetUp: WebRTC enabled");
cpp/core/internal/pcp_manager.cc
+2
View File
@@ -48,7 +48,9 @@ void PcpManager::DisconnectFromEndpointManager() {
}
PcpManager::~PcpManager() {
NEARBY_LOGS(INFO) << "Initiating shutdown of PcpManager.";
DisconnectFromEndpointManager();
NEARBY_LOGS(INFO) << "PcpManager has shut down.";
}
Status PcpManager::StartAdvertising(ClientProxy* client,