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

Fix read error does not cause send to fail.

Browse Source 
PiperOrigin-RevId: 928671777
This commit is contained in:
Francis Tsui
2026-06-08 11:19:50 -07:00
committed by Copybara-Service
parent e7e754611a
commit 3dcbad2dea
7 changed files with 262 additions and 319 deletions
Download Patch File Download Diff File Expand all files Collapse all files
connections/BUILD
+1 -1
View File
@@ -103,8 +103,8 @@ cc_library(
"//proto:connections_enums_cc_proto",
"@com_google_absl//absl/functional:any_invocable",
"@com_google_absl//absl/random",
"@com_google_absl//absl/strings",
"@com_google_absl//absl/time",
"@com_google_absl//absl/types:variant",
],
)
connections/implementation/payload_manager.cc
+142 -189
View File
@@ -27,7 +27,7 @@
#include "absl/functional/any_invocable.h"
#include "absl/functional/bind_front.h"
#include "absl/strings/str_cat.h"
#include "absl/strings/str_format.h"
#include "absl/strings/str_join.h"
#include "absl/time/time.h"
#include "connections/implementation/analytics/analytics_recorder.h"
#include "connections/implementation/client_proxy.h"
@@ -52,8 +52,7 @@
#include "internal/platform/mutex_lock.h"
#include "internal/platform/single_thread_executor.h"
namespace nearby {
namespace connections {
namespace nearby::connections {
namespace {
using ::location::nearby::connections::OfflineFrame;
@@ -65,17 +64,59 @@ using ::location::nearby::proto::connections::PayloadStatus;
using ::nearby::analytics::AnalyticsRecorder;
constexpr absl::Duration kMinTransferUpdateInterval = absl::Milliseconds(50);
std::string EndpointIdsToString(const std::vector<std::string>& endpoint_ids) {
return absl::StrCat(endpoint_ids.size(), ":",
absl::StrJoin(endpoint_ids, ","));
}
PayloadStatus ControlMessageEventToPayloadStatus(
PayloadTransferFrame::ControlMessage::EventType event) {
switch (event) {
case PayloadTransferFrame::ControlMessage::PAYLOAD_ERROR:
return PayloadStatus::REMOTE_ERROR;
case PayloadTransferFrame::ControlMessage::PAYLOAD_CANCELED:
return PayloadStatus::REMOTE_CANCELLATION;
default:
VLOG(1) << "PayloadManager: unknown event=" << event;
return PayloadStatus::UNKNOWN_PAYLOAD_STATUS;
}
}
OperationResultCode ControlMessageEventToOperationResultCode(
PayloadTransferFrame::ControlMessage::EventType event) {
switch (event) {
case PayloadTransferFrame::ControlMessage::PAYLOAD_ERROR:
return OperationResultCode::NEARBY_GENERIC_REMOTE_REPORT_PAYLOADS_ERROR;
case PayloadTransferFrame::ControlMessage::PAYLOAD_CANCELED:
return OperationResultCode::CLIENT_CANCELLATION_REMOTE_CANCEL_PAYLOAD;
default:
VLOG(1) << "PayloadManager: unknown event=" << event;
return OperationResultCode::DETAIL_UNKNOWN;
}
}
PayloadProgressInfo::Status PayloadStatusToTransferUpdateStatus(
PayloadStatus status) {
switch (status) {
case PayloadStatus::LOCAL_CANCELLATION:
case PayloadStatus::REMOTE_CANCELLATION:
return PayloadProgressInfo::Status::kCanceled;
case PayloadStatus::SUCCESS:
return PayloadProgressInfo::Status::kSuccess;
default:
return PayloadProgressInfo::Status::kFailure;
}
}
} // namespace
bool PayloadManager::SendPayloadLoop(
int PayloadManager::SendPayloadLoop(
ClientProxy* client, PendingPayload& pending_payload,
PayloadTransferFrame::PayloadHeader& payload_header,
std::int64_t& next_chunk_offset, size_t resume_offset, int index) {
// in lieu of structured binding:
auto pair = GetAvailableAndUnavailableEndpoints(pending_payload);
const EndpointIds& available_endpoint_ids =
EndpointsToEndpointIds(pair.first);
const Endpoints& unavailable_endpoints = pair.second;
int64_t next_chunk_offset, size_t resume_offset, int index) {
auto [available_endpoint_ids, unavailable_endpoints] =
GetAvailableAndUnavailableEndpoints(pending_payload);
// First, handle any non-available endpoints.
for (const auto& endpoint : unavailable_endpoints) {
@@ -91,7 +132,7 @@ bool PayloadManager::SendPayloadLoop(
<< pending_payload.GetInternalPayload()->GetId()
<< " after sending " << next_chunk_offset
<< " bytes because none of the endpoints are available anymore.";
return false;
return -1;
}
// Check if the payload has been cancelled by the client and, if so,
@@ -104,13 +145,9 @@ bool PayloadManager::SendPayloadLoop(
client, available_endpoint_ids, payload_header, next_chunk_offset,
OperationResultCode::CLIENT_CANCELLATION_LOCAL_CANCEL_PAYLOAD,
PayloadStatus::LOCAL_CANCELLATION);
return false;
return -1;
}
// Update the current offsets for all endpoints still active for this
// payload. For the sake of accuracy, we update the pending payload here
// because it's after all payload terminating events are handled, but
// right before we actually start detaching the next chunk.
if (next_chunk_offset == 0 && resume_offset > 0) {
ExceptionOr<size_t> real_offset =
pending_payload.GetInternalPayload()->SkipToOffset(resume_offset);
@@ -123,13 +160,17 @@ bool PayloadManager::SendPayloadLoop(
payload_header, next_chunk_offset,
OperationResultCode::IO_FILE_READING_ERROR,
PayloadStatus::LOCAL_ERROR);
return false;
return -1;
}
VLOG(1) << "PayloadManager successfully skipped " << real_offset.GetResult()
<< " bytes on payload_id "
<< pending_payload.GetInternalPayload()->GetId();
next_chunk_offset = real_offset.GetResult();
}
// Update the current offsets for all endpoints still active for this
// payload. For the sake of accuracy, we update the pending payload here
// because it's after all payload terminating events are handled, but
// right before we actually start detaching the next chunk.
for (const auto& endpoint_id : available_endpoint_ids) {
pending_payload.SetOffsetForEndpoint(endpoint_id, next_chunk_offset);
}
@@ -139,19 +180,20 @@ bool PayloadManager::SendPayloadLoop(
int chunk_size = GetOptimalChunkSize(available_endpoint_ids);
ByteArray next_chunk =
pending_payload.GetInternalPayload()->DetachNextChunk(chunk_size);
if (shutdown_.Get()) return false;
if (shutdown_.Get()) return -1;
// Save chunk size. We'll need it after we move next_chunk.
auto next_chunk_size = next_chunk.size();
if (!next_chunk_size &&
size_t next_chunk_size = next_chunk.size();
// If there are no more chunks, check if there should be more data to send.
if (next_chunk_size == 0 &&
pending_payload.GetInternalPayload()->GetTotalSize() > 0 &&
pending_payload.GetInternalPayload()->GetTotalSize() <
pending_payload.GetInternalPayload()->GetTotalSize() >
next_chunk_offset) {
VLOG(1) << "Payload xfer failed: payload_id="
<< pending_payload.GetInternalPayload()->GetId();
HandleFinishedOutgoingPayload(
client, available_endpoint_ids, payload_header, next_chunk_offset,
OperationResultCode::IO_FILE_READING_ERROR, PayloadStatus::LOCAL_ERROR);
return false;
return -1;
}
// Only need to handle outgoing data chunk offset, because the offset will be
@@ -160,13 +202,14 @@ bool PayloadManager::SendPayloadLoop(
// happened.
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);
const std::vector<std::string>& failed_endpoint_ids =
endpoint_manager_->SendPayloadChunk(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="
<< payload_header.id() << "; endpoint_ids={"
<< ToString(failed_endpoint_ids) << "}",
<< EndpointIdsToString(failed_endpoint_ids) << "}",
HandleFinishedOutgoingPayload(
client, failed_endpoint_ids, payload_header, next_chunk_offset,
OperationResultCode::CONNECTIVITY_GENERIC_WRITING_CHANNEL_IO_ERROR,
@@ -191,90 +234,36 @@ bool PayloadManager::SendPayloadLoop(
payload_chunk.offset(), payload_chunk.body().size());
}
}
VLOG(1) << "PayloadManager done sending chunk at offset "
<< next_chunk_offset << " of payload_id="
<< pending_payload.GetInternalPayload()->GetId();
next_chunk_offset += next_chunk_size;
if (!next_chunk_size) {
if (next_chunk_size == 0) {
// That was the last chunk, we're outta here.
VLOG(1) << "Payload xfer done: payload_id="
<< pending_payload.GetInternalPayload()->GetId()
<< "; size=" << next_chunk_offset;
return false;
return -1;
} else {
VLOG(1) << "PayloadManager done sending chunk at offset "
<< next_chunk_offset << " of payload_id="
<< pending_payload.GetInternalPayload()->GetId();
}
}
return true;
return next_chunk_size;
}
std::pair<PayloadManager::Endpoints, PayloadManager::Endpoints>
std::pair<std::vector<std::string>, PayloadManager::Endpoints>
PayloadManager::GetAvailableAndUnavailableEndpoints(
const PendingPayload& pending_payload) {
Endpoints available;
Endpoints unavailable;
auto results = std::make_pair(std::vector<std::string>(),
std::vector<const EndpointInfo*>());
for (auto* endpoint_info : pending_payload.GetEndpoints()) {
if (endpoint_info->status.Get() ==
PayloadManager::EndpointInfo::Status::kAvailable) {
available.push_back(endpoint_info);
if (endpoint_info->status.Get() == EndpointInfo::Status::kAvailable) {
results.first.push_back(endpoint_info->id);
} else {
unavailable.push_back(endpoint_info);
results.second.push_back(endpoint_info);
}
}
return std::make_pair(std::move(available), std::move(unavailable));
}
PayloadManager::EndpointIds PayloadManager::EndpointsToEndpointIds(
const Endpoints& endpoints) {
EndpointIds endpoint_ids;
endpoint_ids.reserve(endpoints.size());
for (const auto& item : endpoints) {
if (item) {
endpoint_ids.emplace_back(item->id);
}
}
return endpoint_ids;
}
std::string PayloadManager::ToString(const Endpoints& endpoints) {
std::string endpoints_string = absl::StrCat(endpoints.size(), ": ");
bool first = true;
for (const auto& item : endpoints) {
if (first) {
absl::StrAppend(&endpoints_string, item->id);
first = false;
} else {
absl::StrAppend(&endpoints_string, ", ", item->id);
}
}
return endpoints_string;
}
std::string PayloadManager::ToString(const EndpointIds& endpoint_ids) {
std::string endpoints_string = absl::StrCat(endpoint_ids.size(), ": ");
bool first = true;
for (const auto& id : endpoint_ids) {
if (first) {
absl::StrAppend(&endpoints_string, id);
first = false;
} else {
absl::StrAppend(&endpoints_string, ", ", id);
}
}
return endpoints_string;
}
std::string PayloadManager::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");
}
return results;
}
std::string PayloadManager::ToString(EndpointInfo::Status status) {
@@ -292,7 +281,7 @@ std::string PayloadManager::ToString(EndpointInfo::Status status) {
// Creates and starts tracking a PendingPayload for this Payload.
Payload::Id PayloadManager::CreateOutgoingPayload(
Payload payload, const EndpointIds& endpoint_ids) {
Payload payload, const std::vector<std::string>& endpoint_ids) {
ErrorOr<std::unique_ptr<InternalPayload>> result =
CreateOutgoingInternalPayload(std::move(payload));
if (result.has_error()) {
@@ -389,13 +378,14 @@ bool PayloadManager::NotifyShutdown() {
}
void PayloadManager::SendPayload(ClientProxy* client,
const EndpointIds& endpoint_ids,
const std::vector<std::string>& endpoint_ids,
Payload payload) {
if (shutdown_.Get()) return;
VLOG(1) << "SendPayload: endpoint_ids={" << ToString(endpoint_ids) << "}";
VLOG(1) << "SendPayload: endpoint_ids={" << EndpointIdsToString(endpoint_ids)
<< "}";
// Before transfer to internal payload, retrieves the Payload size for
// analytics.
std::int64_t payload_total_size;
int64_t payload_total_size;
switch (payload.GetType()) {
case connections::PayloadType::kBytes:
payload_total_size = payload.AsBytes().size();
@@ -420,8 +410,7 @@ void PayloadManager::SendPayload(ClientProxy* client,
OperationResultCode::NEARBY_GENERIC_OUTGOING_PAYLOAD_CREATION_FAILURE);
VLOG(1) << "PayloadManager failed to determine the right executor for "
"outgoing payload_id="
<< payload.GetId()
<< ", payload_type=" << ToString(payload.GetType());
<< payload.GetId() << ", payload_type=" << payload.GetType();
return;
}
@@ -450,7 +439,7 @@ void PayloadManager::SendPayload(ClientProxy* client,
NEARBY_GENERIC_OUTGOING_PAYLOAD_CREATION_FAILURE);
VLOG(1) << "PayloadManager failed to create InternalPayload for outgoing "
"payload_id="
<< payload_id << ", payload_type=" << ToString(payload_type)
<< payload_id << ", payload_type=" << payload_type
<< ", aborting sendPayload().";
return;
}
@@ -465,13 +454,19 @@ void PayloadManager::SendPayload(ClientProxy* client,
CreatePayloadHeader(*internal_payload, resume_offset)};
bool should_continue = true;
std::int64_t next_chunk_offset = 0;
int64_t next_chunk_offset = 0;
int index = 0;
while (should_continue && !shutdown_.Get()) {
should_continue =
SendPayloadLoop(client, *pending_payload, payload_header,
next_chunk_offset, resume_offset, index);
int bytes_sent = SendPayloadLoop(client, *pending_payload, payload_header,
next_chunk_offset, resume_offset, index);
should_continue = (bytes_sent >= 0);
if (should_continue) {
if (next_chunk_offset == 0 && resume_offset > 0) {
next_chunk_offset = resume_offset;
}
next_chunk_offset += bytes_sent;
}
index++;
}
@@ -482,8 +477,7 @@ void PayloadManager::SendPayload(ClientProxy* client,
});
});
VLOG(1) << "PayloadManager: xfer scheduled: self=" << this
<< "; payload_id=" << payload_id
<< ", payload_type=" << ToString(payload_type);
<< "; payload_id=" << payload_id << ", payload_type=" << payload_type;
}
PayloadManager::PendingPayloadHandle PayloadManager::GetPayload(
@@ -580,14 +574,14 @@ void PayloadManager::OnEndpointDisconnect(ClientProxy* client,
pending_payloads_.ForEachPayload([&](PendingPayload* pending_payload) {
auto endpoint_info = pending_payload->GetEndpoint(endpoint_id);
if (!endpoint_info) return;
std::int64_t endpoint_offset = endpoint_info->offset;
int64_t endpoint_offset = endpoint_info->offset;
// Stop tracking the endpoint for this payload.
pending_payload->RemoveEndpoints({endpoint_id});
// |endpoint_info| is longer valid after calling
// RemoveEndpoints.
endpoint_info = nullptr;
std::int64_t payload_total_size =
int64_t payload_total_size =
pending_payload->GetInternalPayload()->GetTotalSize();
// If no endpoints are left for this payload, close it.
@@ -669,45 +663,6 @@ OperationResultCode PayloadManager::EndpointInfoStatusToOperationResultCode(
}
}
PayloadStatus PayloadManager::ControlMessageEventToPayloadStatus(
PayloadTransferFrame::ControlMessage::EventType event) {
switch (event) {
case PayloadTransferFrame::ControlMessage::PAYLOAD_ERROR:
return PayloadStatus::REMOTE_ERROR;
case PayloadTransferFrame::ControlMessage::PAYLOAD_CANCELED:
return PayloadStatus::REMOTE_CANCELLATION;
default:
VLOG(1) << "PayloadManager: unknown event=" << event;
return PayloadStatus::UNKNOWN_PAYLOAD_STATUS;
}
}
OperationResultCode PayloadManager::ControlMessageEventToOperationResultCode(
PayloadTransferFrame::ControlMessage::EventType event) {
switch (event) {
case PayloadTransferFrame::ControlMessage::PAYLOAD_ERROR:
return OperationResultCode::NEARBY_GENERIC_REMOTE_REPORT_PAYLOADS_ERROR;
case PayloadTransferFrame::ControlMessage::PAYLOAD_CANCELED:
return OperationResultCode::CLIENT_CANCELLATION_REMOTE_CANCEL_PAYLOAD;
default:
VLOG(1) << "PayloadManager: unknown event=" << event;
return OperationResultCode::DETAIL_UNKNOWN;
}
}
PayloadProgressInfo::Status PayloadManager::PayloadStatusToTransferUpdateStatus(
PayloadStatus status) {
switch (status) {
case PayloadStatus::LOCAL_CANCELLATION:
case PayloadStatus::REMOTE_CANCELLATION:
return PayloadProgressInfo::Status::kCanceled;
case PayloadStatus::SUCCESS:
return PayloadProgressInfo::Status::kSuccess;
default:
return PayloadProgressInfo::Status::kFailure;
}
}
SingleThreadExecutor* PayloadManager::GetOutgoingPayloadExecutor(
PayloadType payload_type) {
switch (payload_type) {
@@ -722,7 +677,8 @@ SingleThreadExecutor* PayloadManager::GetOutgoingPayloadExecutor(
}
}
int PayloadManager::GetOptimalChunkSize(EndpointIds endpoint_ids) {
int PayloadManager::GetOptimalChunkSize(
const std::vector<std::string>& endpoint_ids) {
int minChunkSize = std::numeric_limits<int>::max();
for (const auto& endpoint_id : endpoint_ids) {
minChunkSize = std::min(
@@ -754,8 +710,7 @@ PayloadTransferFrame::PayloadHeader PayloadManager::CreatePayloadHeader(
}
PayloadTransferFrame::PayloadChunk PayloadManager::CreatePayloadChunk(
std::int64_t payload_chunk_offset, ByteArray payload_chunk_body,
int index) {
int64_t payload_chunk_offset, ByteArray payload_chunk_body, int index) {
PayloadTransferFrame::PayloadChunk payload_chunk;
payload_chunk.set_offset(payload_chunk_offset);
@@ -776,9 +731,8 @@ PayloadManager::CreateIncomingPayload(const PayloadTransferFrame& frame,
const std::string& endpoint_id,
const std::string& save_path) {
ErrorOr<std::unique_ptr<InternalPayload>> result =
CreateIncomingInternalPayload(frame, save_path.empty()
? custom_save_path_
: save_path);
CreateIncomingInternalPayload(
frame, save_path.empty() ? custom_save_path_ : save_path);
if (result.has_error()) {
return {result.error()};
}
@@ -788,7 +742,8 @@ PayloadManager::CreateIncomingPayload(const PayloadTransferFrame& frame,
pending_payloads_.StartTrackingPayload(
payload_id,
std::make_unique<PendingPayload>(
std::move(internal_payload), EndpointIds{endpoint_id}, true,
std::move(internal_payload), std::vector<std::string>{endpoint_id},
true,
absl::bind_front(&PayloadManager::OnPendingPayloadDestroy, this)));
return {pending_payloads_.GetPayload(payload_id)};
}
@@ -803,9 +758,9 @@ void PayloadManager::OnPendingPayloadDestroy(const PendingPayload* payload) {
}
void PayloadManager::SendClientCallbacksForFinishedOutgoingPayload(
ClientProxy* client, const EndpointIds& finished_endpoint_ids,
ClientProxy* client, const std::vector<std::string>& finished_endpoint_ids,
const PayloadTransferFrame::PayloadHeader& payload_header,
std::int64_t num_bytes_successfully_transferred, PayloadStatus status,
int64_t num_bytes_successfully_transferred, PayloadStatus status,
OperationResultCode operation_result_code) {
RunOnStatusUpdateThread(
"outgoing-payload-callbacks",
@@ -819,8 +774,7 @@ void PayloadManager::SendClientCallbacksForFinishedOutgoingPayload(
}
PayloadProgressInfo update{
payload_header.id(),
PayloadManager::PayloadStatusToTransferUpdateStatus(status),
payload_header.id(), PayloadStatusToTransferUpdateStatus(status),
payload_header.total_size(), num_bytes_successfully_transferred};
for (const auto& endpoint_id : finished_endpoint_ids) {
// Skip sending notifications if we have stopped tracking this
@@ -855,7 +809,7 @@ void PayloadManager::SendClientCallbacksForFinishedOutgoingPayload(
void PayloadManager::SendClientCallbacksForFinishedIncomingPayload(
ClientProxy* client, const std::string& endpoint_id,
const PayloadTransferFrame::PayloadHeader& payload_header,
std::int64_t offset_bytes, PayloadStatus status,
int64_t offset_bytes, PayloadStatus status,
OperationResultCode operation_result_code) {
RunOnStatusUpdateThread(
"incoming-payload-callbacks",
@@ -870,10 +824,9 @@ void PayloadManager::SendClientCallbacksForFinishedIncomingPayload(
// Unless we never started tracking this payload (meaning we
// failed to even create the InternalPayload), notify the client
// (and close it).
PayloadProgressInfo update{
payload_header.id(),
PayloadManager::PayloadStatusToTransferUpdateStatus(status),
payload_header.total_size(), offset_bytes};
PayloadProgressInfo update{payload_header.id(),
PayloadStatusToTransferUpdateStatus(status),
payload_header.total_size(), offset_bytes};
NotifyClientOfIncomingPayloadProgressInfo(client, endpoint_id, update);
DestroyPendingPayload(payload_header.id());
@@ -884,9 +837,9 @@ void PayloadManager::SendClientCallbacksForFinishedIncomingPayload(
}
void PayloadManager::SendControlMessage(
const EndpointIds& endpoint_ids,
const std::vector<std::string>& endpoint_ids,
const PayloadTransferFrame::PayloadHeader& payload_header,
std::int64_t num_bytes_successfully_transferred,
int64_t num_bytes_successfully_transferred,
PayloadTransferFrame::ControlMessage::EventType event_type) {
PayloadTransferFrame::ControlMessage control_message;
control_message.set_event(event_type);
@@ -922,7 +875,7 @@ bool PayloadManager::WaitForReceivedAck(
ClientProxy* client, const std::string& endpoint_id,
PendingPayload& pending_payload,
const PayloadTransferFrame::PayloadHeader& payload_header,
std::int64_t payload_chunk_offset, bool is_last_chunk) {
int64_t payload_chunk_offset, bool is_last_chunk) {
if (!is_last_chunk ||
!IsPayloadReceivedAckEnabled(client, endpoint_id, pending_payload)) {
return true;
@@ -1025,9 +978,9 @@ bool PayloadManager::IsPayloadReceivedAckEnabled(
}
void PayloadManager::HandleFinishedOutgoingPayload(
ClientProxy* client, const EndpointIds& finished_endpoint_ids,
ClientProxy* client, const std::vector<std::string>& finished_endpoint_ids,
const PayloadTransferFrame::PayloadHeader& payload_header,
std::int64_t num_bytes_successfully_transferred,
int64_t num_bytes_successfully_transferred,
OperationResultCode operation_result_code, PayloadStatus status) {
// This call will destroy a pending payload.
SendClientCallbacksForFinishedOutgoingPayload(
@@ -1071,7 +1024,7 @@ void PayloadManager::HandleFinishedOutgoingPayload(
void PayloadManager::HandleFinishedIncomingPayload(
ClientProxy* client, const std::string& endpoint_id,
const PayloadTransferFrame::PayloadHeader& payload_header,
std::int64_t offset_bytes, PayloadStatus status,
int64_t offset_bytes, PayloadStatus status,
OperationResultCode operation_result_code) {
SendClientCallbacksForFinishedIncomingPayload(client, endpoint_id,
payload_header, offset_bytes,
@@ -1097,8 +1050,8 @@ void PayloadManager::HandleFinishedIncomingPayload(
void PayloadManager::HandleSuccessfulOutgoingChunk(
ClientProxy* client, const std::string& endpoint_id,
const PayloadTransferFrame::PayloadHeader& payload_header,
std::int32_t payload_chunk_flags, std::int64_t payload_chunk_offset,
std::int64_t payload_chunk_body_size) {
int32_t payload_chunk_flags, int64_t payload_chunk_offset,
int64_t payload_chunk_body_size) {
{
MutexLock lock(&chunk_update_mutex_);
++outgoing_chunk_update_count_;
@@ -1192,8 +1145,8 @@ void PayloadManager::DestroyPendingPayload(Payload::Id payload_id) {
void PayloadManager::HandleSuccessfulIncomingChunk(
ClientProxy* client, const std::string& endpoint_id,
const PayloadTransferFrame::PayloadHeader& payload_header,
std::int32_t payload_chunk_flags, std::int64_t payload_chunk_offset,
std::int64_t payload_chunk_body_size) {
int32_t payload_chunk_flags, int64_t payload_chunk_offset,
int64_t payload_chunk_body_size) {
{
MutexLock lock(&chunk_update_mutex_);
++incoming_chunk_update_count_;
@@ -1377,7 +1330,7 @@ void PayloadManager::ProcessDataPacket(
payload_chunk.offset());
// Save size of packet before we move it.
std::int64_t payload_body_size = payload_chunk.body().size();
int64_t payload_body_size = payload_chunk.body().size();
if (pending_payload->GetInternalPayload()
->AttachNextChunk(payload_chunk.body())
@@ -1493,18 +1446,18 @@ void PayloadManager::NotifyClientOfIncomingPayloadProgressInfo(
}
void PayloadManager::RecordPayloadStartedAnalytics(
ClientProxy* client, const EndpointIds& endpoint_ids,
std::int64_t payload_id, PayloadType payload_type, std::int64_t offset,
std::int64_t total_size) {
ClientProxy* client, const std::vector<std::string>& endpoint_ids,
int64_t payload_id, PayloadType payload_type, int64_t offset,
int64_t total_size) {
client->GetAnalyticsRecorder().OnOutgoingPayloadStarted(
endpoint_ids, payload_id, payload_type,
total_size == -1 ? -1 : total_size - offset);
}
void PayloadManager::RecordInvalidPayloadAnalytics(
ClientProxy* client, const EndpointIds& endpoint_ids,
std::int64_t payload_id, PayloadType payload_type, std::int64_t offset,
std::int64_t total_size, OperationResultCode operation_result_code) {
ClientProxy* client, const std::vector<std::string>& endpoint_ids,
int64_t payload_id, PayloadType payload_type, int64_t offset,
int64_t total_size, OperationResultCode operation_result_code) {
RecordPayloadStartedAnalytics(client, endpoint_ids, payload_id, payload_type,
offset, total_size);
@@ -1582,8 +1535,8 @@ bool PayloadManager::EndpointInfo::IsEndpointAvailable(
PayloadManager::PendingPayload::PendingPayload(
std::unique_ptr<InternalPayload> internal_payload,
const EndpointIds& endpoint_ids, bool is_incoming,
DestroyCallback destroy_callback)
const std::vector<std::string>& endpoint_ids, bool is_incoming,
absl::AnyInvocable<void(PendingPayload*) &&> destroy_callback)
: is_incoming_(is_incoming),
internal_payload_(std::move(internal_payload)),
destroy_callback_(std::move(destroy_callback)) {
@@ -1649,7 +1602,7 @@ PayloadManager::EndpointInfo* PayloadManager::PendingPayload::GetEndpoint(
}
void PayloadManager::PendingPayload::RemoveEndpoints(
const EndpointIds& endpoint_ids) {
const std::vector<std::string>& endpoint_ids) {
MutexLock lock(&mutex_);
for (const auto& id : endpoint_ids) {
@@ -1669,7 +1622,7 @@ void PayloadManager::PendingPayload::SetEndpointStatusFromControlMessage(
}
void PayloadManager::PendingPayload::SetOffsetForEndpoint(
const std::string& endpoint_id, std::int64_t offset) {
const std::string& endpoint_id, int64_t offset) {
MutexLock lock(&mutex_);
auto item = endpoints_.find(endpoint_id);
@@ -1783,7 +1736,8 @@ void PayloadManager::PendingPayloads::Release(PendingPayload* payload) {
}
PayloadManager::PendingPayloadHandle::PendingPayloadHandle(
PendingPayload* payload, DestroyCallback destroy_callback)
PendingPayload* payload,
absl::AnyInvocable<void(PendingPayload*) &&> destroy_callback)
: payload_(payload), destroy_callback_(std::move(destroy_callback)) {}
PayloadManager::PendingPayloadHandle::~PendingPayloadHandle() {
@@ -1793,9 +1747,8 @@ PayloadManager::PendingPayloadHandle::~PendingPayloadHandle() {
}
std::string PayloadManager::PendingPayload::ToString() const {
return absl::StrFormat("Payload(%s, %d)",
IsIncoming() ? "incoming" : "outgoing", GetId());
return absl::StrCat("Payload(", IsIncoming() ? "incoming" : "outgoing",
GetId(), ")");
}
} // namespace connections
} // namespace nearby
} // namespace nearby::connections
connections/implementation/payload_manager.h
+56 -76
View File
@@ -42,8 +42,7 @@
#include "internal/platform/mutex.h"
#include "internal/platform/single_thread_executor.h"
namespace nearby {
namespace connections {
namespace nearby::connections {
// Annotations for methods that need to run on PayloadStatusUpdateThread.
// Use only in PayloadManager
@@ -52,13 +51,13 @@ namespace connections {
class PayloadManager : public EndpointManager::FrameProcessor {
public:
using EndpointIds = std::vector<std::string>;
static constexpr absl::Duration kWaitCloseTimeout = absl::Milliseconds(5000);
explicit PayloadManager(EndpointManager& endpoint_manager);
~PayloadManager() override;
void SendPayload(ClientProxy* client, const EndpointIds& endpoint_ids,
void SendPayload(ClientProxy* client,
const std::vector<std::string>& endpoint_ids,
Payload payload);
Status CancelPayload(ClientProxy* client, Payload::Id payload_id);
@@ -103,7 +102,7 @@ class PayloadManager : public EndpointManager::FrameProcessor {
std::string id;
AtomicReference<Status> status{Status::kUnknown};
std::int64_t offset = 0;
int64_t offset = 0;
mutable Mutex payload_received_ack_mutex;
ConditionVariable payload_received_ack_cond{&payload_received_ack_mutex};
bool is_payload_received_ack ABSL_GUARDED_BY(payload_received_ack_mutex) =
@@ -113,10 +112,10 @@ class PayloadManager : public EndpointManager::FrameProcessor {
// Tracks state for an InternalPayload and the endpoints associated with it.
class PendingPayload {
public:
using DestroyCallback = absl::AnyInvocable<void(PendingPayload*) &&>;
PendingPayload(std::unique_ptr<InternalPayload> internal_payload,
const EndpointIds& endpoint_ids, bool is_incoming,
DestroyCallback destroy_callback);
PendingPayload(
std::unique_ptr<InternalPayload> internal_payload,
const std::vector<std::string>& endpoint_ids, bool is_incoming,
absl::AnyInvocable<void(PendingPayload*) &&> destroy_callback);
PendingPayload(PendingPayload&&) = default;
PendingPayload& operator=(PendingPayload&&) = default;
@@ -146,7 +145,7 @@ class PayloadManager : public EndpointManager::FrameProcessor {
ABSL_LOCKS_EXCLUDED(mutex_);
// Removes the given endpoints, e.g. on error.
void RemoveEndpoints(const EndpointIds& endpoint_ids_to_remove)
void RemoveEndpoints(const std::vector<std::string>& endpoint_ids_to_remove)
ABSL_LOCKS_EXCLUDED(mutex_);
// Sets the status for a particular endpoint.
@@ -156,8 +155,8 @@ class PayloadManager : public EndpointManager::FrameProcessor {
ControlMessage& control_message) ABSL_LOCKS_EXCLUDED(mutex_);
// Sets the offset for a particular endpoint.
void SetOffsetForEndpoint(const std::string& endpoint_id,
std::int64_t offset) ABSL_LOCKS_EXCLUDED(mutex_);
void SetOffsetForEndpoint(const std::string& endpoint_id, int64_t offset)
ABSL_LOCKS_EXCLUDED(mutex_);
// Closes internal_payload_.
// Close is called when a pending peyload does not have associated
@@ -173,11 +172,11 @@ class PayloadManager : public EndpointManager::FrameProcessor {
private:
mutable Mutex mutex_;
bool is_incoming_;
const bool is_incoming_;
AtomicBoolean is_locally_canceled_{false};
AtomicBoolean is_closed_;
std::unique_ptr<InternalPayload> internal_payload_;
DestroyCallback destroy_callback_;
const std::unique_ptr<InternalPayload> internal_payload_;
absl::AnyInvocable<void(PendingPayload*) &&> destroy_callback_;
absl::flat_hash_map<std::string, EndpointInfo> endpoints_
ABSL_GUARDED_BY(mutex_);
int refcount_ = 0;
@@ -188,10 +187,10 @@ class PayloadManager : public EndpointManager::FrameProcessor {
// Create instances with `GetPayload(Payload::Id)`.
class PendingPayloadHandle {
public:
using DestroyCallback = absl::AnyInvocable<void(PendingPayload*) &&>;
PendingPayloadHandle() = default;
PendingPayloadHandle(PendingPayload* payload,
DestroyCallback destroy_callback);
PendingPayloadHandle(
PendingPayload* payload,
absl::AnyInvocable<void(PendingPayload*) &&> destroy_callback);
PendingPayloadHandle(const PendingPayloadHandle&) = delete;
PendingPayloadHandle(PendingPayloadHandle&& other) {
payload_ = other.payload_;
@@ -217,7 +216,7 @@ class PayloadManager : public EndpointManager::FrameProcessor {
private:
PendingPayload* payload_ = nullptr;
DestroyCallback destroy_callback_;
absl::AnyInvocable<void(PendingPayload*) &&> destroy_callback_;
};
// Tracks and manages PendingPayload objects in a synchronized manner.
@@ -256,31 +255,26 @@ class PayloadManager : public EndpointManager::FrameProcessor {
};
using Endpoints = std::vector<const EndpointInfo*>;
static std::string ToString(const EndpointIds& endpoint_ids);
static std::string ToString(const Endpoints& endpoints);
static std::string ToString(PayloadType type);
static std::string ToString(EndpointInfo::Status status);
// Splits the endpoints for this payload by availability.
// Returns a pair of lists of EndpointInfo*, with the first being the list
// of still-available endpoints, and the second for unavailable endpoints.
static std::pair<Endpoints, Endpoints> GetAvailableAndUnavailableEndpoints(
const PendingPayload& pending_payload);
// Returns a pair of lists, with the first being the list of still-available
// endpoint ids, and the second for unavailable endpoints.
static std::pair<std::vector<std::string>, Endpoints>
GetAvailableAndUnavailableEndpoints(const PendingPayload& pending_payload);
// Converts list of EndpointInfo to list of Endpoint ids.
// Returns list of endpoint ids.
static EndpointIds EndpointsToEndpointIds(const Endpoints& endpoints);
bool SendPayloadLoop(
// Returns the number of bytes sent. 0 bytes sent indicates end of payload.
// Returns -1 on error.
int SendPayloadLoop(
ClientProxy* client, PendingPayload& pending_payload,
location::nearby::connections::PayloadTransferFrame::PayloadHeader&
payload_header,
std::int64_t& next_chunk_offset, size_t resume_offset, int index);
int64_t next_chunk_offset, size_t resume_offset, int index);
void SendClientCallbacksForFinishedIncomingPayloadRunnable(
ClientProxy* client, const std::string& endpoint_id,
const location::nearby::connections::PayloadTransferFrame::PayloadHeader&
payload_header,
std::int64_t offset_bytes,
int64_t offset_bytes,
location::nearby::proto::connections::PayloadStatus status,
location::nearby::proto::connections::OperationResultCode
operation_result_code);
@@ -292,27 +286,14 @@ class PayloadManager : public EndpointManager::FrameProcessor {
EndpointInfoStatusToPayloadStatus(EndpointInfo::Status status);
static location::nearby::proto::connections::OperationResultCode
EndpointInfoStatusToOperationResultCode(EndpointInfo::Status status);
// Converts a ControlMessage::EventType for a particular payload to a
// PayloadStatus. Called when we've received a ControlMessage with this
// event from a remote endpoint; thus the PayloadStatuses are REMOTE_*.
static location::nearby::proto::connections::PayloadStatus
ControlMessageEventToPayloadStatus(
location::nearby::connections::PayloadTransferFrame::ControlMessage::
EventType event);
static location::nearby::proto::connections::OperationResultCode
ControlMessageEventToOperationResultCode(
location::nearby::connections::PayloadTransferFrame::ControlMessage::
EventType event);
static PayloadProgressInfo::Status PayloadStatusToTransferUpdateStatus(
location::nearby::proto::connections::PayloadStatus status);
int GetOptimalChunkSize(EndpointIds endpoint_ids);
int GetOptimalChunkSize(const std::vector<std::string>& endpoint_ids);
location::nearby::connections::PayloadTransferFrame::PayloadHeader
CreatePayloadHeader(const InternalPayload& internal_payload, size_t offset);
location::nearby::connections::PayloadTransferFrame::PayloadChunk
CreatePayloadChunk(std::int64_t offset, ByteArray body, int index);
CreatePayloadChunk(int64_t offset, ByteArray body, int index);
bool IsLastChunk(
location::nearby::connections::PayloadTransferFrame::PayloadChunk
payload_chunk) {
@@ -326,18 +307,19 @@ class PayloadManager : public EndpointManager::FrameProcessor {
// path set in `SetCustomSavePath()`.
ErrorOr<PendingPayloadHandle> CreateIncomingPayload(
const location::nearby::connections::PayloadTransferFrame& frame,
const std::string& endpoint_id,
const std::string& save_path) ABSL_LOCKS_EXCLUDED(mutex_);
const std::string& endpoint_id, const std::string& save_path)
ABSL_LOCKS_EXCLUDED(mutex_);
Payload::Id CreateOutgoingPayload(Payload payload,
const EndpointIds& endpoint_ids)
Payload::Id CreateOutgoingPayload(
Payload payload, const std::vector<std::string>& endpoint_ids)
ABSL_LOCKS_EXCLUDED(mutex_);
void SendClientCallbacksForFinishedOutgoingPayload(
ClientProxy* client, const EndpointIds& finished_endpoint_ids,
ClientProxy* client,
const std::vector<std::string>& finished_endpoint_ids,
const location::nearby::connections::PayloadTransferFrame::PayloadHeader&
payload_header,
std::int64_t num_bytes_successfully_transferred,
int64_t num_bytes_successfully_transferred,
location::nearby::proto::connections::PayloadStatus status,
location::nearby::proto::connections::OperationResultCode
operation_result_code);
@@ -345,16 +327,16 @@ class PayloadManager : public EndpointManager::FrameProcessor {
ClientProxy* client, const std::string& endpoint_id,
const location::nearby::connections::PayloadTransferFrame::PayloadHeader&
payload_header,
std::int64_t offset_bytes,
int64_t offset_bytes,
location::nearby::proto::connections::PayloadStatus status,
location::nearby::proto::connections::OperationResultCode
operation_result_code);
void SendControlMessage(
const EndpointIds& endpoint_ids,
const std::vector<std::string>& endpoint_ids,
const location::nearby::connections::PayloadTransferFrame::PayloadHeader&
payload_header,
std::int64_t num_bytes_successfully_transferred,
int64_t num_bytes_successfully_transferred,
location::nearby::connections::PayloadTransferFrame::ControlMessage::
EventType event_type);
@@ -368,7 +350,7 @@ class PayloadManager : public EndpointManager::FrameProcessor {
PendingPayload& pending_payload,
const location::nearby::connections::PayloadTransferFrame::PayloadHeader&
payload_header,
std::int64_t payload_chunk_offset, bool is_last_chunk);
int64_t payload_chunk_offset, bool is_last_chunk);
bool IsPayloadReceivedAckEnabled(ClientProxy* client,
const std::string& endpoint_id,
PendingPayload& pending_payload);
@@ -376,10 +358,11 @@ class PayloadManager : public EndpointManager::FrameProcessor {
// Handles a finished outgoing payload for the given endpointIds. All
// statuses except for SUCCESS are handled here.
void HandleFinishedOutgoingPayload(
ClientProxy* client, const EndpointIds& finished_endpoint_ids,
ClientProxy* client,
const std::vector<std::string>& finished_endpoint_ids,
const location::nearby::connections::PayloadTransferFrame::PayloadHeader&
payload_header,
std::int64_t num_bytes_successfully_transferred,
int64_t num_bytes_successfully_transferred,
location::nearby::proto::connections::OperationResultCode
operation_result_code,
location::nearby::proto::connections::PayloadStatus status = location::
@@ -388,7 +371,7 @@ class PayloadManager : public EndpointManager::FrameProcessor {
ClientProxy* client, const std::string& endpoint_id,
const location::nearby::connections::PayloadTransferFrame::PayloadHeader&
payload_header,
std::int64_t offset_bytes,
int64_t offset_bytes,
location::nearby::proto::connections::PayloadStatus status,
location::nearby::proto::connections::OperationResultCode
operation_result_code);
@@ -397,14 +380,14 @@ class PayloadManager : public EndpointManager::FrameProcessor {
ClientProxy* client, const std::string& endpoint_id,
const location::nearby::connections::PayloadTransferFrame::PayloadHeader&
payload_header,
std::int32_t payload_chunk_flags, std::int64_t payload_chunk_offset,
std::int64_t payload_chunk_body_size);
int32_t payload_chunk_flags, int64_t payload_chunk_offset,
int64_t payload_chunk_body_size);
void HandleSuccessfulIncomingChunk(
ClientProxy* client, const std::string& endpoint_id,
const location::nearby::connections::PayloadTransferFrame::PayloadHeader&
payload_header,
std::int32_t payload_chunk_flags, std::int64_t payload_chunk_offset,
std::int64_t payload_chunk_body_size);
int32_t payload_chunk_flags, int64_t payload_chunk_offset,
int64_t payload_chunk_body_size);
void ProcessDataPacket(ClientProxy* to_client,
const std::string& from_endpoint_id,
@@ -436,16 +419,14 @@ class PayloadManager : public EndpointManager::FrameProcessor {
ABSL_LOCKS_EXCLUDED(mutex_);
void CancelAllPayloads() ABSL_LOCKS_EXCLUDED(mutex_);
void RecordPayloadStartedAnalytics(ClientProxy* client,
const EndpointIds& endpoint_ids,
std::int64_t payload_id,
PayloadType payload_type,
std::int64_t offset,
std::int64_t total_size);
void RecordPayloadStartedAnalytics(
ClientProxy* client, const std::vector<std::string>& endpoint_ids,
int64_t payload_id, PayloadType payload_type, int64_t offset,
int64_t total_size);
void RecordInvalidPayloadAnalytics(
ClientProxy* client, const EndpointIds& endpoint_ids,
std::int64_t payload_id, PayloadType payload_type, std::int64_t offset,
std::int64_t total_size,
ClientProxy* client, const std::vector<std::string>& endpoint_ids,
int64_t payload_id, PayloadType payload_type, int64_t offset,
int64_t total_size,
location::nearby::proto::connections::OperationResultCode
operation_result_code);
@@ -480,7 +461,6 @@ class PayloadManager : public EndpointManager::FrameProcessor {
ABSL_GUARDED_BY(chunk_update_mutex_) = absl::InfinitePast();
};
} // namespace connections
} // namespace nearby
} // namespace nearby::connections
#endif // CORE_INTERNAL_PAYLOAD_MANAGER_H_
connections/payload.cc
+3 -4
View File
@@ -28,9 +28,9 @@
#include "internal/platform/byte_array.h"
#include "internal/platform/file.h"
#include "internal/platform/input_stream.h"
#include "internal/platform/logging.h"
namespace nearby {
namespace connections {
namespace nearby::connections {
namespace {
@@ -151,5 +151,4 @@ const std::string& Payload::GetParentFolder() const { return parent_folder_; }
const std::string& Payload::GetFileName() const { return file_name_; }
} // namespace connections
} // namespace nearby
} // namespace nearby::connections
connections/payload.h
+3 -10
View File
@@ -15,26 +15,20 @@
#ifndef CORE_PAYLOAD_H_
#define CORE_PAYLOAD_H_
#include <cstdint>
#include <functional>
#include <cstddef>
#include <memory>
#include <string>
#include <utility>
#include <variant>
#include "absl/time/clock.h"
#include "absl/time/time.h"
#include "absl/types/variant.h"
#include "connections/payload_type.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/file.h"
#include "internal/platform/input_stream.h"
#include "internal/platform/logging.h"
#include "internal/platform/payload_id.h"
#include "internal/platform/prng.h"
namespace nearby {
namespace connections {
namespace nearby::connections {
// Payload is default-constructible, and moveable, but not copyable container
// that holds at most one instance of one of:
@@ -124,7 +118,6 @@ class Payload {
Content content_;
};
} // namespace connections
} // namespace nearby
} // namespace nearby::connections
#endif // CORE_PAYLOAD_H_
connections/payload_test.cc
+19 -4
View File
@@ -21,6 +21,7 @@
#include <utility>
#include "gtest/gtest.h"
#include "absl/strings/str_cat.h"
#include "absl/strings/string_view.h"
#include "connections/payload_type.h"
#include "internal/platform/byte_array.h"
@@ -28,8 +29,7 @@
#include "internal/platform/input_stream.h"
#include "internal/platform/pipe.h"
namespace nearby {
namespace connections {
namespace nearby::connections {
TEST(PayloadTest, DefaultPayloadHasUnknownType) {
Payload payload;
@@ -132,5 +132,20 @@ TEST(PayloadTest, PayloadIsNotCopyable) {
EXPECT_FALSE(std::is_copy_assignable_v<Payload>);
}
} // namespace connections
} // namespace nearby
TEST(PayloadTypeTest, Stringify) {
EXPECT_EQ(absl::StrCat(PayloadType::kUnknown), "Unknown");
EXPECT_EQ(absl::StrCat(PayloadType::kBytes), "Bytes");
EXPECT_EQ(absl::StrCat(PayloadType::kFile), "File");
EXPECT_EQ(absl::StrCat(PayloadType::kStream), "Stream");
}
TEST(PayloadDirectionTest, Stringify) {
EXPECT_EQ(absl::StrCat(PayloadDirection::UNKNOWN_DIRECTION_PAYLOAD),
"UNKNOWN_DIRECTION_PAYLOAD");
EXPECT_EQ(absl::StrCat(PayloadDirection::INCOMING_PAYLOAD),
"INCOMING_PAYLOAD");
EXPECT_EQ(absl::StrCat(PayloadDirection::OUTGOING_PAYLOAD),
"OUTGOING_PAYLOAD");
}
} // namespace nearby::connections
connections/payload_type.h
+38 -35
View File
@@ -16,59 +16,62 @@
#define CORE_PAYLOAD_TYPE_H_
#include <ostream>
#include "absl/strings/str_cat.h"
namespace nearby {
namespace connections {
namespace nearby::connections {
enum class PayloadType { kUnknown = 0, kBytes = 1, kFile = 2, kStream = 3 };
// Support logging of PayloadType.
template <typename Sink>
void AbslStringify(Sink& sink, PayloadType payload_type) {
switch (payload_type) {
case PayloadType::kBytes:
sink.Append("Bytes");
break;
case PayloadType::kStream:
sink.Append("Stream");
break;
case PayloadType::kFile:
sink.Append("File");
break;
case PayloadType::kUnknown:
sink.Append("Unknown");
break;
}
}
inline std::ostream& operator<<(std::ostream& os, PayloadType payload_type) {
return os << absl::StrCat(payload_type);
}
enum class PayloadDirection {
UNKNOWN_DIRECTION_PAYLOAD = 0,
INCOMING_PAYLOAD = 1,
OUTGOING_PAYLOAD = 2,
};
inline std::ostream& operator<<(std::ostream& os, PayloadType payload_type) {
switch (payload_type) {
case PayloadType::kUnknown:
os << "kUnknown";
// Support logging of PayloadDirection.
template <typename Sink>
void AbslStringify(Sink& sink, PayloadDirection payload_direction) {
switch (payload_direction) {
case PayloadDirection::UNKNOWN_DIRECTION_PAYLOAD:
sink.Append("UNKNOWN_DIRECTION_PAYLOAD");
break;
case PayloadType::kBytes:
os << "kBytes";
case PayloadDirection::INCOMING_PAYLOAD:
sink.Append("INCOMING_PAYLOAD");
break;
case PayloadType::kFile:
os << "kFile";
break;
case PayloadType::kStream:
os << "kStream";
break;
default:
os << "Invalid PayloadType";
case PayloadDirection::OUTGOING_PAYLOAD:
sink.Append("OUTGOING_PAYLOAD");
break;
}
return os;
}
inline std::ostream& operator<<(std::ostream& os,
PayloadDirection payload_direction) {
switch (payload_direction) {
case PayloadDirection::UNKNOWN_DIRECTION_PAYLOAD:
os << "UNKNOWN_DIRECTION_PAYLOAD";
break;
case PayloadDirection::INCOMING_PAYLOAD:
os << "INCOMING_PAYLOAD";
break;
case PayloadDirection::OUTGOING_PAYLOAD:
os << "OUTGOING_PAYLOAD";
break;
default:
os << "Invalid PayloadDirection";
break;
}
return os;
PayloadDirection payload_direction) {
return os << absl::StrCat(payload_direction);
}
} // namespace connections
} // namespace nearby
} // namespace nearby::connections
#endif // CORE_PAYLOAD_TYPE_H_