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

analytics: Add operation result code for Payload analytics, III

Browse Source 
PiperOrigin-RevId: 704259848
This commit is contained in:
Edwin Wu
2024-12-09 06:20:21 -08:00
committed by Copybara-Service
parent f0fb9662e2
commit c60b6c46f6
5 changed files with 40 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
connections/implementation/client_proxy.cc
+14 -6
View File
@@ -66,6 +66,7 @@
namespace nearby {
namespace connections {
namespace {
using ::location::nearby::connections::OsInfo;
@@ -79,13 +80,8 @@ bool IsFeatureUseStableEndpointIdEnabled() {
connections::config_package_nearby::nearby_connections_feature::
kUseStableEndpointId);
}
} // namespace
// The definition is necessary before C++17.
constexpr absl::Duration
ClientProxy::kHighPowerAdvertisementEndpointIdCacheTimeout;
ClientProxy::ClientProxy(::nearby::analytics::EventLogger* event_logger)
: client_id_(Prng().NextInt64()) {
NEARBY_LOGS(INFO) << "ClientProxy ctor event_logger=" << event_logger;
@@ -558,8 +554,9 @@ void ClientProxy::OnBandwidthChanged(const std::string& endpoint_id,
NEARBY_LOGS(INFO) << "ClientProxy [BandwidthChanged]: id=" << endpoint_id;
MutexLock lock(&mutex_);
const ConnectionPair* item = LookupConnection(endpoint_id);
ConnectionPair* item = LookupConnection(endpoint_id);
if (item != nullptr) {
item->first.connected_medium = new_medium;
item->first.connection_listener.bandwidth_changed_cb(endpoint_id,
new_medium);
NEARBY_LOGS(INFO) << "ClientProxy [reporting onBandwidthChanged]: client="
@@ -600,6 +597,17 @@ bool ClientProxy::ConnectionStatusMatches(const std::string& endpoint_id,
return false;
}
Medium ClientProxy::GetConnectedMedium(const std::string& endpoint_id) const {
MutexLock lock(&mutex_);
const ConnectionPair* item = LookupConnection(endpoint_id);
if (item != nullptr) {
return item->first.connected_medium;
}
return Medium::UNKNOWN_MEDIUM;
}
BooleanMediumSelector ClientProxy::GetUpgradeMediums(
const std::string& endpoint_id) const {
MutexLock lock(&mutex_);
connections/implementation/client_proxy.h
+4
View File
@@ -179,6 +179,9 @@ class ClientProxy final {
// ConnectionListener.disconnected_cb() callback.
void OnDisconnected(const std::string& endpoint_id, bool notify);
// Returns the medium we're currently connected to the endpoint over, or
// UNKNOWN if we don't know or don't have a connection.
Medium GetConnectedMedium(const std::string& endpoint_id) const;
// Returns all mediums eligible for upgrade.
BooleanMediumSelector GetUpgradeMediums(const std::string& endpoint_id) const;
// Returns if this endpoint support 5G for WIFI.
@@ -363,6 +366,7 @@ class ClientProxy final {
kConnected = 1 << 4,
};
bool is_incoming{false};
Medium connected_medium{Medium::UNKNOWN_MEDIUM};
Status status{kPending};
ConnectionListener connection_listener;
ConnectionOptions connection_options;
connections/implementation/endpoint_manager.cc
+1 -3
View File
@@ -30,7 +30,6 @@
#include "connections/implementation/endpoint_channel.h"
#include "connections/implementation/endpoint_channel_manager.h"
#include "connections/implementation/offline_frames.h"
#include "connections/implementation/payload_manager.h"
#include "connections/implementation/proto/offline_wire_formats.pb.h"
#include "connections/implementation/service_id_constants.h"
#include "connections/listeners.h"
@@ -57,9 +56,8 @@ using ::location::nearby::analytics::proto::ConnectionsLog;
using ::location::nearby::connections::OfflineFrame;
using ::location::nearby::connections::PayloadTransferFrame;
using ::location::nearby::connections::V1Frame;
using ::location::nearby::proto::connections::DisconnectionReason;
using ::nearby::analytics::PacketMetaData;
using DisconnectionReason =
::location::nearby::proto::connections::DisconnectionReason;
// We set this to 11s to provide sufficient time for an in-progress WebRTC
// bandwidth upgrade to resolve. This is chosen to be slightly longer than the
connections/implementation/payload_manager.cc
+19 -14
View File
@@ -48,6 +48,7 @@
#include "internal/platform/byte_array.h"
#include "internal/platform/count_down_latch.h"
#include "internal/platform/exception.h"
#include "internal/platform/expected.h"
#include "internal/platform/feature_flags.h"
#include "internal/platform/logging.h"
#include "internal/platform/mutex_lock.h"
@@ -631,8 +632,10 @@ void PayloadManager::OnEndpointDisconnect(ClientProxy* client,
case DisconnectionReason::IO_ERROR:
default:
payload_status = PayloadStatus::ENDPOINT_IO_ERROR;
// TODO(edwinwu): Add for return result code
operation_result_code = OperationResultCode::DETAIL_UNKNOWN;
operation_result_code =
client->GetAnalyticsRecorder()
.GetChannelIoErrorResultCodeFromMedium(
client->GetConnectedMedium(endpoint_id));
break;
}
@@ -783,14 +786,14 @@ PayloadTransferFrame::PayloadChunk PayloadManager::CreatePayloadChunk(
return payload_chunk;
}
std::pair<PayloadManager::PendingPayloadHandle, OperationResultCode>
ErrorOr<PayloadManager::PendingPayloadHandle>
PayloadManager::CreateIncomingPayload(const PayloadTransferFrame& frame,
const std::string& endpoint_id) {
// TODO(edwinwu): Add for return result code
auto internal_payload =
// TODO(edwinwu): Add for return result code in CreateIncomingInternalPayload.
std::unique_ptr<InternalPayload> internal_payload =
CreateIncomingInternalPayload(frame, custom_save_path_);
if (!internal_payload) {
return {PendingPayloadHandle(), OperationResultCode::DETAIL_UNKNOWN};
return {Error(OperationResultCode::DETAIL_UNKNOWN)};
}
Payload::Id payload_id = internal_payload->GetId();
@@ -800,8 +803,7 @@ PayloadManager::CreateIncomingPayload(const PayloadTransferFrame& frame,
std::make_unique<PendingPayload>(
std::move(internal_payload), EndpointIds{endpoint_id}, true,
absl::bind_front(&PayloadManager::OnPendingPayloadDestroy, this)));
return {pending_payloads_.GetPayload(payload_id),
OperationResultCode::DETAIL_UNKNOWN};
return {pending_payloads_.GetPayload(payload_id)};
}
void PayloadManager::OnPendingPayloadDestroy(const PendingPayload* payload) {
@@ -847,13 +849,14 @@ void PayloadManager::SendClientCallbacksForFinishedOutgoingPayload(
// Notify the client.
client->OnPayloadProgress(endpoint_id, update);
// TODO(edwinwu): Add for return result code
// Mark this payload as done for analytics.
client->GetAnalyticsRecorder().OnOutgoingPayloadDone(
endpoint_id, payload_header.id(), status,
(operation_result_code == OperationResultCode::DETAIL_UNKNOWN &&
status == PayloadStatus::ENDPOINT_IO_ERROR)
? OperationResultCode::DETAIL_UNKNOWN
? client->GetAnalyticsRecorder()
.GetChannelIoErrorResultCodeFromMedium(
client->GetConnectedMedium(endpoint_id))
: operation_result_code);
}
@@ -1334,17 +1337,17 @@ void PayloadManager::ProcessDataPacket(
payload_header.total_size());
});
std::pair<PendingPayloadHandle, OperationResultCode> result =
ErrorOr<PendingPayloadHandle> result =
CreateIncomingPayload(payload_transfer_frame, from_endpoint_id);
pending_payload = std::move(result.first);
OperationResultCode operation_result_code = result.second;
if (!pending_payload) {
if (result.has_error()) {
LOG(WARNING) << "PayloadManager failed to create InternalPayload from "
"PayloadTransferFrame with payload_id="
<< payload_header.id() << " and type "
<< payload_header.type() << ", aborting receipt.";
// Analyticize.
OperationResultCode operation_result_code =
result.error().operation_result_code().value();
RunOnStatusUpdateThread(
"process-data-packet",
[to_client, from_endpoint_id, payload_header, operation_result_code]()
@@ -1359,6 +1362,8 @@ void PayloadManager::ProcessDataPacket(
payload_chunk.offset(),
PayloadTransferFrame::ControlMessage::PAYLOAD_ERROR);
return;
} else {
pending_payload = std::move(result.value());
}
// Also, let the client know of this new incoming payload.
RunOnStatusUpdateThread(
connections/implementation/payload_manager.h
+2 -3
View File
@@ -39,6 +39,7 @@
#include "internal/platform/byte_array.h"
#include "internal/platform/condition_variable.h"
#include "internal/platform/count_down_latch.h"
#include "internal/platform/expected.h"
#include "internal/platform/mutex.h"
#include "internal/platform/single_thread_executor.h"
@@ -324,9 +325,7 @@ class PayloadManager : public EndpointManager::FrameProcessor {
LAST_CHUNK) != 0);
}
std::pair<PendingPayloadHandle,
location::nearby::proto::connections::OperationResultCode>
CreateIncomingPayload(
ErrorOr<PendingPayloadHandle> CreateIncomingPayload(
const location::nearby::connections::PayloadTransferFrame& frame,
const std::string& endpoint_id) ABSL_LOCKS_EXCLUDED(mutex_);