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

Remove unnecessary code.

Browse Source 
PiperOrigin-RevId: 908359047
This commit is contained in:
Francis Tsui
2026-04-30 13:55:42 -07:00
committed by Copybara-Service
parent ea61e9cefa
commit 6f52ec53dd
25 changed files with 153 additions and 1185 deletions
Download Patch File Download Diff File Expand all files Collapse all files
connections/implementation/BUILD
+3 -4
View File
@@ -214,6 +214,7 @@ cc_library(
"fake_bwu_handler.h",
"fake_endpoint_channel.h",
"mock_device.h",
"mock_endpoint_channel.h",
"mock_service_controller.h",
"mock_service_controller_router.h",
"offline_simulation_user.h",
@@ -440,8 +441,8 @@ cc_test(
],
deps = [
":internal",
":internal_test",
"//connections:core_types",
"//connections/implementation/analytics",
"//connections/implementation/flags:connections_flags",
"//internal/flags:nearby_flags",
"//internal/platform:base",
@@ -491,13 +492,11 @@ cc_test(
],
deps = [
":internal",
"//connections/implementation/analytics",
":internal_test",
"//internal/platform:base",
"//internal/platform/implementation/g3", # build_cleaner: keep
"//proto:connections_enums_cc_proto",
"@com_github_protobuf_matchers//protobuf-matchers",
"@com_google_absl//absl/strings:string_view",
"@com_google_absl//absl/time",
"@com_google_googletest//:gtest_main",
],
)
connections/implementation/analytics/BUILD
-4
View File
@@ -20,15 +20,12 @@ cc_library(
name = "analytics",
srcs = [
"analytics_recorder.cc",
"throughput_recorder.cc",
],
hdrs = [
"advertising_metadata_params.h",
"analytics_recorder.h",
"connection_attempt_metadata_params.h",
"discovery_metadata_params.h",
"packet_meta_data.h",
"throughput_recorder.h",
],
copts = ["-DCORE_ADAPTER_DLL"],
visibility = ["//connections:__subpackages__"],
@@ -58,7 +55,6 @@ cc_test(
size = "small",
srcs = [
"analytics_recorder_test.cc",
"throughput_recorder_test.cc",
],
shard_count = 16,
deps = [
connections/implementation/analytics/packet_meta_data.h
-101
View File
@@ -1,101 +0,0 @@
// Copyright 2022-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 NEARBY_CONNECTIONS_IMPLEMENTATION_ANALYTICS_PACKET_META_DATA_H_
#define NEARBY_CONNECTIONS_IMPLEMENTATION_ANALYTICS_PACKET_META_DATA_H_
#include <cstdint>
#include "absl/time/time.h"
#include "internal/platform/implementation/system_clock.h"
namespace nearby {
namespace analytics {
struct PacketMetaData {
int packet_size;
absl::Time file_io_start_time;
absl::Time file_io_end_time;
absl::Time encryption_start_time;
absl::Time encryption_end_time;
absl::Time socket_io_start_time;
absl::Time socket_io_end_time;
void Reset() {
file_io_start_time = SystemClock::ElapsedRealtime();
encryption_start_time = SystemClock::ElapsedRealtime();
socket_io_start_time = SystemClock::ElapsedRealtime();
packet_size = 0;
}
void SetPacketSize(int packet_size) {
this->packet_size = packet_size;
}
int GetPacketSize() const {
return packet_size;
}
void StartFileIo() {
file_io_start_time = SystemClock::ElapsedRealtime();
}
void StopFileIo() {
file_io_end_time = SystemClock::ElapsedRealtime();
}
void StartEncryption() {
encryption_start_time = SystemClock::ElapsedRealtime();
}
void StopEncryption() {
encryption_end_time = SystemClock::ElapsedRealtime();
}
void StartSocketIo() {
socket_io_start_time = SystemClock::ElapsedRealtime();
}
void StopSocketIo() {
socket_io_end_time = SystemClock::ElapsedRealtime();
}
int64_t GetEncryptionTimeInMillis() const {
if (encryption_end_time > encryption_start_time) {
return absl::ToInt64Milliseconds(encryption_end_time -
encryption_start_time);
}
return 0;
}
int64_t GetFileIoTimeInMillis() const {
if (file_io_end_time > file_io_start_time) {
return absl::ToInt64Milliseconds(file_io_end_time - file_io_start_time);
}
return 0;
}
int64_t GetSocketIoTimeInMillis() const {
if (socket_io_end_time > socket_io_start_time) {
return absl::ToInt64Milliseconds(socket_io_end_time -
socket_io_start_time);
}
return 0;
}
};
} // namespace analytics
} // namespace nearby
#endif // NEARBY_CONNECTIONS_IMPLEMENTATION_ANALYTICS_PACKET_META_DATA_H_
connections/implementation/analytics/throughput_recorder.cc
-362
View File
@@ -1,362 +0,0 @@
// Copyright 2022-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.
#include "connections/implementation/analytics/throughput_recorder.h"
#include <cstdint>
#include <memory>
#include <string>
#include <utility>
#include "absl/base/no_destructor.h"
#include "absl/container/flat_hash_map.h"
#include "absl/strings/str_format.h"
#include "absl/time/time.h"
#include "connections/implementation/analytics/packet_meta_data.h"
#include "connections/payload_type.h"
#include "internal/platform/implementation/system_clock.h"
#include "internal/platform/logging.h"
#include "internal/platform/mutex_lock.h"
namespace nearby {
namespace analytics {
using Medium = ::location::nearby::proto::connections::Medium;
using ::nearby::connections::PayloadDirection;
using ::nearby::connections::PayloadType;
namespace {
constexpr int kDefaultThroughoutKbps = 0;
constexpr int kKbInBytes = 1024;
constexpr int kSecInMs = 1000;
int64_t CalculateThroughputKBps(int64_t total_byte_size, int64_t total_millis) {
if (total_millis > 0) {
return total_byte_size * kSecInMs / kKbInBytes / total_millis;
}
return kDefaultThroughoutKbps;
}
int64_t CalculateThroughputMBps(int64_t throughputKBps) {
return throughputKBps / kKbInBytes;
}
std::string ToString(PayloadType type) {
switch (type) {
case PayloadType::kBytes:
return std::string("Bytes");
case PayloadType::kStream:
return std::string("Stream");
case PayloadType::kFile:
return std::string("File");
case PayloadType::kUnknown:
return std::string("Unknown");
}
}
} // namespace
ThroughputRecorderContainer& ThroughputRecorderContainer::GetInstance() {
static absl::NoDestructor<ThroughputRecorderContainer> instance;
return *instance;
}
ThroughputRecorderContainer::ThroughputRecorder::ThroughputRecorder(
int64_t payload_id, PayloadDirection payload_direction,
PayloadType payload_type)
: payload_id_(payload_id),
payload_direction_(payload_direction),
payload_type_(payload_type) {
LOG_IF(DFATAL, payload_type_ == PayloadType::kUnknown)
<< "Invalid payload type";
}
void ThroughputRecorderContainer::ThroughputRecorder::Start() {
if (VLOG_IS_ON(1)) {
std::string direction =
(payload_direction_ == PayloadDirection::INCOMING_PAYLOAD) ? "; Receive"
: "; Send";
VLOG(1) << "Start TP profiling for payload_id:" << payload_id_ << direction;
}
start_timestamp_ = SystemClock::ElapsedRealtime();
}
bool ThroughputRecorderContainer::ThroughputRecorder::Stop() {
VLOG(1) << "Stop TP profiling for payload_id:" << payload_id_;
{
absl::Time stop_timestamp = SystemClock::ElapsedRealtime();
int64_t total_byte_size = 0;
int medium_size = throughputs_.size();
if (!success_) {
if (!throughputs_.empty()) {
for (auto& tp : throughputs_) {
tp.second.SetLastTimestamp(stop_timestamp);
}
}
}
for (auto& tp : throughputs_) {
tp.second.dump(payload_direction_, payload_type_);
total_byte_size += tp.second.GetTotalByteSize();
}
throughputs_.clear();
int64_t total_millis =
absl::ToInt64Milliseconds(stop_timestamp - start_timestamp_);
throughput_kbps_ = CalculateThroughputKBps(total_byte_size, total_millis);
int64_t throughput_mbps = CalculateThroughputMBps(throughput_kbps_);
if (medium_size > 1) {
if (throughput_kbps_ != kDefaultThroughoutKbps) {
std::string dump_content = absl::StrFormat(
"%s %s data(%lld bytes) %s, overall used %lld milliseconds, "
"throughput "
"is %lld MB/s (%lld KB/s), File IO takes %lld ms, %s takes %lld "
"ms, "
"Socket IO takes %lld ms",
(payload_direction_ == PayloadDirection::INCOMING_PAYLOAD)
? "Received"
: "Sent",
ToString(payload_type_), total_byte_size,
success_ ? "SUCCEEDED" : "FAILED", total_millis, throughput_mbps,
throughput_kbps_, file_io_time_,
(payload_direction_ == PayloadDirection::INCOMING_PAYLOAD)
? "Decryption"
: "Encryption",
encryption_time_, socket_io_time_);
LOG(INFO) << dump_content;
}
}
}
return true;
}
void ThroughputRecorderContainer::ThroughputRecorder::MarkAsSuccess() {
success_ = true;
}
void ThroughputRecorderContainer::ThroughputRecorder::Throughput::Add(
int frame_size, int64_t file_io_time, int64_t encryption_time,
int64_t socket_io_time) {
total_byte_size_ += frame_size;
last_timestamp_ = SystemClock::ElapsedRealtime();
file_io_time_ += file_io_time;
encryption_time_ += encryption_time;
socket_io_time_ += socket_io_time;
}
bool ThroughputRecorderContainer::ThroughputRecorder::Throughput::dump(
PayloadDirection payload_direction, PayloadType payload_type) {
int64_t total_millis =
absl::ToInt64Milliseconds(last_timestamp_ - start_timestamp_);
int64_t throughput_kbps =
CalculateThroughputKBps(total_byte_size_, total_millis);
if (throughput_kbps == kDefaultThroughoutKbps) {
return false;
}
int64_t throughput_mbps = CalculateThroughputMBps(throughput_kbps);
int64_t other =
total_millis - file_io_time_ - encryption_time_ - socket_io_time_;
std::string dump_content = absl::StrFormat(
"%s %s data(%lld bytes) via %s used %lld ms, throughput is %lld "
"MB/s (%lld KB/s), File IO takes %lld ms, %s takes %lld ms, "
"Socket IO takes %lld ms, "
"Other takes %lld ms",
(payload_direction == PayloadDirection::INCOMING_PAYLOAD) ? "Received"
: "Sent",
ToString(payload_type), total_byte_size_,
location::nearby::proto::connections::Medium_Name(medium_), total_millis,
throughput_mbps, throughput_kbps, file_io_time_,
(payload_direction == PayloadDirection::INCOMING_PAYLOAD) ? "Decryption"
: "Encryption",
encryption_time_, socket_io_time_, other);
LOG(INFO) << dump_content;
return true;
}
ThroughputRecorderContainer::ThroughputRecorder::Throughput&
ThroughputRecorderContainer::ThroughputRecorder::GetThroughput(
Medium medium, int64_t duration_millis) {
auto it = throughputs_.find(medium);
if (it == throughputs_.end()) {
throughputs_.emplace(
medium, Throughput(medium, SystemClock::ElapsedRealtime() -
absl::Milliseconds(duration_millis)));
return throughputs_.find(medium)->second;
}
return it->second;
}
int ThroughputRecorderContainer::ThroughputRecorder::GetThroughputsSize()
const {
return throughputs_.size();
}
int64_t ThroughputRecorderContainer::ThroughputRecorder::GetThroughputKbps()
const {
return throughput_kbps_;
}
int64_t ThroughputRecorderContainer::ThroughputRecorder::GetDurationMillis()
const {
return duration_millis_;
}
void ThroughputRecorderContainer::ThroughputRecorder::UpdateFrameData(
Medium medium, PacketMetaData& packetMetaData) {
duration_millis_ = packetMetaData.GetEncryptionTimeInMillis() +
packetMetaData.GetFileIoTimeInMillis() +
packetMetaData.GetSocketIoTimeInMillis();
GetThroughput(medium, duration_millis_)
.Add(packetMetaData.packet_size, packetMetaData.GetFileIoTimeInMillis(),
packetMetaData.GetEncryptionTimeInMillis(),
packetMetaData.GetSocketIoTimeInMillis());
CalculateDurationTimes(packetMetaData);
}
void ThroughputRecorderContainer::ThroughputRecorder::CalculateDurationTimes(
const PacketMetaData& packetMetaData) {
encryption_time_ += packetMetaData.GetEncryptionTimeInMillis();
socket_io_time_ += packetMetaData.GetSocketIoTimeInMillis();
file_io_time_ += packetMetaData.GetFileIoTimeInMillis();
}
// Implementation for ThroughputRecorderContainer
void ThroughputRecorderContainer::Start(int64_t payload_id,
PayloadDirection payload_direction,
PayloadType payload_type) {
if (payload_type == PayloadType::kUnknown) {
return;
}
MutexLock lock(&mutex_);
auto it = throughput_recorders_.find(
std::pair<int64_t, PayloadDirection>(payload_id, payload_direction));
if (it == throughput_recorders_.end()) {
auto instance = std::make_unique<ThroughputRecorder>(
payload_id, payload_direction, payload_type);
instance->Start();
throughput_recorders_.emplace(
std::pair<int64_t, PayloadDirection>(payload_id, payload_direction),
std::move(instance));
} else {
it->second->Start();
}
}
void ThroughputRecorderContainer::UpdateFrameData(
int64_t payload_id, PayloadDirection payload_direction, Medium medium,
PacketMetaData& packet_meta_data) {
MutexLock lock(&mutex_);
auto it = throughput_recorders_.find(
std::pair<int64_t, PayloadDirection>(payload_id, payload_direction));
if (it != throughput_recorders_.end()) {
it->second->UpdateFrameData(medium, packet_meta_data);
}
}
void ThroughputRecorderContainer::MarkAsSuccess(
int64_t payload_id, PayloadDirection payload_direction) {
MutexLock lock(&mutex_);
auto it = throughput_recorders_.find(
std::pair<int64_t, PayloadDirection>(payload_id, payload_direction));
if (it != throughput_recorders_.end()) {
it->second->MarkAsSuccess();
}
}
int64_t ThroughputRecorderContainer::StopTPRecorder(
int64_t payload_id, PayloadDirection payload_direction) {
MutexLock lock(&mutex_);
auto it = throughput_recorders_.find(
std::pair<int64_t, PayloadDirection>(payload_id, payload_direction));
if (it != throughput_recorders_.end()) {
it->second->Stop();
int64_t throughput_kbps = it->second->GetThroughputKbps();
throughput_recorders_.erase(it);
return throughput_kbps;
}
return 0;
}
int ThroughputRecorderContainer::GetSize() {
MutexLock lock(&mutex_);
return throughput_recorders_.size();
}
void ThroughputRecorderContainer::ClearForTest() {
MutexLock lock(&mutex_);
throughput_recorders_.clear();
}
int64_t ThroughputRecorderContainer::GetTotalByteSizeForTesting(
int64_t payload_id, PayloadDirection payload_direction, Medium medium) {
MutexLock lock(&mutex_);
auto it = throughput_recorders_.find(
std::pair<int64_t, PayloadDirection>(payload_id, payload_direction));
if (it != throughput_recorders_.end()) {
return it->second->GetThroughput(medium, 0).GetTotalByteSize();
}
return 0;
}
int ThroughputRecorderContainer::GetThroughputsSizeForTesting(
int64_t payload_id, PayloadDirection payload_direction) {
MutexLock lock(&mutex_);
auto it = throughput_recorders_.find(
std::pair<int64_t, PayloadDirection>(payload_id, payload_direction));
if (it != throughput_recorders_.end()) {
return it->second->GetThroughputsSize();
}
return 0;
}
int64_t ThroughputRecorderContainer::GetDurationMillisForTesting(
int64_t payload_id, PayloadDirection payload_direction) {
MutexLock lock(&mutex_);
auto it = throughput_recorders_.find(
std::pair<int64_t, PayloadDirection>(payload_id, payload_direction));
if (it != throughput_recorders_.end()) {
return it->second->GetDurationMillis();
}
return 0;
}
int64_t ThroughputRecorderContainer::GetThroughputKbpsForTesting(
int64_t payload_id, PayloadDirection payload_direction) {
MutexLock lock(&mutex_);
auto it = throughput_recorders_.find(
std::pair<int64_t, PayloadDirection>(payload_id, payload_direction));
if (it != throughput_recorders_.end()) {
return it->second->GetThroughputKbps();
}
return 0;
}
bool ThroughputRecorderContainer::DumpForTesting(
int64_t payload_id, PayloadDirection payload_direction, Medium medium) {
MutexLock lock(&mutex_);
auto it = throughput_recorders_.find(
std::pair<int64_t, PayloadDirection>(payload_id, payload_direction));
if (it != throughput_recorders_.end()) {
return it->second->GetThroughput(medium, 0).dump(
payload_direction, it->second->GetPayloadType());
}
return false;
}
} // namespace analytics
} // namespace nearby
connections/implementation/analytics/throughput_recorder.h
-178
View File
@@ -1,178 +0,0 @@
// Copyright 2022-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 NEARBY_CONNECTIONS_IMPLEMENTATION_ANALYTICS_THROUGHPUT_RECORDER_H_
#define NEARBY_CONNECTIONS_IMPLEMENTATION_ANALYTICS_THROUGHPUT_RECORDER_H_
#include <cstdint>
#include <memory>
#include <utility>
#include "absl/base/no_destructor.h"
#include "absl/base/thread_annotations.h"
#include "absl/container/flat_hash_map.h"
#include "absl/time/time.h"
#include "connections/implementation/analytics/packet_meta_data.h"
#include "connections/payload_type.h"
#include "internal/platform/mutex.h"
namespace nearby {
namespace analytics {
// Container class to manage ThroughputRecorder instances.
// This class is a singleton and provides thread-safe proxy methods to record
// throughput for different payloads.
class ThroughputRecorderContainer {
public:
static ThroughputRecorderContainer& GetInstance();
// Records the start of a payload transfer.
void Start(int64_t payload_id,
connections::PayloadDirection payload_direction,
connections::PayloadType payload_type) ABSL_LOCKS_EXCLUDED(mutex_);
// Records when a frame is sent or received.
void UpdateFrameData(int64_t payload_id,
connections::PayloadDirection payload_direction,
location::nearby::proto::connections::Medium medium,
PacketMetaData& packet_meta_data)
ABSL_LOCKS_EXCLUDED(mutex_);
// Marks a payload transfer as successful.
void MarkAsSuccess(int64_t payload_id,
connections::PayloadDirection payload_direction)
ABSL_LOCKS_EXCLUDED(mutex_);
// Stops and removes the throughput recorder for a given payload.
// This calculates and logs the final throughput statistics.
// Returns the throughput in KBps.
int64_t StopTPRecorder(int64_t payload_id,
connections::PayloadDirection payload_direction)
ABSL_LOCKS_EXCLUDED(mutex_);
// Returns the number of active recorder instances.
int GetSize() ABSL_LOCKS_EXCLUDED(mutex_);
// Clear all recorders. Used for testing.
void ClearForTest() ABSL_LOCKS_EXCLUDED(mutex_);
// Testing proxy methods
int64_t GetTotalByteSizeForTesting(
int64_t payload_id, connections::PayloadDirection payload_direction,
location::nearby::proto::connections::Medium medium)
ABSL_LOCKS_EXCLUDED(mutex_);
int GetThroughputsSizeForTesting(
int64_t payload_id, connections::PayloadDirection payload_direction)
ABSL_LOCKS_EXCLUDED(mutex_);
int64_t GetDurationMillisForTesting(
int64_t payload_id, connections::PayloadDirection payload_direction)
ABSL_LOCKS_EXCLUDED(mutex_);
int64_t GetThroughputKbpsForTesting(
int64_t payload_id, connections::PayloadDirection payload_direction)
ABSL_LOCKS_EXCLUDED(mutex_);
bool DumpForTesting(int64_t payload_id,
connections::PayloadDirection payload_direction,
location::nearby::proto::connections::Medium medium)
ABSL_LOCKS_EXCLUDED(mutex_);
private:
friend class absl::NoDestructor<ThroughputRecorderContainer>;
class ThroughputRecorder {
public:
ThroughputRecorder(int64_t payload_id,
connections::PayloadDirection payload_direction,
connections::PayloadType payload_type);
~ThroughputRecorder() = default;
void Start();
bool Stop() ABSL_LOCKS_EXCLUDED(mutex_);
class Throughput {
public:
Throughput() = default;
~Throughput() = default;
Throughput(location::nearby::proto::connections::Medium medium,
absl::Time start_timestamp)
: medium_(medium), start_timestamp_(start_timestamp) {}
void Add(int frame_size, int64_t file_io_time, int64_t encryption_time,
int64_t socket_io_time);
void SetLastTimestamp(absl::Time time_stamp) {
last_timestamp_ = time_stamp;
}
int64_t GetTotalByteSize() const { return total_byte_size_; }
bool dump(connections::PayloadDirection payload_direction,
connections::PayloadType payload_type);
private:
const ::location::nearby::proto::connections::Medium medium_;
const absl::Time start_timestamp_;
int64_t total_byte_size_ = 0;
absl::Time last_timestamp_;
int64_t file_io_time_ = 0;
int64_t encryption_time_ = 0;
int64_t socket_io_time_ = 0;
};
Throughput& GetThroughput(
location::nearby::proto::connections::Medium medium,
int64_t duration_millis);
int GetThroughputsSize() const;
int64_t GetThroughputKbps() const;
int64_t GetDurationMillis() const;
void UpdateFrameData(location::nearby::proto::connections::Medium medium,
PacketMetaData& packetMetaData);
void MarkAsSuccess();
connections::PayloadType GetPayloadType() const { return payload_type_; }
private:
void CalculateDurationTimes(const PacketMetaData& packetMetaData);
const int64_t payload_id_;
const connections::PayloadDirection payload_direction_;
const connections::PayloadType payload_type_;
absl::Time start_timestamp_;
absl::flat_hash_map<location::nearby::proto::connections::Medium,
Throughput>
throughputs_;
bool success_ = false;
int64_t file_io_time_ = 0;
int64_t encryption_time_ = 0;
int64_t socket_io_time_ = 0;
int64_t duration_millis_ = 0;
int64_t throughput_kbps_ = 0;
};
ThroughputRecorderContainer() = default;
ThroughputRecorderContainer(const ThroughputRecorderContainer&) = delete;
ThroughputRecorderContainer& operator=(const ThroughputRecorderContainer&) =
delete;
~ThroughputRecorderContainer() = default;
Mutex mutex_;
// std::pair<int64_t, PayloadDirection> for <payload id, payload direction>
absl::flat_hash_map<std::pair<int64_t, connections::PayloadDirection>,
std::unique_ptr<ThroughputRecorder>>
throughput_recorders_ ABSL_GUARDED_BY(mutex_);
};
} // namespace analytics
} // namespace nearby
#endif // NEARBY_CONNECTIONS_IMPLEMENTATION_ANALYTICS_THROUGHPUT_RECORDER_H_
connections/implementation/analytics/throughput_recorder_test.cc
-241
View File
@@ -1,241 +0,0 @@
// Copyright 2022-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.
#include "connections/implementation/analytics/throughput_recorder.h"
#include <stdint.h>
#include <string>
#include "gtest/gtest.h"
#include "absl/time/clock.h"
#include "absl/time/time.h"
#include "connections/implementation/analytics/packet_meta_data.h"
#include "connections/payload_type.h"
#include "internal/platform/logging.h"
#include "proto/connections_enums.pb.h"
namespace nearby {
namespace analytics {
namespace {
constexpr int64_t kPayloadIdA = 123456789;
constexpr int64_t kPayloadIdB = 987654321;
constexpr int kFrameSize = 10 * 64 * 1024;
class ThroughputRecorderTest : public testing::TestWithParam<bool> {
protected:
ThroughputRecorderTest() = default;
~ThroughputRecorderTest() override {
ThroughputRecorderContainer::GetInstance().ClearForTest();
}
ThroughputRecorderContainer& tp_recorder_container_ =
ThroughputRecorderContainer::GetInstance();
};
INSTANTIATE_TEST_SUITE_P(ParametrisedTestThroughputRecorderTest,
ThroughputRecorderTest, testing::Values(true, false));
TEST(ThroughputRecorderContainer, InstanceCreate_ContainerSize) {
ThroughputRecorderContainer& TPRecorderContainer =
ThroughputRecorderContainer::GetInstance();
TPRecorderContainer.Start(kPayloadIdA,
connections::PayloadDirection::OUTGOING_PAYLOAD,
connections::PayloadType::kFile);
TPRecorderContainer.Start(kPayloadIdB,
connections::PayloadDirection::INCOMING_PAYLOAD,
connections::PayloadType::kFile);
EXPECT_EQ(ThroughputRecorderContainer::GetInstance().GetSize(), 2);
ThroughputRecorderContainer::GetInstance().ClearForTest();
EXPECT_EQ(ThroughputRecorderContainer::GetInstance().GetSize(), 0);
}
TEST_F(ThroughputRecorderTest, OnFrameSentSaveTransferredSize) {
tp_recorder_container_.Start(kPayloadIdA,
connections::PayloadDirection::OUTGOING_PAYLOAD,
connections::PayloadType::kFile);
PacketMetaData packet_meta_data;
packet_meta_data.SetPacketSize(kFrameSize);
tp_recorder_container_.UpdateFrameData(
kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD,
location::nearby::proto::connections::BLE, packet_meta_data);
tp_recorder_container_.UpdateFrameData(
kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD,
location::nearby::proto::connections::BLE, packet_meta_data);
tp_recorder_container_.UpdateFrameData(
kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD,
location::nearby::proto::connections::BLE, packet_meta_data);
EXPECT_EQ(tp_recorder_container_.GetTotalByteSizeForTesting(
kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD,
location::nearby::proto::connections::BLE),
kFrameSize * 3);
}
TEST_F(ThroughputRecorderTest, OnIgnoreUnkownPaylaodType) {
tp_recorder_container_.Start(kPayloadIdA,
connections::PayloadDirection::OUTGOING_PAYLOAD,
connections::PayloadType::kUnknown);
PacketMetaData packet_meta_data;
tp_recorder_container_.UpdateFrameData(
kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD,
location::nearby::proto::connections::BLE, packet_meta_data);
EXPECT_EQ(tp_recorder_container_.GetThroughputsSizeForTesting(
kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD),
0);
tp_recorder_container_.Start(kPayloadIdA,
connections::PayloadDirection::INCOMING_PAYLOAD,
connections::PayloadType::kUnknown);
tp_recorder_container_.UpdateFrameData(
kPayloadIdA, connections::PayloadDirection::INCOMING_PAYLOAD,
location::nearby::proto::connections::BLE, packet_meta_data);
EXPECT_EQ(tp_recorder_container_.GetThroughputsSizeForTesting(
kPayloadIdA, connections::PayloadDirection::INCOMING_PAYLOAD),
0);
}
TEST_P(ThroughputRecorderTest, OnFrameSentStopAndDump) {
tp_recorder_container_.Start(kPayloadIdA,
connections::PayloadDirection::OUTGOING_PAYLOAD,
connections::PayloadType::kFile);
PacketMetaData packet_meta_data;
packet_meta_data.SetPacketSize(kFrameSize);
packet_meta_data.StartFileIo();
absl::SleepFor(absl::Milliseconds(5));
packet_meta_data.StopFileIo();
packet_meta_data.StartEncryption();
absl::SleepFor(absl::Milliseconds(6));
packet_meta_data.StopEncryption();
packet_meta_data.StartSocketIo();
absl::SleepFor(absl::Milliseconds(7));
packet_meta_data.StopSocketIo();
tp_recorder_container_.UpdateFrameData(
kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD,
location::nearby::proto::connections::BLE, packet_meta_data);
EXPECT_EQ(tp_recorder_container_.GetDurationMillisForTesting(
kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD),
packet_meta_data.GetEncryptionTimeInMillis() +
packet_meta_data.GetFileIoTimeInMillis() +
packet_meta_data.GetSocketIoTimeInMillis());
packet_meta_data.SetPacketSize(kFrameSize);
packet_meta_data.StartFileIo();
absl::SleepFor(absl::Milliseconds(15));
packet_meta_data.StopFileIo();
packet_meta_data.StartEncryption();
absl::SleepFor(absl::Milliseconds(16));
packet_meta_data.StopEncryption();
packet_meta_data.StartSocketIo();
absl::SleepFor(absl::Milliseconds(17));
packet_meta_data.StopSocketIo();
tp_recorder_container_.UpdateFrameData(
kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD,
location::nearby::proto::connections::BLE, packet_meta_data);
if (GetParam() == true) {
LOG(INFO) << "MarkAsSuccess";
tp_recorder_container_.MarkAsSuccess(
kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD);
}
int throughput_kbps = tp_recorder_container_.StopTPRecorder(
kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD);
EXPECT_NE(throughput_kbps, 0);
EXPECT_EQ(tp_recorder_container_.GetSize(), 0);
}
TEST_F(ThroughputRecorderTest, OnFrameSentStopAndDumpForMultiMeadium) {
tp_recorder_container_.Start(kPayloadIdA,
connections::PayloadDirection::OUTGOING_PAYLOAD,
connections::PayloadType::kFile);
PacketMetaData packet_meta_data1;
packet_meta_data1.SetPacketSize(kFrameSize);
packet_meta_data1.StartFileIo();
absl::SleepFor(absl::Milliseconds(5));
packet_meta_data1.StopFileIo();
packet_meta_data1.StartEncryption();
absl::SleepFor(absl::Milliseconds(6));
packet_meta_data1.StopEncryption();
packet_meta_data1.StartSocketIo();
absl::SleepFor(absl::Milliseconds(7));
packet_meta_data1.StopSocketIo();
tp_recorder_container_.UpdateFrameData(
kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD,
location::nearby::proto::connections::BLE, packet_meta_data1);
PacketMetaData packet_meta_data2;
packet_meta_data2.SetPacketSize(kFrameSize);
packet_meta_data2.StartFileIo();
absl::SleepFor(absl::Milliseconds(15));
packet_meta_data2.StopFileIo();
packet_meta_data2.StartEncryption();
absl::SleepFor(absl::Milliseconds(16));
packet_meta_data2.StopEncryption();
packet_meta_data2.StartSocketIo();
absl::SleepFor(absl::Milliseconds(17));
packet_meta_data2.StopSocketIo();
tp_recorder_container_.UpdateFrameData(
kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD,
location::nearby::proto::connections::WIFI_LAN, packet_meta_data2);
tp_recorder_container_.MarkAsSuccess(
kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD);
int throughput_kbps = tp_recorder_container_.StopTPRecorder(
kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD);
EXPECT_NE(throughput_kbps, 0);
}
TEST_F(ThroughputRecorderTest, OnFrameReceivedCheckDurationMillis) {
tp_recorder_container_.Start(kPayloadIdA,
connections::PayloadDirection::INCOMING_PAYLOAD,
connections::PayloadType::kFile);
PacketMetaData packet_meta_data;
packet_meta_data.SetPacketSize(kFrameSize);
packet_meta_data.StartFileIo();
absl::SleepFor(absl::Milliseconds(5));
packet_meta_data.StopFileIo();
packet_meta_data.StartEncryption();
absl::SleepFor(absl::Milliseconds(6));
packet_meta_data.StopEncryption();
packet_meta_data.StartSocketIo();
absl::SleepFor(absl::Milliseconds(7));
packet_meta_data.StopSocketIo();
tp_recorder_container_.UpdateFrameData(
kPayloadIdA, connections::PayloadDirection::INCOMING_PAYLOAD,
location::nearby::proto::connections::BLE, packet_meta_data);
EXPECT_EQ(tp_recorder_container_.GetDurationMillisForTesting(
kPayloadIdA, connections::PayloadDirection::INCOMING_PAYLOAD),
packet_meta_data.GetEncryptionTimeInMillis() +
packet_meta_data.GetFileIoTimeInMillis() +
packet_meta_data.GetSocketIoTimeInMillis());
}
TEST_F(ThroughputRecorderTest, OnTPRecorderNotStarted) {
tp_recorder_container_.Start(kPayloadIdA,
connections::PayloadDirection::OUTGOING_PAYLOAD,
connections::PayloadType::kUnknown);
EXPECT_FALSE(tp_recorder_container_.DumpForTesting(
kPayloadIdA, connections::PayloadDirection::OUTGOING_PAYLOAD,
location::nearby::proto::connections::BLE));
}
} // namespace
} // namespace analytics
} // namespace nearby
connections/implementation/base_endpoint_channel.cc
-22
View File
@@ -92,17 +92,10 @@ BaseEndpointChannel::BaseEndpointChannel(
try_count_(try_count) {}
ExceptionOr<ByteArray> BaseEndpointChannel::Read() {
PacketMetaData packet_meta_data;
return Read(packet_meta_data);
}
ExceptionOr<ByteArray> BaseEndpointChannel::Read(
PacketMetaData& packet_meta_data) {
ByteArray result;
{
MutexLock lock(&reader_mutex_);
packet_meta_data.StartSocketIo();
ExceptionOr<std::int32_t> read_int;
if (NearbyFlags::GetInstance().GetBoolFlag(
config_package_nearby::nearby_connections_feature::
@@ -133,8 +126,6 @@ ExceptionOr<ByteArray> BaseEndpointChannel::Read(
if (!read_bytes.ok()) {
return read_bytes;
}
packet_meta_data.StopSocketIo();
packet_meta_data.SetPacketSize(read_int.result() + sizeof(std::int32_t));
result = std::move(read_bytes.result());
}
@@ -144,7 +135,6 @@ ExceptionOr<ByteArray> BaseEndpointChannel::Read(
if (IsEncryptionEnabledLocked()) {
// If encryption is enabled, decode the message.
std::string input(std::move(result));
packet_meta_data.StartEncryption();
std::unique_ptr<std::string> decrypted_data =
crypto_context_->DecodeMessageFromPeer(input);
if (decrypted_data) {
@@ -175,7 +165,6 @@ ExceptionOr<ByteArray> BaseEndpointChannel::Read(
<< ": Unable to parse data as unencrypted message.";
}
}
packet_meta_data.StopEncryption();
if (result.Empty()) {
LOG(WARNING) << __func__ << ": Unable to parse read result.";
return ExceptionOr<ByteArray>(message_exception);
@@ -191,12 +180,6 @@ ExceptionOr<ByteArray> BaseEndpointChannel::Read(
}
Exception BaseEndpointChannel::Write(absl::string_view data) {
PacketMetaData packet_meta_data;
return Write(data, packet_meta_data);
}
Exception BaseEndpointChannel::Write(absl::string_view data,
PacketMetaData& packet_meta_data) {
{
MutexLock pause_lock(&is_paused_mutex_);
if (is_paused_) {
@@ -217,9 +200,7 @@ Exception BaseEndpointChannel::Write(absl::string_view data,
MutexLock crypto_lock(&crypto_mutex_);
if (IsEncryptionEnabledLocked()) {
// If encryption is enabled, encode the message.
packet_meta_data.StartEncryption();
encrypted = crypto_context_->EncodeMessageToPeer(data);
packet_meta_data.StopEncryption();
if (!encrypted) {
LOG(WARNING) << __func__ << ": Failed to encrypt data.";
return {Exception::kIo};
@@ -235,7 +216,6 @@ Exception BaseEndpointChannel::Write(absl::string_view data,
return {Exception::kIo};
}
packet_meta_data.StartSocketIo();
Exception write_exception;
if (NearbyFlags::GetInstance().GetBoolFlag(
config_package_nearby::nearby_connections_feature::
@@ -262,8 +242,6 @@ Exception BaseEndpointChannel::Write(absl::string_view data,
<< ": Failed to flush writer: " << flush_exception.value;
return flush_exception;
}
packet_meta_data.StopSocketIo();
packet_meta_data.SetPacketSize(data_size + sizeof(std::uint32_t));
}
{
connections/implementation/base_endpoint_channel.h
+2 -7
View File
@@ -23,7 +23,6 @@
#include "absl/strings/string_view.h"
#include "absl/time/time.h"
#include "connections/implementation/analytics/analytics_recorder.h"
#include "connections/implementation/analytics/packet_meta_data.h"
#include "connections/implementation/endpoint_channel.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/condition_variable.h"
@@ -35,8 +34,6 @@
namespace nearby {
namespace connections {
using analytics::PacketMetaData;
class BaseEndpointChannel : public EndpointChannel {
public:
BaseEndpointChannel(const std::string& service_id,
@@ -51,12 +48,10 @@ class BaseEndpointChannel : public EndpointChannel {
~BaseEndpointChannel() override = default;
// EndpointChannel:
ExceptionOr<ByteArray> Read() override;
ExceptionOr<ByteArray> Read(PacketMetaData& packet_meta_data)
ExceptionOr<ByteArray> Read()
ABSL_LOCKS_EXCLUDED(reader_mutex_, crypto_mutex_,
last_read_mutex_) override;
Exception Write(absl::string_view data) override;
Exception Write(absl::string_view data, PacketMetaData& packet_meta_data)
Exception Write(absl::string_view data)
ABSL_LOCKS_EXCLUDED(writer_mutex_, crypto_mutex_) override;
void Close() ABSL_LOCKS_EXCLUDED(is_paused_mutex_) override;
void Close(location::nearby::proto::connections::DisconnectionReason reason)
connections/implementation/base_pcp_handler.cc
+1 -2
View File
@@ -1666,8 +1666,7 @@ Status BasePcpHandler::RejectConnection(ClientProxy* client,
void BasePcpHandler::OnIncomingFrame(
OfflineFrame& frame, const std::string& endpoint_id, ClientProxy* client,
location::nearby::proto::connections::Medium medium,
PacketMetaData& packet_meta_data) {
location::nearby::proto::connections::Medium medium) {
CountDownLatch latch(1);
RunOnPcpHandlerThread(
"incoming-frame",
connections/implementation/base_pcp_handler.h
+4 -5
View File
@@ -31,7 +31,6 @@
#include "connections/advertising_options.h"
#include "connections/connection_options.h"
#include "connections/discovery_options.h"
#include "connections/implementation/analytics/packet_meta_data.h"
#include "connections/implementation/bwu_manager.h"
#include "connections/implementation/client_proxy.h"
#include "connections/implementation/encryption_runner.h"
@@ -158,10 +157,10 @@ class BasePcpHandler : public PcpHandler,
const std::string& endpoint_id) override;
// @EndpointManagerReaderThread
void OnIncomingFrame(location::nearby::connections::OfflineFrame& frame,
const std::string& endpoint_id, ClientProxy* client,
location::nearby::proto::connections::Medium medium,
analytics::PacketMetaData& packet_meta_data) override;
void OnIncomingFrame(
location::nearby::connections::OfflineFrame& frame,
const std::string& endpoint_id, ClientProxy* client,
location::nearby::proto::connections::Medium medium) override;
// Called when an endpoint disconnects while we're waiting for both sides to
// approve/reject the connection.
connections/implementation/base_pcp_handler_test.cc
+1 -3
View File
@@ -32,7 +32,6 @@
#include "connections/advertising_options.h"
#include "connections/connection_options.h"
#include "connections/discovery_options.h"
#include "connections/implementation/analytics/packet_meta_data.h"
#include "connections/implementation/base_endpoint_channel.h"
#include "connections/implementation/bwu_manager.h"
#include "connections/implementation/client_proxy.h"
@@ -1591,7 +1590,6 @@ TEST_P(BasePcpHandlerTest, OnIncomingFrameChangesState) {
EndpointManager em(&ecm);
BwuManager bwu(m, em, ecm, {}, {});
MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
analytics::PacketMetaData packet_meta_data;
StartDiscovery(client_.get(), &pcp_handler);
auto mediums = pcp_handler.GetDiscoveryMediums(client_.get());
auto connect_medium = mediums[mediums.size() - 1];
@@ -1615,7 +1613,7 @@ TEST_P(BasePcpHandlerTest, OnIncomingFrameChangesState) {
Status::kSuccess, os_info, /*multiplex_socket_bitmask=*/0));
EXPECT_CALL(mock_connection_listener_.bandwidth_changed_cb, Call).Times(1);
pcp_handler.OnIncomingFrame(frame.result(), endpoint_id, client_.get(),
connect_medium, packet_meta_data);
connect_medium);
LOG(INFO) << "Closing connection: id=" << endpoint_id;
channel_b->Close();
bwu.Shutdown();
connections/implementation/bwu_manager.cc
+2 -3
View File
@@ -400,8 +400,7 @@ void BwuManager::InitiateBwuForEndpoint(ClientProxy* client,
void BwuManager::OnIncomingFrame(OfflineFrame& frame,
const std::string& endpoint_id,
ClientProxy* client, Medium medium,
PacketMetaData& packet_meta_data) {
ClientProxy* client, Medium medium) {
V1Frame::FrameType frame_type = parser::GetFrameType(frame);
if (frame_type != V1Frame::BANDWIDTH_UPGRADE_NEGOTIATION) return;
@@ -548,7 +547,7 @@ BwuHandler* BwuManager::GetHandlerForMedium(Medium medium) const {
}
void BwuManager::OnBwuNegotiationFrame(
ClientProxy* client, const BandwidthUpgradeNegotiationFrame frame,
ClientProxy* client, const BandwidthUpgradeNegotiationFrame& frame,
const std::string& endpoint_id) {
LOG(INFO) << "OnBwuNegotiationFrame: processing incoming "
<< BandwidthUpgradeNegotiationFrame::EventType_Name(
connections/implementation/bwu_manager.h
+2 -3
View File
@@ -92,8 +92,7 @@ class BwuManager : public EndpointManager::FrameProcessor {
// @EndpointManagerReaderThread
void OnIncomingFrame(location::nearby::connections::OfflineFrame& frame,
const std::string& endpoint_id, ClientProxy* client,
Medium medium,
PacketMetaData& packet_meta_data) override;
Medium medium) override;
// Cleans up in-progress upgrades after endpoint disconnection.
// @EndpointManagerReaderThread
@@ -144,7 +143,7 @@ class BwuManager : public EndpointManager::FrameProcessor {
// upgrade.
void OnBwuNegotiationFrame(
ClientProxy* client,
const location::nearby::connections::BandwidthUpgradeNegotiationFrame
const location::nearby::connections::BandwidthUpgradeNegotiationFrame&
frame,
const string& endpoint_id);
connections/implementation/bwu_manager_test.cc
+11 -12
View File
@@ -188,12 +188,12 @@ class BwuManagerTest : public ::testing::Test {
parser::FromBytes(parser::ForBwuLastWrite());
bwu_manager_->OnIncomingFrame(last_write_frame.result(),
std::string(endpoint_id), &client_,
initial_medium, packet_meta_data_);
initial_medium);
ExceptionOr<OfflineFrame> safe_to_close_frame =
parser::FromBytes(parser::ForBwuSafeToClose());
bwu_manager_->OnIncomingFrame(safe_to_close_frame.result(),
std::string(endpoint_id), &client_,
initial_medium, packet_meta_data_);
initial_medium);
return upgraded_channel;
}
@@ -209,7 +209,6 @@ class BwuManagerTest : public ::testing::Test {
FakeBwuHandler* fake_wifi_direct_bwu_handler_ = nullptr;
FakeBwuHandler* fake_wifi_hotspot_bwu_handler_ = nullptr;
std::unique_ptr<BwuManager> bwu_manager_;
PacketMetaData packet_meta_data_;
};
TEST(BwuManagerBaseTest, AllowToUpgradeMedium) {
@@ -370,12 +369,12 @@ TEST_P(BwuManagerTestParam, InitiateBwu_Success) {
parser::FromBytes(parser::ForBwuLastWrite());
bwu_manager_->OnIncomingFrame(last_write_frame.result(),
std::string(kEndpointId1), &client_,
Medium::BLUETOOTH, packet_meta_data_);
Medium::BLUETOOTH);
ExceptionOr<OfflineFrame> safe_to_close_frame =
parser::FromBytes(parser::ForBwuSafeToClose());
bwu_manager_->OnIncomingFrame(safe_to_close_frame.result(),
std::string(kEndpointId1), &client_,
Medium::BLUETOOTH, packet_meta_data_);
Medium::BLUETOOTH);
// Confirm that upgrade channel is resumed after initial channel is shut down.
// Note: If we didn't grab the shared initial channel pointer above, this
@@ -871,7 +870,7 @@ TEST_F(BwuManagerTest, InitiateBwu_Revert_OnUpgradeFailure_FlagEnabled) {
parser::FromBytes(parser::ForBwuFailure(info));
bwu_manager_->OnIncomingFrame(upgrade_failure.result(),
std::string(kEndpointId3), &client_,
Medium::WEB_RTC, packet_meta_data_);
Medium::WEB_RTC);
// With the flag enabled, we can safely revert WebRTC just for service B
// because service B has no active WebRTC endpoints.
@@ -908,7 +907,7 @@ TEST_F(BwuManagerTest, InitiateBwu_Revert_OnUpgradeFailure_FlagDisabled) {
parser::FromBytes(parser::ForBwuFailure(info));
bwu_manager_->OnIncomingFrame(upgrade_failure.result(),
std::string(kEndpointId3), &client_,
Medium::WEB_RTC, packet_meta_data_);
Medium::WEB_RTC);
// With the flag disabled, we don't revert if there are still connected
// endpoints for _any_ service. We don't have service-level bookkeeping; we
@@ -938,7 +937,7 @@ TEST_F(BwuManagerTest, InitiateBwu_Revert_OnDisconnect_WifiDirect) {
upgrade_path_info = sub_frame->mutable_upgrade_path_info();
upgrade_path_info->set_supports_client_introduction_ack(false);
bwu_manager_->OnIncomingFrame(frame, std::string(kEndpointId1), &client_,
Medium::BLUETOOTH, packet_meta_data_);
Medium::BLUETOOTH);
CountDownLatch latch(1);
bwu_manager_->OnEndpointDisconnect(&client_, (std::string)kServiceIdA,
std::string(kEndpointId1), latch,
@@ -973,7 +972,7 @@ TEST_F(BwuManagerTest, InitiateBwu_Revert_OnDisconnect_Hotspot) {
upgrade_path_info->set_supports_client_introduction_ack(false);
upgrade_path_info->set_supports_disabling_encryption(true);
bwu_manager_->OnIncomingFrame(frame, std::string(kEndpointId1), &client_,
Medium::BLUETOOTH, packet_meta_data_);
Medium::BLUETOOTH);
CountDownLatch latch(1);
bwu_manager_->OnEndpointDisconnect(&client_, (std::string)kServiceIdA,
std::string(kEndpointId1), latch,
@@ -1001,7 +1000,7 @@ TEST_F(BwuManagerTest, InitiateBwu_Revert_OnDisconnect_Wlan) {
upgrade_path_info->set_supports_client_introduction_ack(false);
bwu_manager_->OnIncomingFrame(frame, std::string(kEndpointId1), &client_,
Medium::BLUETOOTH, packet_meta_data_);
Medium::BLUETOOTH);
CountDownLatch latch(1);
bwu_manager_->OnEndpointDisconnect(&client_, (std::string)kServiceIdA,
std::string(kEndpointId1), latch,
@@ -1040,7 +1039,7 @@ TEST_F(BwuManagerTest, BlockBwuFrameBeforeAccept) {
upgrade_path_info2->set_supports_client_introduction_ack(false);
upgrade_path_info2->set_supports_disabling_encryption(true);
bwu_manager_->OnIncomingFrame(frame2, std::string(kEndpointId2), &client_,
Medium::BLUETOOTH, packet_meta_data_);
Medium::BLUETOOTH);
CountDownLatch latch2(1);
// The BWU frame should be drop, so the inProgressUpgrades should be empty.
ASSERT_EQ(bwu_manager_->IsUpgradeOngoing(std::string(kEndpointId2)), false);
@@ -1084,7 +1083,7 @@ TEST_F(BwuManagerTest, BlockBwuFrameFromAdvertiser) {
EXPECT_TRUE(client_.IsConnectedToEndpoint(std::string(kEndpointId2)));
bwu_manager_->OnIncomingFrame(frame, std::string(kEndpointId2), &client_,
Medium::BLUETOOTH, packet_meta_data_);
Medium::BLUETOOTH);
CountDownLatch latch2(1);
// The BWU frame should be drop, so the IsUpgradeOngoing should be empty.
ASSERT_EQ(bwu_manager_->IsUpgradeOngoing(std::string(kEndpointId2)), false);
connections/implementation/connections_authentication_transport_test.cc
+12 -63
View File
@@ -14,7 +14,6 @@
#include "connections/implementation/connections_authentication_transport.h"
#include <cstdint>
#include <memory>
#include <string>
#include <vector>
@@ -23,12 +22,9 @@
#include "protobuf-matchers/protocol-buffer-matchers.h"
#include "gtest/gtest.h"
#include "absl/strings/string_view.h"
#include "absl/time/time.h"
#include "connections/implementation/analytics/analytics_recorder.h"
#include "connections/implementation/endpoint_channel.h"
#include "connections/implementation/mock_endpoint_channel.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/exception.h"
#include "proto/connections_enums.pb.h"
namespace nearby {
namespace connections {
@@ -36,86 +32,39 @@ namespace {
using ::testing::_;
class MockEndpointChannel : public EndpointChannel {
public:
MOCK_METHOD(ExceptionOr<ByteArray>, Read, (), (override));
MOCK_METHOD(ExceptionOr<ByteArray>, Read, (PacketMetaData&), (override));
MOCK_METHOD(Exception, Write, (absl::string_view data), (override));
MOCK_METHOD(Exception, Write, (absl::string_view data, PacketMetaData&),
(override));
MOCK_METHOD(void, Close, (), (override));
MOCK_METHOD(
void, Close,
(location::nearby::proto::connections::DisconnectionReason reason),
(override));
MOCK_METHOD(void, Close,
(location::nearby::proto::connections::DisconnectionReason reason,
location::nearby::analytics::proto::ConnectionsLog::
EstablishedConnection::SafeDisconnectionResult result),
(override));
MOCK_METHOD(bool, IsClosed, (), (const, override));
MOCK_METHOD(std::string, GetType, (), (const, override));
MOCK_METHOD(std::string, GetServiceId, (), (const, override));
MOCK_METHOD(std::string, GetName, (), (const, override));
MOCK_METHOD(location::nearby::proto::connections::Medium, GetMedium, (),
(const, override));
MOCK_METHOD(location::nearby::proto::connections::ConnectionTechnology,
GetTechnology, (), (const, override));
MOCK_METHOD(location::nearby::proto::connections::ConnectionBand, GetBand, (),
(const, override));
MOCK_METHOD(int, GetFrequency, (), (const, override));
MOCK_METHOD(int, GetTryCount, (), (const, override));
MOCK_METHOD(int, GetMaxTransmitPacketSize, (), (const, override));
MOCK_METHOD(void, EnableEncryption, (std::shared_ptr<EncryptionContext>),
(override));
MOCK_METHOD(void, DisableEncryption, (), (override));
MOCK_METHOD(bool, IsEncrypted, (), (override));
MOCK_METHOD(ExceptionOr<ByteArray>, TryDecrypt, (const ByteArray& data),
(override));
MOCK_METHOD(bool, IsPaused, (), (const, override));
MOCK_METHOD(void, Pause, (), (override));
MOCK_METHOD(void, Resume, (), (override));
MOCK_METHOD(absl::Time, GetLastReadTimestamp, (), (const, override));
MOCK_METHOD(absl::Time, GetLastWriteTimestamp, (), (const, override));
MOCK_METHOD(uint32_t, GetNextKeepAliveSeqNo, (), (const, override));
MOCK_METHOD(void, SetAnalyticsRecorder,
(analytics::AnalyticsRecorder*, const std::string&), (override));
std::vector<std::string> messages_;
};
TEST(ConnectionsAuthenticationTransportTest, TestWriteMessage) {
std::vector<std::string> messages;
auto channel = std::make_shared<MockEndpointChannel>();
auto* channel_ptr = channel.get();
ConnectionsAuthenticationTransport transport(channel);
EXPECT_CALL(*channel, Write(_))
.WillOnce([channel_ptr](absl::string_view data) {
channel_ptr->messages_.push_back(std::string(data));
.WillOnce([&messages](absl::string_view data) {
messages.push_back(std::string(data));
return Exception{
.value = Exception::Value::kSuccess,
};
});
transport.WriteMessage("hello world");
EXPECT_THAT(channel_ptr->messages_, testing::ElementsAre("hello world"));
EXPECT_THAT(messages, testing::ElementsAre("hello world"));
}
TEST(ConnectionsAuthenticationTransportTest, TestReadMessage) {
std::vector<std::string> messages;
auto channel = std::make_shared<MockEndpointChannel>();
auto* channel_ptr = channel.get();
ConnectionsAuthenticationTransport transport(channel);
channel_ptr->messages_.push_back("hello world");
EXPECT_CALL(*channel, Read()).WillOnce([channel_ptr]() {
std::string ret = channel_ptr->messages_[0];
channel_ptr->messages_.erase(channel_ptr->messages_.begin());
messages.push_back("hello world");
EXPECT_CALL(*channel, Read()).WillOnce([&messages]() {
std::string ret = messages[0];
messages.erase(messages.begin());
return ExceptionOr<ByteArray>(ByteArray(ret));
});
EXPECT_EQ(transport.ReadMessage(), "hello world");
}
TEST(ConnectionsAuthenticationTransportTest, TestReadMessageFail) {
std::vector<std::string> messages;
auto channel = std::make_shared<MockEndpointChannel>();
ConnectionsAuthenticationTransport transport(channel);
channel->messages_.push_back("hello world");
messages.push_back("hello world");
EXPECT_CALL(*channel, Read()).WillOnce([]() {
return ExceptionOr<ByteArray>(Exception::Value::kIo);
});
connections/implementation/encryption_runner_test.cc
+1 -8
View File
@@ -52,18 +52,11 @@ class FakeEndpointChannel : public EndpointChannel {
read_timestamp_ = SystemClock::ElapsedRealtime();
return in_ ? in_->Read(kChunkSize) : ExceptionOr<ByteArray>{Exception::kIo};
}
ExceptionOr<ByteArray> Read(PacketMetaData& packet_meta_data) override {
read_timestamp_ = SystemClock::ElapsedRealtime();
return in_ ? in_->Read(kChunkSize) : ExceptionOr<ByteArray>{Exception::kIo};
}
Exception Write(absl::string_view data) override {
write_timestamp_ = SystemClock::ElapsedRealtime();
return out_ ? out_->Write(data) : Exception{Exception::kIo};
}
Exception Write(absl::string_view data,
PacketMetaData& packet_meta_data) override {
return Write(data);
}
void Close() override {
if (in_) in_->Close();
if (out_) out_->Close();
connections/implementation/endpoint_channel.h
-9
View File
@@ -23,15 +23,12 @@
#include "absl/strings/string_view.h"
#include "absl/time/time.h"
#include "connections/implementation/analytics/analytics_recorder.h"
#include "connections/implementation/analytics/packet_meta_data.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/exception.h"
namespace nearby {
namespace connections {
using analytics::PacketMetaData;
class EndpointChannel {
public:
virtual ~EndpointChannel() = default;
@@ -41,15 +38,9 @@ class EndpointChannel {
virtual ExceptionOr<ByteArray>
Read() = 0; // throws Exception::IO, Exception::INTERRUPTED
virtual ExceptionOr<ByteArray> Read(PacketMetaData& packet_meta_data) = 0;
virtual Exception Write(absl::string_view data) = 0; // throws Exception::IO
virtual Exception Write(
absl::string_view data,
PacketMetaData& packet_meta_data) = 0; // throws Exception::IO
// Closes this EndpointChannel, without tracking the closure in analytics.
virtual void Close() = 0;
// Closes this EndpointChannel and records the closure with the given reason.
connections/implementation/endpoint_manager.cc
+8 -23
View File
@@ -24,8 +24,6 @@
#include "absl/functional/any_invocable.h"
#include "absl/time/time.h"
#include "connections/connection_options.h"
#include "connections/implementation/analytics/packet_meta_data.h"
#include "connections/implementation/analytics/throughput_recorder.h"
#include "connections/implementation/client_proxy.h"
#include "connections/implementation/endpoint_channel.h"
#include "connections/implementation/endpoint_channel_manager.h"
@@ -34,7 +32,6 @@
#include "connections/implementation/service_id_constants.h"
#include "connections/listeners.h"
#include "connections/medium_selector.h"
#include "connections/payload_type.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/count_down_latch.h"
#include "internal/platform/exception.h"
@@ -58,7 +55,6 @@ 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;
// 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
@@ -235,8 +231,7 @@ ExceptionOr<bool> EndpointManager::HandleData(
// a replacement for this endpoint since we last checked with the
// EndpointChannelManager.
while (true) {
PacketMetaData packet_meta_data;
ExceptionOr<ByteArray> bytes = endpoint_channel->Read(packet_meta_data);
ExceptionOr<ByteArray> bytes = endpoint_channel->Read();
if (!bytes.ok()) {
LOG(INFO) << "Stop reading on read-time exception: " << bytes.exception();
// Treat kNoData as kIo.
@@ -317,8 +312,7 @@ ExceptionOr<bool> EndpointManager::HandleData(
}
frame_processor->OnIncomingFrame(frame, endpoint_id, client,
endpoint_channel->GetMedium(),
packet_meta_data);
endpoint_channel->GetMedium());
}
}
@@ -657,8 +651,7 @@ int EndpointManager::GetMaxTransmitPacketSize(const std::string& endpoint_id) {
std::vector<std::string> EndpointManager::SendPayloadChunk(
const PayloadTransferFrame::PayloadHeader& payload_header,
const PayloadTransferFrame::PayloadChunk& payload_chunk,
const std::vector<std::string>& endpoint_ids,
PacketMetaData& packet_meta_data) {
const std::vector<std::string>& endpoint_ids) {
std::string bytes =
parser::ForDataPayloadTransfer(payload_header, payload_chunk);
@@ -666,8 +659,7 @@ std::vector<std::string> EndpointManager::SendPayloadChunk(
endpoint_ids, bytes, payload_header.id(),
/*offset=*/payload_chunk.offset(),
/*packet_type=*/
PayloadTransferFrame::PacketType_Name(PayloadTransferFrame::DATA),
packet_meta_data);
PayloadTransferFrame::PacketType_Name(PayloadTransferFrame::DATA));
}
// Designed to run asynchronously. It is called from IO thread pools, and
@@ -735,14 +727,12 @@ std::vector<std::string> EndpointManager::SendControlMessage(
const PayloadTransferFrame::ControlMessage& control,
const std::vector<std::string>& endpoint_ids) {
std::string bytes = parser::ForControlPayloadTransfer(header, control);
PacketMetaData packet_meta_data;
return SendTransferFrameBytes(
endpoint_ids, bytes, header.id(),
/*offset=*/control.offset(),
/*packet_type=*/
PayloadTransferFrame::PacketType_Name(PayloadTransferFrame::CONTROL),
packet_meta_data);
PayloadTransferFrame::PacketType_Name(PayloadTransferFrame::CONTROL));
}
// @EndpointManagerThread
@@ -912,20 +902,18 @@ CountDownLatch EndpointManager::NotifyFrameProcessorsOnEndpointDisconnect(
std::vector<std::string> EndpointManager::SendPayloadAck(
std::int64_t payload_id, const std::vector<std::string>& endpoint_ids) {
std::string bytes = parser::ForPayloadAckPayloadTransfer(payload_id);
PacketMetaData packet_meta_data;
return SendTransferFrameBytes(
endpoint_ids, bytes, payload_id,
/* offset= */ -1,
/*packet_type=*/
PayloadTransferFrame::PacketType_Name(PayloadTransferFrame::PAYLOAD_ACK),
packet_meta_data);
PayloadTransferFrame::PacketType_Name(PayloadTransferFrame::PAYLOAD_ACK));
}
std::vector<std::string> EndpointManager::SendTransferFrameBytes(
const std::vector<std::string>& endpoint_ids, const std::string& bytes,
std::int64_t payload_id, std::int64_t offset,
const std::string& packet_type, PacketMetaData& packet_meta_data) {
const std::string& packet_type) {
std::vector<std::string> failed_endpoint_ids;
for (const std::string& endpoint_id : endpoint_ids) {
std::shared_ptr<EndpointChannel> channel =
@@ -944,15 +932,12 @@ std::vector<std::string> EndpointManager::SendTransferFrameBytes(
continue;
}
Exception write_exception = channel->Write(bytes, packet_meta_data);
Exception write_exception = channel->Write(bytes);
if (!write_exception.Ok()) {
failed_endpoint_ids.push_back(endpoint_id);
LOG(INFO) << "Failed to send packet; endpoint_id=" << endpoint_id;
continue;
}
analytics::ThroughputRecorderContainer::GetInstance().UpdateFrameData(
payload_id, PayloadDirection::OUTGOING_PAYLOAD, channel->GetMedium(),
packet_meta_data);
}
return failed_endpoint_ids;
connections/implementation/endpoint_manager.h
+3 -7
View File
@@ -26,7 +26,6 @@
#include "absl/functional/any_invocable.h"
#include "absl/time/time.h"
#include "connections/connection_options.h"
#include "connections/implementation/analytics/packet_meta_data.h"
#include "connections/implementation/client_proxy.h"
#include "connections/implementation/endpoint_channel.h"
#include "connections/implementation/endpoint_channel_manager.h"
@@ -80,8 +79,7 @@ class EndpointManager {
virtual void OnIncomingFrame(
location::nearby::connections::OfflineFrame& offline_frame,
const std::string& from_endpoint_id, ClientProxy* to_client,
location::nearby::proto::connections::Medium current_medium,
analytics::PacketMetaData& packet_meta_data) = 0;
location::nearby::proto::connections::Medium current_medium) = 0;
// Implementations must call barrier.CountDown() once
// they're done. This parallelizes the disconnection event across all frame
@@ -134,8 +132,7 @@ class EndpointManager {
payload_header,
const location::nearby::connections::PayloadTransferFrame::PayloadChunk&
payload_chunk,
const std::vector<std::string>& endpoint_ids,
analytics::PacketMetaData& packet_meta_data);
const std::vector<std::string>& endpoint_ids);
std::vector<std::string> SendControlMessage(
const location::nearby::connections::PayloadTransferFrame::PayloadHeader&
payload_header,
@@ -285,8 +282,7 @@ class EndpointManager {
std::vector<std::string> SendTransferFrameBytes(
const std::vector<std::string>& endpoint_ids,
const std::string& payload_transfer_frame_bytes, std::int64_t payload_id,
std::int64_t offset, const std::string& packet_type,
analytics::PacketMetaData& packet_meta_data);
std::int64_t offset, const std::string& packet_type);
// Executes all jobs sequentially, on a serial_executor_.
void RunOnEndpointManagerThread(const std::string& name, Runnable runnable);
connections/implementation/endpoint_manager_test.cc
+19 -72
View File
@@ -29,11 +29,10 @@
#include "absl/time/clock.h"
#include "absl/time/time.h"
#include "connections/connection_options.h"
#include "connections/implementation/analytics/analytics_recorder.h"
#include "connections/implementation/client_proxy.h"
#include "connections/implementation/endpoint_channel.h"
#include "connections/implementation/endpoint_channel_manager.h"
#include "connections/implementation/flags/nearby_connections_feature_flags.h"
#include "connections/implementation/mock_endpoint_channel.h"
#include "connections/implementation/offline_frames.h"
#include "connections/listeners.h"
#include "connections/status.h"
@@ -63,68 +62,12 @@ using ::testing::MockFunction;
using ::testing::Return;
using ::testing::StrictMock;
class MockEndpointChannel : public EndpointChannel {
public:
MOCK_METHOD(ExceptionOr<ByteArray>, Read, (), (override));
MOCK_METHOD(ExceptionOr<ByteArray>, Read, (PacketMetaData & packet_meta_data),
(override));
MOCK_METHOD(Exception, Write, (absl::string_view data), (override));
MOCK_METHOD(Exception, Write,
(absl::string_view data, PacketMetaData& packet_meta_data),
(override));
MOCK_METHOD(void, Close, (), (override));
MOCK_METHOD(void, Close, (DisconnectionReason reason), (override));
MOCK_METHOD(void, Close,
(DisconnectionReason reason,
location::nearby::analytics::proto::ConnectionsLog::
EstablishedConnection::SafeDisconnectionResult result),
(override));
MOCK_METHOD(location::nearby::proto::connections::ConnectionTechnology,
GetTechnology, (), (const, override));
MOCK_METHOD(location::nearby::proto::connections::ConnectionBand, GetBand, (),
(const, override));
MOCK_METHOD(int, GetFrequency, (), (const, override));
MOCK_METHOD(int, GetTryCount, (), (const, override));
MOCK_METHOD(std::string, GetType, (), (const, override));
MOCK_METHOD(std::string, GetServiceId, (), (const, override));
MOCK_METHOD(std::string, GetName, (), (const, override));
MOCK_METHOD(Medium, GetMedium, (), (const, override));
MOCK_METHOD(int, GetMaxTransmitPacketSize, (), (const, override));
MOCK_METHOD(void, EnableEncryption,
(std::shared_ptr<EncryptionContext> context), (override));
MOCK_METHOD(void, DisableEncryption, (), (override));
MOCK_METHOD(bool, IsPaused, (), (const, override));
MOCK_METHOD(bool, IsEncrypted, (), (override));
MOCK_METHOD(ExceptionOr<ByteArray>, TryDecrypt, (const ByteArray& data),
(override));
MOCK_METHOD(void, Pause, (), (override));
MOCK_METHOD(void, Resume, (), (override));
MOCK_METHOD(absl::Time, GetLastReadTimestamp, (), (const, override));
MOCK_METHOD(absl::Time, GetLastWriteTimestamp, (), (const, override));
MOCK_METHOD(uint32_t, GetNextKeepAliveSeqNo, (), (const, override));
MOCK_METHOD(void, SetAnalyticsRecorder,
(analytics::AnalyticsRecorder*, const std::string&), (override));
bool IsClosed() const override {
absl::MutexLock lock(mutex_);
return closed_;
}
void DoClose() {
absl::MutexLock lock(mutex_);
closed_ = true;
}
private:
mutable absl::Mutex mutex_;
bool closed_ = false;
};
class MockFrameProcessor : public EndpointManager::FrameProcessor {
public:
MOCK_METHOD(void, OnIncomingFrame,
(OfflineFrame & offline_frame,
const std::string& from_endpoint_id, ClientProxy* to_client,
Medium current_medium, PacketMetaData& packet_meta_data),
Medium current_medium),
(override));
MOCK_METHOD(void, OnEndpointDisconnect,
@@ -281,7 +224,7 @@ TEST_F(EndpointManagerTest, RegisterFrameProcessorWorks) {
parser::ForConnectionRequestConnections({}, connection_info);
EXPECT_CALL(*connect_request, OnIncomingFrame);
EXPECT_CALL(*connect_request, OnEndpointDisconnect);
EXPECT_CALL(*endpoint_channel, Read(_))
EXPECT_CALL(*endpoint_channel, Read())
.WillOnce(Return(ExceptionOr<ByteArray>(ByteArray(read_data))))
.WillRepeatedly(Return(ExceptionOr<ByteArray>(Exception::kIo)));
EXPECT_CALL(*endpoint_channel, Write(_))
@@ -319,6 +262,8 @@ TEST_F(EndpointManagerTest, UnregisterFrameProcessorWorks) {
TEST_F(EndpointManagerTest, SendControlMessageAndPayloadAckWorks) {
auto endpoint_channel = std::make_unique<MockEndpointChannel>();
absl::Mutex close_mutex;
bool closed = false;
PayloadTransferFrame::PayloadHeader header;
PayloadTransferFrame::ControlMessage control;
header.set_id(12345);
@@ -327,22 +272,24 @@ TEST_F(EndpointManagerTest, SendControlMessageAndPayloadAckWorks) {
control.set_offset(150);
control.set_event(PayloadTransferFrame::ControlMessage::PAYLOAD_CANCELED);
ON_CALL(*endpoint_channel, Read(_))
.WillByDefault([channel = endpoint_channel.get()]() {
if (channel->IsClosed()) return ExceptionOr<ByteArray>(Exception::kIo);
ON_CALL(*endpoint_channel, Read())
.WillByDefault([&, channel = endpoint_channel.get()]() {
absl::MutexLock lock(close_mutex);
if (closed) return ExceptionOr<ByteArray>(Exception::kIo);
LOG(INFO) << "Simulate read delay: wait";
absl::SleepFor(absl::Milliseconds(100));
LOG(INFO) << "Simulate read delay: done";
if (channel->IsClosed()) return ExceptionOr<ByteArray>(Exception::kIo);
if (closed) return ExceptionOr<ByteArray>(Exception::kIo);
return ExceptionOr<ByteArray>(ByteArray{});
});
ON_CALL(*endpoint_channel, Close(_))
.WillByDefault(
[channel = endpoint_channel.get()](DisconnectionReason reason) {
channel->DoClose();
[&, channel = endpoint_channel.get()](DisconnectionReason reason) {
absl::MutexLock lock(close_mutex);
closed = true;
LOG(INFO) << "Channel closed";
});
EXPECT_CALL(*endpoint_channel, Write(_, _))
EXPECT_CALL(*endpoint_channel, Write(_))
.WillRepeatedly(Return(Exception{Exception::kSuccess}));
RegisterEndpoint(std::move(endpoint_channel), false);
@@ -359,7 +306,7 @@ TEST_F(EndpointManagerTest, SendControlMessageAndPayloadAckWorks) {
TEST_F(EndpointManagerTest, SingleReadOnReadError) {
auto endpoint_channel = std::make_unique<MockEndpointChannel>();
EXPECT_CALL(*endpoint_channel, Read(_))
EXPECT_CALL(*endpoint_channel, Read())
.WillOnce(
Return(ExceptionOr<ByteArray>(Exception::kInvalidProtocolBuffer)));
EXPECT_CALL(*endpoint_channel, Write(_))
@@ -377,7 +324,7 @@ TEST_F(EndpointManagerTest, ReadInvalidUnencryptedPayloadIgnoresFrame) {
CountDownLatch latch(1);
const ByteArray payload("not a valid frame");
auto endpoint_channel = std::make_unique<MockEndpointChannel>();
EXPECT_CALL(*endpoint_channel, Read(_))
EXPECT_CALL(*endpoint_channel, Read())
.WillOnce(Return(ExceptionOr<ByteArray>(payload)))
.WillRepeatedly(Return(ExceptionOr<ByteArray>(Exception::kIo)));
EXPECT_CALL(*endpoint_channel, TryDecrypt(Eq(payload)))
@@ -401,7 +348,7 @@ class EndpointManagerFuzzTest
// too.
// 4. Invalid frame is ignored. No bad side effects.
auto endpoint_channel = std::make_unique<MockEndpointChannel>();
EXPECT_CALL(*endpoint_channel, Read(_))
EXPECT_CALL(*endpoint_channel, Read())
.WillOnce(Return(ExceptionOr<ByteArray>(payload)))
.WillRepeatedly(Return(ExceptionOr<ByteArray>(Exception::kIo)));
EXPECT_CALL(*endpoint_channel, TryDecrypt(Eq(payload)))
@@ -417,7 +364,7 @@ class EndpointManagerFuzzTest
// 2. EndpointManager receives an invalid encrypted frame.
// 3. No calls to TryDecrypt.
auto endpoint_channel = std::make_unique<MockEndpointChannel>();
EXPECT_CALL(*endpoint_channel, Read(_))
EXPECT_CALL(*endpoint_channel, Read())
.WillOnce(Return(ExceptionOr<ByteArray>(payload)))
.WillRepeatedly(Return(ExceptionOr<ByteArray>(Exception::kIo)));
EXPECT_CALL(*endpoint_channel, IsEncrypted()).WillRepeatedly(Return(true));
@@ -468,7 +415,7 @@ TEST_F(EndpointManagerTest, TryDecrypt) {
parser::ForConnectionRequestConnections({}, connection_info);
EXPECT_CALL(*connect_request, OnIncomingFrame);
EXPECT_CALL(*connect_request, OnEndpointDisconnect);
EXPECT_CALL(*endpoint_channel, Read(_))
EXPECT_CALL(*endpoint_channel, Read())
.WillOnce(Return(ExceptionOr<ByteArray>(payload)))
.WillRepeatedly(Return(ExceptionOr<ByteArray>(Exception::kIo)));
EXPECT_CALL(*endpoint_channel, TryDecrypt(Eq(payload)))
connections/implementation/fake_endpoint_channel.h
-9
View File
@@ -46,19 +46,10 @@ class FakeEndpointChannel : public EndpointChannel {
read_timestamp_ = SystemClock::ElapsedRealtime();
return read_output_;
}
ExceptionOr<ByteArray> Read(PacketMetaData& packet_meta_data) override {
read_timestamp_ = SystemClock::ElapsedRealtime();
return read_output_;
}
Exception Write(absl::string_view data) override {
write_timestamp_ = SystemClock::ElapsedRealtime();
return write_output_;
}
Exception Write(absl::string_view data,
PacketMetaData& packet_meta_data) override {
write_timestamp_ = SystemClock::ElapsedRealtime();
return write_output_;
}
void Close() override { is_closed_ = true; }
void Close(location::nearby::proto::connections::DisconnectionReason reason)
override {
connections/implementation/mock_endpoint_channel.h
+77
View File
@@ -0,0 +1,77 @@
// Copyright 2026 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_IMPLEMENTATION_MOCK_ENDPOINT_CHANNEL_H_
#define THIRD_PARTY_NEARBY_CONNECTIONS_IMPLEMENTATION_MOCK_ENDPOINT_CHANNEL_H_
#include <cstdint>
#include <memory>
#include <string>
#include "gmock/gmock.h"
#include "absl/strings/string_view.h"
#include "absl/time/time.h"
#include "connections/implementation/analytics/analytics_recorder.h"
#include "connections/implementation/endpoint_channel.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/exception.h"
namespace nearby::connections {
class MockEndpointChannel : public EndpointChannel {
public:
MOCK_METHOD(ExceptionOr<ByteArray>, Read, (), (override));
MOCK_METHOD(Exception, Write, (absl::string_view data),
(override));
MOCK_METHOD(void, Close, (), (override));
MOCK_METHOD(
void, Close,
(location::nearby::proto::connections::DisconnectionReason reason),
(override));
MOCK_METHOD(void, Close,
(location::nearby::proto::connections::DisconnectionReason reason,
location::nearby::analytics::proto::ConnectionsLog::
EstablishedConnection::SafeDisconnectionResult result),
(override));
MOCK_METHOD(bool, IsClosed, (), (const, override));
MOCK_METHOD(std::string, GetType, (), (const, override));
MOCK_METHOD(std::string, GetServiceId, (), (const, override));
MOCK_METHOD(std::string, GetName, (), (const, override));
MOCK_METHOD(location::nearby::proto::connections::Medium, GetMedium, (),
(const, override));
MOCK_METHOD(location::nearby::proto::connections::ConnectionTechnology,
GetTechnology, (), (const, override));
MOCK_METHOD(location::nearby::proto::connections::ConnectionBand, GetBand, (),
(const, override));
MOCK_METHOD(int, GetFrequency, (), (const, override));
MOCK_METHOD(int, GetTryCount, (), (const, override));
MOCK_METHOD(int, GetMaxTransmitPacketSize, (), (const, override));
MOCK_METHOD(void, EnableEncryption, (std::shared_ptr<EncryptionContext>),
(override));
MOCK_METHOD(void, DisableEncryption, (), (override));
MOCK_METHOD(bool, IsEncrypted, (), (override));
MOCK_METHOD(ExceptionOr<ByteArray>, TryDecrypt, (const ByteArray& data),
(override));
MOCK_METHOD(bool, IsPaused, (), (const, override));
MOCK_METHOD(void, Pause, (), (override));
MOCK_METHOD(void, Resume, (), (override));
MOCK_METHOD(absl::Time, GetLastReadTimestamp, (), (const, override));
MOCK_METHOD(absl::Time, GetLastWriteTimestamp, (), (const, override));
MOCK_METHOD(uint32_t, GetNextKeepAliveSeqNo, (), (const, override));
MOCK_METHOD(void, SetAnalyticsRecorder,
(analytics::AnalyticsRecorder*, const std::string&), (override));
};
} // namespace nearby::connections
#endif // THIRD_PARTY_NEARBY_CONNECTIONS_IMPLEMENTATION_MOCK_ENDPOINT_CHANNEL_H_
connections/implementation/payload_manager.cc
+4 -37
View File
@@ -29,8 +29,6 @@
#include "absl/strings/str_cat.h"
#include "absl/strings/str_format.h"
#include "absl/time/time.h"
#include "connections/implementation/analytics/packet_meta_data.h"
#include "connections/implementation/analytics/throughput_recorder.h"
#include "connections/implementation/client_proxy.h"
#include "connections/implementation/endpoint_channel_manager.h"
#include "connections/implementation/endpoint_manager.h"
@@ -65,8 +63,6 @@ using ::location::nearby::connections::V1Frame;
using ::location::nearby::proto::connections::Medium;
using ::location::nearby::proto::connections::OperationResultCode;
using ::location::nearby::proto::connections::PayloadStatus;
using PacketMetaData = ::nearby::analytics::PacketMetaData;
using ::nearby::analytics::ThroughputRecorderContainer;
using PayloadDirection = ::nearby::connections::PayloadDirection;
constexpr absl::Duration kMinTransferUpdateInterval = absl::Milliseconds(50);
@@ -81,7 +77,6 @@ bool PayloadManager::SendPayloadLoop(
const EndpointIds& available_endpoint_ids =
EndpointsToEndpointIds(pair.first);
const Endpoints& unavailable_endpoints = pair.second;
PacketMetaData packet_meta_data;
// First, handle any non-available endpoints.
for (const auto& endpoint : unavailable_endpoints) {
@@ -143,10 +138,8 @@ bool PayloadManager::SendPayloadLoop(
// This will block if there is no data to transfer.
// It will resume when new data arrives, or if Close() is called.
int chunk_size = GetOptimalChunkSize(available_endpoint_ids);
packet_meta_data.StartFileIo();
ByteArray next_chunk =
pending_payload.GetInternalPayload()->DetachNextChunk(chunk_size);
packet_meta_data.StopFileIo();
if (shutdown_.Get()) return false;
// Save chunk size. We'll need it after we move next_chunk.
auto next_chunk_size = next_chunk.size();
@@ -169,7 +162,7 @@ bool PayloadManager::SendPayloadLoop(
PayloadTransferFrame::PayloadChunk payload_chunk(CreatePayloadChunk(
next_chunk_offset - resume_offset, std::move(next_chunk), index));
const EndpointIds& failed_endpoint_ids = endpoint_manager_->SendPayloadChunk(
payload_header, payload_chunk, available_endpoint_ids, packet_meta_data);
payload_header, payload_chunk, available_endpoint_ids);
// Check whether at least one endpoint failed.
if (!failed_endpoint_ids.empty()) {
VLOG(1) << "Payload xfer: endpoints failed: payload_id="
@@ -209,9 +202,6 @@ bool PayloadManager::SendPayloadLoop(
VLOG(1) << "Payload xfer done: payload_id="
<< pending_payload.GetInternalPayload()->GetId()
<< "; size=" << next_chunk_offset;
ThroughputRecorderContainer::GetInstance().MarkAsSuccess(
pending_payload.GetInternalPayload()->GetId(),
PayloadDirection::OUTGOING_PAYLOAD);
return false;
}
}
@@ -479,8 +469,6 @@ void PayloadManager::SendPayload(ClientProxy* client,
std::int64_t next_chunk_offset = 0;
int index = 0;
ThroughputRecorderContainer::GetInstance().Start(
payload_id, PayloadDirection::OUTGOING_PAYLOAD, payload_type);
while (should_continue && !shutdown_.Get()) {
should_continue =
SendPayloadLoop(client, *pending_payload, payload_header,
@@ -528,8 +516,7 @@ Status PayloadManager::CancelPayload(ClientProxy* client,
void PayloadManager::OnIncomingFrame(OfflineFrame& offline_frame,
const std::string& from_endpoint_id,
ClientProxy* to_client,
Medium current_medium,
PacketMetaData& packet_meta_data) {
Medium current_medium) {
PayloadTransferFrame& frame =
*offline_frame.mutable_v1()->mutable_payload_transfer();
@@ -560,8 +547,7 @@ void PayloadManager::OnIncomingFrame(OfflineFrame& offline_frame,
ProcessControlPacket(to_client, from_endpoint_id, frame);
break;
case PayloadTransferFrame::DATA:
ProcessDataPacket(to_client, from_endpoint_id, frame, current_medium,
packet_meta_data);
ProcessDataPacket(to_client, from_endpoint_id, frame, current_medium);
break;
case PayloadTransferFrame::PAYLOAD_ACK:
VLOG(1) << "[safe-to-disconnect][PAYLOAD_RECEIVED_ACK] sender "
@@ -812,10 +798,6 @@ PayloadManager::CreateIncomingPayload(const PayloadTransferFrame& frame,
void PayloadManager::OnPendingPayloadDestroy(const PendingPayload* payload) {
VLOG(1) << "PayloadManager: destroying " << payload->ToString()
<< " self=" << this;
ThroughputRecorderContainer::GetInstance().StopTPRecorder(
payload->GetId(), payload->IsIncoming()
? PayloadDirection::INCOMING_PAYLOAD
: PayloadDirection::OUTGOING_PAYLOAD);
if (payload->IsIncoming()) return;
RunOnStatusUpdateThread(
"~PendingPayload",
@@ -1301,8 +1283,7 @@ void PayloadManager::HandleSuccessfulIncomingChunk(
// @EndpointManagerDataPool
void PayloadManager::ProcessDataPacket(
ClientProxy* to_client, const std::string& from_endpoint_id,
PayloadTransferFrame& payload_transfer_frame, Medium medium,
PacketMetaData& packet_meta_data) {
PayloadTransferFrame& payload_transfer_frame, Medium medium) {
PayloadTransferFrame::PayloadHeader& payload_header =
*payload_transfer_frame.mutable_payload_header();
PayloadTransferFrame::PayloadChunk& payload_chunk =
@@ -1323,10 +1304,6 @@ void PayloadManager::ProcessDataPacket(
Payload::Id payload_id = payload_header.id();
PendingPayloadHandle pending_payload;
if (payload_chunk.offset() == 0) {
ThroughputRecorderContainer::GetInstance().Start(
payload_id, PayloadDirection::INCOMING_PAYLOAD,
(PayloadType)payload_header.type());
packet_meta_data.Reset();
RunOnStatusUpdateThread(
"process-data-packet", [to_client, from_endpoint_id, payload_header,
this]() RUN_ON_PAYLOAD_STATUS_UPDATE_THREAD() {
@@ -1413,7 +1390,6 @@ void PayloadManager::ProcessDataPacket(
// Save size of packet before we move it.
std::int64_t payload_body_size = payload_chunk.body().size();
packet_meta_data.StartFileIo();
if (pending_payload->GetInternalPayload()
->AttachNextChunk(payload_chunk.body())
.Raised()) {
@@ -1425,7 +1401,6 @@ void PayloadManager::ProcessDataPacket(
PayloadStatus::LOCAL_ERROR, OperationResultCode::IO_FILE_WRITING_ERROR);
return;
}
packet_meta_data.StopFileIo();
bool is_last_chunk = (payload_chunk.flags() &
PayloadTransferFrame::PayloadChunk::LAST_CHUNK) != 0;
SendPayloadReceivedAck(to_client, *pending_payload, from_endpoint_id,
@@ -1434,14 +1409,6 @@ void PayloadManager::ProcessDataPacket(
HandleSuccessfulIncomingChunk(to_client, from_endpoint_id, payload_header,
payload_chunk.flags(), payload_chunk.offset(),
payload_body_size);
ThroughputRecorderContainer::GetInstance().UpdateFrameData(
payload_header.id(), PayloadDirection::INCOMING_PAYLOAD, medium,
packet_meta_data);
if (is_last_chunk) {
ThroughputRecorderContainer::GetInstance().MarkAsSuccess(
payload_header.id(), PayloadDirection::INCOMING_PAYLOAD);
}
}
// @EndpointManagerDataPool
connections/implementation/payload_manager.h
+2 -5
View File
@@ -26,7 +26,6 @@
#include "absl/container/flat_hash_map.h"
#include "absl/functional/any_invocable.h"
#include "absl/time/time.h"
#include "connections/implementation/analytics/packet_meta_data.h"
#include "connections/implementation/client_proxy.h"
#include "connections/implementation/endpoint_manager.h"
#include "connections/implementation/internal_payload.h"
@@ -67,8 +66,7 @@ class PayloadManager : public EndpointManager::FrameProcessor {
void OnIncomingFrame(
location::nearby::connections::OfflineFrame& offline_frame,
const std::string& from_endpoint_id, ClientProxy* to_client,
location::nearby::proto::connections::Medium current_medium,
analytics::PacketMetaData& packet_meta_data) override;
location::nearby::proto::connections::Medium current_medium) override;
// @EndpointManagerThread
void OnEndpointDisconnect(
@@ -412,8 +410,7 @@ class PayloadManager : public EndpointManager::FrameProcessor {
const std::string& from_endpoint_id,
location::nearby::connections::PayloadTransferFrame&
payload_transfer_frame,
location::nearby::proto::connections::Medium medium,
analytics::PacketMetaData& packet_meta_data);
location::nearby::proto::connections::Medium medium);
void ProcessControlPacket(ClientProxy* to_client,
const std::string& from_endpoint_id,
location::nearby::connections::PayloadTransferFrame&
connections/implementation/payload_manager_test.cc
+1 -5
View File
@@ -21,7 +21,6 @@
#include "gtest/gtest.h"
#include "absl/strings/string_view.h"
#include "absl/time/time.h"
#include "connections/implementation/analytics/packet_meta_data.h"
#include "connections/implementation/offline_frames.h"
#include "connections/implementation/simulation_user.h"
#include "connections/listeners.h"
@@ -43,7 +42,6 @@ namespace {
using ::location::nearby::connections::OfflineFrame;
using ::location::nearby::connections::PayloadTransferFrame;
using ::location::nearby::proto::connections::Medium;
using ::nearby::analytics::PacketMetaData;
constexpr size_t kChunkSize = 64 * 1024;
constexpr absl::string_view kServiceId = "service-id";
@@ -116,10 +114,8 @@ class PayloadSimulationUser : public SimulationUser {
std::string bytes = parser::ForDataPayloadTransfer(header, chunk);
offline_frame.ParseFromString(bytes);
PacketMetaData packet_meta_data;
pm_.OnIncomingFrame(offline_frame, from_payload_id, &client_,
Medium::WIFI_HOTSPOT, packet_meta_data);
Medium::WIFI_HOTSPOT);
}
Status CancelPayload() {