Refactor Input/Output file to use const char* instead of standard string, and to use parent_folder and file_name for input and output files. This also incorporates the move from nearby_connections to nearby.

PiperOrigin-RevId: 415587802
This commit is contained in:
jfcarroll
2021-12-10 12:49:50 -08:00
committed by Copybara-Service
parent 5ba6623eac
commit f12bb6c405
48 changed files with 540 additions and 377 deletions
+1
View File
@@ -190,6 +190,7 @@ cc_test(
deps = [
":internal",
":internal_test",
"//file/util:temp_path",
"//testing/base/public:gunit",
"//testing/base/public:gunit_main",
"//absl/container:flat_hash_set",
+21 -16
View File
@@ -1478,6 +1478,7 @@ void BasePcpHandler::LogConnectionAttemptFailure(
}
}
// TODO(jfcarroll): FIXME!!!!
void BasePcpHandler::LogConnectionAttemptSuccess(
const std::string& endpoint_id,
const PendingConnectionInfo& connection_info) {
@@ -1497,22 +1498,26 @@ void BasePcpHandler::LogConnectionAttemptSuccess(
"LogConnectionAttemptSuccess. Bail out.");
return;
}
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,
connections_attempt_metadata_params.get());
} 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,
connections_attempt_metadata_params.get());
}
// TODO(jfcarroll): Something in the below code is coming up null
// causing a crash. I can't debug this locally, and as a TVC I'm
// not able to debug using ciderd.
// 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,
// connections_attempt_metadata_params.get());
//} 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,
// connections_attempt_metadata_params.get());
//}
}
bool BasePcpHandler::Cancelled(ClientProxy* client,
+50 -10
View File
@@ -189,7 +189,7 @@ class OutgoingFileInternalPayload : public InternalPayload {
std::int64_t GetTotalSize() const override { return total_size_; }
ByteArray DetachNextChunk(int chunk_size) override {
InputFile* file = payload_.AsFile();
const InputFile* file = payload_.AsFile();
if (!file) return {};
ExceptionOr<ByteArray> bytes_read = file->Read(chunk_size);
@@ -215,7 +215,7 @@ class OutgoingFileInternalPayload : public InternalPayload {
ExceptionOr<size_t> SkipToOffset(size_t offset) override {
NEARBY_LOGS(INFO) << "SkipToOffset " << offset;
InputFile* file = payload_.AsFile();
const InputFile* file = payload_.AsFile();
if (!file) {
return {Exception::kIo};
}
@@ -236,7 +236,7 @@ class OutgoingFileInternalPayload : public InternalPayload {
}
void Close() override {
InputFile* file = payload_.AsFile();
const InputFile* file = payload_.AsFile();
if (file) file->Close();
}
@@ -292,10 +292,6 @@ std::unique_ptr<InternalPayload> CreateOutgoingInternalPayload(
return absl::make_unique<BytesInternalPayload>(std::move(payload));
case Payload::Type::kFile: {
InputFile* file = payload.AsFile();
const PayloadId file_payload_id = file ? file->GetPayloadId() : 0;
const PayloadId payload_id = payload.GetId();
CHECK(payload_id == file_payload_id);
return absl::make_unique<OutgoingFileInternalPayload>(std::move(payload));
}
@@ -309,6 +305,22 @@ std::unique_ptr<InternalPayload> CreateOutgoingInternalPayload(
}
}
std::string make_path(std::string parent_folder, std::string file_name) {
if (parent_folder.find_last_of('/') == std::string::npos) {
parent_folder.append("/");
}
return parent_folder.append(file_name);
}
std::string make_path(std::string parent_folder, int64_t id) {
if (parent_folder.find_last_of('/') == std::string::npos) {
parent_folder.append("/");
}
return parent_folder.append(std::to_string(id));
}
std::unique_ptr<InternalPayload> CreateIncomingInternalPayload(
const PayloadTransferFrame& frame) {
if (frame.packet_type() != PayloadTransferFrame::DATA) {
@@ -334,11 +346,39 @@ std::unique_ptr<InternalPayload> CreateIncomingInternalPayload(
}
case PayloadTransferFrame::PayloadHeader::FILE: {
std::int64_t total_size = frame.payload_header().total_size();
std::string file_path;
int64_t total_size;
if (frame.payload_header().has_parent_folder()) {
file_path = frame.payload_header().parent_folder();
}
if (!frame.payload_header().has_file_name()) {
file_path = make_path(file_path, frame.payload_header().id());
} else {
file_path = make_path(file_path, frame.payload_header().file_name());
}
if (frame.payload_header().has_total_size()) {
total_size = frame.payload_header().total_size();
}
// These are ordered, the output file must be created first otherwise
// there will be no input file to open.
OutputFile outputFile(
location::nearby::api::ImplementationPlatform::GetDownloadPath(
std::make_unique<std::string>(file_path))
->c_str());
InputFile inputFile(
location::nearby::api::ImplementationPlatform::GetDownloadPath(
std::make_unique<std::string>(file_path))
->c_str());
return absl::make_unique<IncomingFileInternalPayload>(
Payload(payload_id, InputFile(payload_id, total_size)),
OutputFile(payload_id), total_size);
Payload(payload_id, std::move(inputFile)), std::move(outputFile),
frame.payload_header().total_size());
}
default:
DCHECK(false); // This should never happen.
return {};
@@ -14,9 +14,13 @@
#include "core/internal/internal_payload_factory.h"
#include <filesystem>
#include <fstream>
#include <memory>
#include <string>
#include <utility>
#include "file/util/temp_path.h"
#include "gmock/gmock.h"
#include "gtest/gtest.h"
#include "core/internal/offline_frames.h"
@@ -30,8 +34,28 @@ namespace connections {
namespace {
constexpr char kText[] = "data chunk";
#define TEST_FILE_NAME std::string("testfilename.txt")
#define TEST_FILE_PARENT_FOLDER std::string("")
TEST(InternalPayloadFActoryTest, CanCreateIternalPayloadFromBytePayload) {
class InternalPayloadFActoryTest : public ::testing::Test {
protected:
void SetUp() override {
temp_path_ = std::make_unique<TempPath>(TempPath::Local);
path_ = temp_path_->path() + "/" + TEST_FILE_NAME;
file_ = std::fstream(path_, std::fstream::out | std::fstream::trunc);
file_ << "This is a test file with a minimum of 101 characters. This is "
"used to verify the InputFile in the payload_test google test.";
file_.close();
}
void TearDown() override { std::filesystem::remove(path_.c_str()); }
std::fstream file_;
std::unique_ptr<TempPath> temp_path_;
std::string path_;
};
TEST_F(InternalPayloadFActoryTest, CanCreateIternalPayloadFromBytePayload) {
ByteArray data(kText);
std::unique_ptr<InternalPayload> internal_payload =
CreateOutgoingInternalPayload(Payload{data});
@@ -42,7 +66,7 @@ TEST(InternalPayloadFActoryTest, CanCreateIternalPayloadFromBytePayload) {
EXPECT_EQ(payload.AsBytes(), ByteArray(kText));
}
TEST(InternalPayloadFActoryTest, CanCreateIternalPayloadFromStreamPayload) {
TEST_F(InternalPayloadFActoryTest, CanCreateIternalPayloadFromStreamPayload) {
auto pipe = std::make_shared<Pipe>();
std::unique_ptr<InternalPayload> internal_payload =
CreateOutgoingInternalPayload(Payload{[pipe]() -> InputStream& {
@@ -55,21 +79,21 @@ TEST(InternalPayloadFActoryTest, CanCreateIternalPayloadFromStreamPayload) {
EXPECT_EQ(payload.AsBytes(), ByteArray());
}
TEST(InternalPayloadFActoryTest, CanCreateIternalPayloadFromFilePayload) {
Payload::Id payload_id = Payload::GenerateId();
TEST_F(InternalPayloadFActoryTest, CanCreateIternalPayloadFromFilePayload) {
std::unique_ptr<InternalPayload> internal_payload =
CreateOutgoingInternalPayload(
Payload{payload_id, InputFile(payload_id, 512)});
CreateOutgoingInternalPayload(Payload{
path_.c_str(), TEST_FILE_NAME.c_str(), InputFile(path_.c_str())});
EXPECT_NE(internal_payload, nullptr);
Payload payload = internal_payload->ReleasePayload();
EXPECT_NE(payload.AsFile(), nullptr);
EXPECT_EQ(payload.AsStream(), nullptr);
EXPECT_EQ(payload.AsBytes(), ByteArray());
EXPECT_EQ(payload.GetId(), payload_id);
EXPECT_EQ(payload.AsFile()->GetPayloadId(), payload_id);
EXPECT_EQ(payload.AsFile()->GetFilePath(), path_);
payload.AsFile()->Close();
std::filesystem::remove(path_);
}
TEST(InternalPayloadFActoryTest, CanCreateIternalPayloadFromByteMessage) {
TEST_F(InternalPayloadFActoryTest, CanCreateIternalPayloadFromByteMessage) {
PayloadTransferFrame frame;
frame.set_packet_type(PayloadTransferFrame::DATA);
std::int64_t payload_chunk_offset = 0;
@@ -92,7 +116,7 @@ TEST(InternalPayloadFActoryTest, CanCreateIternalPayloadFromByteMessage) {
EXPECT_EQ(payload.AsBytes(), ByteArray(kText));
}
TEST(InternalPayloadFActoryTest, CanCreateIternalPayloadFromStreamMessage) {
TEST_F(InternalPayloadFActoryTest, CanCreateIternalPayloadFromStreamMessage) {
PayloadTransferFrame frame;
frame.set_packet_type(PayloadTransferFrame::DATA);
auto& header = *frame.mutable_payload_header();
@@ -109,7 +133,7 @@ TEST(InternalPayloadFActoryTest, CanCreateIternalPayloadFromStreamMessage) {
EXPECT_EQ(payload.GetType(), Payload::Type::kStream);
}
TEST(InternalPayloadFActoryTest, CanCreateIternalPayloadFromFileMessage) {
TEST_F(InternalPayloadFActoryTest, CanCreateIternalPayloadFromFileMessage) {
PayloadTransferFrame frame;
frame.set_packet_type(PayloadTransferFrame::DATA);
auto& header = *frame.mutable_payload_header();
@@ -124,25 +148,32 @@ TEST(InternalPayloadFActoryTest, CanCreateIternalPayloadFromFileMessage) {
EXPECT_EQ(payload.AsStream(), nullptr);
EXPECT_EQ(payload.AsBytes(), ByteArray());
EXPECT_EQ(payload.GetType(), Payload::Type::kFile);
EXPECT_EQ(payload.GetId(), payload.AsFile()->GetPayloadId());
}
void CreateFileWithContents(Payload::Id payload_id, const ByteArray& contents) {
OutputFile file(payload_id);
EXPECT_TRUE(file.Write(contents).Ok());
EXPECT_TRUE(file.Close().Ok());
void CreateFileWithContents(const char* file_path, const ByteArray& contents) {
std::unique_ptr<OutputFile> file = std::make_unique<OutputFile>(file_path);
EXPECT_TRUE(file->Write(contents).Ok());
EXPECT_TRUE(file->Close().Ok());
}
TEST(InternalPayloadFActoryTest,
SkipToOffset_FilePayloadValidOffset_SkipsOffset) {
TEST_F(InternalPayloadFActoryTest,
SkipToOffset_FilePayloadValidOffset_SkipsOffset) {
ByteArray contents("0123456789");
constexpr size_t kOffset = 4;
size_t size_after_skip = contents.size() - kOffset;
NEARBY_LOGS(INFO)
<< "SkipToOffset_FilePayloadValidOffset_SkipsOffset: file path = "
<< path_.c_str() << "\n";
NEARBY_LOGS(INFO)
<< "SkipToOffset_FilePayloadValidOffset_SkipsOffset: contents = "
<< contents.data() << "\n";
CreateFileWithContents(path_.c_str(), contents);
Payload::Id payload_id = Payload::GenerateId();
CreateFileWithContents(payload_id, contents);
std::unique_ptr<InputFile> inputFile =
std::make_unique<InputFile>(path_.c_str());
std::unique_ptr<InternalPayload> internal_payload =
CreateOutgoingInternalPayload(
Payload{payload_id, InputFile(payload_id, contents.size())});
CreateOutgoingInternalPayload(Payload{payload_id, std::move(*inputFile)});
EXPECT_NE(internal_payload, nullptr);
ExceptionOr<size_t> result = internal_payload->SkipToOffset(kOffset);
@@ -153,10 +184,12 @@ TEST(InternalPayloadFActoryTest,
ByteArray contents_after_skip =
internal_payload->DetachNextChunk(size_after_skip);
EXPECT_EQ(contents_after_skip, ByteArray("456789"));
internal_payload = nullptr;
std::filesystem::remove(path_);
}
TEST(InternalPayloadFActoryTest,
SkipToOffset_StreamPayloadValidOffset_SkipsOffset) {
TEST_F(InternalPayloadFActoryTest,
SkipToOffset_StreamPayloadValidOffset_SkipsOffset) {
ByteArray contents("0123456789");
constexpr size_t kOffset = 6;
auto pipe = std::make_shared<Pipe>();
@@ -24,6 +24,13 @@ namespace location {
namespace nearby {
namespace connections {
namespace parser {
bool Validate(std::string toBeValidated,
std::vector<std::string> illegalPatterns) {
return !std::any_of(illegalPatterns.begin(), illegalPatterns.end(),
[&toBeValidated](const auto& s) {
return toBeValidated.find(s) != std::string::npos;
});
}
namespace {
using PayloadChunk = PayloadTransferFrame::PayloadChunk;
@@ -120,6 +127,25 @@ Exception EnsureValidPayloadTransferFrame(const PayloadTransferFrame& frame) {
frame.payload_header().total_size() !=
InternalPayload::kIndeterminateSize))
return {Exception::kInvalidProtocolBuffer};
if (frame.payload_header().has_type() &&
frame.payload_header().type() ==
PayloadTransferFrame::PayloadHeader::FILE) {
if (frame.payload_header().has_file_name()) {
if (!Validate(frame.payload_header().file_name(),
ILLEGAL_FILENAME_PATTERNS)) {
return {Exception::kFailed};
}
}
if (frame.payload_header().has_parent_folder()) {
if (!Validate(frame.payload_header().file_name(),
ILLEGAL_PARENT_FOLDER_PATTERNS)) {
return {Exception::kFailed};
}
}
}
if (!frame.has_packet_type()) return {Exception::kInvalidProtocolBuffer};
switch (frame.packet_type()) {
@@ -23,8 +23,19 @@ namespace nearby {
namespace connections {
namespace parser {
const std::vector<std::string> ILLEGAL_FILENAME_PATTERNS{
"/", "\\", "?", "*", "\"", "<", ">", "|", "[", "]",
":", ",", ";", "..", "\0", "\n", "\r", "\t", "\f"};
const std::vector<std::string> ILLEGAL_PARENT_FOLDER_PATTERNS{
"\\", "?", "*", "\"", "<", ">", "|", "[", "]",
":", ",", ";", "..", "\0", "\n", "\r", "\t", "\f"};
Exception EnsureValidOfflineFrame(const OfflineFrame& offline_frame);
bool Validate(std::string toBeValidated,
std::vector<std::string> illegalPatterns);
} // namespace parser
} // namespace connections
} // namespace nearby
+17 -5
View File
@@ -259,6 +259,7 @@ Payload::Id PayloadManager::CreateOutgoingPayload(
Payload::Id payload_id = internal_payload->GetId();
NEARBY_LOGS(INFO) << "CreateOutgoingPayload: payload_id=" << payload_id;
MutexLock lock(&mutex_);
pending_payloads_.StartTrackingPayload(
payload_id, absl::make_unique<PendingPayload>(std::move(internal_payload),
endpoint_ids,
@@ -354,6 +355,7 @@ void PayloadManager::SendPayload(ClientProxy* client,
// Before transfer to internal payload, retrieves the Payload size for
// analytics.
std::int64_t payload_total_size;
switch (payload.GetType()) {
case connections::Payload::Type::kBytes:
payload_total_size = payload.AsBytes().size();
@@ -392,11 +394,15 @@ void PayloadManager::SendPayload(ClientProxy* client,
? payload.GetOffset()
: 0;
std::string file_name("");
std::string parent_folder("");
Payload::Id payload_id =
CreateOutgoingPayload(std::move(payload), endpoint_ids);
executor->Execute(
"send-payload", [this, client, endpoint_ids, payload_id, payload_type,
resume_offset, payload_total_size]() {
"send-payload",
[this, client, endpoint_ids, payload_id, payload_type, resume_offset,
payload_total_size, file_name, parent_folder]() {
if (shutdown_.Get()) return;
PendingPayload* pending_payload = GetPayload(payload_id);
if (!pending_payload) {
@@ -417,10 +423,12 @@ void PayloadManager::SendPayload(ClientProxy* client,
payload_type, resume_offset,
internal_payload->GetTotalSize());
PayloadTransferFrame::PayloadHeader payload_header{
CreatePayloadHeader(*internal_payload, resume_offset)};
NEARBY_LOG(INFO, "Creating payload header (JFC)");
PayloadTransferFrame::PayloadHeader payload_header{CreatePayloadHeader(
*internal_payload, resume_offset, parent_folder, file_name)};
bool should_continue = true;
std::int64_t next_chunk_offset = 0;
NEARBY_LOG(INFO, "Entering send payload loop (JFC)");
while (should_continue && !shutdown_.Get()) {
should_continue =
SendPayloadLoop(client, *pending_payload, payload_header,
@@ -610,12 +618,15 @@ int PayloadManager::GetOptimalChunkSize(EndpointIds endpoint_ids) {
}
PayloadTransferFrame::PayloadHeader PayloadManager::CreatePayloadHeader(
const InternalPayload& internal_payload, size_t offset) {
const InternalPayload& internal_payload, size_t offset,
std::string parent_folder, std::string file_name) {
PayloadTransferFrame::PayloadHeader payload_header;
size_t payload_size = internal_payload.GetTotalSize();
payload_header.set_id(internal_payload.GetId());
payload_header.set_type(internal_payload.GetType());
payload_header.set_file_name(file_name);
payload_header.set_parent_folder(parent_folder);
payload_header.set_total_size(payload_size ==
InternalPayload::kIndeterminateSize
? InternalPayload::kIndeterminateSize
@@ -1170,6 +1181,7 @@ PayloadManager::PendingPayload::PendingPayload(
// Later on some may become canceled, some may experience data transfer
// failures. Any of these situations will cause endpoint to be marked as
// unavailable.
for (const auto& id : endpoint_ids) {
EndpointInfo endpoint_info{};
endpoint_info.id = id;
+2 -1
View File
@@ -211,7 +211,8 @@ class PayloadManager : public EndpointManager::FrameProcessor {
int GetOptimalChunkSize(EndpointIds endpoint_ids);
PayloadTransferFrame::PayloadHeader CreatePayloadHeader(
const InternalPayload& payload, size_t offset);
const InternalPayload& payload, size_t offset, std::string parent_folder,
std::string file_name);
PayloadTransferFrame::PayloadChunk CreatePayloadChunk(std::int64_t offset,
ByteArray body);
+2 -1
View File
@@ -258,7 +258,7 @@ TEST_P(PayloadManagerTest, CanCancelPayloadOnSenderSide) {
PayloadSimulationUser user_a(kDeviceA, GetParam());
PayloadSimulationUser user_b(kDeviceB, GetParam());
ASSERT_TRUE(SetupConnection(user_a, user_b));
NEARBY_LOG(INFO, "User a and user b have been setup. JFC");
auto pipe = std::make_shared<Pipe>();
OutputStream& tx = pipe->GetOutputStream();
@@ -266,6 +266,7 @@ TEST_P(PayloadManagerTest, CanCancelPayloadOnSenderSide) {
const ByteArray message{std::string(kMessage)};
tx.Write(message);
NEARBY_LOG(INFO, "Sending payload from user_b. JFC");
user_b.SendPayload(Payload([pipe]() -> InputStream& {
return pipe->GetInputStream(); // NOLINT
}));