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

Change code to follow payload ack send/receive frame format change.

Browse Source 
PiperOrigin-RevId: 601486151
This commit is contained in:
hai007
2024-01-25 10:17:23 -08:00
committed by Copybara-Service
parent e9ae655916
commit f112d9ff71
11 changed files with 181 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
connections/implementation/endpoint_manager.cc
+20 -1
View File
@@ -15,6 +15,7 @@
#include "connections/implementation/endpoint_manager.h"
#include <algorithm>
#include <cstdint>
#include <functional>
#include <memory>
#include <string>
@@ -30,6 +31,7 @@
#include "connections/implementation/payload_manager.h"
#include "connections/implementation/proto/offline_wire_formats.pb.h"
#include "connections/implementation/service_id_constants.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/count_down_latch.h"
#include "internal/platform/exception.h"
#include "internal/platform/feature_flags.h"
@@ -836,7 +838,9 @@ bool EndpointManager::ApplySafeToDisconnect(const std::string& endpoint_id,
NEARBY_LOGS(WARNING) << "[safe-to-disconnect] Wait for "
<< (is_wait_for_ack ? "ack" : "disconnection")
<< ", timeout in " << timeout_millis;
<< " from endpoint: " << endpoint_id
<< " for reason: " << reason << ", timeout in "
<< timeout_millis;
bool state = channel_manager_->CreateNewTimeoutDisconnectedState(
endpoint_id, timeout_millis);
if (!state) return is_safe_disconnection;
@@ -901,6 +905,21 @@ CountDownLatch EndpointManager::NotifyFrameProcessorsOnEndpointDisconnect(
return barrier;
}
std::vector<std::string> EndpointManager::SendPayloadAck(
std::int64_t payload_id,
const std::vector<std::string>& endpoint_ids) {
ByteArray 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);
}
std::vector<std::string> EndpointManager::SendTransferFrameBytes(
const std::vector<std::string>& endpoint_ids, const ByteArray& bytes,
std::int64_t payload_id, std::int64_t offset,
connections/implementation/endpoint_manager.h
+4 -1
View File
@@ -141,7 +141,10 @@ class EndpointManager {
const location::nearby::connections::PayloadTransferFrame::ControlMessage&
control_message,
const std::vector<std::string>& endpoint_ids);
// Receiver sends this frame when all the packets are received. Returns the
// list of endpoints to which sending this frame failed.
std::vector<std::string> SendPayloadAck(
std::int64_t payload_id, const std::vector<std::string>& endpoint_ids);
// Called when we internally want to get rid of the endpoint, without the
// client directly telling us to. For example...
// a) We failed to read from the endpoint in its dedicated reader thread.
connections/implementation/endpoint_manager_test.cc
+27 -7
View File
@@ -15,6 +15,7 @@
#include "connections/implementation/endpoint_manager.h"
#include <atomic>
#include <cstdint>
#include <memory>
#include <string>
#include <utility>
@@ -31,12 +32,15 @@
#include "connections/implementation/endpoint_channel_manager.h"
#include "connections/implementation/flags/nearby_connections_feature_flags.h"
#include "connections/implementation/offline_frames.h"
#include "connections/listeners.h"
#include "connections/status.h"
#include "internal/flags/nearby_flags.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/count_down_latch.h"
#include "internal/platform/exception.h"
// #include "internal/platform/feature_flags.h"
#include "internal/platform/logging.h"
#include "internal/platform/single_thread_executor.h"
#include "internal/test/fake_single_thread_executor.h"
#include "proto/connections_enums.pb.h"
@@ -122,11 +126,24 @@ class MockFrameProcessor : public EndpointManager::FrameProcessor {
class SetSafeToDisconnect {
public:
explicit SetSafeToDisconnect(bool safe_to_disconnect) {
SetSafeToDisconnect(bool safe_to_disconnect, bool auto_reconnect,
bool payload_received_ack,
std::int32_t safe_to_disconnect_version) {
NearbyFlags::GetInstance().OverrideBoolFlagValue(
config_package_nearby::nearby_connections_feature::
kEnableSafeToDisconnect,
safe_to_disconnect);
NearbyFlags::GetInstance().OverrideBoolFlagValue(
config_package_nearby::nearby_connections_feature::kEnableAutoReconnect,
auto_reconnect);
NearbyFlags::GetInstance().OverrideBoolFlagValue(
config_package_nearby::nearby_connections_feature::
kEnablePayloadReceivedAck,
payload_received_ack);
NearbyFlags::GetInstance().OverrideInt64FlagValue(
config_package_nearby::nearby_connections_feature::
kSafeToDisconnectVersion,
safe_to_disconnect_version);
}
};
@@ -155,12 +172,12 @@ class EndpointManagerTest : public ::testing::Test {
EXPECT_CALL(mock_listener_.initiated_cb, Call).Times(1);
em_.RegisterEndpoint(client_.get(), endpoint_id_, info_,
connection_options_, std::move(channel), listener_,
connection_token);
connection_token_);
if (should_close) {
EXPECT_TRUE(done.Await(absl::Milliseconds(1000)).result());
}
}
SetSafeToDisconnect set_safe_to_disconnect_{true};
SetSafeToDisconnect set_safe_to_disconnect_{true, false, true, 5};
std::unique_ptr<ClientProxy> client_ = std::make_unique<ClientProxy>();
ConnectionOptions connection_options_{
.keep_alive_interval_millis = 5000,
@@ -198,7 +215,7 @@ class EndpointManagerTest : public ::testing::Test {
.bandwidth_changed_cb =
mock_listener_.bandwidth_changed_cb.AsStdFunction(),
};
std::string connection_token = "conntokn";
std::string connection_token_ = "conntokn";
absl::Time start_time_{absl::Now()};
};
@@ -293,7 +310,7 @@ TEST_F(EndpointManagerTest, UnregisterFrameProcessorWorks) {
em_.UnregisterEndpoint(client_.get(), endpoint_id_);
}
TEST_F(EndpointManagerTest, SendControlMessageWorks) {
TEST_F(EndpointManagerTest, SendControlMessageAndPayloadAckWorks) {
auto endpoint_channel = std::make_unique<MockEndpointChannel>();
PayloadTransferFrame::PayloadHeader header;
PayloadTransferFrame::ControlMessage control;
@@ -322,9 +339,12 @@ TEST_F(EndpointManagerTest, SendControlMessageWorks) {
.WillRepeatedly(Return(Exception{Exception::kSuccess}));
RegisterEndpoint(std::move(endpoint_channel), false);
auto failed_ids =
auto failed_ids_1 =
em_.SendControlMessage(header, control, std::vector{endpoint_id_});
EXPECT_EQ(failed_ids, std::vector<std::string>{});
EXPECT_EQ(failed_ids_1, std::vector<std::string>{});
auto failed_ids_2 = em_.SendPayloadAck(header.id(),
std::vector<std::string>{endpoint_id_});
EXPECT_EQ(failed_ids_2, std::vector<std::string>{});
NEARBY_LOG(INFO, "Will unregister endpoint now");
em_.UnregisterEndpoint(client_.get(), endpoint_id_);
NEARBY_LOG(INFO, "Will call destructors now");
connections/implementation/offline_frames.cc
+18
View File
@@ -21,6 +21,7 @@
#include <vector>
#include "connections/implementation/flags/nearby_connections_feature_flags.h"
#include "connections/implementation/internal_payload.h"
#include "connections/implementation/offline_frames_validator.h"
#include "connections/implementation/proto/offline_wire_formats.pb.h"
#include "connections/medium_selector.h"
@@ -227,6 +228,23 @@ ByteArray ForControlPayloadTransfer(
return ToBytes(std::move(frame));
}
ByteArray ForPayloadAckPayloadTransfer(std::int64_t payload_id) {
OfflineFrame frame;
frame.set_version(OfflineFrame::V1);
auto* v1_frame = frame.mutable_v1();
v1_frame->set_type(V1Frame::PAYLOAD_TRANSFER);
auto* sub_frame = v1_frame->mutable_payload_transfer();
sub_frame->set_packet_type(PayloadTransferFrame::PAYLOAD_ACK);
PayloadTransferFrame::PayloadHeader header;
header.set_id(payload_id);
header.set_total_size(InternalPayload::kIndeterminateSize);
*sub_frame->mutable_payload_header() = header;
return ToBytes(std::move(frame));
}
ByteArray ForBwuWifiHotspotPathAvailable(const std::string& ssid,
const std::string& password,
std::int32_t port,
connections/implementation/offline_frames.h
+1
View File
@@ -66,6 +66,7 @@ ByteArray ForControlPayloadTransfer(
header,
const location::nearby::connections::PayloadTransferFrame::ControlMessage&
control);
ByteArray ForPayloadAckPayloadTransfer(std::int64_t payload_id);
// Builds Bandwidth Upgrade [BWU] messages.
ByteArray ForBwuIntroduction(const std::string& endpoint_id,
connections/implementation/offline_frames_test.cc
+18 -1
View File
@@ -17,7 +17,6 @@
#include <array>
#include <memory>
#include <string>
#include <utility>
#include <vector>
#include "gmock/gmock.h"
@@ -337,6 +336,24 @@ TEST(OfflineFramesTest, CanGenerateDataPayloadTransfer) {
EXPECT_THAT(message, EqualsProto(kExpected));
}
TEST(OfflineFramesTest, CanGeneratePayloadAckPayloadTransfer) {
constexpr absl::string_view kExpected =
R"pb(
version: V1
v1: <
type: PAYLOAD_TRANSFER
payload_transfer: <
packet_type: PAYLOAD_ACK,
payload_header: < id: 12345 total_size: -1 >
>
>)pb";
ByteArray bytes = ForPayloadAckPayloadTransfer(12345);
auto response = FromBytes(bytes);
ASSERT_TRUE(response.ok());
OfflineFrame message = response.result();
EXPECT_THAT(message, EqualsProto(kExpected));
}
TEST(OfflineFramesTest, CanGenerateBwuWifiHotspotPathAvailable) {
constexpr absl::string_view kExpected =
R"pb(
connections/implementation/offline_frames_validator.cc
+6
View File
@@ -21,6 +21,7 @@
#include "connections/implementation/internal_payload.h"
#include "connections/implementation/offline_frames.h"
#include "connections/implementation/proto/offline_wire_formats.pb.h"
#include "internal/platform/exception.h"
#include "internal/platform/implementation/platform.h"
#include "internal/platform/logging.h"
@@ -126,6 +127,11 @@ Exception EnsureValidPayloadTransferControlFrame(
Exception EnsureValidPayloadTransferFrame(const PayloadTransferFrame& frame) {
if (!frame.has_payload_header()) return {Exception::kInvalidProtocolBuffer};
if (frame.packet_type() == PayloadTransferFrame::PAYLOAD_ACK) {
// Phone side code doesn't set "total_size" for "payload_header", so skip
// checking it.
return {Exception::kSuccess};
}
if (!frame.payload_header().has_total_size() ||
(frame.payload_header().total_size() < 0 &&
frame.payload_header().total_size() !=
connections/implementation/payload_manager.cc
+75 -27
View File
@@ -15,6 +15,7 @@
#include "connections/implementation/payload_manager.h"
#include <algorithm>
#include <cstddef>
#include <cstdint>
#include <limits>
#include <memory>
@@ -32,8 +33,11 @@
#include "connections/implementation/endpoint_channel_manager.h"
#include "connections/implementation/flags/nearby_connections_feature_flags.h"
#include "connections/implementation/internal_payload_factory.h"
#include "connections/implementation/proto/offline_wire_formats.pb.h"
#include "connections/listeners.h"
#include "connections/payload_type.h"
#include "internal/flags/nearby_flags.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/count_down_latch.h"
#include "internal/platform/exception.h"
#include "internal/platform/feature_flags.h"
@@ -59,7 +63,7 @@ constexpr absl::Duration PayloadManager::kWaitCloseTimeout;
bool PayloadManager::SendPayloadLoop(
ClientProxy* client, PendingPayload& pending_payload,
PayloadTransferFrame::PayloadHeader& payload_header,
std::int64_t& next_chunk_offset, size_t resume_offset) {
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 =
@@ -151,7 +155,7 @@ bool PayloadManager::SendPayloadLoop(
// In other cases, the offset should only be used in both side logs when error
// happened.
PayloadTransferFrame::PayloadChunk payload_chunk(CreatePayloadChunk(
next_chunk_offset - resume_offset, std::move(next_chunk)));
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);
// Check whether at least one endpoint failed.
@@ -349,6 +353,7 @@ PayloadManager::~PayloadManager() {
bytes_payload_executor_.Shutdown();
stream_payload_executor_.Shutdown();
file_payload_executor_.Shutdown();
send_payload_ack_executor_.Shutdown();
CountDownLatch stop_latch(1);
// Clear our tracked pending payloads.
@@ -459,6 +464,7 @@ void PayloadManager::SendPayload(ClientProxy* client,
bool should_continue = true;
std::int64_t next_chunk_offset = 0;
int index = 0;
ThroughputRecorderContainer::GetInstance()
.GetTPRecorder(payload_id, PayloadDirection::OUTGOING_PAYLOAD)
@@ -466,7 +472,8 @@ void PayloadManager::SendPayload(ClientProxy* client,
while (should_continue && !shutdown_.Get()) {
should_continue =
SendPayloadLoop(client, *pending_payload, payload_header,
next_chunk_offset, resume_offset);
next_chunk_offset, resume_offset, index);
index++;
}
RunOnStatusUpdateThread("destroy-payload",
@@ -525,6 +532,11 @@ void PayloadManager::OnIncomingFrame(
ProcessDataPacket(to_client, from_endpoint_id, frame, current_medium,
packet_meta_data);
break;
case PayloadTransferFrame::PAYLOAD_ACK:
NEARBY_LOGS(INFO) << "[safe-to-disconnect][PAYLOAD_RECEIVED_ACK] sender "
"received payload ack from " << from_endpoint_id;
ProcessPayloadAckPacket(from_endpoint_id, frame);
break;
default:
NEARBY_LOGS(WARNING)
<< "PayloadManager: invalid frame; remote endpoint: self=" << this
@@ -694,7 +706,8 @@ PayloadTransferFrame::PayloadHeader PayloadManager::CreatePayloadHeader(
}
PayloadTransferFrame::PayloadChunk PayloadManager::CreatePayloadChunk(
std::int64_t payload_chunk_offset, ByteArray payload_chunk_body) {
std::int64_t payload_chunk_offset, ByteArray payload_chunk_body,
int index) {
PayloadTransferFrame::PayloadChunk payload_chunk;
payload_chunk.set_offset(payload_chunk_offset);
@@ -705,6 +718,7 @@ PayloadTransferFrame::PayloadChunk PayloadManager::CreatePayloadChunk(
payload_chunk.set_flags(payload_chunk.flags() |
PayloadTransferFrame::PayloadChunk::LAST_CHUNK);
}
payload_chunk.set_index(index);
return payload_chunk;
}
@@ -831,21 +845,24 @@ void PayloadManager::SendControlMessage(
void PayloadManager::SendPayloadReceivedAck(
ClientProxy* client, PendingPayload& pending_payload,
const std::string& endpoint_id,
const PayloadTransferFrame::PayloadHeader& payload_header,
std::int64_t chunk_size, bool is_last_chunk) {
const std::string& endpoint_id, bool is_last_chunk) {
if (!is_last_chunk ||
!IsPayloadReceivedAckEnabled(client, endpoint_id, pending_payload)) {
return;
}
// Send the PAYLOAD_RECEIVED_ACK to the remote endpoint for the sender asap.
NEARBY_LOGS(INFO)
<< "[PAYLOAD_RECEIVED_ACK] isLastChunk, receiver send ack to "
<< endpoint_id;
SendControlMessage(
{endpoint_id}, payload_header, chunk_size,
PayloadTransferFrame::ControlMessage::PAYLOAD_RECEIVED_ACK);
send_payload_ack_executor_.Execute(
"send_payload_ack", [this, &pending_payload, endpoint_id]() {
endpoint_manager_->SendPayloadAck(pending_payload.GetId(),
{endpoint_id});
NEARBY_LOGS(INFO) << "[safe-to-disconnect] Send "
"PAYLOAD_RECEIVED_ACK frame to: "
<< endpoint_id << " done";
});
// Send the PAYLOAD_RECEIVED_ACK to the remote endpoint for the sender asap.
NEARBY_LOGS(INFO) << "[safe-to-disconnect] [PAYLOAD_RECEIVED_ACK] "
"isLastChunk, receiver send payload ack to "
<< endpoint_id;
}
bool PayloadManager::WaitForReceivedAck(
@@ -866,11 +883,17 @@ bool PayloadManager::WaitForReceivedAck(
GetPayload(payload_header.id());
// Make sure we're still tracking this payload and its associated endpoint.
if (!latest_pending_payload) {
NEARBY_LOGS(INFO) << "[safe-to-disconnect] short-circuiting "
"latest_pending_payload is null for "
<< payload_header.id() << ", stop wait ack.";
return false;
}
auto* endpoint_info = latest_pending_payload->GetEndpoint(endpoint_id);
if (endpoint_info == nullptr) {
NEARBY_LOGS(INFO) << "[safe-to-disconnect] short-circuiting "
"endpointInfo is null for "
<< payload_header.id() << ", stop wait ack.";
return false;
}
@@ -880,6 +903,9 @@ bool PayloadManager::WaitForReceivedAck(
payload_chunk_offset,
location::nearby::proto::connections::
PayloadStatus::LOCAL_CANCELLATION);
NEARBY_LOGS(INFO) << "[safe-to-disconnect] short-circuiting local "
"payload cancellation for "
<< payload_header.id() << ", stop wait ack.";
return false;
}
// Remote payload cancellation, etc
@@ -888,6 +914,9 @@ bool PayloadManager::WaitForReceivedAck(
HandleFinishedOutgoingPayload(
client, {endpoint_id}, payload_header, payload_chunk_offset,
EndpointInfoStatusToPayloadStatus(endpoint_info->status.Get()));
NEARBY_LOGS(INFO) << "[safe-to-disconnect] short-circuiting remote "
"payload cancellation for "
<< payload_header.id() << ", stop wait ack.";
return false;
}
{
@@ -902,8 +931,15 @@ bool PayloadManager::WaitForReceivedAck(
.wait_payload_received_ack_millis);
endpoint_info->is_payload_received_ack = false;
if (!wait_exception.Ok()) {
NEARBY_LOGS(INFO)
<< "[safe-to-disconnect][PAYLOAD_RECEIVED_ACK] sender wait for "
"received payload ack from "
<< endpoint_id << " end with exception: " << wait_exception.value;
return false;
}
NEARBY_LOGS(INFO) << "[safe-to-disconnect][PAYLOAD_RECEIVED_ACK] sender "
"received payload ack from "
<< endpoint_id;
return true;
}
}
@@ -1261,8 +1297,7 @@ void PayloadManager::ProcessDataPacket(
bool is_last_chunk = (payload_chunk.flags() &
PayloadTransferFrame::PayloadChunk::LAST_CHUNK) != 0;
SendPayloadReceivedAck(
to_client, *pending_payload, from_endpoint_id, payload_header,
payload_chunk.offset() + payload_body_size, is_last_chunk);
to_client, *pending_payload, from_endpoint_id, is_last_chunk);
HandleSuccessfulIncomingChunk(to_client, from_endpoint_id, payload_header,
payload_chunk.flags(), payload_chunk.offset(),
@@ -1330,17 +1365,6 @@ void PayloadManager::ProcessControlPacket(
control_message);
}
break;
case PayloadTransferFrame::ControlMessage::PAYLOAD_RECEIVED_ACK:
if (!pending_payload->IsIncoming() &&
IsPayloadReceivedAckEnabled(to_client, from_endpoint_id,
*pending_payload)) {
NEARBY_LOGS(INFO) << "[safe-to-disconnect]Sender received "
"PAYLOAD_RECEIVED_ACK frame with id:"
<< pending_payload->GetInternalPayload()->GetId()
<< " from endpoint_id=" << from_endpoint_id;
pending_payload->MarkReceivedAckFromEndpoint(from_endpoint_id);
}
break;
default:
NEARBY_LOGS(INFO) << "Unhandled control message "
<< control_message.event() << " for payload_id="
@@ -1349,6 +1373,30 @@ void PayloadManager::ProcessControlPacket(
}
}
void PayloadManager::ProcessPayloadAckPacket(
const std::string& from_endpoint_id,
PayloadTransferFrame& payload_transfer_frame) {
auto payload_header = payload_transfer_frame.payload_header();
PendingPayloadHandle pending_payload = GetPayload(payload_header.id());
if (!pending_payload) {
NEARBY_LOGS(INFO) << "[safe-to-disconnect][PAYLOAD_RECEIVED_ACK] "
"short-circuiting got payload "
"ack for unknown payload "
<< payload_header.id() << ", ignoring";
return;
}
if (pending_payload->IsIncoming()) {
NEARBY_LOGS(INFO) << "[safe-to-disconnect][PAYLOAD_RECEIVED_ACK] "
"short-circuiting got Payload "
"ack for incoming payload "
<< payload_header.id() << ", ignoring";
}
NEARBY_LOGS(INFO)
<< "[safe-to-disconnect][PAYLOAD_RECEIVED_ACK] sender received payload "
<< payload_header.id() << " ack from " << from_endpoint_id;
pending_payload->MarkReceivedAckFromEndpoint(from_endpoint_id);
}
// @PayloadManagerStatusUpdateThread
void PayloadManager::NotifyClientOfIncomingPayloadProgressInfo(
ClientProxy* client, const std::string& endpoint_id,
connections/implementation/payload_manager.h
+10 -5
View File
@@ -31,6 +31,7 @@
#include "connections/implementation/internal_payload.h"
#include "connections/listeners.h"
#include "connections/payload.h"
#include "connections/payload_type.h"
#include "connections/status.h"
#include "internal/platform/atomic_boolean.h"
#include "internal/platform/atomic_reference.h"
@@ -38,6 +39,7 @@
#include "internal/platform/condition_variable.h"
#include "internal/platform/count_down_latch.h"
#include "internal/platform/mutex.h"
#include "internal/platform/single_thread_executor.h"
namespace nearby {
namespace connections {
@@ -270,7 +272,8 @@ class PayloadManager : public EndpointManager::FrameProcessor {
bool SendPayloadLoop(ClientProxy* client, PendingPayload& pending_payload,
PayloadTransferFrame::PayloadHeader& payload_header,
std::int64_t& next_chunk_offset, size_t resume_offset);
std::int64_t& next_chunk_offset, size_t resume_offset,
int index);
void SendClientCallbacksForFinishedIncomingPayloadRunnable(
ClientProxy* client, const std::string& endpoint_id,
const PayloadTransferFrame::PayloadHeader& payload_header,
@@ -298,7 +301,8 @@ class PayloadManager : public EndpointManager::FrameProcessor {
const std::string& parent_folder, const std::string& file_name);
PayloadTransferFrame::PayloadChunk CreatePayloadChunk(std::int64_t offset,
ByteArray body);
ByteArray body,
int index);
bool IsLastChunk(PayloadTransferFrame::PayloadChunk payload_chunk) {
return ((payload_chunk.flags() &
PayloadTransferFrame::PayloadChunk::LAST_CHUNK) != 0);
@@ -331,9 +335,7 @@ class PayloadManager : public EndpointManager::FrameProcessor {
void SendPayloadReceivedAck(
ClientProxy* client, PendingPayload& pending_payload,
const std::string& endpoint_id,
const PayloadTransferFrame::PayloadHeader& payload_header,
std::int64_t chunk_size, bool is_last_chunk);
const std::string& endpoint_id, bool is_last_chunk);
bool WaitForReceivedAck(
ClientProxy* client, const std::string& endpoint_id,
@@ -377,6 +379,8 @@ class PayloadManager : public EndpointManager::FrameProcessor {
void ProcessControlPacket(ClientProxy* to_client,
const std::string& from_endpoint_id,
PayloadTransferFrame& payload_transfer_frame);
void ProcessPayloadAckPacket(const std::string& from_endpoint_id,
PayloadTransferFrame& payload_transfer_frame);
void NotifyClientOfIncomingPayloadProgressInfo(
ClientProxy* client, const std::string& endpoint_id,
@@ -420,6 +424,7 @@ class PayloadManager : public EndpointManager::FrameProcessor {
SingleThreadExecutor file_payload_executor_;
SingleThreadExecutor stream_payload_executor_;
SingleThreadExecutor payload_status_update_executor_;
SingleThreadExecutor send_payload_ack_executor_;
PendingPayloads pending_payloads_;
EndpointManager* endpoint_manager_;
connections/implementation/simulation_user.h
+1 -1
View File
@@ -81,7 +81,7 @@ class SimulationUser {
SimulationUser(const std::string& device_name,
BooleanMediumSelector allowed = BooleanMediumSelector(),
SetSafeToDisconnect set_safe_to_disconnect =
SetSafeToDisconnect(true, false, true, 2))
SetSafeToDisconnect(true, false, true, 5))
: info_{ByteArray{device_name}},
advertising_options_{
{
internal/platform/feature_flags.h
+1
View File
@@ -62,6 +62,7 @@ class FeatureFlags {
// requested service id before attempting to connect over rfcomm. SDP fails
// on Windows when connecting to FP service id but the rfcomm is successful.
bool skip_service_discovery_before_connecting_to_rfcomm = false;
std::int32_t min_nc_version_supports_safe_to_disconnect = 1;
std::int32_t min_nc_version_supports_auto_reconnect = 3;
absl::Duration auto_reconnect_retry_delay_millis = absl::Milliseconds(5000);