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

analytics: 3p NC: Implement ConnectionAttempt.

Browse Source 
PiperOrigin-RevId: 392794192
This commit is contained in:
edwinwu
2021-08-24 18:44:26 -07:00
committed by Copybara-Service
parent 3e40931286
commit 62e7c01cd3
7 changed files with 467 additions and 231 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cpp/core/internal/base_pcp_handler.cc
+75 -16
View File
@@ -338,7 +338,9 @@ void BasePcpHandler::OnEncryptionSuccessRunnable(
if (!ukey2) {
// Fail early, if there is no crypto context.
ProcessPreConnectionInitiationFailure(
endpoint_id, connection_info.channel.get(), {Status::kEndpointIoError},
connection_info.client, connection_info.channel->GetMedium(),
endpoint_id, connection_info.channel.get(), connection_info.is_incoming,
connection_info.start_time, {Status::kEndpointIoError},
connection_info.result.lock().get());
return;
}
@@ -388,6 +390,21 @@ void BasePcpHandler::OnEncryptionSuccessRunnable(
std::move(connection_info.channel), connection_info.listener,
connection_info.connection_token);
if (connection_info.is_incoming) {
connection_info.client->GetAnalyticsRecorder().OnIncomingConnectionAttempt(
proto::connections::INITIAL, connection_info.channel->GetMedium(),
proto::connections::RESULT_SUCCESS,
SystemClock::ElapsedRealtime() - connection_info.start_time,
connection_info.connection_token);
} else {
connection_info.client->GetAnalyticsRecorder().OnOutgoingConnectionAttempt(
endpoint_id, proto::connections::INITIAL,
connection_info.channel->GetMedium(),
proto::connections::RESULT_SUCCESS,
SystemClock::ElapsedRealtime() - connection_info.start_time,
connection_info.connection_token);
}
if (auto future_status = connection_info.result.lock()) {
NEARBY_LOGS(INFO) << "Connection established; Finalising future OK.";
future_status->Set({Status::kSuccess});
@@ -420,9 +437,10 @@ void BasePcpHandler::OnEncryptionFailureRunnable(
return;
}
ProcessPreConnectionInitiationFailure(endpoint_id, info.channel.get(),
{Status::kEndpointIoError},
info.result.lock().get());
ProcessPreConnectionInitiationFailure(
info.client, info.channel->GetMedium(), endpoint_id, info.channel.get(),
info.is_incoming, info.start_time, {Status::kEndpointIoError},
info.result.lock().get());
}
Status BasePcpHandler::RequestConnection(ClientProxy* client,
@@ -486,6 +504,7 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client,
ConnectImplResult connect_impl_result;
for (auto connect_endpoint : discovered_endpoints) {
absl::Time connect_start_time = SystemClock::ElapsedRealtime();
if (!MediumSupportedByClientOptions(connect_endpoint->medium,
options))
continue;
@@ -493,15 +512,20 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client,
if (connect_impl_result.status.Ok()) {
channel = std::move(connect_impl_result.endpoint_channel);
break;
} else {
LogConnectionAttempt(client, connect_endpoint->medium,
connect_endpoint->endpoint_id,
/* is_incoming = */ false, connect_start_time);
}
}
if (channel == nullptr) {
NEARBY_LOGS(INFO)
<< "Endpoint channel not available: endpoint_id=" << endpoint_id;
ProcessPreConnectionInitiationFailure(endpoint_id, channel.get(),
connect_impl_result.status,
result.get());
ProcessPreConnectionInitiationFailure(
client, channel->GetMedium(), endpoint_id, channel.get(),
/* is_incoming = */ false, start_time, connect_impl_result.status,
result.get());
return;
}
@@ -522,9 +546,10 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client,
if (!write_exception.Ok()) {
NEARBY_LOGS(INFO) << "Failed to send connection request: endpoint_id="
<< endpoint_id;
ProcessPreConnectionInitiationFailure(endpoint_id, channel.get(),
{Status::kEndpointIoError},
result.get());
ProcessPreConnectionInitiationFailure(
client, channel->GetMedium(), endpoint_id, channel.get(),
/* is_incoming = */ false, start_time, {Status::kEndpointIoError},
result.get());
return;
}
@@ -682,8 +707,9 @@ Exception BasePcpHandler::WriteConnectionRequestFrame(
}
void BasePcpHandler::ProcessPreConnectionInitiationFailure(
const std::string& endpoint_id, EndpointChannel* channel, Status status,
Future<Status>* result) {
ClientProxy* client, Medium medium, const std::string& endpoint_id,
EndpointChannel* channel, bool is_incoming, absl::Time start_time,
Status status, Future<Status>* result) {
if (channel != nullptr) {
channel->Close();
}
@@ -693,6 +719,7 @@ void BasePcpHandler::ProcessPreConnectionInitiationFailure(
result->Set(status);
}
LogConnectionAttempt(client, medium, endpoint_id, is_incoming, start_time);
// result is hold inside a swapper, and saved in PendingConnectionInfo.
// PendingConnectionInfo destructor will clear the memory of SettableFuture
// shared_ptr for result.
@@ -1051,7 +1078,9 @@ Exception BasePcpHandler::OnIncomingConnection(
<< "Failed to parse incoming connection request; client="
<< client->GetClientId()
<< "; device=" << absl::BytesToHexString(remote_endpoint_info.data());
ProcessPreConnectionInitiationFailure("", channel.get(), {Status::kError},
ProcessPreConnectionInitiationFailure(client, medium, "", channel.get(),
/* is_incoming= */ false,
start_time, {Status::kError},
nullptr);
return {Exception::kSuccess};
}
@@ -1217,9 +1246,10 @@ bool BasePcpHandler::BreakTie(ClientProxy* client,
void BasePcpHandler::ProcessTieBreakLoss(
ClientProxy* client, const std::string& endpoint_id,
BasePcpHandler::PendingConnectionInfo* info) {
ProcessPreConnectionInitiationFailure(endpoint_id, info->channel.get(),
{Status::kEndpointIoError},
info->result.lock().get());
ProcessPreConnectionInitiationFailure(
client, info->channel->GetMedium(), endpoint_id, info->channel.get(),
info->is_incoming, info->start_time, {Status::kEndpointIoError},
info->result.lock().get());
ProcessPreConnectionResultFailure(client, endpoint_id);
}
@@ -1430,6 +1460,35 @@ std::string BasePcpHandler::GetHashedConnectionToken(
.substr(0, kConnectionTokenLength);
}
void BasePcpHandler::LogConnectionAttempt(ClientProxy* client, Medium medium,
const std::string& endpoint_id,
bool is_incoming,
absl::Time start_time) {
proto::connections::ConnectionAttemptResult result =
Cancelled(client, endpoint_id) ? proto::connections::RESULT_CANCELLED
: proto::connections::RESULT_ERROR;
if (is_incoming) {
client->GetAnalyticsRecorder().OnIncomingConnectionAttempt(
proto::connections::INITIAL, medium, result,
SystemClock::ElapsedRealtime() - start_time,
/* connection_token= */ "");
} else {
client->GetAnalyticsRecorder().OnOutgoingConnectionAttempt(
endpoint_id, proto::connections::INITIAL, medium, result,
SystemClock::ElapsedRealtime() - start_time,
/* connection_token= */ "");
}
}
bool BasePcpHandler::Cancelled(ClientProxy* client,
const std::string& endpoint_id) {
if (endpoint_id.empty()) {
return false;
}
return client->GetCancellationFlag(endpoint_id)->Cancelled();
}
///////////////////// BasePcpHandler::PendingConnectionInfo ///////////////////
void BasePcpHandler::PendingConnectionInfo::SetCryptoContext(
cpp/core/internal/base_pcp_handler.h
+12 -4
View File
@@ -427,10 +427,10 @@ class BasePcpHandler : public PcpHandler,
bool AppendWebRTCEndpoint(const std::string& endpoint_id,
const ConnectionOptions& local_discovery_options);
void ProcessPreConnectionInitiationFailure(const std::string& endpoint_id,
EndpointChannel* channel,
Status status,
Future<Status>* result);
void ProcessPreConnectionInitiationFailure(
ClientProxy* client, Medium medium, const std::string& endpoint_id,
EndpointChannel* channel, bool is_incoming, absl::Time start_time,
Status status, Future<Status>* result);
void ProcessPreConnectionResultFailure(ClientProxy* client,
const std::string& endpoint_id);
@@ -454,6 +454,14 @@ class BasePcpHandler : public PcpHandler,
// array.
std::string GetHashedConnectionToken(const ByteArray& token_bytes);
static void LogConnectionAttempt(ClientProxy* client, Medium medium,
const std::string& endpoint_id,
bool is_incoming, absl::Time start_time);
// Returns true if the client cancels the operation in progress through the
// endpoint id. This is done by CancellationFlag.
static bool Cancelled(ClientProxy* client, const std::string& endpoint_id);
void WaitForLatch(const std::string& method_name, CountDownLatch* latch);
Status WaitForResult(const std::string& method_name, std::int64_t client_id,
Future<Status>* future);
cpp/core/internal/bwu_manager.cc
+25
View File
@@ -367,6 +367,8 @@ void BwuManager::OnIncomingConnection(
mutable_connection.release());
RunOnBwuManagerThread(
"bwu-on-incoming-connection", [this, client, connection]() {
absl::Time connection_attempt_start_time =
SystemClock::ElapsedRealtime();
EndpointChannel* channel = connection->channel.get();
if (channel == nullptr) {
NEARBY_LOG(
@@ -422,6 +424,13 @@ void BwuManager::OnIncomingConnection(
CHECK(client == mapped_client);
// The ConnectionAttempt has now succeeded, so record it as such.
client->GetAnalyticsRecorder().OnIncomingConnectionAttempt(
proto::connections::UPGRADE, channel->GetMedium(),
proto::connections::RESULT_SUCCESS,
SystemClock::ElapsedRealtime() - connection_attempt_start_time,
client->GetConnectionToken(endpoint_id));
// Use the introductory client information sent over to run the upgrade
// protocol.
RunUpgradeProtocol(mapped_client, endpoint_id,
@@ -537,8 +546,24 @@ void BwuManager::ProcessBwuPathAvailableEvent(
return;
}
absl::Time connection_attempt_start_time = SystemClock::ElapsedRealtime();
auto channel = ProcessBwuPathAvailableEventInternal(client, endpoint_id,
upgrade_path_info);
proto::connections::ConnectionAttemptResult connection_attempt_result;
if (channel != nullptr) {
connection_attempt_result = proto::connections::RESULT_SUCCESS;
} else if (client->GetCancellationFlag(endpoint_id)->Cancelled()) {
connection_attempt_result = proto::connections::RESULT_CANCELLED;
} else {
connection_attempt_result = proto::connections::RESULT_ERROR;
}
client->GetAnalyticsRecorder().OnOutgoingConnectionAttempt(
endpoint_id, proto::connections::UPGRADE, channel->GetMedium(),
connection_attempt_result,
SystemClock::ElapsedRealtime() - connection_attempt_start_time,
client->GetConnectionToken(endpoint_id));
if (channel == nullptr) {
NEARBY_LOG(INFO, "Failed to get new channel.");
RunUpgradeFailedProtocol(client, endpoint_id, upgrade_path_info);
cpp/core/internal/client_proxy.h
+4
View File
@@ -57,6 +57,10 @@ class ClientProxy final {
std::string GetLocalEndpointId();
analytics::AnalyticsRecorder& GetAnalyticsRecorder() const {
return *analytics_recorder_;
}
std::string GetConnectionToken(const std::string& endpoint_id);
// Clears all the runtime state of this client.