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

Internal change to AWDL APIs

Browse Source 
PiperOrigin-RevId: 755602469
This commit is contained in:
Guogang Li
2025-05-06 18:12:01 -07:00
committed by Copybara-Service
parent b57c23bcc1
commit 62413e4de9
5 changed files with 39 additions and 339 deletions
Download Patch File Download Diff File Expand all files Collapse all files
connections/implementation/mediums/awdl.cc
+21 -210
View File
@@ -56,18 +56,7 @@ Awdl::~Awdl() {
while (!advertising_info_.nsd_service_infos.empty()) {
StopAdvertising(advertising_info_.nsd_service_infos.begin()->first);
}
{
MutexLock lock(&mutex_);
if (is_multiplex_enabled_) {
LOG(INFO) << "Closing multiplex sockets for " << multiplex_sockets_.size()
<< " IPs";
for (auto& [ip_addr, multiplex_socket] : multiplex_sockets_) {
LOG(INFO) << "Closing multiplex sockets for: " << ip_addr;
multiplex_socket->~MultiplexSocket();
}
multiplex_sockets_.clear();
}
}
// All the AcceptLoopRunnable objects in here should already have gotten an
// opportunity to shut themselves down cleanly in the calls to
// StopAcceptingConnections() above.
@@ -274,25 +263,13 @@ ErrorOr<bool> Awdl::StartAcceptingConnections(
server_sockets_.insert({service_id, std::move(server_socket)})
.first->second;
// Register the callback to listen for incoming multiplex virtual socket.
if (is_multiplex_enabled_) {
MultiplexSocket::ListenForIncomingConnection(
service_id, Medium::AWDL,
[&callback](const std::string& listening_service_id,
MediumSocket* virtual_socket) mutable {
if (callback) {
callback(listening_service_id,
*(down_cast<AwdlSocket*>(virtual_socket)));
}
});
}
// Start the accept loop on a dedicated thread - this stays alive and
// listening for new incoming connections until StopAcceptingConnections() is
// invoked.
accept_loops_runner_.Execute(
"wifi-lan-accept", [callback = std::move(callback),
server_socket = std::move(owned_server_socket),
service_id, this]() mutable {
"awdl-accept",
[callback = std::move(callback),
server_socket = std::move(owned_server_socket), service_id]() mutable {
while (true) {
AwdlSocket client_socket = server_socket.Accept();
if (!client_socket.IsValid()) {
@@ -300,55 +277,7 @@ ErrorOr<bool> Awdl::StartAcceptingConnections(
break;
}
LOG(INFO) << "Accepted connection for " << service_id;
bool callback_called = false;
{
MutexLock lock(&mutex_);
if (is_multiplex_enabled_) {
// Observed from the log that when the sender tries to connect to
// the receiver's server socket, the server side will somehow
// receive 3 connection request events(don’t know what’s happening
// in Windows’s lower layer code). The 2nd normally is the real
// one. The other two will result in a failed data receiving in
// Windows platform layer. To avoid creating multiplex
// IncomingSocket, we will check if the first read is successful
// or not. If not, discard it. If yes, save that packet
// content(the first frame length), then create the multiplex
// socket, then feed that content to that multiplex socket.
ExceptionOr<std::int32_t> read_int =
Base64Utils::ReadInt(&client_socket.GetInputStream());
if (!read_int.ok()) {
LOG(WARNING)
<< __func__
<< "Failed to read. Exception:" << read_int.exception()
<< "Discard the connection.";
continue;
}
AwdlSocket client_socket_bak = client_socket;
auto physical_socket_ptr =
std::make_shared<AwdlSocket>(client_socket_bak);
MultiplexSocket* multiplex_socket =
MultiplexSocket::CreateIncomingSocket(
physical_socket_ptr, service_id, read_int.result());
if (multiplex_socket != nullptr &&
multiplex_socket->GetVirtualSocket(service_id)) {
multiplex_sockets_.emplace(server_socket.GetIPAddress(),
multiplex_socket);
MultiplexSocket::StopListeningForIncomingConnection(
service_id, Medium::AWDL);
LOG(INFO) << "Multiplex virtaul socket created for "
<< server_socket.GetIPAddress();
if (callback) {
callback(
service_id,
*(down_cast<AwdlSocket*>(
multiplex_socket->GetVirtualSocket(service_id))));
callback_called = true;
}
}
}
}
if (callback && !callback_called) {
if (callback) {
LOG(INFO) << "Call back triggered for physical socket.";
callback(service_id, std::move(client_socket));
}
@@ -373,10 +302,6 @@ bool Awdl::StopAcceptingConnections(const std::string& service_id) {
<< " because it was never started.";
return false;
}
if (is_multiplex_enabled_) {
MultiplexSocket::StopListeningForIncomingConnection(service_id,
Medium::AWDL);
}
// Closing the AwdlServerSocket will kick off the suicide of the thread
// in accept_loops_thread_pool_ that blocks on AwdlServerSocket.accept().
@@ -437,10 +362,12 @@ ErrorOr<AwdlSocket> Awdl::Connect(const std::string& service_id,
CLIENT_CANCELLATION_CANCEL_LAN_OUTGOING_CONNECTION)};
}
ExceptionOr<AwdlSocket> virtual_socket =
ConnectWithMultiplexSocketLocked(service_id, service_info.GetIPAddress());
if (virtual_socket.ok()) {
return virtual_socket.result();
if (service_info.GetServiceName().empty() ||
service_info.GetServiceType().empty()) {
LOG(INFO) << "Can't create client Awdl socket due to invalid service "
"information.";
return {
Error(OperationResultCode::CONNECTIVITY_WIFI_LAN_INVALID_CREDENTIAL)};
}
socket = medium_.ConnectToService(service_info, cancellation_flag);
@@ -448,15 +375,6 @@ ErrorOr<AwdlSocket> Awdl::Connect(const std::string& service_id,
LOG(INFO) << "Failed to Connect via Awdl [service_id=" << service_id << "]";
return {Error(
OperationResultCode::CONNECTIVITY_LAN_CLIENT_SOCKET_CREATION_FAILURE)};
} else {
ExceptionOr<AwdlSocket> virtual_socket =
CreateOutgoingMultiplexSocketLocked(socket, service_id,
service_info.GetIPAddress());
if (virtual_socket.ok()) {
LOG(INFO) << "Successfully connected via Multiplex Awdl [service_id="
<< service_id << "]";
return virtual_socket.result();
}
}
LOG(INFO) << "Successfully connected via Awdl [service_id=" << service_id
@@ -464,125 +382,18 @@ ErrorOr<AwdlSocket> Awdl::Connect(const std::string& service_id,
return socket;
}
ErrorOr<AwdlSocket> Awdl::Connect(const std::string& service_id,
const std::string& ip_address, int port,
CancellationFlag* cancellation_flag) {
Awdl::AwdlCredential Awdl::GetCredentials(const std::string& service_id) {
MutexLock lock(&mutex_);
// Socket to return. To allow for NRVO to work, it has to be a single object.
AwdlSocket socket;
if (service_id.empty()) {
LOG(INFO) << "Refusing to create client Awdl socket because "
"service_id is empty.";
return {Error(OperationResultCode::NEARBY_LOCAL_CLIENT_STATE_WRONG)};
AwdlCredential credential{};
NsdServiceInfo* service_info = advertising_info_.GetServiceInfo(service_id);
if (service_info == nullptr || !service_info->IsValid() ||
service_info->GetServiceName().empty() ||
service_info->GetServiceType().empty()) {
return credential;
}
if (!IsAvailableLocked()) {
LOG(INFO) << "Can't create client Awdl socket [service_id=" << service_id
<< "]; Awdl isn't available.";
return {Error(OperationResultCode::MEDIUM_UNAVAILABLE_LAN_NOT_AVAILABLE)};
}
if (cancellation_flag->Cancelled()) {
LOG(INFO) << "Can't create client Awdl socket due to cancel.";
return {Error(OperationResultCode::
CLIENT_CANCELLATION_CANCEL_LAN_OUTGOING_CONNECTION)};
}
ExceptionOr<AwdlSocket> virtual_socket =
ConnectWithMultiplexSocketLocked(service_id, ip_address);
if (virtual_socket.ok()) {
return virtual_socket.result();
}
socket = medium_.ConnectToService(ip_address, port, cancellation_flag);
if (!socket.IsValid()) {
LOG(INFO) << "Failed to Connect via Awdl [service_id=" << service_id << "]";
return {Error(
OperationResultCode::CONNECTIVITY_LAN_CLIENT_SOCKET_CREATION_FAILURE)};
} else {
ExceptionOr<AwdlSocket> virtual_socket =
CreateOutgoingMultiplexSocketLocked(socket, service_id, ip_address);
if (virtual_socket.ok()) {
LOG(INFO) << "Successfully connected via Multiplex Awdl [service_id="
<< service_id << "]";
return virtual_socket.result();
}
}
LOG(INFO) << "Successfully connected via Awdl [service_id=" << service_id
<< "]";
return socket;
}
ExceptionOr<AwdlSocket> Awdl::ConnectWithMultiplexSocketLocked(
const std::string& service_id, const std::string& ip_address) {
if (is_multiplex_enabled_) {
LOG(INFO) << "multiplex_sockets_ size:" << multiplex_sockets_.size();
auto it = multiplex_sockets_.find(ip_address);
if (it != multiplex_sockets_.end()) {
MultiplexSocket* multiplex_socket = it->second;
if (multiplex_socket->IsShutdown()) {
LOG(INFO) << "Erase multiplex_socket(already shutdown) for ip_address: "
<< WifiUtils::GetHumanReadableIpAddress(ip_address);
multiplex_socket->~MultiplexSocket();
multiplex_sockets_.erase(it);
return ExceptionOr<AwdlSocket>(Exception::kFailed);
}
if (multiplex_socket->IsEnabled()) {
auto* virtual_socket =
multiplex_socket->EstablishVirtualSocket(service_id);
// Should not happen.
auto* wlan_socket = down_cast<AwdlSocket*>(virtual_socket);
if (wlan_socket == nullptr) {
LOG(INFO) << "Failed to cast to AwdlSocket for " << service_id
<< " with ip_address: "
<< WifiUtils::GetHumanReadableIpAddress(ip_address);
return ExceptionOr<AwdlSocket>(Exception::kFailed);
}
return ExceptionOr<AwdlSocket>(*wlan_socket);
}
}
}
return ExceptionOr<AwdlSocket>(Exception::kFailed);
}
ExceptionOr<AwdlSocket> Awdl::CreateOutgoingMultiplexSocketLocked(
AwdlSocket& socket, const std::string& service_id,
const std::string& ip_address) {
if (is_multiplex_enabled_) {
// Create MultiplexSocket, but set it to be disabled as default. It will be
// enabled if both side support multiplex for WIFI_LAN
auto physical_socket_ptr = std::make_shared<AwdlSocket>(socket);
MultiplexSocket* multiplex_socket =
MultiplexSocket::CreateOutgoingSocket(physical_socket_ptr, service_id);
auto* virtual_socket = multiplex_socket->GetVirtualSocket(service_id);
// Should not happen.
auto* wlan_socket = down_cast<AwdlSocket*>(virtual_socket);
if (wlan_socket == nullptr) {
LOG(INFO) << "Failed to cast to AwdlSocket for " << service_id
<< " with ip_address: "
<< WifiUtils::GetHumanReadableIpAddress(ip_address);
return ExceptionOr<AwdlSocket>(Exception::kFailed);
}
LOG(INFO) << "Multiplex socket created for ip_address: "
<< WifiUtils::GetHumanReadableIpAddress(ip_address);
multiplex_sockets_.emplace(ip_address, multiplex_socket);
return ExceptionOr<AwdlSocket>(*wlan_socket);
}
return ExceptionOr<AwdlSocket>(Exception::kFailed);
}
std::pair<std::string, int> Awdl::GetCredentials(
const std::string& service_id) {
MutexLock lock(&mutex_);
const auto& it = server_sockets_.find(service_id);
if (it == server_sockets_.end()) {
return std::pair<std::string, int>();
}
return std::pair<std::string, int>(it->second.GetIPAddress(),
it->second.GetPort());
credential.service_name = service_info->GetServiceName();
credential.service_type = service_info->GetServiceType();
return credential;
}
std::string Awdl::GenerateServiceType(const std::string& service_id) {
connections/implementation/mediums/awdl.h
+9 -35
View File
@@ -26,13 +26,13 @@
#include "connections/implementation/flags/nearby_connections_feature_flags.h"
#include "connections/implementation/mediums/multiplex/multiplex_socket.h"
#include "internal/flags/nearby_flags.h"
#include "internal/platform/awdl.h"
#include "internal/platform/cancellation_flag.h"
#include "internal/platform/exception.h"
#include "internal/platform/expected.h"
#include "internal/platform/multi_thread_executor.h"
#include "internal/platform/mutex.h"
#include "internal/platform/nsd_service_info.h"
#include "internal/platform/awdl.h"
namespace nearby {
namespace connections {
@@ -45,6 +45,11 @@ class Awdl {
using AcceptedConnectionCallback = absl::AnyInvocable<void(
const std::string& service_id, AwdlSocket socket)>;
struct AwdlCredential {
std::string service_name;
std::string service_type;
};
Awdl() = default;
~Awdl();
@@ -96,23 +101,15 @@ class Awdl {
// Blocks until connection is established, or server-side is terminated.
// Returns socket instance. On success, AwdlSocket.IsValid() return true.
ErrorOr<AwdlSocket> Connect(const std::string& service_id,
const NsdServiceInfo& service_info,
CancellationFlag* cancellation_flag)
ABSL_LOCKS_EXCLUDED(mutex_);
// Establishes connection to Awdl service by ip address and port for
// bandwidth upgradation.
// Returns socket instance. On success, AwdlSocket.IsValid() return true.
ErrorOr<AwdlSocket> Connect(const std::string& service_id,
const std::string& ip_address, int port,
CancellationFlag* cancellation_flag)
const NsdServiceInfo& service_info,
CancellationFlag* cancellation_flag)
ABSL_LOCKS_EXCLUDED(mutex_);
// Gets ip address + port for remote services on the network to identify and
// connect to this service.
//
// Credential is for the currently-hosted Wifi ServerSocket (if any).
std::pair<std::string, int> GetCredentials(const std::string& service_id)
AwdlCredential GetCredentials(const std::string& service_id)
ABSL_LOCKS_EXCLUDED(mutex_);
private:
@@ -156,18 +153,6 @@ class Awdl {
static constexpr int kMaxConcurrentAcceptLoops = 5;
// Establishes connection to Awdl service by ip address through
// MultiplexSocket.
ExceptionOr<AwdlSocket> ConnectWithMultiplexSocketLocked(
const std::string& service_id, const std::string& ip_address)
ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
// Creates a MultiplexSocket for outgoing connection based on connected
// AwdlSocket physical socket for specific service_id and ip address.
ExceptionOr<AwdlSocket> CreateOutgoingMultiplexSocketLocked(
AwdlSocket& socket, const std::string& service_id,
const std::string& ip_address) ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
// Same as IsAvailable(), but must be called with mutex_ held.
bool IsAvailableLocked() const ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
@@ -205,17 +190,6 @@ class Awdl {
// and thus require pointer stability.
absl::flat_hash_map<std::string, AwdlServerSocket> server_sockets_
ABSL_GUARDED_BY(mutex_);
// Whether the multiplex feature is enabled.
bool is_multiplex_enabled_ = NearbyFlags::GetInstance().GetBoolFlag(
config_package_nearby::nearby_connections_feature::kEnableMultiplex) &&
NearbyFlags::GetInstance().GetBoolFlag(
config_package_nearby::nearby_connections_feature::
kEnableMultiplexAwdl);
// A map of IpAddress -> MultiplexSocket.
absl::flat_hash_map<std::string, mediums::multiplex::MultiplexSocket*>
multiplex_sockets_ ABSL_GUARDED_BY(mutex_);
};
} // namespace connections
connections/implementation/mediums/awdl_test.cc
-75
View File
@@ -114,81 +114,6 @@ TEST_P(AwdlTest, CanConnect) {
env_.Stop();
}
TEST_P(AwdlTest, CanConnectWithMultiplex) {
bool is_multiplex_enabled = NearbyFlags::GetInstance().GetBoolFlag(
config_package_nearby::nearby_connections_feature::kEnableMultiplex);
NearbyFlags::GetInstance().OverrideBoolFlagValue(
config_package_nearby::nearby_connections_feature::kEnableMultiplex,
true);
bool is_multiplex_enabled_awdl = NearbyFlags::GetInstance().GetBoolFlag(
config_package_nearby::nearby_connections_feature::kEnableMultiplexAwdl);
NearbyFlags::GetInstance().OverrideBoolFlagValue(
config_package_nearby::nearby_connections_feature::kEnableMultiplexAwdl,
true);
FeatureFlags feature_flags = GetParam();
env_.SetFeatureFlags(feature_flags);
env_.Start();
Awdl awdl_client;
Awdl awdl_server;
std::string service_id(kServiceID);
std::string service_info_name(kServiceInfoName);
std::string endpoint_info_name(kEndpointName);
CountDownLatch discovered_latch(1);
CountDownLatch accept_latch(1);
AwdlSocket socket_for_server;
EXPECT_TRUE(awdl_server.StartAcceptingConnections(
service_id, [&](const std::string& service_id, AwdlSocket socket) {
socket_for_server = std::move(socket);
accept_latch.CountDown();
}));
NsdServiceInfo nsd_service_info;
nsd_service_info.SetServiceName(service_info_name);
nsd_service_info.SetTxtRecord(std::string(kEndpointInfoKey),
endpoint_info_name);
awdl_server.StartAdvertising(service_id, nsd_service_info);
AwdlSocket socket_for_client;
SingleThreadExecutor client_executor;
client_executor.Execute([&]() {
NsdServiceInfo discovered_service_info;
awdl_client.StartDiscovery(
service_id, {
.service_discovered_cb =
[&discovered_latch, &discovered_service_info](
NsdServiceInfo service_info,
const std::string& service_id) {
NEARBY_LOGS(INFO) << "Discovered service_info="
<< &service_info;
discovered_service_info = service_info;
discovered_latch.CountDown();
},
});
discovered_latch.Await(kWaitDuration).result();
ASSERT_TRUE(discovered_service_info.IsValid());
CancellationFlag flag;
ErrorOr<AwdlSocket> socket_for_client_result =
awdl_client.Connect(service_id, discovered_service_info, &flag);
socket_for_client = std::move(socket_for_client_result.value());
Base64Utils::WriteInt(&socket_for_client_result.value().GetOutputStream(),
4);
});
EXPECT_TRUE(accept_latch.Await(kWaitDuration).result());
EXPECT_TRUE(awdl_server.StopAcceptingConnections(service_id));
EXPECT_TRUE(awdl_server.StopAdvertising(service_id));
EXPECT_TRUE(socket_for_server.IsValid());
EXPECT_TRUE(socket_for_client.IsValid());
env_.Stop();
NearbyFlags::GetInstance().OverrideBoolFlagValue(
config_package_nearby::nearby_connections_feature::kEnableMultiplex,
is_multiplex_enabled);
NearbyFlags::GetInstance().OverrideBoolFlagValue(
config_package_nearby::nearby_connections_feature::kEnableMultiplexAwdl,
is_multiplex_enabled_awdl);
}
TEST_P(AwdlTest, CanCancelConnect) {
FeatureFlags feature_flags = GetParam();
env_.SetFeatureFlags(feature_flags);
internal/platform/awdl.cc
+7 -14
View File
@@ -21,12 +21,12 @@
#include "absl/container/flat_hash_map.h"
#include "absl/container/flat_hash_set.h"
#include "internal/platform/cancellation_flag.h"
#include "internal/platform/implementation/awdl.h"
#include "internal/platform/logging.h"
#include "internal/platform/mutex_lock.h"
#include "internal/platform/nsd_service_info.h"
#include "internal/platform/output_stream.h"
#include "internal/platform/socket.h"
#include "internal/platform/implementation/wifi_utils.h"
namespace nearby {
using location::nearby::proto::connections::Medium;
@@ -66,8 +66,8 @@ bool AwdlMedium::StopAdvertising(const NsdServiceInfo& nsd_service_info) {
}
bool AwdlMedium::StartDiscovery(const std::string& service_id,
const std::string& service_type,
DiscoveredServiceCallback callback) {
const std::string& service_type,
DiscoveredServiceCallback callback) {
{
MutexLock lock(&mutex_);
if (service_type_to_callback_map_.contains(service_type)) {
@@ -193,20 +193,13 @@ bool AwdlMedium::StopDiscovery(const std::string& service_type) {
AwdlSocket AwdlMedium::ConnectToService(
const NsdServiceInfo& remote_service_info,
CancellationFlag* cancellation_flag) {
NEARBY_LOGS(INFO) << "AwdlMedium::ConnectToService: remote_service_name="
<< remote_service_info.GetServiceName();
NEARBY_LOGS(INFO) << "AwdlMedium::ConnectToService {service_name="
<< remote_service_info.GetServiceName()
<< ", service_type=" << remote_service_info.GetServiceType()
<< "}";
return AwdlSocket(
impl_->ConnectToService(remote_service_info, cancellation_flag));
}
AwdlSocket AwdlMedium::ConnectToService(
const std::string& ip_address, int port,
CancellationFlag* cancellation_flag) {
NEARBY_LOGS(INFO) << "AwdlMedium::ConnectToService: ip address="
<< WifiUtils::GetHumanReadableIpAddress(ip_address)
<< ", port=" << port;
return AwdlSocket(
impl_->ConnectToService(ip_address, port, cancellation_flag));
}
} // namespace nearby
internal/platform/awdl.h
+2 -5
View File
@@ -19,7 +19,9 @@
#include <string>
#include <utility>
#include "absl/base/thread_annotations.h"
#include "absl/container/flat_hash_map.h"
#include "absl/container/flat_hash_set.h"
#include "absl/functional/any_invocable.h"
#include "absl/types/optional.h"
#include "internal/platform/blocking_queue_stream.h"
@@ -238,11 +240,6 @@ class AwdlMedium {
AwdlSocket ConnectToService(const NsdServiceInfo& remote_service_info,
CancellationFlag* cancellation_flag);
// Returns a new AwdlSocket by ip address and port.
// On Success, AwdlSocket::IsValid()returns true.
AwdlSocket ConnectToService(const std::string& ip_address, int port,
CancellationFlag* cancellation_flag);
// Returns a new AwdlServerSocket.
// On Success, AwdlServerSocket::IsValid() returns true.
AwdlServerSocket ListenForService(int port = 0) {