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

Implement Auto-reconnect after disconnection [2]

Browse Source 
PiperOrigin-RevId: 689040466
This commit is contained in:
hai007
2024-10-23 11:30:32 -07:00
committed by Copybara-Service
parent ad2e6e3436
commit df6c4fadef
8 changed files with 287 additions and 177 deletions
Download Patch File Download Diff File Expand all files Collapse all files
connections/implementation/BUILD
+1
View File
@@ -461,6 +461,7 @@ cc_test(
":internal_test",
"//connections:core_types",
"//connections/implementation/mediums",
"//internal/platform:base",
"//internal/platform:test_util",
"//internal/platform:types",
"//internal/platform/implementation/g3", # build_cleaner: keep
connections/implementation/offline_frames.cc
+1 -2
View File
@@ -503,14 +503,13 @@ ByteArray ForAutoReconnectIntroduction(const std::string& endpoint_id) {
return ToBytes(std::move(frame));
}
ByteArray ForAutoReconnectIntroductionAck(const std::string& endpoint_id) {
ByteArray ForAutoReconnectIntroductionAck() {
OfflineFrame frame;
frame.set_version(OfflineFrame::V1);
auto* v1_frame = frame.mutable_v1();
v1_frame->set_type(V1Frame::AUTO_RECONNECT);
auto* auto_reconnect = v1_frame->mutable_auto_reconnect();
auto_reconnect->set_endpoint_id(endpoint_id);
auto_reconnect->set_event_type(AutoReconnectFrame::CLIENT_INTRODUCTION_ACK);
return ToBytes(std::move(frame));
connections/implementation/offline_frames.h
+1 -1
View File
@@ -104,7 +104,7 @@ ByteArray ForKeepAlive();
ByteArray ForDisconnection(bool request_safe_to_disconnect,
bool ack_safe_to_disconnect);
ByteArray ForAutoReconnectIntroduction(const std::string& endpoint_id);
ByteArray ForAutoReconnectIntroductionAck(const std::string& endpoint_id);
ByteArray ForAutoReconnectIntroductionAck();
UpgradePathInfo::Medium MediumToUpgradePathInfoMedium(Medium medium);
Medium UpgradePathInfoMediumToMedium(UpgradePathInfo::Medium medium);
connections/implementation/offline_frames_test.cc
+1 -2
View File
@@ -610,10 +610,9 @@ TEST(OfflineFramesTest, CanGenerateAutoReconnectIntroductionAck) {
type: AUTO_RECONNECT
auto_reconnect: <
event_type: CLIENT_INTRODUCTION_ACK
endpoint_id: "ABC"
>
>)pb";
ByteArray bytes = ForAutoReconnectIntroductionAck(std::string(kEndpointId));
ByteArray bytes = ForAutoReconnectIntroductionAck();
auto response = FromBytes(bytes);
ASSERT_TRUE(response.ok());
OfflineFrame message = response.result();
connections/implementation/reconnect_manager.cc
+262 -162
View File
@@ -20,6 +20,7 @@
#include <utility>
#include "securegcm/ukey2_handshake.h"
#include "absl/container/flat_hash_set.h"
#include "absl/functional/any_invocable.h"
#include "absl/functional/bind_front.h"
#include "absl/strings/str_cat.h"
@@ -36,6 +37,7 @@
#include "internal/platform/bluetooth_classic.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/cancelable_alarm.h"
#include "internal/platform/cancellation_flag.h"
#include "internal/platform/cancellation_flag_listener.h"
#include "internal/platform/count_down_latch.h"
#include "internal/platform/exception.h"
@@ -56,17 +58,17 @@ ReconnectManager::ReconnectManager(Mediums& mediums,
ReconnectManager::~ReconnectManager() { Shutdown(); }
bool ReconnectManager::AutoReconnect(
ClientProxy* client, const std::string& endpoint_id,
AutoReconnectCallback& callback,
bool send_disconnection_notification,
DisconnectionReason disconnection_reason) {
bool ReconnectManager::AutoReconnect(ClientProxy* client,
const std::string& endpoint_id,
AutoReconnectCallback& callback,
bool send_disconnection_notification,
DisconnectionReason disconnection_reason) {
if (!client->IsAutoReconnectEnabled(endpoint_id)) {
return false;
}
if (resumed_endpoints_.contains(endpoint_id)) {
NEARBY_LOGS(INFO) << TAG << "AutoReconnect is not needed for endpoint_id = "
LOG(INFO) << TAG << "AutoReconnect is not needed for endpoint_id = "
<< endpoint_id
<< ", since it's just reconnected successfully.";
return true;
@@ -74,7 +76,7 @@ bool ReconnectManager::AutoReconnect(
auto endpoint_channel = channel_manager_->GetChannelForEndpoint(endpoint_id);
if (endpoint_channel == nullptr) {
NEARBY_LOGS(INFO)
LOG(INFO)
<< TAG << " endpoint_channel shouldn't be null for endpoint_id = "
<< endpoint_id;
return false;
@@ -83,7 +85,7 @@ bool ReconnectManager::AutoReconnect(
bool is_incoming = client->IsIncomingConnection(endpoint_id);
if (is_incoming == client->IsOutgoingConnection(endpoint_id)) {
NEARBY_LOGS(INFO)
LOG(INFO)
<< TAG << " autoReconnect failed for medium: "
<< location::nearby::proto::connections::Medium_Name(medium)
<< " because there is no existing incoming/outgoing connection, "
@@ -99,19 +101,20 @@ bool ReconnectManager::AutoReconnect(
ReconnectMetadata(is_incoming, std::move(callback),
send_disconnection_notification, disconnection_reason,
reconnect_service_id));
NEARBY_LOGS(INFO) << TAG << "add a new endpoint_id " << endpoint_id
LOG(INFO) << TAG << "add a new endpoint_id " << endpoint_id
<< " into metadata_by_service_id_map.";
if (Start(is_incoming, client, endpoint_id, reconnect_service_id, medium)) {
resumed_endpoints_.emplace(endpoint_id);
auto time_out = FeatureFlags::GetInstance()
.GetFlags()
.auto_reconnect_skip_duplicated_endpoint_duration;
auto time_out =
FeatureFlags::GetInstance()
.GetFlags()
.safe_to_disconnect_reconnect_skip_duplicated_endpoint_duration;
std::make_unique<CancelableAlarm>(
absl::StrCat("RemoveSuccessfulResumedEndpointId for ", endpoint_id),
[this, endpoint_id, time_out]() {
NEARBY_LOGS(INFO)
LOG(INFO)
<< TAG << "Timeout after " << time_out
<< "ms. RemoveSuccessfulResumedEndpointId for " << endpoint_id;
resumed_endpoints_.erase(endpoint_id);
@@ -130,24 +133,25 @@ bool ReconnectManager::Start(bool is_incoming, ClientProxy* client,
const std::string& reconnect_service_id,
Medium medium) {
auto retry_delay_millis =
FeatureFlags::GetInstance().GetFlags().auto_reconnect_retry_delay_millis;
auto reconnect_retry_num =
FeatureFlags::GetInstance().GetFlags().auto_reconnect_retry_attempts;
NEARBY_LOGS(INFO) << TAG << " " << (is_incoming ? "rehost" : "reconnect")
FeatureFlags::GetInstance()
.GetFlags()
.safe_to_disconnect_reconnect_retry_delay_millis;
auto reconnect_retry_num = FeatureFlags::GetInstance()
.GetFlags()
.safe_to_disconnect_reconnect_retry_attempts;
LOG(INFO) << TAG << " " << (is_incoming ? "rehost" : "reconnect")
<< " for medium: "
<< location::nearby::proto::connections::Medium_Name(
medium)
<< location::nearby::proto::connections::Medium_Name(medium)
<< " for endpoint_id " << endpoint_id << " started...";
bool final_result = false;
CountDownLatch latch(1);
reconnect_executor_.Execute(
"reconnect-start",
[this, &final_result, is_incoming, client, endpoint_id,
&reconnect_service_id, retry_delay_millis, reconnect_retry_num, medium,
&latch]() mutable {
"reconnect-start", [this, &final_result, is_incoming, client, endpoint_id,
&reconnect_service_id, retry_delay_millis,
reconnect_retry_num, medium, &latch]() mutable {
for (int i = 0; i < reconnect_retry_num; ++i) {
if (client->GetCancellationFlag(endpoint_id)->Cancelled()) {
NEARBY_LOGS(INFO)
LOG(INFO)
<< TAG << " Stop retry, Endpoint connection is cancelled";
break;
}
@@ -158,11 +162,14 @@ bool ReconnectManager::Start(bool is_incoming, ClientProxy* client,
}
SystemClock::Sleep(retry_delay_millis);
}
NEARBY_LOGS(INFO) << "Reconnect "
LOG(INFO) << "Reconnect "
<< (final_result ? "succeeded" : "failed");
latch.CountDown();
});
latch.Await();
if (!final_result) {
client->GetCancellationFlag(endpoint_id)->Cancel();
}
return final_result;
}
@@ -174,13 +181,13 @@ bool ReconnectManager::RunOnce(bool is_incoming, ClientProxy* client,
switch (medium) {
case Medium::BLUETOOTH: {
BluetoothImpl bluetooth_impl(client, endpoint_id, reconnect_service_id,
is_incoming, medium, mediums_,
channel_manager_, *this);
is_incoming, medium, mediums_,
channel_manager_, *this);
result = bluetooth_impl.Run();
} break;
default:
NEARBY_LOGS(INFO) << "AutoReconnect not implemented yet for "
LOG(INFO) << "AutoReconnect not implemented yet for "
<< location::nearby::proto::connections::Medium_Name(
medium);
@@ -201,13 +208,13 @@ void ReconnectManager::ClearReconnectData(
item.second.disconnection_reason);
}
}
NEARBY_LOGS(INFO) << TAG << "erase endpoint_id " << item.first;
LOG(INFO) << TAG << "erase endpoint_id " << item.first;
endpoint_id_metadata_map_.erase(item.first);
}
}
void ReconnectManager::Shutdown() {
NEARBY_LOGS(INFO) << TAG << "Initiating shutdown of ReconnectManager.";
LOG(INFO) << TAG << "Initiating shutdown of ReconnectManager.";
{
MutexLock lock(&mutex_);
listen_timeout_alarm_by_service_id_.clear();
@@ -220,12 +227,12 @@ void ReconnectManager::Shutdown() {
reconnect_executor_.Shutdown();
encryption_cb_executor_.Shutdown();
incoming_connection_cb_executor_.Shutdown();
NEARBY_LOGS(INFO) << TAG << "ReconnectManager has shut down.";
LOG(INFO) << TAG << "ReconnectManager has shut down.";
}
bool ReconnectManager::BaseMediumImpl::Run() {
if (!IsMediumRadioOn()) {
NEARBY_LOGS(INFO) << TAG
LOG(INFO) << TAG
<< location::nearby::proto::connections::Medium_Name(
medium_)
<< " radio is turned off, try later";
@@ -233,7 +240,7 @@ bool ReconnectManager::BaseMediumImpl::Run() {
}
if (client_->IsConnectedToEndpoint(endpoint_id_)) {
NEARBY_LOGS(INFO) << TAG
LOG(INFO) << TAG
<< "ReconnectBluetooth is not needed since it's already "
"connected to the RemoteDevice: ";
return true;
@@ -241,7 +248,7 @@ bool ReconnectManager::BaseMediumImpl::Run() {
auto previou_channel = channel_manager_->GetChannelForEndpoint(endpoint_id_);
if (previou_channel == nullptr) {
NEARBY_LOGS(INFO)
LOG(INFO)
<< TAG
<< "ReconnectionManager didn't find a previous EndpointChannel "
"for "
@@ -251,48 +258,77 @@ bool ReconnectManager::BaseMediumImpl::Run() {
previou_channel->Close(
DisconnectionReason::PREV_CHANNEL_DISCONNECTION_IN_RECONNECT);
return is_incoming_ ? RehostForIncomingConnections()
return is_incoming_ ? RehostForIncomingConnections(/*is_last_medium*/ true)
: ReconnectToRemoteDevice();
}
bool ReconnectManager::BaseMediumImpl::RehostForIncomingConnections() {
auto time_out =
FeatureFlags::GetInstance().GetFlags().auto_reconnect_timeout_millis;
bool ReconnectManager::BaseMediumImpl::RehostForIncomingConnections(
bool is_last_medium) {
auto time_out = FeatureFlags::GetInstance()
.GetFlags()
.safe_to_disconnect_reconnect_timeout_millis;
auto cancellation_flag = client_->GetCancellationFlag(endpoint_id_);
if (!IsListeningForIncomingConnections()) {
NEARBY_LOGS(INFO) << "Start rehosting for: " << reconnect_service_id_;
if (!StartListeningForIncomingConnections()) {
NEARBY_LOGS(ERROR)
<< TAG
<< "Rehost failed since "
"StartListeningForIncomingConnections return false.";
return false;
}
{
MutexLock lock(&reconnect_manager_.mutex_);
reconnect_manager_
.listen_timeout_alarm_by_service_id_[reconnect_service_id_] =
std::make_unique<CancelableAlarm>(
absl::StrCat("Rehost listen timeout for ", reconnect_service_id_),
[this, time_out]() {
NEARBY_LOGS(INFO)
<< "Timeout after " << time_out
<< "ms. Stop listening for incoming "
"Connections for serviceId "
<< reconnect_service_id_ << " for rehost, initiated by "
<< endpoint_id_
<< ", unregister all still not connected endpointIds.";
StopListeningIfAllConnected(
reconnect_service_id_,
[this]() { StopListeningForIncomingConnections(); },
/* forceStop= */ true);
},
time_out, &reconnect_manager_.alarm_executor_);
}
} else {
NEARBY_LOGS(INFO) << "Rehosting is not needed since it's already "
if (IsListeningForIncomingConnections()) {
LOG(INFO) << "Rehosting is not needed since it's already "
"rehosts for: "
<< reconnect_service_id_;
if (cancellation_flag == nullptr) {
return true;
}
if (cancellation_flag->Cancelled()) {
StopListeningIfAllConnected(
reconnect_service_id_,
[this]() { StopListeningForIncomingConnections(); },
/* forceStop= */ false);
return false;
} else {
auto cancellation_listener =
std::make_unique<nearby::CancellationFlagListener>(
cancellation_flag, [this]() {
LOG(INFO) << "Calling CancellationFlagListener.";
ProcessFailedReconnection(endpoint_id_, [this]() {
StopListeningForIncomingConnections();
});
});
std::make_unique<CancelableAlarm>(
absl::StrCat(TAG, " unregisterOnCancelListener"),
[cancellation_listener = std::move(cancellation_listener)]() mutable {
// clean up the listener after auto reconnect is done.
cancellation_listener.reset();
},
time_out, &reconnect_manager_.alarm_executor_);
}
return true;
}
LOG(INFO) << "Start rehosting for: " << reconnect_service_id_;
if (!StartListeningForIncomingConnections()) {
LOG(ERROR) << TAG
<< "Rehost failed since "
"StartListeningForIncomingConnections return false.";
return false;
}
{
MutexLock lock(&reconnect_manager_.mutex_);
reconnect_manager_
.listen_timeout_alarm_by_service_id_[reconnect_service_id_] =
std::make_unique<CancelableAlarm>(
absl::StrCat("Rehost listen timeout for ", reconnect_service_id_),
[this, time_out, is_last_medium]() {
LOG(INFO)
<< "Timeout after " << time_out
<< "ms. Stop listening for incoming "
"Connections for serviceId "
<< reconnect_service_id_ << " for rehost, initiated by "
<< endpoint_id_
<< ", unregister all still not connected endpointIds.";
StopListeningIfAllConnected(
reconnect_service_id_,
[this]() { StopListeningForIncomingConnections(); },
/* forceStop= */ is_last_medium);
},
time_out, &reconnect_manager_.alarm_executor_);
}
if (cancellation_flag == nullptr) {
@@ -308,7 +344,7 @@ bool ReconnectManager::BaseMediumImpl::RehostForIncomingConnections() {
auto cancellation_listener =
std::make_unique<nearby::CancellationFlagListener>(
cancellation_flag, [this]() {
NEARBY_LOGS(INFO) << "Calling CancellationFlagListener.";
LOG(INFO) << "Calling CancellationFlagListener.";
ProcessFailedReconnection(endpoint_id_, [this]() {
StopListeningForIncomingConnections();
});
@@ -326,37 +362,37 @@ bool ReconnectManager::BaseMediumImpl::RehostForIncomingConnections() {
bool ReconnectManager::BaseMediumImpl::ReconnectToRemoteDevice() {
if (!ConnectOverMedium()) {
NEARBY_LOGS(INFO) << TAG << "Connect over medium "
LOG(INFO) << TAG << "Connect over medium "
<< location::nearby::proto::connections::Medium_Name(
medium_)
<< " failed.";
return false;
}
NEARBY_LOGS(INFO) << TAG << "Write CLIENT_INTRODUCTION frame";
LOG(INFO) << TAG << "Write CLIENT_INTRODUCTION frame";
Exception write_exception = reconnect_channel_->Write(
parser::ForAutoReconnectIntroduction(endpoint_id_));
parser::ForAutoReconnectIntroduction(client_->GetLocalEndpointId()));
if (!write_exception.Ok()) {
NEARBY_LOGS(ERROR)
LOG(ERROR)
<< TAG << "Failed to write forAutoReconnectClientIntroductionEvent.";
QuietlyCloseChannelAndSocket();
return false;
}
if (!ReadClientIntroductionAckFrame(reconnect_channel_.get())) {
NEARBY_LOGS(ERROR) << TAG << "Failed to read ClientIntroductionAck frame.";
LOG(ERROR) << TAG << "Failed to read ClientIntroductionAck frame.";
QuietlyCloseChannelAndSocket();
return false;
}
if (ReplaceChannelForEndpoint(client_, endpoint_id_,
std::move(reconnect_channel_),
SupportEncryptionDisabled(), nullptr)) {
NEARBY_LOGS(INFO) << TAG
LOG(INFO) << TAG
<< " successfully rebuild the outgoing connection with "
<< location::nearby::proto::connections::Medium_Name(
medium_)
<< " for the endpointId:" << endpoint_id_;
return true;
}
NEARBY_LOGS(INFO)
LOG(INFO)
<< TAG << " ReplaceChannelForEndpoint for the outgoing connection with "
<< location::nearby::proto::connections::Medium_Name(medium_)
<< " for the endpointId:" << endpoint_id_ << " failed. Please retry";
@@ -365,42 +401,46 @@ bool ReconnectManager::BaseMediumImpl::ReconnectToRemoteDevice() {
void ReconnectManager::BaseMediumImpl::OnIncomingConnection(
const std::string& reconnect_service_id) {
NEARBY_LOGS(INFO) << TAG << "Received reconnection successfully";
LOG(INFO) << TAG << "Received reconnection successfully";
reconnect_manager_.incoming_connection_cb_executor_.Execute(
"OnIncomingConnection", [this]() {
auto incoming_endpoin_id =
auto incoming_endpoint_id =
ReadClientIntroductionFrame(reconnect_channel_.get());
if (incoming_endpoin_id.empty()) {
NEARBY_LOGS(ERROR) << TAG << "read ClientIntroductionFrame failed";
if (incoming_endpoint_id.empty()) {
LOG(ERROR) << TAG << "read ClientIntroductionFrame failed";
QuietlyCloseChannelAndSocket();
return;
}
NEARBY_LOGS(INFO) << TAG << "Write CLIENT_INTRODUCTION_ACK frame";
Exception write_exception = reconnect_channel_->Write(
parser::ForAutoReconnectIntroductionAck(endpoint_id_));
parser::ForAutoReconnectIntroductionAck());
if (!write_exception.Ok()) {
NEARBY_LOGS(ERROR)
LOG(ERROR)
<< TAG
<< "Failed to write forAutoReconnectClientIntroductionAckEvent.";
QuietlyCloseChannelAndSocket();
return;
}
LOG(INFO)
<< TAG << "successfully read ClientIntroductionFrame and write"
" ClientIntroductionAckFrame with"
<< location::nearby::proto::connections::Medium_Name(medium_)
<< " for the incoming endpointId " << incoming_endpoint_id;
if (ReplaceChannelForEndpoint(
client_, endpoint_id_, std::move(reconnect_channel_),
client_, incoming_endpoint_id, std::move(reconnect_channel_),
SupportEncryptionDisabled(),
[this]() { StopListeningForIncomingConnections(); })) {
NEARBY_LOGS(INFO)
LOG(INFO)
<< TAG << " successfully rebuild the incoming connection with "
<< location::nearby::proto::connections::Medium_Name(medium_)
<< " for the endpointId:" << endpoint_id_;
<< " for the endpointId:" << incoming_endpoint_id;
return;
}
QuietlyCloseChannelAndSocket();
NEARBY_LOGS(INFO)
LOG(INFO)
<< TAG
<< " ReplaceChannelForEndpoint for the incoming connection with "
<< location::nearby::proto::connections::Medium_Name(medium_)
<< " for the endpointId:" << endpoint_id_
<< " for the endpointId:" << incoming_endpoint_id
<< " failed. Please retry";
return;
});
@@ -408,7 +448,7 @@ void ReconnectManager::BaseMediumImpl::OnIncomingConnection(
std::string ReconnectManager::BaseMediumImpl::ReadClientIntroductionFrame(
EndpointChannel* endpoint_channel) {
NEARBY_LOGS(INFO) << TAG << "Read CLIENT_INTRODUCTION frame";
LOG(INFO) << TAG << "Read CLIENT_INTRODUCTION frame";
auto timeout = FeatureFlags::GetInstance()
.GetFlags()
@@ -416,7 +456,7 @@ std::string ReconnectManager::BaseMediumImpl::ReadClientIntroductionFrame(
CancelableAlarm timeout_alarm(
"ReconnectManager::ReadClientIntroductionFrame",
[timeout, endpoint_channel]() {
NEARBY_LOGS(ERROR) << "In ReconnectManager, failed to read the "
LOG(ERROR) << "In ReconnectManager, failed to read the "
"ClientIntroductionFrame after "
<< timeout
<< ". Timing out and closing EndpointChannel "
@@ -428,7 +468,7 @@ std::string ReconnectManager::BaseMediumImpl::ReadClientIntroductionFrame(
auto data = endpoint_channel->Read();
timeout_alarm.Cancel();
if (!data.ok()) {
NEARBY_LOGS(ERROR)
LOG(ERROR)
<< "Data read fail when expecting a ClientIntroductionFrame from "
"EndpointChannel "
<< endpoint_channel->GetType();
@@ -436,7 +476,7 @@ std::string ReconnectManager::BaseMediumImpl::ReadClientIntroductionFrame(
}
auto transfer(parser::FromBytes(data.result()));
if (!transfer.ok()) {
NEARBY_LOGS(ERROR) << "Attempted to read a ClientIntroductionFrame from "
LOG(ERROR) << "Attempted to read a ClientIntroductionFrame from "
"EndpointChannel "
<< endpoint_channel->GetType()
<< ", but was unable to obtain any OfflineFrame.";
@@ -444,14 +484,14 @@ std::string ReconnectManager::BaseMediumImpl::ReadClientIntroductionFrame(
}
OfflineFrame frame = transfer.result();
if (!frame.has_v1() || !frame.v1().has_auto_reconnect()) {
NEARBY_LOGS(ERROR) << "In ReadClientIntroductionFrame(), eExpected a "
LOG(ERROR) << "In ReadClientIntroductionFrame(), eExpected a "
"AUTO_RECONNECT v1 OfflineFrame but got a "
<< parser::GetFrameType(frame) << " frame instead.";
return {};
}
if (frame.v1().auto_reconnect().event_type() !=
AutoReconnectFrame::CLIENT_INTRODUCTION) {
NEARBY_LOGS(ERROR) << "In ReadClientIntroductionFrame(), expected a "
LOG(ERROR) << "In ReadClientIntroductionFrame(), expected a "
"CLIENT_INTRODUCTION "
"v1 OfflineFrame but got a AUTO_RECONNECT frame "
"with eventType "
@@ -464,7 +504,7 @@ std::string ReconnectManager::BaseMediumImpl::ReadClientIntroductionFrame(
bool ReconnectManager::BaseMediumImpl::ReadClientIntroductionAckFrame(
EndpointChannel* endpoint_channel) {
NEARBY_LOGS(INFO) << TAG << "Read CLIENT_INTRODUCTION_ACK frame";
LOG(INFO) << TAG << "Read CLIENT_INTRODUCTION_ACK frame";
auto timeout = FeatureFlags::GetInstance()
.GetFlags()
@@ -472,7 +512,7 @@ bool ReconnectManager::BaseMediumImpl::ReadClientIntroductionAckFrame(
CancelableAlarm timeout_alarm(
"ReconnectManager::ReadClientIntroductionAckFrame",
[timeout, endpoint_channel]() {
NEARBY_LOGS(ERROR) << "In ReconnectManager, failed to read the "
LOG(ERROR) << "In ReconnectManager, failed to read the "
"ClientIntroductionAckFrame after "
<< timeout
<< ". Timing out and closing EndpointChannel "
@@ -486,7 +526,7 @@ bool ReconnectManager::BaseMediumImpl::ReadClientIntroductionAckFrame(
if (!data.ok()) return false;
auto transfer(parser::FromBytes(data.result()));
if (!transfer.ok()) {
NEARBY_LOGS(ERROR) << "Attempted to read a ClientIntroductionAckFrame from "
LOG(ERROR) << "Attempted to read a ClientIntroductionAckFrame from "
"EndpointChannel "
<< endpoint_channel->GetType()
<< ", but was unable to obtain any OfflineFrame.";
@@ -494,14 +534,14 @@ bool ReconnectManager::BaseMediumImpl::ReadClientIntroductionAckFrame(
}
OfflineFrame frame = transfer.result();
if (!frame.has_v1() || !frame.v1().has_auto_reconnect()) {
NEARBY_LOGS(ERROR) << "In ReadClientIntroductionAckFrame(), eExpected a "
LOG(ERROR) << "In ReadClientIntroductionAckFrame(), eExpected a "
"AUTO_RECONNECT v1 OfflineFrame but got a "
<< parser::GetFrameType(frame) << " frame instead.";
return false;
}
if (frame.v1().auto_reconnect().event_type() !=
AutoReconnectFrame::CLIENT_INTRODUCTION_ACK) {
NEARBY_LOGS(ERROR) << "In ReadClientIntroductionAckFrame(), expected a "
LOG(ERROR) << "In ReadClientIntroductionAckFrame(), expected a "
"CLIENT_INTRODUCTION_ACK "
"v1 OfflineFrame but got a AUTO_RECONNECT frame "
"with eventType "
@@ -520,7 +560,7 @@ bool ReconnectManager::BaseMediumImpl::ReplaceChannelForEndpoint(
auto& endpoint_id_metadata_map = reconnect_manager_.endpoint_id_metadata_map_;
auto reconnect_metadata = endpoint_id_metadata_map.find(endpoint_id);
if (reconnect_metadata == endpoint_id_metadata_map.end()) {
NEARBY_LOGS(ERROR) << TAG << "ReconnectMetadata is null for endpointId: "
LOG(ERROR) << TAG << "ReconnectMetadata is null for endpointId: "
<< endpoint_id << " ,please retry!";
return false;
}
@@ -529,34 +569,54 @@ bool ReconnectManager::BaseMediumImpl::ReplaceChannelForEndpoint(
reconnect_manager_.new_endpoint_channels_
.emplace(endpoint_id, std::move(new_channel))
.first->second.get();
replace_channel_succeed_ = false;
wait_encryption_to_finish_ = std::make_unique<CountDownLatch>(1);
if (reconnect_metadata->second.is_incoming) {
reconnect_manager_.encryption_runner_.StartServer(
client, endpoint_id, endpoint_channel, GetResultListener());
} else {
reconnect_manager_.encryption_runner_.StartClient(
client, endpoint_id, endpoint_channel, GetResultListener());
}
wait_encryption_to_finish_->Await(
FeatureFlags::GetInstance().GetFlags().auto_reconnect_timeout_millis);
{
MutexLock lock(&mutex_);
replace_channel_succeed_ = false;
wait_encryption_to_finish_ = std::make_unique<CountDownLatch>(1);
if (reconnect_metadata->second.is_incoming) {
reconnect_manager_.encryption_runner_.StartServer(
client, endpoint_id, endpoint_channel, GetResultListener());
} else {
reconnect_manager_.encryption_runner_.StartClient(
client, endpoint_id, endpoint_channel, GetResultListener());
}
auto cancellation_flag = client_->GetCancellationFlag(endpoint_id_);
std::unique_ptr<nearby::CancellationFlagListener> cancellation_listener;
if (cancellation_flag != nullptr) {
cancellation_listener =
std::make_unique<nearby::CancellationFlagListener>(
cancellation_flag, [this]() {
LOG(INFO) << "Calling CancellationFlagListener for "
"stopping wait_encryption_to_finish";
MutexLock lock(&mutex_);
replace_channel_succeed_ = false;
wait_encryption_to_finish_->CountDown();
});
}
NEARBY_LOGS(INFO) << TAG
<< "replace_channel_succeed_: " << replace_channel_succeed_
<< " for endpointId: " << endpoint_id;
wait_encryption_to_finish_->Await(
FeatureFlags::GetInstance()
.GetFlags()
.safe_to_disconnect_reconnect_timeout_millis);
if (replace_channel_succeed_) {
ProcessSuccessfulReconnection(
endpoint_id, [this]() { StopListeningForIncomingConnections(); });
client->GetAnalyticsRecorder().OnConnectionEstablished(
endpoint_id, endpoint_channel->GetMedium(),
client->GetConnectionToken(endpoint_id));
} else {
ProcessFailedReconnection(
endpoint_id, [this]() { StopListeningForIncomingConnections(); });
LOG(INFO) << TAG << "replace_channel_succeed_: "
<< replace_channel_succeed_
<< " for endpointId: " << endpoint_id;
if (replace_channel_succeed_) {
ProcessSuccessfulReconnection(
endpoint_id, [this]() { StopListeningForIncomingConnections(); });
client->GetAnalyticsRecorder().OnConnectionEstablished(
endpoint_id, endpoint_channel->GetMedium(),
client->GetConnectionToken(endpoint_id));
} else {
ProcessFailedReconnection(
endpoint_id, [this]() { StopListeningForIncomingConnections(); });
}
cancellation_listener.reset();
reconnect_manager_.new_endpoint_channels_.erase(endpoint_id);
return replace_channel_succeed_;
}
reconnect_manager_.new_endpoint_channels_.erase(endpoint_id);
return replace_channel_succeed_;
}
EncryptionRunner::ResultListener
@@ -582,7 +642,7 @@ ReconnectManager::BaseMediumImpl::GetResultListener() {
[this](const std::string& endpoint_id, EndpointChannel* channel) {
reconnect_manager_.encryption_cb_executor_.Execute(
"encryption-failure", [this, endpoint_id, channel]() mutable {
NEARBY_LOGS(ERROR)
LOG(ERROR)
<< "Encryption failed for endpoint_id=" << endpoint_id
<< " on medium="
<< location::nearby::proto::connections::Medium_Name(
@@ -600,14 +660,14 @@ void ReconnectManager::BaseMediumImpl::OnEncryptionSuccessRunnable(
const std::string& auth_token, const ByteArray& raw_auth_token) {
auto item = reconnect_manager_.new_endpoint_channels_.find(endpoint_id);
if (item == reconnect_manager_.new_endpoint_channels_.end()) {
NEARBY_LOGS(INFO) << "TAG"
LOG(INFO) << "TAG"
<< "OnEncryptionSuccess failed, new_endpoint_channel is "
"null for Endpoint:"
<< endpoint_id;
return;
}
if (!ukey2) {
NEARBY_LOGS(INFO)
LOG(INFO)
<< "TAG"
<< "OnEncryptionSuccess failed, ukey2 is null for Endpoint:"
<< endpoint_id;
@@ -624,7 +684,7 @@ void ReconnectManager::BaseMediumImpl::OnEncryptionSuccessRunnable(
if (!reconnect_manager_.channel_manager_->EncryptChannelForEndpoint(
endpoint_id, std::move(context))) {
NEARBY_LOGS(INFO) << "TAG"
LOG(INFO) << "TAG"
<< "new_endpoint_channel failed to update "
"EncryptionContext for Endpoint:"
<< endpoint_id;
@@ -633,7 +693,7 @@ void ReconnectManager::BaseMediumImpl::OnEncryptionSuccessRunnable(
auto previous_channel =
reconnect_manager_.channel_manager_->GetChannelForEndpoint(endpoint_id);
if (previous_channel == nullptr) {
NEARBY_LOGS(INFO)
LOG(INFO)
<< "TAG"
<< "ReconnectionManager didn't find a previous EndpointChannel for "
<< endpoint_id
@@ -644,12 +704,15 @@ void ReconnectManager::BaseMediumImpl::OnEncryptionSuccessRunnable(
reconnect_manager_.channel_manager_->ReplaceChannelForEndpoint(
client_, endpoint_id, std::move(item->second),
SupportEncryptionDisabled());
replace_channel_succeed_ = true;
{
MutexLock lock(&mutex_);
replace_channel_succeed_ = true;
}
}
void ReconnectManager::BaseMediumImpl::OnEncryptionFailureRunnable(
const std::string& endpoint_id, EndpointChannel* endpoint_channel) {
NEARBY_LOGS(INFO)
LOG(INFO)
<< "TAG"
<< "new_endpoint_channel failed to use encryption for Endpoint:"
<< endpoint_id;
@@ -661,7 +724,7 @@ void ReconnectManager::BaseMediumImpl::ProcessSuccessfulReconnection(
auto& endpoint_id_metadata_map = reconnect_manager_.endpoint_id_metadata_map_;
auto reconnect_metadata = endpoint_id_metadata_map.find(endpoint_id);
if (reconnect_metadata == endpoint_id_metadata_map.end()) {
NEARBY_LOGS(ERROR) << TAG
LOG(ERROR) << TAG
<< "when ProcessSuccessfulReconnection, endpoint_id: "
<< endpoint_id
<< " is already removed fromendpoint_id_metadata_map.";
@@ -674,14 +737,13 @@ void ReconnectManager::BaseMediumImpl::ProcessSuccessfulReconnection(
if (callback.on_reconnect_success_cb) {
callback.on_reconnect_success_cb(client_, endpoint_id);
} else {
NEARBY_LOGS(ERROR) << TAG
LOG(ERROR) << TAG
<< "when ProcessSuccessfulReconnection, endpoint_id: "
<< endpoint_id
<< " callback.on_reconnect_success_cb is null";
}
if (medatdata.is_incoming &&
stop_listening_incoming_connection) {
if (medatdata.is_incoming && stop_listening_incoming_connection) {
StopListeningIfAllConnected(medatdata.reconnect_service_id,
std::move(stop_listening_incoming_connection),
/* forceStop= */ false);
@@ -689,19 +751,51 @@ void ReconnectManager::BaseMediumImpl::ProcessSuccessfulReconnection(
}
void ReconnectManager::BaseMediumImpl::ProcessFailedReconnection(
const std::string& endpoint_id,
absl::AnyInvocable<void(void)> stop_listening_incoming_connection) {}
absl::AnyInvocable<void(void)> stop_listening_incoming_connection) {
auto& endpoint_id_metadata_map = reconnect_manager_.endpoint_id_metadata_map_;
auto reconnect_metadata = endpoint_id_metadata_map.find(endpoint_id);
if (reconnect_metadata == endpoint_id_metadata_map.end()) {
LOG(ERROR) << TAG << "when ProcessFailedReconnection, endpoint_id: "
<< endpoint_id
<< " is already removed fromendpoint_id_metadata_map.";
return;
}
auto medatdata = std::move(reconnect_metadata->second);
auto& callback = medatdata.reconnect_cb;
if (medatdata.is_incoming) {
endpoint_id_metadata_map.erase(reconnect_metadata);
if (callback.on_reconnect_failure_cb) {
callback.on_reconnect_failure_cb(
client_, endpoint_id, medatdata.send_disconnection_notification,
medatdata.disconnection_reason);
} else {
LOG(ERROR) << TAG
<< "when ProcessFailedReconnection, endpoint_id: "
<< endpoint_id
<< " callback.on_reconnect_success_cb is null";
}
if (stop_listening_incoming_connection) {
StopListeningIfAllConnected(medatdata.reconnect_service_id,
std::move(stop_listening_incoming_connection),
/* forceStop= */ false);
}
}
}
void ReconnectManager::BaseMediumImpl::StopListeningIfAllConnected(
const std::string& reconnect_service_id,
absl::AnyInvocable<void(void)> stop_listening_incoming_connection,
bool force_stop) {
const std::string& reconnect_service_id,
absl::AnyInvocable<void(void)> stop_listening_incoming_connection,
bool force_stop) {
if (!force_stop && HasPendingIncomingConnections(reconnect_service_id)) {
return;
}
CancelClearHostTimeoutAlarm(reconnect_service_id);
stop_listening_incoming_connection();
ClearReconnectData(reconnect_service_id, /* is_incoming= */ true);
NEARBY_LOGS(INFO) << TAG
LOG(INFO) << TAG
<< " No more pending incoming connections, "
"stop_listening_incoming_connection for "
<< reconnect_service_id << " before timeout.";
@@ -718,8 +812,8 @@ bool ReconnectManager::BaseMediumImpl::HasPendingIncomingConnections(
return false;
}
void ReconnectManager::BaseMediumImpl::
CancelClearHostTimeoutAlarm(const std::string& service_id) {
void ReconnectManager::BaseMediumImpl::CancelClearHostTimeoutAlarm(
const std::string& service_id) {
MutexLock lock(&reconnect_manager_.mutex_);
auto item =
reconnect_manager_.listen_timeout_alarm_by_service_id_.find(service_id);
@@ -732,21 +826,27 @@ void ReconnectManager::BaseMediumImpl::
}
reconnect_manager_.listen_timeout_alarm_by_service_id_.erase(item);
}
void ReconnectManager::BaseMediumImpl::
ClearReconnectData(const std::string& service_id, bool is_incoming) {
void ReconnectManager::BaseMediumImpl::ClearReconnectData(
const std::string& service_id, bool is_incoming) {
absl::flat_hash_set<std::string> endpoint_ids_pending_removal;
auto& metadata_map = reconnect_manager_.endpoint_id_metadata_map_;
for (auto item = metadata_map.begin(); item != metadata_map.end(); ) {
for (auto item = metadata_map.begin(); item != metadata_map.end();) {
if (item->second.reconnect_service_id == service_id && is_incoming) {
auto& callback = item->second.reconnect_cb.on_reconnect_failure_cb;
if (callback)
callback(client_, item->first,
item->second.send_disconnection_notification,
item->second.disconnection_reason);
metadata_map.erase(item);
endpoint_ids_pending_removal.insert(item->first);
} else {
++item;
}
}
for (const auto& endpoint_id : endpoint_ids_pending_removal) {
metadata_map.erase(endpoint_id);
}
}
bool ReconnectManager::BluetoothImpl::IsMediumRadioOn() const {
@@ -760,11 +860,11 @@ bool ReconnectManager::BluetoothImpl::IsListeningForIncomingConnections()
bool ReconnectManager::BluetoothImpl::StartListeningForIncomingConnections() {
if (!bluetooth_medium_.StartAcceptingConnections(
reconnect_service_id_,
absl::bind_front(
&ReconnectManager::BluetoothImpl::OnIncomingBluetoothConnection, this,
client_))) {
NEARBY_LOGS(ERROR)
reconnect_service_id_,
absl::bind_front(
&ReconnectManager::BluetoothImpl::OnIncomingBluetoothConnection,
this, client_))) {
LOG(ERROR)
<< "ReconnectManager::BluetoothImpl couldn't initiate the "
"BLUETOOTH reconnect for endpoint "
<< endpoint_id_
@@ -772,7 +872,7 @@ bool ReconnectManager::BluetoothImpl::StartListeningForIncomingConnections() {
"incoming Bluetooth connections.";
return false;
}
NEARBY_LOGS(INFO) << "ReconnectManager::BluetoothImpl successfully started "
LOG(INFO) << "ReconnectManager::BluetoothImpl successfully started "
"listening for incoming "
"reconnection on service_id="
<< reconnect_service_id_ << " for endpoint "
@@ -782,11 +882,11 @@ bool ReconnectManager::BluetoothImpl::StartListeningForIncomingConnections() {
void ReconnectManager::BluetoothImpl::OnIncomingBluetoothConnection(
ClientProxy* client, const std::string& upgrade_service_id,
BluetoothSocket socket) {
BluetoothSocket socket) {
reconnect_channel_ = std::make_unique<BluetoothEndpointChannel>(
upgrade_service_id, /*channel_name=*/upgrade_service_id, socket);
if (reconnect_channel_ == nullptr) {
NEARBY_LOGS(ERROR) << TAG
LOG(ERROR) << TAG
<< "Create new endpointChannel for incoming socket "
"failed, close the socket";
@@ -795,7 +895,7 @@ void ReconnectManager::BluetoothImpl::OnIncomingBluetoothConnection(
}
bluetooth_socket_ = std::move(socket);
NEARBY_LOGS(INFO)
LOG(INFO)
<< TAG << "Create new endpointChannel successfully for incoming socket.";
OnIncomingConnection(upgrade_service_id);
@@ -809,7 +909,7 @@ bool ReconnectManager::BluetoothImpl::ConnectOverMedium() {
std::optional<std::string> remote_mac_address =
client_->GetBluetoothMacAddress(endpoint_id_);
if (!remote_mac_address.has_value()) {
NEARBY_LOGS(INFO)
LOG(INFO)
<< "ReconnectBluetooth failed since remoteMacAddress is empty";
return false;
}
@@ -817,7 +917,7 @@ bool ReconnectManager::BluetoothImpl::ConnectOverMedium() {
BluetoothDevice remote_bluetooth_device =
bluetooth_medium.GetRemoteDevice(remote_mac_address.value());
if (!remote_bluetooth_device.IsValid()) {
NEARBY_LOGS(INFO)
LOG(INFO)
<< "ReconnectBluetooth failed since remoteBluetoothDevice is null: "
<< remote_mac_address.value();
return false;
@@ -828,7 +928,7 @@ bool ReconnectManager::BluetoothImpl::ConnectOverMedium() {
client_->GetCancellationFlag(endpoint_id_));
if (!bluetooth_socket_.IsValid()) {
NEARBY_LOGS(ERROR) << "Failed to reconnect to Bluetooth device "
LOG(ERROR) << "Failed to reconnect to Bluetooth device "
<< remote_bluetooth_device.GetName()
<< " for endpoint(id=" << endpoint_id_ << ").";
return false;
@@ -838,7 +938,7 @@ bool ReconnectManager::BluetoothImpl::ConnectOverMedium() {
UnWrapInitiatorReconnectServiceId(reconnect_service_id_),
/*channel_name=*/endpoint_id_, bluetooth_socket_);
if (reconnect_channel_ == nullptr) {
NEARBY_LOGS(ERROR) << "ReconnectBluetooth Failed to get the Bluetooth "
LOG(ERROR) << "ReconnectBluetooth Failed to get the Bluetooth "
"channel, please retry ";
bluetooth_socket_.Close();
return false;
connections/implementation/reconnect_manager.h
+3 -2
View File
@@ -137,7 +137,7 @@ class ReconnectManager {
void OnIncomingConnection(const std::string& reconnect_service_id);
private:
bool RehostForIncomingConnections();
bool RehostForIncomingConnections(bool is_last_medium);
bool ReconnectToRemoteDevice();
std::string ReadClientIntroductionFrame(EndpointChannel* endpoint_channel);
@@ -168,8 +168,9 @@ class ReconnectManager {
void CancelClearHostTimeoutAlarm(const std::string& service_id);
void ClearReconnectData(const std::string& service_id, bool is_incoming);
mutable Mutex mutex_;
std::unique_ptr<CountDownLatch> wait_encryption_to_finish_;
bool replace_channel_succeed_;
bool replace_channel_succeed_ ABSL_GUARDED_BY(mutex_)= false;
};
class BluetoothImpl : public BaseMediumImpl {
connections/implementation/reconnect_manager_test.cc
+11 -4
View File
@@ -16,6 +16,7 @@
#include <memory>
#include <string>
#include <tuple>
#include "gtest/gtest.h"
#include "absl/strings/string_view.h"
@@ -26,6 +27,7 @@
#include "connections/implementation/simulation_user.h"
#include "connections/medium_selector.h"
#include "internal/platform/count_down_latch.h"
#include "internal/platform/feature_flags.h"
#include "internal/platform/logging.h"
#include "internal/platform/medium_environment.h"
@@ -63,7 +65,7 @@ class ReconnectSimulatorUser : public SimulationUser {
};
class ReconnectManagerTest
: public ::testing::TestWithParam<BooleanMediumSelector> {
: public ::testing::TestWithParam<std::tuple<BooleanMediumSelector, bool>> {
protected:
bool SetupConnection(ReconnectSimulatorUser& user_a,
ReconnectSimulatorUser& user_b) {
@@ -96,9 +98,13 @@ class ReconnectManagerTest
};
TEST_P(ReconnectManagerTest, AllowReconnect) {
FeatureFlags::Flags feature_flags = {
.enable_cancellation_flag = std::get<1>(GetParam())};
env_.SetFeatureFlags(feature_flags);
env_.Start();
ReconnectSimulatorUser user_a(kDeviceA, GetParam());
ReconnectSimulatorUser user_b(kDeviceB, GetParam());
ReconnectSimulatorUser user_a(kDeviceA, std::get<0>(GetParam()));
ReconnectSimulatorUser user_b(kDeviceB, std::get<0>(GetParam()));
ASSERT_TRUE(SetupConnection(user_a, user_b));
Mediums mediums;
@@ -140,7 +146,8 @@ TEST_P(ReconnectManagerTest, AllowReconnect) {
}
INSTANTIATE_TEST_SUITE_P(ParametrisedReconnectManagerTest, ReconnectManagerTest,
::testing::ValuesIn(kTestCases));
::testing::Combine(::testing::ValuesIn(kTestCases),
::testing::Bool()));
// More test will be added later.
internal/platform/feature_flags.h
+7 -4
View File
@@ -75,11 +75,14 @@ class FeatureFlags {
// auto-resume 5. non-distance-constraint-recovery 6. payload_ack
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);
absl::Duration auto_reconnect_timeout_millis = absl::Milliseconds(30000);
std::int32_t auto_reconnect_retry_attempts = 3;
absl::Duration auto_reconnect_skip_duplicated_endpoint_duration =
absl::Duration safe_to_disconnect_reconnect_retry_delay_millis =
absl::Milliseconds(4000);
absl::Duration safe_to_disconnect_reconnect_timeout_millis =
absl::Milliseconds(15000);
std::int32_t safe_to_disconnect_reconnect_retry_attempts = 3;
absl::Duration
safe_to_disconnect_reconnect_skip_duplicated_endpoint_duration =
absl::Milliseconds(2000);
// Android code won't be able to launch "payload_received_ack" feature for
// in near future, so change "payload_received_ack" version from "2" to "5"
// after auto-reconnect and auto-resume.