From 190bb141cc2319c209e55ca486a569b560f539de Mon Sep 17 00:00:00 2001 From: hai007 Date: Tue, 1 Apr 2025 19:09:45 -0700 Subject: [PATCH] Implement AWDL in connection layer (part 1) PiperOrigin-RevId: 742925127 --- Package.swift | 1 + .../flags/nearby_connections_feature_flags.h | 3 + connections/implementation/mediums/BUILD | 3 + connections/implementation/mediums/awdl.cc | 615 ++++++++++++++++++ connections/implementation/mediums/awdl.h | 224 +++++++ .../implementation/mediums/awdl_test.cc | 417 ++++++++++++ connections/implementation/mediums/mediums.cc | 3 + connections/implementation/mediums/mediums.h | 5 + .../mediums/multiplex/multiplex_socket.cc | 2 + internal/platform/implementation/g3/BUILD | 2 + internal/platform/implementation/g3/awdl.cc | 309 +++++++++ internal/platform/implementation/g3/awdl.h | 254 ++++++++ .../platform/implementation/g3/platform.cc | 6 + .../implementation/windows/platform.cc | 5 + internal/platform/medium_environment.cc | 164 +++++ internal/platform/medium_environment.h | 46 ++ 16 files changed, 2059 insertions(+) create mode 100644 connections/implementation/mediums/awdl.cc create mode 100644 connections/implementation/mediums/awdl.h create mode 100644 connections/implementation/mediums/awdl_test.cc create mode 100644 internal/platform/implementation/g3/awdl.cc create mode 100644 internal/platform/implementation/g3/awdl.h diff --git a/Package.swift b/Package.swift index 4a597d5f..bb95974a 100644 --- a/Package.swift +++ b/Package.swift @@ -458,6 +458,7 @@ let package = Package( "connections/implementation/mediums/advertisements/data_element_test.cc", "connections/implementation/mediums/advertisements/dct_advertisement_test.cc", "connections/implementation/mediums/advertisements/advertisement_util_test.cc", + "connections/implementation/mediums/awdl_test.cc", "connections/implementation/mediums/ble_v2_test.cc", "connections/implementation/mediums/ble_v2/bloom_filter_test.cc", "connections/implementation/mediums/ble_v2/ble_l2cap_packet_test.cc", diff --git a/connections/implementation/flags/nearby_connections_feature_flags.h b/connections/implementation/flags/nearby_connections_feature_flags.h index 773500db..79828a65 100644 --- a/connections/implementation/flags/nearby_connections_feature_flags.h +++ b/connections/implementation/flags/nearby_connections_feature_flags.h @@ -89,6 +89,9 @@ constexpr auto kUseStableEndpointId = // When true, disable instant on lost on BLE without extended feature. constexpr auto kDisableInstantOnLostOnBleWithoutExtended = flags::Flag(kConfigPackage, "45687098", true); +// When true, enable multiplexing in NC for AWDL. +constexpr auto kEnableMultiplexAwdl = + flags::Flag(kConfigPackage, "45696647", true); } // namespace nearby_connections_feature } // namespace config_package_nearby diff --git a/connections/implementation/mediums/BUILD b/connections/implementation/mediums/BUILD index bbaa6ad0..b73e5908 100644 --- a/connections/implementation/mediums/BUILD +++ b/connections/implementation/mediums/BUILD @@ -16,6 +16,7 @@ licenses(["notice"]) cc_library( name = "mediums", srcs = [ + "awdl.cc", "ble.cc", "ble_v2.cc", "bluetooth_classic.cc", @@ -28,6 +29,7 @@ cc_library( "wifi_lan.cc", ], hdrs = [ + "awdl.h", "ble.h", "ble_v2.h", "bluetooth_classic.h", @@ -140,6 +142,7 @@ cc_test( name = "core_internal_mediums_test", size = "small", srcs = [ + "awdl_test.cc", "ble_test.cc", "ble_v2_test.cc", "bluetooth_classic_test.cc", diff --git a/connections/implementation/mediums/awdl.cc b/connections/implementation/mediums/awdl.cc new file mode 100644 index 00000000..1a2f4914 --- /dev/null +++ b/connections/implementation/mediums/awdl.cc @@ -0,0 +1,615 @@ +// Copyright 2025 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include "connections/implementation/mediums/awdl.h" + +#include +#include +#include +#include + +#include "absl/strings/str_cat.h" +#include "absl/strings/str_format.h" +#include "connections/implementation/mediums/multiplex/multiplex_socket.h" +#include "connections/implementation/mediums/utils.h" +#include "connections/medium_selector.h" +#include "internal/platform/awdl.h" +#include "internal/platform/base64_utils.h" +#include "internal/platform/byte_array.h" +#include "internal/platform/cancellation_flag.h" +#include "internal/platform/exception.h" +#include "internal/platform/expected.h" +#include "internal/platform/implementation/wifi_utils.h" +#include "internal/platform/logging.h" +#include "internal/platform/mutex_lock.h" +#include "internal/platform/nsd_service_info.h" +#include "internal/platform/socket.h" +#include "internal/platform/types.h" + +namespace nearby { +namespace connections { + +namespace { +using MultiplexSocket = mediums::multiplex::MultiplexSocket; +using location::nearby::proto::connections::OperationResultCode; +} // namespace + +Awdl::~Awdl() { + // Destructor is not taking locks, but methods it is calling are. + while (!discovering_info_.service_ids.empty()) { + StopDiscovery(*discovering_info_.service_ids.begin()); + } + while (!server_sockets_.empty()) { + StopAcceptingConnections(server_sockets_.begin()->first); + } + 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. + accept_loops_runner_.Shutdown(); +} + +bool Awdl::IsAvailable() const { + MutexLock lock(&mutex_); + + return IsAvailableLocked(); +} + +bool Awdl::IsAvailableLocked() const { return medium_.IsValid(); } + +ErrorOr Awdl::StartAdvertising(const std::string& service_id, + NsdServiceInfo& nsd_service_info) { + MutexLock lock(&mutex_); + + if (!IsAvailableLocked()) { + LOG(INFO) << "Can't turn on Awdl advertising. Awdl is not available."; + return {Error(OperationResultCode::MEDIUM_UNAVAILABLE_LAN_NOT_AVAILABLE)}; + } + + if (!nsd_service_info.IsValid()) { + LOG(INFO) + << "Refusing to turn on Awdl advertising. nsd_service_info is not " + "valid."; + return {Error(OperationResultCode::MEDIUM_UNAVAILABLE_NSD_NOT_AVAILABLE)}; + } + + if (IsAdvertisingLocked(service_id)) { + LOG(INFO) << "Failed to Awdl advertise because we're already advertising."; + return {Error(OperationResultCode::CLIENT_WIFI_LAN_DUPLICATE_ADVERTISING)}; + } + + if (!IsAcceptingConnectionsLocked(service_id)) { + LOG(INFO) << "Failed to turn on Awdl advertising with nsd_service_info=" + << &nsd_service_info + << ", service_name=" << nsd_service_info.GetServiceName() + << ", service_id=" << service_id + << ". Should accept connections before advertising."; + return {Error(OperationResultCode:: + CLIENT_DUPLICATE_ACCEPTING_LAN_CONNECTION_REQUEST)}; + } + + nsd_service_info.SetServiceType(GenerateServiceType(service_id)); + const auto& it = server_sockets_.find(service_id); + if (it != server_sockets_.end()) { + nsd_service_info.SetIPAddress(it->second.GetIPAddress()); + nsd_service_info.SetPort(it->second.GetPort()); + } + if (!medium_.StartAdvertising(nsd_service_info)) { + LOG(INFO) << "Failed to turn on Awdl advertising with nsd_service_info=" + << &nsd_service_info + << ", service_name=" << nsd_service_info.GetServiceName() + << ", service_id=" << service_id; + return {Error( + OperationResultCode::CONNECTIVITY_WIFI_LAN_START_ADVERTISING_FAILURE)}; + } + + LOG(INFO) << "Turned on Awdl advertising with nsd_service_info=" + << &nsd_service_info + << ", service_name=" << nsd_service_info.GetServiceName() + << ", service_id=" << service_id; + advertising_info_.Add(service_id, std::move(nsd_service_info)); + return {true}; +} + +bool Awdl::StopAdvertising(const std::string& service_id) { + MutexLock lock(&mutex_); + + if (!IsAdvertisingLocked(service_id)) { + LOG(INFO) << "Can't turn off Awdl advertising; it is already off"; + return false; + } + + LOG(INFO) << "Turned off Awdl advertising with service_id=" << service_id; + bool ret = + medium_.StopAdvertising(*advertising_info_.GetServiceInfo(service_id)); + // Reset our bundle of advertising state to mark that we're no longer + // advertising for specific service_id. + advertising_info_.Remove(service_id); + return ret; +} + +bool Awdl::IsAdvertising(const std::string& service_id) { + MutexLock lock(&mutex_); + + return IsAdvertisingLocked(service_id); +} + +bool Awdl::IsAdvertisingLocked(const std::string& service_id) { + return advertising_info_.Existed(service_id); +} + +ErrorOr Awdl::StartDiscovery(const std::string& service_id, + DiscoveredServiceCallback callback) { + MutexLock lock(&mutex_); + + if (service_id.empty()) { + LOG(INFO) << "Refusing to start Awdl discovering with empty service_id."; + return {Error(OperationResultCode::NEARBY_LOCAL_CLIENT_STATE_WRONG)}; + } + + if (!IsAvailableLocked()) { + LOG(INFO) << "Can't discover Awdl services because Awdl isn't available."; + return {Error( + OperationResultCode::MEDIUM_UNAVAILABLE_WIFI_AWARE_NOT_AVAILABLE)}; + } + + if (IsDiscoveringLocked(service_id)) { + LOG(INFO) << "Refusing to start discovery of Awdl services because another " + "discovery is already in-progress."; + return {Error(OperationResultCode::CLIENT_WIFI_LAN_DUPLICATE_DISCOVERING)}; + } + + std::string service_type = GenerateServiceType(service_id); + bool ret = + medium_.StartDiscovery(service_id, service_type, std::move(callback)); + if (!ret) { + LOG(INFO) << "Failed to start discovery of Awdl services."; + return {Error( + OperationResultCode::CONNECTIVITY_WIFI_LAN_START_DISCOVERY_FAILURE)}; + } + + LOG(INFO) << "Turned on Awdl discovering with service_id=" << service_id; + // Mark the fact that we're currently performing a Awdl discovering. + discovering_info_.Add(service_id); + return {true}; +} + +bool Awdl::StopDiscovery(const std::string& service_id) { + MutexLock lock(&mutex_); + + if (!IsDiscoveringLocked(service_id)) { + LOG(INFO) << "Can't turn off Awdl discovering because we never started " + "discovering."; + return false; + } + + std::string service_type = GenerateServiceType(service_id); + LOG(INFO) << "Turned off Awdl discovering with service_id=" << service_id + << ", service_type=" << service_type; + bool ret = medium_.StopDiscovery(service_type); + discovering_info_.Remove(service_id); + return ret; +} + +bool Awdl::IsDiscovering(const std::string& service_id) { + MutexLock lock(&mutex_); + return IsDiscoveringLocked(service_id); +} + +bool Awdl::IsDiscoveringLocked(const std::string& service_id) { + return discovering_info_.Existed(service_id); +} + +ErrorOr Awdl::StartAcceptingConnections( + const std::string& service_id, AcceptedConnectionCallback callback) { + MutexLock lock(&mutex_); + + if (service_id.empty()) { + LOG(INFO) << "Refusing to start accepting Awdl connections; " + "service_id is empty."; + return {Error(OperationResultCode::NEARBY_LOCAL_CLIENT_STATE_WRONG)}; + } + + if (!IsAvailableLocked()) { + LOG(INFO) << "Can't start accepting Awdl connections [service_id=" + << service_id << "]; Awdl not available."; + return {Error( + OperationResultCode::MEDIUM_UNAVAILABLE_WIFI_AWARE_NOT_AVAILABLE)}; + } + + if (IsAcceptingConnectionsLocked(service_id)) { + LOG(INFO) << "Refusing to start accepting Awdl connections [service=" + << service_id + << "]; Awdl server is already in-progress with the same name."; + return {Error(OperationResultCode:: + CLIENT_DUPLICATE_ACCEPTING_LAN_CONNECTION_REQUEST)}; + } + + auto port_range = medium_.GetDynamicPortRange(); + // Generate an exact port here on server socket; if platform doesn't provide + // range of port then assign 0 to let platform decide it. + int port = 0; + if (port_range.has_value() && + (port_range->first > 0 && port_range->first <= 65535 && + port_range->second > 0 && port_range->second <= 65535 && + port_range->first <= port_range->second)) { + port = GeneratePort(service_id, port_range.value()); + } + AwdlServerSocket server_socket = medium_.ListenForService(port); + if (!server_socket.IsValid()) { + LOG(INFO) << "Failed to start accepting Awdl connections for service_id=" + << service_id; + return {Error(OperationResultCode:: + CLIENT_CANCELLATION_WIFI_LAN_SERVER_SOCKET_CREATION)}; + } + + // Mark the fact that there's an in-progress Awdl server accepting + // connections. + auto owned_server_socket = + 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(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 { + while (true) { + AwdlSocket client_socket = server_socket.Accept(); + if (!client_socket.IsValid()) { + server_socket.Close(); + 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 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(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( + multiplex_socket->GetVirtualSocket(service_id)))); + callback_called = true; + } + } + } + } + if (callback && !callback_called) { + LOG(INFO) << "Call back triggered for physical socket."; + callback(service_id, std::move(client_socket)); + } + } + }); + + return {true}; +} + +bool Awdl::StopAcceptingConnections(const std::string& service_id) { + MutexLock lock(&mutex_); + + if (service_id.empty()) { + LOG(INFO) << "Unable to stop accepting Awdl connections because " + "the service_id is empty."; + return false; + } + + const auto& it = server_sockets_.find(service_id); + if (it == server_sockets_.end()) { + LOG(INFO) << "Can't stop accepting Awdl connections for " << 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(). + // That may take some time to complete, but there's no particular reason to + // wait around for it. + auto item = server_sockets_.extract(it); + + // Store a handle to the AwdlServerSocket, so we can use it after + // removing the entry from server_sockets_; making it scoped + // is a bonus that takes care of deallocation before we leave this method. + AwdlServerSocket& listening_socket = item.mapped(); + + // Regardless of whether or not we fail to close the existing + // AwdlServerSocket, remove it from server_sockets_ so that it + // frees up this service for another round. + + // Finally, close the AwdlServerSocket. + if (!listening_socket.Close().Ok()) { + LOG(INFO) << "Failed to close Awdl server socket for service_id=" + << service_id; + return false; + } + + return true; +} + +bool Awdl::IsAcceptingConnections(const std::string& service_id) { + MutexLock lock(&mutex_); + return IsAcceptingConnectionsLocked(service_id); +} + +bool Awdl::IsAcceptingConnectionsLocked(const std::string& service_id) { + return server_sockets_.find(service_id) != server_sockets_.end(); +} + +ErrorOr Awdl::Connect(const std::string& service_id, + const NsdServiceInfo& service_info, + CancellationFlag* cancellation_flag) { + 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)}; + } + + 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 virtual_socket = + ConnectWithMultiplexSocketLocked(service_id, service_info.GetIPAddress()); + if (virtual_socket.ok()) { + return virtual_socket.result(); + } + + socket = medium_.ConnectToService(service_info, 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 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 + << "]"; + return socket; +} + +ErrorOr Awdl::Connect(const std::string& service_id, + const std::string& ip_address, int port, + CancellationFlag* cancellation_flag) { + 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)}; + } + + 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 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 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 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(Exception::kFailed); + } + if (multiplex_socket->IsEnabled()) { + auto* virtual_socket = + multiplex_socket->EstablishVirtualSocket(service_id); + // Should not happen. + auto* wlan_socket = down_cast(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(Exception::kFailed); + } + return ExceptionOr(*wlan_socket); + } + } + } + return ExceptionOr(Exception::kFailed); +} + +ExceptionOr 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(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(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(Exception::kFailed); + } + LOG(INFO) << "Multiplex socket created for ip_address: " + << WifiUtils::GetHumanReadableIpAddress(ip_address); + multiplex_sockets_.emplace(ip_address, multiplex_socket); + return ExceptionOr(*wlan_socket); + } + return ExceptionOr(Exception::kFailed); +} + +std::pair 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(); + } + return std::pair(it->second.GetIPAddress(), + it->second.GetPort()); +} + +std::string Awdl::GenerateServiceType(const std::string& service_id) { + std::string service_id_hash_string; + + const ByteArray service_id_hash = Utils::Sha256Hash( + service_id, NsdServiceInfo::kTypeFromServiceIdHashLength); + for (auto byte : std::string(service_id_hash)) { + absl::StrAppend(&service_id_hash_string, absl::StrFormat("%02X", byte)); + } + + return absl::StrFormat(NsdServiceInfo::kNsdTypeFormat, + service_id_hash_string); +} + +int Awdl::GeneratePort(const std::string& service_id, + std::pair port_range) { + const std::string service_id_hash = + std::string(Utils::Sha256Hash(service_id, 4)); + + std::uint32_t uint_of_service_id_hash = + service_id_hash[0] << 24 | service_id_hash[1] << 16 | + service_id_hash[2] << 8 | service_id_hash[3]; + + return port_range.first + + (uint_of_service_id_hash % (port_range.second - port_range.first)); +} + +} // namespace connections +} // namespace nearby diff --git a/connections/implementation/mediums/awdl.h b/connections/implementation/mediums/awdl.h new file mode 100644 index 00000000..4ca00f4d --- /dev/null +++ b/connections/implementation/mediums/awdl.h @@ -0,0 +1,224 @@ +// Copyright 2020 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#ifndef CORE_INTERNAL_MEDIUMS_AWDL_H_ +#define CORE_INTERNAL_MEDIUMS_AWDL_H_ + +#include +#include +#include + +#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 "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/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 { + +class Awdl { + public: + using DiscoveredServiceCallback = AwdlMedium::DiscoveredServiceCallback; + + // Callback that is invoked when a new connection is accepted. + using AcceptedConnectionCallback = absl::AnyInvocable; + + Awdl() = default; + ~Awdl(); + + // Returns true, if Awdl communications are supported by a platform. + bool IsAvailable() const ABSL_LOCKS_EXCLUDED(mutex_); + + // Sets custom service info name, endpoint info name in NsdServiceInfo and + // then enables Awdl advertising. + // Returns true, if NsdServiceInfo is successfully set, and false otherwise. + ErrorOr StartAdvertising(const std::string& service_id, + NsdServiceInfo& nsd_service_info) + ABSL_LOCKS_EXCLUDED(mutex_); + + // Disables Awdl advertising. + // Returns false if no successful call StartAdvertising() was previously + // made, otherwise returns true. + bool StopAdvertising(const std::string& service_id) + ABSL_LOCKS_EXCLUDED(mutex_); + + bool IsAdvertising(const std::string& service_id) ABSL_LOCKS_EXCLUDED(mutex_); + + // Enables Awdl discovery. Will report any discoverable services + // through a callback. + // Returns true, if discovery was enabled, false otherwise. + ErrorOr StartDiscovery(const std::string& service_id, + DiscoveredServiceCallback callback) + ABSL_LOCKS_EXCLUDED(mutex_); + + // Disables Awdl discovery. + bool StopDiscovery(const std::string& service_id) ABSL_LOCKS_EXCLUDED(mutex_); + + bool IsDiscovering(const std::string& service_id) ABSL_LOCKS_EXCLUDED(mutex_); + + // Starts a worker thread, creates a Awdl socket, associates it with a + // service id. + ErrorOr StartAcceptingConnections(const std::string& service_id, + AcceptedConnectionCallback callback) + ABSL_LOCKS_EXCLUDED(mutex_); + + // Closes socket corresponding to a service id. + bool StopAcceptingConnections(const std::string& service_id) + ABSL_LOCKS_EXCLUDED(mutex_); + + bool IsAcceptingConnections(const std::string& service_id) + ABSL_LOCKS_EXCLUDED(mutex_); + + // Establishes connection to Awdl service that was might be started on + // another service with StartAcceptingConnections() using the same service_id. + // Blocks until connection is established, or server-side is terminated. + // Returns socket instance. On success, AwdlSocket.IsValid() return true. + ErrorOr 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 Connect(const std::string& service_id, + const std::string& ip_address, int port, + 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 GetCredentials(const std::string& service_id) + ABSL_LOCKS_EXCLUDED(mutex_); + + private: + struct AdvertisingInfo { + bool Empty() const { return nsd_service_infos.empty(); } + void Clear() { nsd_service_infos.clear(); } + void Add(const std::string& service_id, + const NsdServiceInfo& nsd_service_info) { + nsd_service_infos.insert({service_id, nsd_service_info}); + } + void Remove(const std::string& service_id) { + nsd_service_infos.erase(service_id); + } + bool Existed(const std::string& service_id) const { + return nsd_service_infos.contains(service_id); + } + NsdServiceInfo* GetServiceInfo(const std::string& service_id) { + const auto& it = nsd_service_infos.find(service_id); + if (it == nsd_service_infos.end()) { + return nullptr; + } + return &it->second; + } + + absl::flat_hash_map nsd_service_infos; + }; + + struct DiscoveringInfo { + bool Empty() const { return service_ids.empty(); } + void Clear() { service_ids.clear(); } + void Add(const std::string& service_id) { service_ids.insert(service_id); } + void Remove(const std::string& service_id) { + service_ids.erase(service_id); + } + bool Existed(const std::string& service_id) const { + return service_ids.contains(service_id); + } + + absl::flat_hash_set service_ids; + }; + + static constexpr int kMaxConcurrentAcceptLoops = 5; + + // Establishes connection to Awdl service by ip address through + // MultiplexSocket. + ExceptionOr 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 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_); + + // Same as IsAdvertising(), but must be called with mutex_ held. + bool IsAdvertisingLocked(const std::string& service_id) + ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + + // Same as IsDiscovering(), but must be called with mutex_ held. + bool IsDiscoveringLocked(const std::string& service_id) + ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + + // Same as IsAcceptingConnections(), but must be called with mutex_ held. + bool IsAcceptingConnectionsLocked(const std::string& service_id) + ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + + // Generates mDNS type. + std::string GenerateServiceType(const std::string& service_id); + + // Generates port number based on port_range_. + int GeneratePort(const std::string& service_id, + std::pair port_range); + + mutable Mutex mutex_; + AwdlMedium medium_ ABSL_GUARDED_BY(mutex_); + AdvertisingInfo advertising_info_ ABSL_GUARDED_BY(mutex_); + DiscoveringInfo discovering_info_ ABSL_GUARDED_BY(mutex_); + + // A thread pool dedicated to running all the accept loops from + // StartAcceptingConnections(). + MultiThreadExecutor accept_loops_runner_{kMaxConcurrentAcceptLoops}; + + // A map of service_id -> ServerSocket. If map is non-empty, we + // are currently listening for incoming connections. + // AwdlServerSocket instances are used from accept_loops_runner_, + // and thus require pointer stability. + absl::flat_hash_map 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 + multiplex_sockets_ ABSL_GUARDED_BY(mutex_); +}; + +} // namespace connections +} // namespace nearby + +#endif // CORE_INTERNAL_MEDIUMS_AWDL_H_ diff --git a/connections/implementation/mediums/awdl_test.cc b/connections/implementation/mediums/awdl_test.cc new file mode 100644 index 00000000..2b24b516 --- /dev/null +++ b/connections/implementation/mediums/awdl_test.cc @@ -0,0 +1,417 @@ +// Copyright 2020 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include "connections/implementation/mediums/awdl.h" + +#include +#include + +#include "gtest/gtest.h" +#include "absl/strings/string_view.h" +#include "absl/time/time.h" +#include "connections/implementation/flags/nearby_connections_feature_flags.h" +#include "internal/flags/nearby_flags.h" +#include "internal/platform/awdl.h" +#include "internal/platform/base64_utils.h" +#include "internal/platform/cancellation_flag.h" +#include "internal/platform/count_down_latch.h" +#include "internal/platform/expected.h" +#include "internal/platform/feature_flags.h" +#include "internal/platform/logging.h" +#include "internal/platform/medium_environment.h" +#include "internal/platform/nsd_service_info.h" +#include "internal/platform/single_thread_executor.h" + +namespace nearby { +namespace connections { +namespace { + +using FeatureFlags = FeatureFlags::Flags; + +constexpr FeatureFlags kTestCases[] = { + FeatureFlags{ + .enable_cancellation_flag = true, + }, + FeatureFlags{ + .enable_cancellation_flag = false, + }, +}; + +constexpr absl::Duration kWaitDuration = absl::Milliseconds(1000); +constexpr absl::string_view kServiceID{"com.google.location.nearby.apps.test"}; +constexpr absl::string_view kServiceInfoName{"ServiceInfoName"}; +constexpr absl::string_view kEndpointName{"EndpointName"}; +constexpr absl::string_view kEndpointInfoKey{"n"}; + +class AwdlTest : public ::testing::TestWithParam { + protected: + using DiscoveredServiceCallback = AwdlMedium::DiscoveredServiceCallback; + + MediumEnvironment& env_{MediumEnvironment::Instance()}; +}; + +TEST_P(AwdlTest, CanConnect) { + 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); + + 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 socket_for_client_result = + awdl_client.Connect(service_id, discovered_service_info, &flag); + 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_result.has_value()); + EXPECT_TRUE(socket_for_client_result.value().IsValid()); + 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 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); + 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); + + 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(); + }, + }); + EXPECT_TRUE(discovered_latch.Await(kWaitDuration).result()); + ASSERT_TRUE(discovered_service_info.IsValid()); + + CancellationFlag flag(true); + ErrorOr socket_for_client_result = + awdl_client.Connect(service_id, discovered_service_info, &flag); + // If FeatureFlag is disabled, Cancelled is false as no-op. + if (!feature_flags.enable_cancellation_flag) { + 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_result.has_value()); + EXPECT_TRUE(socket_for_client_result.value().IsValid()); + } else { + EXPECT_FALSE(accept_latch.Await(kWaitDuration).result()); + EXPECT_TRUE(awdl_server.StopAcceptingConnections(service_id)); + EXPECT_TRUE(awdl_server.StopAdvertising(service_id)); + EXPECT_FALSE(socket_for_server.IsValid()); + EXPECT_TRUE(socket_for_client_result.has_error()); + } + env_.Stop(); +} + +INSTANTIATE_TEST_SUITE_P(ParametrisedAwdlTest, AwdlTest, + ::testing::ValuesIn(kTestCases)); + +TEST_F(AwdlTest, CanConstructValidObject) { + env_.Start(); + Awdl awdl_a; + Awdl awdl_b; + std::string service_id(kServiceID); + + EXPECT_TRUE(awdl_a.IsAvailable()); + EXPECT_TRUE(awdl_b.IsAvailable()); + env_.Stop(); +} + +TEST_F(AwdlTest, CanStartAdvertising) { + env_.Start(); + Awdl awdl_a; + std::string service_id(kServiceID); + std::string service_info_name(kServiceInfoName); + std::string endpoint_info_name(kEndpointName); + + EXPECT_TRUE(awdl_a.StartAcceptingConnections(service_id, {})); + + NsdServiceInfo nsd_service_info; + nsd_service_info.SetServiceName(service_info_name); + nsd_service_info.SetTxtRecord(std::string(kEndpointInfoKey), + endpoint_info_name); + EXPECT_TRUE(awdl_a.StartAdvertising(service_id, nsd_service_info)); + EXPECT_TRUE(awdl_a.StopAdvertising(service_id)); + env_.Stop(); +} + +TEST_F(AwdlTest, CanStartMultipleAdvertising) { + env_.Start(); + Awdl awdl_a; + std::string service_id_1(kServiceID); + std::string service_id_2("com.google.location.nearby.apps.test_1"); + std::string service_info_name_1(kServiceInfoName); + std::string service_info_name_2("ServiceInfoName_1"); + std::string endpoint_info_name(kEndpointName); + + EXPECT_TRUE(awdl_a.StartAcceptingConnections(service_id_1, {})); + EXPECT_TRUE(awdl_a.StartAcceptingConnections(service_id_2, {})); + + NsdServiceInfo nsd_service_info_1; + nsd_service_info_1.SetServiceName(service_info_name_1); + nsd_service_info_1.SetTxtRecord(std::string(kEndpointInfoKey), + endpoint_info_name); + NsdServiceInfo nsd_service_info_2; + nsd_service_info_2.SetServiceName(service_info_name_2); + nsd_service_info_2.SetTxtRecord(std::string(kEndpointInfoKey), + endpoint_info_name); + EXPECT_TRUE(awdl_a.StartAdvertising(service_id_1, nsd_service_info_1)); + EXPECT_TRUE(awdl_a.StartAdvertising(service_id_2, nsd_service_info_2)); + EXPECT_TRUE(awdl_a.StopAdvertising(service_id_1)); + EXPECT_TRUE(awdl_a.StopAdvertising(service_id_2)); + EXPECT_TRUE(awdl_a.StopAcceptingConnections(service_id_1)); + EXPECT_TRUE(awdl_a.StopAcceptingConnections(service_id_2)); + env_.Stop(); +} + +TEST_F(AwdlTest, CanStartDiscovery) { + env_.Start(); + Awdl awdl_a; + std::string service_id(kServiceID); + + EXPECT_TRUE(awdl_a.StartDiscovery(service_id, DiscoveredServiceCallback{})); + EXPECT_TRUE(awdl_a.StopDiscovery(service_id)); + env_.Stop(); +} + +TEST_F(AwdlTest, CanStartMultipleDiscovery) { + env_.Start(); + Awdl awdl_a; + std::string service_id_1(kServiceID); + std::string service_id_2("com.google.location.nearby.apps.test_1"); + + EXPECT_TRUE(awdl_a.StartDiscovery(service_id_1, DiscoveredServiceCallback{})); + + EXPECT_TRUE(awdl_a.StartDiscovery(service_id_2, DiscoveredServiceCallback{})); + EXPECT_TRUE(awdl_a.StopDiscovery(service_id_1)); + EXPECT_TRUE(awdl_a.StopDiscovery(service_id_2)); + env_.Stop(); +} + +TEST_F(AwdlTest, CanAdvertiseThatOtherMediumDiscover) { + env_.Start(); + Awdl awdl_a; + Awdl awdl_b; + std::string service_id(kServiceID); + std::string service_info_name(kServiceInfoName); + std::string endpoint_info_name(kEndpointName); + CountDownLatch discovered_latch(1); + CountDownLatch lost_latch(1); + + awdl_b.StartDiscovery( + service_id, DiscoveredServiceCallback{ + .service_discovered_cb = + [&discovered_latch](NsdServiceInfo service_info, + const std::string& service_id) { + discovered_latch.CountDown(); + }, + .service_lost_cb = + [&lost_latch](NsdServiceInfo service_info, + const std::string& service_id) { + lost_latch.CountDown(); + }, + }); + + EXPECT_TRUE(awdl_a.StartAcceptingConnections(service_id, {})); + + NsdServiceInfo nsd_service_info; + nsd_service_info.SetServiceName(service_info_name); + nsd_service_info.SetTxtRecord(std::string(kEndpointInfoKey), + endpoint_info_name); + EXPECT_TRUE(awdl_a.StartAdvertising(service_id, nsd_service_info)); + EXPECT_TRUE(discovered_latch.Await(kWaitDuration).result()); + EXPECT_TRUE(awdl_a.StopAdvertising(service_id)); + EXPECT_TRUE(lost_latch.Await(kWaitDuration).result()); + EXPECT_TRUE(awdl_b.StopDiscovery(service_id)); + env_.Stop(); +} + +TEST_F(AwdlTest, CanDiscoverThatOtherMediumAdvertise) { + env_.Start(); + Awdl awdl_a; + Awdl awdl_b; + std::string service_id(kServiceID); + std::string service_info_name(kServiceInfoName); + std::string endpoint_info_name(kEndpointName); + CountDownLatch discovered_latch(1); + CountDownLatch lost_latch(1); + + EXPECT_TRUE(awdl_b.StartAcceptingConnections(service_id, {})); + + NsdServiceInfo nsd_service_info; + nsd_service_info.SetServiceName(service_info_name); + nsd_service_info.SetTxtRecord(std::string(kEndpointInfoKey), + endpoint_info_name); + awdl_b.StartAdvertising(service_id, nsd_service_info); + + EXPECT_TRUE(awdl_a.StartDiscovery( + service_id, DiscoveredServiceCallback{ + .service_discovered_cb = + [&discovered_latch](NsdServiceInfo service_info, + const std::string& service_id) { + discovered_latch.CountDown(); + }, + .service_lost_cb = + [&lost_latch](NsdServiceInfo service_info, + const std::string& service_id) { + lost_latch.CountDown(); + }, + })); + EXPECT_TRUE(discovered_latch.Await(kWaitDuration).result()); + EXPECT_TRUE(awdl_b.StopAdvertising(service_id)); + EXPECT_TRUE(lost_latch.Await(kWaitDuration).result()); + EXPECT_TRUE(awdl_a.StopDiscovery(service_id)); + env_.Stop(); +} + +} // namespace +} // namespace connections +} // namespace nearby diff --git a/connections/implementation/mediums/mediums.cc b/connections/implementation/mediums/mediums.cc index 52cded28..131e451e 100644 --- a/connections/implementation/mediums/mediums.cc +++ b/connections/implementation/mediums/mediums.cc @@ -13,6 +13,7 @@ // limitations under the License. #include "connections/implementation/mediums/mediums.h" +#include "connections/implementation/mediums/awdl.h" namespace nearby { namespace connections { @@ -35,5 +36,7 @@ WifiDirect& Mediums::GetWifiDirect() { return wifi_direct_; } mediums::WebRtc& Mediums::GetWebRtc() { return webrtc_; } +Awdl& Mediums::GetAwdl() { return awdl_; } + } // namespace connections } // namespace nearby diff --git a/connections/implementation/mediums/mediums.h b/connections/implementation/mediums/mediums.h index 9eb1608c..37187c7f 100644 --- a/connections/implementation/mediums/mediums.h +++ b/connections/implementation/mediums/mediums.h @@ -15,6 +15,7 @@ #ifndef CORE_INTERNAL_MEDIUMS_MEDIUMS_H_ #define CORE_INTERNAL_MEDIUMS_MEDIUMS_H_ +#include "connections/implementation/mediums/awdl.h" #include "connections/implementation/mediums/ble.h" #include "connections/implementation/mediums/ble_v2.h" #include "connections/implementation/mediums/bluetooth_classic.h" @@ -65,6 +66,9 @@ class Mediums { // Returns a handle to the WebRtc medium. mediums::WebRtc& GetWebRtc(); + // Returns a handle to the Awdl medium. + Awdl& GetAwdl(); + private: // The order of declaration is critical for both construction and // destruction. @@ -83,6 +87,7 @@ class Mediums { WifiHotspot wifi_hotspot_; WifiDirect wifi_direct_; mediums::WebRtc webrtc_; + Awdl awdl_; }; } // namespace connections diff --git a/connections/implementation/mediums/multiplex/multiplex_socket.cc b/connections/implementation/mediums/multiplex/multiplex_socket.cc index cd9ea191..cc00ec7e 100644 --- a/connections/implementation/mediums/multiplex/multiplex_socket.cc +++ b/connections/implementation/mediums/multiplex/multiplex_socket.cc @@ -124,6 +124,7 @@ MultiplexSocket* MultiplexSocket::CreateIncomingSocket( new (&storage_ble) MultiplexSocket(physical_socket); break; case Medium::WIFI_LAN: + case Medium::AWDL: alignas( MultiplexSocket) static char storage_wlan[sizeof(MultiplexSocket)]; multiplex_incoming_socket = @@ -167,6 +168,7 @@ MultiplexSocket* MultiplexSocket::CreateOutgoingSocket( new (&storage_ble) MultiplexSocket(physical_socket); break; case Medium::WIFI_LAN: + case Medium::AWDL: alignas( MultiplexSocket) static char storage_wlan[sizeof(MultiplexSocket)]; multiplex_outgoing_socket = diff --git a/internal/platform/implementation/g3/BUILD b/internal/platform/implementation/g3/BUILD index c79cf420..e7fb7fc2 100644 --- a/internal/platform/implementation/g3/BUILD +++ b/internal/platform/implementation/g3/BUILD @@ -62,6 +62,7 @@ cc_library( name = "comm", testonly = True, srcs = [ + "awdl.cc", "ble.cc", "ble_v2.cc", "bluetooth_adapter.cc", @@ -72,6 +73,7 @@ cc_library( "wifi_lan.cc", ], hdrs = [ + "awdl.h", "ble.h", "ble_v2.h", "bluetooth_adapter.h", diff --git a/internal/platform/implementation/g3/awdl.cc b/internal/platform/implementation/g3/awdl.cc new file mode 100644 index 00000000..e6df71b3 --- /dev/null +++ b/internal/platform/implementation/g3/awdl.cc @@ -0,0 +1,309 @@ +// Copyright 2020 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include "internal/platform/implementation/g3/awdl.h" + +#include +#include +#include +#include + +#include "absl/log/check.h" +#include "absl/strings/str_cat.h" +#include "absl/strings/str_format.h" +#include "absl/synchronization/mutex.h" +#include "internal/platform/cancellation_flag.h" +#include "internal/platform/cancellation_flag_listener.h" +#include "internal/platform/exception.h" +#include "internal/platform/implementation/awdl.h" +#include "internal/platform/logging.h" +#include "internal/platform/medium_environment.h" +#include "internal/platform/nsd_service_info.h" + +namespace nearby { +namespace g3 { + +std::string AwdlServerSocket::GetName(const std::string& ip_address, + int port) { + std::string dot_delimited_string; + if (!ip_address.empty()) { + for (auto byte : ip_address) { + if (!dot_delimited_string.empty()) + absl::StrAppend(&dot_delimited_string, "."); + absl::StrAppend(&dot_delimited_string, absl::StrFormat("%d", byte)); + } + } + std::string out = absl::StrCat(dot_delimited_string, ":", port); + return out; +} + +std::unique_ptr AwdlServerSocket::Accept() { + absl::MutexLock lock(&mutex_); + while (!closed_ && pending_sockets_.empty()) { + cond_.Wait(&mutex_); + } + // whether or not we were running in the wait loop, return early if closed. + if (closed_) return {}; + auto* remote_socket = + pending_sockets_.extract(pending_sockets_.begin()).value(); + CHECK(remote_socket); + + auto local_socket = std::make_unique(); + local_socket->Connect(*remote_socket); + remote_socket->Connect(*local_socket); + cond_.SignalAll(); + return local_socket; +} + +bool AwdlServerSocket::Connect(AwdlSocket& socket) { + absl::MutexLock lock(&mutex_); + if (closed_) return false; + if (socket.IsConnected()) { + NEARBY_LOGS(ERROR) + << "Failed to connect to Awdl server socket: already connected"; + return true; // already connected. + } + // add client socket to the pending list + pending_sockets_.insert(&socket); + cond_.SignalAll(); + while (!socket.IsConnected()) { + cond_.Wait(&mutex_); + if (closed_) return false; + } + return true; +} + +void AwdlServerSocket::SetCloseNotifier( + absl::AnyInvocable notifier) { + absl::MutexLock lock(&mutex_); + close_notifier_ = std::move(notifier); +} + +AwdlServerSocket::~AwdlServerSocket() { + absl::MutexLock lock(&mutex_); + DoClose(); +} + +Exception AwdlServerSocket::Close() { + absl::MutexLock lock(&mutex_); + return DoClose(); +} + +Exception AwdlServerSocket::DoClose() { + bool should_notify = !closed_; + closed_ = true; + if (should_notify) { + cond_.SignalAll(); + if (close_notifier_) { + auto notifier = std::move(close_notifier_); + mutex_.Unlock(); + // Notifier may contain calls to public API, and may cause deadlock, if + // mutex_ is held during the call. + notifier(); + mutex_.Lock(); + } + } + return {Exception::kSuccess}; +} + +AwdlMedium::AwdlMedium() { + auto& env = MediumEnvironment::Instance(); + env.RegisterAwdlMedium(*this); +} + +AwdlMedium::~AwdlMedium() { + auto& env = MediumEnvironment::Instance(); + env.UnregisterAwdlMedium(*this); +} + +bool AwdlMedium::StartAdvertising(const NsdServiceInfo& nsd_service_info) { + std::string service_type = nsd_service_info.GetServiceType(); + NEARBY_LOGS(INFO) << "G3 Awdl StartAdvertising: nsd_service_info=" + << &nsd_service_info + << ", service_name=" << nsd_service_info.GetServiceName() + << ", service_type=" << service_type; + { + absl::MutexLock lock(&mutex_); + if (advertising_info_.Existed(service_type)) { + NEARBY_LOGS(INFO) + << "G3 Awdl StartAdvertising: Can't start advertising because " + "service_type=" + << service_type << ", has started already."; + return false; + } + } + auto& env = MediumEnvironment::Instance(); + env.UpdateAwdlMediumForAdvertising(*this, nsd_service_info, + /*enabled=*/true); + { + absl::MutexLock lock(&mutex_); + advertising_info_.Add(service_type); + } + return true; +} + +bool AwdlMedium::StopAdvertising(const NsdServiceInfo& nsd_service_info) { + std::string service_type = nsd_service_info.GetServiceType(); + NEARBY_LOGS(INFO) << "G3 Awdl StopAdvertising: nsd_service_info=" + << &nsd_service_info + << ", service_name=" << nsd_service_info.GetServiceName() + << ", service_type=" << service_type; + { + absl::MutexLock lock(&mutex_); + if (!advertising_info_.Existed(service_type)) { + NEARBY_LOGS(INFO) + << "G3 Awdl StopAdvertising: Can't stop advertising because " + "we never started advertising for service_type=" + << service_type; + return false; + } + advertising_info_.Remove(service_type); + } + auto& env = MediumEnvironment::Instance(); + env.UpdateAwdlMediumForAdvertising(*this, nsd_service_info, + /*enabled=*/false); + return true; +} + +bool AwdlMedium::StartDiscovery(const std::string& service_type, + DiscoveredServiceCallback callback) { + NEARBY_LOGS(INFO) << "G3 Awdl StartDiscovery: service_type=" + << service_type; + { + absl::MutexLock lock(&mutex_); + if (discovering_info_.Existed(service_type)) { + NEARBY_LOGS(INFO) + << "G3 Awdl StartDiscovery: Can't start discovery because " + "service_type=" + << service_type << " has started already."; + return false; + } + } + auto& env = MediumEnvironment::Instance(); + env.UpdateAwdlMediumForDiscovery(*this, std::move(callback), service_type, + true); + { + absl::MutexLock lock(&mutex_); + discovering_info_.Add(service_type); + } + return true; +} + +bool AwdlMedium::StopDiscovery(const std::string& service_type) { + NEARBY_LOGS(INFO) << "G3 Awdl StopDiscovery: service_type=" + << service_type; + { + absl::MutexLock lock(&mutex_); + if (!discovering_info_.Existed(service_type)) { + NEARBY_LOGS(INFO) + << "G3 Awdl StopDiscovery: Can't stop discovering because we " + "never started discovering."; + return false; + } + discovering_info_.Remove(service_type); + } + auto& env = MediumEnvironment::Instance(); + env.UpdateAwdlMediumForDiscovery(*this, {}, service_type, false); + return true; +} + +std::unique_ptr AwdlMedium::ConnectToService( + const NsdServiceInfo& remote_service_info, + CancellationFlag* cancellation_flag) { + std::string service_type = remote_service_info.GetServiceType(); + NEARBY_LOGS(INFO) << "G3 Awdl ConnectToService [self]: medium=" << this + << ", service_type=" << service_type; + return ConnectToService(remote_service_info.GetIPAddress(), + remote_service_info.GetPort(), cancellation_flag); +} + +std::unique_ptr AwdlMedium::ConnectToService( + const std::string& ip_address, int port, + CancellationFlag* cancellation_flag) { + std::string socket_name = AwdlServerSocket::GetName(ip_address, port); + NEARBY_LOGS(INFO) << "G3 Awdl ConnectToService [self]: medium=" << this + << ", ip address + port=" << socket_name; + // First, find an instance of remote medium, that exposed this service. + auto& env = MediumEnvironment::Instance(); + auto* remote_medium = + static_cast(env.GetAwdlMedium(ip_address, port)); + if (!remote_medium) { + return {}; + } + + AwdlServerSocket* server_socket = nullptr; + NEARBY_LOGS(INFO) << "G3 Awdl ConnectToService [peer]: medium=" + << remote_medium + << ", remote ip address + port=" << socket_name; + // Then, find our server socket context in this medium. + { + absl::MutexLock medium_lock(&remote_medium->mutex_); + auto item = remote_medium->server_sockets_.find(socket_name); + server_socket = + item != remote_medium->server_sockets_.end() ? item->second : nullptr; + if (server_socket == nullptr) { + NEARBY_LOGS(ERROR) + << "G3 Awdl Failed to find Awdl Server socket: socket_name=" + << socket_name; + return {}; + } + } + + if (cancellation_flag->Cancelled()) { + NEARBY_LOGS(ERROR) << "G3 Awdl Connect: Has been cancelled: socket_name=" + << socket_name; + return {}; + } + + CancellationFlagListener listener(cancellation_flag, [&server_socket]() { + NEARBY_LOGS(INFO) << "G3 Awdl Cancel Connect."; + if (server_socket != nullptr) { + server_socket->Close(); + } + }); + + auto socket = std::make_unique(); + // Finally, Request to connect to this socket. + if (!server_socket->Connect(*socket)) { + NEARBY_LOGS(ERROR) << "G3 Awdl Failed to connect to existing Awdl " + "Server socket: name=" + << socket_name; + return {}; + } + NEARBY_LOGS(INFO) << "G3 Awdl ConnectToService: connected: socket=" + << socket.get(); + return socket; +} + +std::unique_ptr AwdlMedium::ListenForService( + int port) { + auto& env = MediumEnvironment::Instance(); + auto server_socket = std::make_unique(); + server_socket->SetIPAddress(env.GetFakeIPAddress()); + server_socket->SetPort(port == 0 ? env.GetFakePort() : port); + std::string socket_name = AwdlServerSocket::GetName( + server_socket->GetIPAddress(), server_socket->GetPort()); + server_socket->SetCloseNotifier([this, socket_name]() { + absl::MutexLock lock(&mutex_); + server_sockets_.erase(socket_name); + }); + NEARBY_LOGS(INFO) << "G3 Awdl Adding server socket: medium=" << this + << ", socket_name=" << socket_name; + absl::MutexLock lock(&mutex_); + server_sockets_.insert({socket_name, server_socket.get()}); + return server_socket; +} + +} // namespace g3 +} // namespace nearby diff --git a/internal/platform/implementation/g3/awdl.h b/internal/platform/implementation/g3/awdl.h new file mode 100644 index 00000000..b86dacc9 --- /dev/null +++ b/internal/platform/implementation/g3/awdl.h @@ -0,0 +1,254 @@ +// Copyright 2020 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#ifndef PLATFORM_IMPL_G3_AWDL_H_ +#define PLATFORM_IMPL_G3_AWDL_H_ + +#include +#include +#include + +#include "absl/container/flat_hash_map.h" +#include "absl/container/flat_hash_set.h" +#include "absl/synchronization/mutex.h" +#include "internal/platform/byte_array.h" +#include "internal/platform/implementation/g3/multi_thread_executor.h" +#include "internal/platform/implementation/g3/socket_base.h" +#include "internal/platform/implementation/awdl.h" +#include "internal/platform/input_stream.h" +#include "internal/platform/nsd_service_info.h" +#include "internal/platform/output_stream.h" + +namespace nearby { +namespace g3 { + +class AwdlMedium; + +class AwdlSocket : public api::AwdlSocket, public SocketBase { + public: + // Returns the InputStream of this connected AwdlSocket. + InputStream& GetInputStream() override { + return SocketBase::GetInputStream(); + } + + // Returns the OutputStream of this connected AwdlSocket. + // This stream is for local side to write. + OutputStream& GetOutputStream() override { + return SocketBase::GetOutputStream(); + } + + // Returns address of a remote AwdlSocket or nullptr. + AwdlSocket* GetRemoteSocket() { + return static_cast(SocketBase::GetRemoteSocket()); + } + + // Returns Exception::kIo on error, Exception::kSuccess otherwise. + Exception Close() override { return SocketBase::Close(); } +}; + +class AwdlServerSocket : public api::AwdlServerSocket { + public: + static std::string GetName(const std::string& ip_address, int port); + + ~AwdlServerSocket() override; + + // Gets ip address. + std::string GetIPAddress() const override ABSL_LOCKS_EXCLUDED(mutex_) { + absl::MutexLock lock(&mutex_); + return ip_address_; + } + + // Sets the ip address. + void SetIPAddress(const std::string& ip_address) ABSL_LOCKS_EXCLUDED(mutex_) { + absl::MutexLock lock(&mutex_); + ip_address_ = ip_address; + } + + // Gets the port. + int GetPort() const override ABSL_LOCKS_EXCLUDED(mutex_) { + absl::MutexLock lock(&mutex_); + return port_; + } + + // Sets the port. + void SetPort(int port) ABSL_LOCKS_EXCLUDED(mutex_) { + absl::MutexLock lock(&mutex_); + port_ = port; + } + + // Blocks until either: + // - at least one incoming connection request is available, or + // - ServerSocket is closed. + // On success, returns connected socket, ready to exchange data. + // Returns nullptr on error. + // Once error is reported, it is permanent, and ServerSocket has to be closed. + // + // Called by the server side of a connection. + // Returns AwdlSocket to the server side. + // If not null, returned socket is connected to its remote (client-side) peer. + std::unique_ptr Accept() override + ABSL_LOCKS_EXCLUDED(mutex_); + + // Blocks until either: + // - connection is available, or + // - server socket is closed, or + // - error happens. + // + // Called by the client side of a connection. + // Returns true, if socket is successfully connected. + bool Connect(AwdlSocket& socket) ABSL_LOCKS_EXCLUDED(mutex_); + + // Called by the server side of a connection before passing ownership of + // AwdlServerSocker to user, to track validity of a pointer to this + // server socket. + void SetCloseNotifier(absl::AnyInvocable notifier) + ABSL_LOCKS_EXCLUDED(mutex_); + + // Returns Exception::kIo on error, Exception::kSuccess otherwise. + // Calls close_notifier if it was previously set, and marks socket as closed. + Exception Close() override ABSL_LOCKS_EXCLUDED(mutex_); + + private: + Exception DoClose() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + + mutable absl::Mutex mutex_; + std::string ip_address_ ABSL_GUARDED_BY(mutex_); + int port_ ABSL_GUARDED_BY(mutex_); + absl::CondVar cond_; + absl::flat_hash_set pending_sockets_ ABSL_GUARDED_BY(mutex_); + absl::AnyInvocable close_notifier_ ABSL_GUARDED_BY(mutex_); + bool closed_ ABSL_GUARDED_BY(mutex_) = false; +}; + +// Container of operations that can be performed over the Awdl medium. +class AwdlMedium : public api::AwdlMedium { + public: + AwdlMedium(); + ~AwdlMedium() override; + + // Check if a network connection to a primary router exist. + bool IsNetworkConnected() const override { return true; } + + // Starts Awdl advertising. + // + // nsd_service_info - NsdServiceInfo data that's advertised through mDNS + // service. + // On success if the service is now advertising. + // On error if the service cannot start to advertise or the service type in + // NsdServiceInfo has been passed previously which StopAdvertising is not + // been called. + bool StartAdvertising(const NsdServiceInfo& nsd_service_info) override + ABSL_LOCKS_EXCLUDED(mutex_); + + // Stops Awdl advertising. + // + // nsd_service_info - NsdServiceInfo data that's advertised through mDNS + // service. + // On success if the service stops advertising. + // On error if the service cannot stop advertising or the service type in + // NsdServiceInfo cannot be found. + bool StopAdvertising(const NsdServiceInfo& nsd_service_info) override + ABSL_LOCKS_EXCLUDED(mutex_); + + // Starts the discovery of nearby Awdl services. + // + // Returns true once the Awdl discovery has been initiated. The + // service_type is associated with callback. + bool StartDiscovery(const std::string& service_type, + DiscoveredServiceCallback callback) override + ABSL_LOCKS_EXCLUDED(mutex_); + + // Stops the discovery of nearby Awdl services. + // + // service_type - The one assigend in StartDiscovery. + // On success if service_type is matched to the callback and will be removed + // from the list. If list is empty then stops the Awdl discovery + // service. + // On error if the service_type is not existed, then return immediately. + bool StopDiscovery(const std::string& service_type) override + ABSL_LOCKS_EXCLUDED(mutex_); + + // Connects to a Awdl service. + // On success, returns a new AwdlSocket. + // On error, returns nullptr. + std::unique_ptr ConnectToService( + const NsdServiceInfo& remote_service_info, + CancellationFlag* cancellation_flag) override ABSL_LOCKS_EXCLUDED(mutex_); + + // Connects to a Awdl service by ip address and port. + // On success, returns a new AwdlSocket. + // On error, returns nullptr. + std::unique_ptr ConnectToService( + const std::string& ip_address, int port, + CancellationFlag* cancellation_flag) override ABSL_LOCKS_EXCLUDED(mutex_); + + // Listens for incoming connection. + // + // port - A port number. + // 0 : use a random port. + // 1~65536 : open a server socket on that exact port. + // On success, returns a new AwdlServerSocket. + // On error, returns nullptr. + std::unique_ptr ListenForService(int port) override + ABSL_LOCKS_EXCLUDED(mutex_); + + // Returns the port range as a pair of min and max port. + absl::optional> GetDynamicPortRange() + override { + return std::make_pair(49152, 65535); + } + + private: + struct AdvertisingInfo { + bool Empty() const { return service_types.empty(); } + void Clear() { service_types.clear(); } + void Add(const std::string& service_type) { + service_types.insert(service_type); + } + void Remove(const std::string& service_type) { + service_types.erase(service_type); + } + bool Existed(const std::string& service_type) const { + return service_types.contains(service_type); + } + + absl::flat_hash_set service_types; + }; + struct DiscoveringInfo { + bool Empty() const { return service_types.empty(); } + void Clear() { service_types.clear(); } + void Add(const std::string& service_type) { + service_types.insert(service_type); + } + void Remove(const std::string& service_type) { + service_types.erase(service_type); + } + bool Existed(const std::string& service_type) const { + return service_types.contains(service_type); + } + + absl::flat_hash_set service_types; + }; + + absl::Mutex mutex_; + AdvertisingInfo advertising_info_ ABSL_GUARDED_BY(mutex_); + DiscoveringInfo discovering_info_ ABSL_GUARDED_BY(mutex_); + absl::flat_hash_map server_sockets_ + ABSL_GUARDED_BY(mutex_); +}; + +} // namespace g3 +} // namespace nearby + +#endif // PLATFORM_IMPL_G3_AWDL_H_ diff --git a/internal/platform/implementation/g3/platform.cc b/internal/platform/implementation/g3/platform.cc index 72e60604..9de46713 100644 --- a/internal/platform/implementation/g3/platform.cc +++ b/internal/platform/implementation/g3/platform.cc @@ -28,6 +28,7 @@ #include "internal/base/files.h" #include "internal/platform/implementation/atomic_boolean.h" #include "internal/platform/implementation/atomic_reference.h" +#include "internal/platform/implementation/awdl.h" #include "internal/platform/implementation/ble.h" #include "internal/platform/implementation/ble_v2.h" #include "internal/platform/implementation/bluetooth_adapter.h" @@ -36,6 +37,7 @@ #include "internal/platform/implementation/count_down_latch.h" #include "internal/platform/implementation/credential_storage.h" #include "internal/platform/implementation/device_info.h" +#include "internal/platform/implementation/g3/awdl.h" #include "internal/platform/implementation/http_loader.h" #include "internal/platform/implementation/input_file.h" #include "internal/platform/implementation/log_message.h" @@ -213,6 +215,10 @@ std::unique_ptr ImplementationPlatform::CreateWifiLanMedium() { return std::make_unique(); } +std::unique_ptr ImplementationPlatform::CreateAwdlMedium() { + return std::make_unique(); +} + std::unique_ptr ImplementationPlatform::CreateWifiHotspotMedium() { return std::make_unique(); diff --git a/internal/platform/implementation/windows/platform.cc b/internal/platform/implementation/windows/platform.cc index 78080d46..b64a1820 100644 --- a/internal/platform/implementation/windows/platform.cc +++ b/internal/platform/implementation/windows/platform.cc @@ -39,6 +39,7 @@ #include "internal/base/files.h" #include "internal/platform/implementation/atomic_boolean.h" #include "internal/platform/implementation/atomic_reference.h" +#include "internal/platform/implementation/awdl.h" #include "internal/platform/implementation/ble.h" #include "internal/platform/implementation/ble_v2.h" #include "internal/platform/implementation/bluetooth_adapter.h" @@ -303,6 +304,10 @@ std::unique_ptr ImplementationPlatform::CreateWifiLanMedium() { return std::make_unique(); } +std::unique_ptr ImplementationPlatform::CreateAwdlMedium() { + return nullptr; +} + std::unique_ptr ImplementationPlatform::CreateWifiHotspotMedium() { return std::make_unique(); diff --git a/internal/platform/medium_environment.cc b/internal/platform/medium_environment.cc index 14518655..de8d091e 100644 --- a/internal/platform/medium_environment.cc +++ b/internal/platform/medium_environment.cc @@ -33,6 +33,7 @@ #include "internal/platform/byte_array.h" #include "internal/platform/count_down_latch.h" #include "internal/platform/feature_flags.h" +#include "internal/platform/implementation/awdl.h" #include "internal/platform/implementation/ble.h" #include "internal/platform/implementation/ble_v2.h" #include "internal/platform/implementation/bluetooth_adapter.h" @@ -93,6 +94,7 @@ void MediumEnvironment::Reset() { webrtc_signaling_complete_callback_.clear(); #endif wifi_lan_mediums_.clear(); + awdl_mediums_.clear(); { MutexLock lock(&mutex_); wifi_direct_mediums_.clear(); @@ -399,6 +401,64 @@ void MediumEnvironment::OnWifiLanServiceStateChanged( } } +void MediumEnvironment::OnAwdlServiceStateChanged( + AwdlMediumContext& info, const NsdServiceInfo& service_info, bool enabled) { + if (!enabled_) return; + std::string service_name = service_info.GetServiceName(); + std::string service_type = service_info.GetServiceType(); + auto item = info.discovered_services.find(service_name); + if (item == info.discovered_services.end()) { + NEARBY_LOGS(INFO) << "OnAwdlServiceStateChanged; context=" << &info + << "; service_type=" << service_type + << "; enabled=" << enabled + << "; notify=" << enable_notifications_.load(); + if (enabled) { + // Find advertising service with matched service_type. Report it as + // discovered. + NsdServiceInfo discovered_service_info(service_info); + info.discovered_services.insert({service_name, discovered_service_info}); + if (enable_notifications_) { + RunOnMediumEnvironmentThread( + [&info, discovered_service_info, service_type]() { + auto item = info.discovered_callbacks.find(service_type); + if (item != info.discovered_callbacks.end()) { + item->second.service_discovered_cb(discovered_service_info); + } + }); + } + } + } else { + NEARBY_LOGS(INFO) << "OnAwdlServiceStateChanged: exisitng service; context=" + << &info << "; service_type=" << service_type + << "; enabled=" << enabled + << "; notify=" << enable_notifications_.load(); + if (enabled) { + if (enable_notifications_) { + RunOnMediumEnvironmentThread( + [&info, service_info = service_info, service_type]() { + auto item = info.discovered_callbacks.find(service_type); + if (item != info.discovered_callbacks.end()) { + item->second.service_discovered_cb(service_info); + } + }); + } + } else { + // Known service is off. + // Erase it from the map, and report as lost. + if (enable_notifications_) { + RunOnMediumEnvironmentThread( + [&info, service_info = service_info, service_type]() { + auto item = info.discovered_callbacks.find(service_type); + if (item != info.discovered_callbacks.end()) { + item->second.service_lost_cb(service_info); + } + }); + } + info.discovered_services.erase(item); + } + } +} + void MediumEnvironment::RunOnMediumEnvironmentThread(Runnable runnable) { job_count_++; executor_.Execute(std::move(runnable)); @@ -876,6 +936,14 @@ void MediumEnvironment::RegisterWifiLanMedium(api::WifiLanMedium& medium) { }); } +void MediumEnvironment::RegisterAwdlMedium(api::AwdlMedium& medium) { + if (!enabled_) return; + RunOnMediumEnvironmentThread([this, &medium]() { + awdl_mediums_.insert({&medium, AwdlMediumContext{}}); + NEARBY_LOGS(INFO) << "Registered: Awdl medium:" << &medium; + }); +} + void MediumEnvironment::UpdateWifiLanMediumForAdvertising( api::WifiLanMedium& medium, const NsdServiceInfo& service_info, bool enabled) { @@ -907,6 +975,36 @@ void MediumEnvironment::UpdateWifiLanMediumForAdvertising( }); } +void MediumEnvironment::UpdateAwdlMediumForAdvertising( + api::AwdlMedium& medium, const NsdServiceInfo& service_info, bool enabled) { + if (!enabled_) return; + RunOnMediumEnvironmentThread( + [this, &medium, service_info = service_info, enabled]() { + std::string service_name = service_info.GetServiceName(); + std::string service_type = service_info.GetServiceType(); + NEARBY_LOGS(INFO) << "Update Awdl medium for advertising: this=" << this + << "; medium=" << &medium + << "; service_name=" << service_name + << "; service_type=" << service_type + << ", enabled=" << enabled; + for (auto& medium_info : awdl_mediums_) { + auto& local_medium = medium_info.first; + auto& info = medium_info.second; + // Do not send notification to the same medium but update + // service info map. + if (local_medium == &medium) { + if (enabled) { + info.advertising_services.insert({service_name, service_info}); + } else { + info.advertising_services.erase(service_name); + } + continue; + } + OnAwdlServiceStateChanged(info, service_info, enabled); + } + }); +} + void MediumEnvironment::UpdateWifiLanMediumForDiscovery( api::WifiLanMedium& medium, WifiLanDiscoveredServiceCallback callback, const std::string& service_type, bool enabled) { @@ -940,6 +1038,39 @@ void MediumEnvironment::UpdateWifiLanMediumForDiscovery( }); } +void MediumEnvironment::UpdateAwdlMediumForDiscovery( + api::AwdlMedium& medium, AwdlDiscoveredServiceCallback callback, + const std::string& service_type, bool enabled) { + if (!enabled_) return; + RunOnMediumEnvironmentThread([this, &medium, callback = std::move(callback), + service_type, enabled]() mutable { + auto item = awdl_mediums_.find(&medium); + if (item == awdl_mediums_.end()) { + NEARBY_LOGS(INFO) + << "UpdateAwdlMediumForDiscovery failed. There is no medium " + "registered."; + return; + } + auto& context = item->second; + context.discovered_callbacks.insert({service_type, std::move(callback)}); + NEARBY_LOGS(INFO) << "Update Awdl medium for discovery: this=" << this + << "; medium=" << &medium + << "; service_type=" << service_type + << "; enabled=" << enabled; + for (auto& medium_info : awdl_mediums_) { + auto& local_medium = medium_info.first; + auto& info = medium_info.second; + // Do not send notification to the same medium. + if (local_medium == &medium) continue; + // Search advertising services and send notification. + for (auto& advertising_service : info.advertising_services) { + auto& service_info = advertising_service.second; + OnAwdlServiceStateChanged(context, service_info, /*enabled=*/true); + } + } + }); +} + void MediumEnvironment::UnregisterWifiLanMedium(api::WifiLanMedium& medium) { if (!enabled_) return; RunOnMediumEnvironmentThread([this, &medium]() { @@ -949,6 +1080,15 @@ void MediumEnvironment::UnregisterWifiLanMedium(api::WifiLanMedium& medium) { }); } +void MediumEnvironment::UnregisterAwdlMedium(api::AwdlMedium& medium) { + if (!enabled_) return; + RunOnMediumEnvironmentThread([this, &medium]() { + auto item = awdl_mediums_.extract(&medium); + if (item.empty()) return; + NEARBY_LOGS(INFO) << "Unregistered Awdl medium:" << &medium; + }); +} + api::WifiLanMedium* MediumEnvironment::GetWifiLanMedium( const std::string& ip_address, int port) { api::WifiLanMedium* result = nullptr; @@ -973,6 +1113,30 @@ api::WifiLanMedium* MediumEnvironment::GetWifiLanMedium( return result; } +api::AwdlMedium* MediumEnvironment::GetAwdlMedium(const std::string& ip_address, + int port) { + api::AwdlMedium* result = nullptr; + CountDownLatch latch(1); + RunOnMediumEnvironmentThread([&]() { + for (auto& medium_info : awdl_mediums_) { + auto* medium_found = medium_info.first; + auto& info = medium_info.second; + for (auto& advertising_service : info.advertising_services) { + auto& service_info = advertising_service.second; + if (ip_address == service_info.GetIPAddress() && + port == service_info.GetPort()) { + result = medium_found; + latch.CountDown(); + return; + } + } + } + latch.CountDown(); + }); + latch.Await(); + return result; +} + void MediumEnvironment::RegisterWifiDirectMedium( api::WifiDirectMedium& medium) { if (!enabled_) return; diff --git a/internal/platform/medium_environment.h b/internal/platform/medium_environment.h index bad815a4..6eb93f1e 100644 --- a/internal/platform/medium_environment.h +++ b/internal/platform/medium_environment.h @@ -29,6 +29,7 @@ #include "absl/types/optional.h" #include "internal/base/observer_list.h" #include "internal/platform/borrowable.h" +#include "internal/platform/implementation/awdl.h" #include "internal/platform/implementation/ble.h" #include "internal/platform/implementation/ble_v2.h" #include "internal/platform/implementation/bluetooth_adapter.h" @@ -87,6 +88,8 @@ class MediumEnvironment { #endif using WifiLanDiscoveredServiceCallback = api::WifiLanMedium::DiscoveredServiceCallback; + using AwdlDiscoveredServiceCallback = + api::AwdlMedium::DiscoveredServiceCallback; struct BleV2MediumStatus { bool is_advertising; @@ -275,11 +278,21 @@ class MediumEnvironment { // expects they should communicate. void RegisterWifiLanMedium(api::WifiLanMedium& medium); + // Adds medium-related info to allow for discovery/advertising to work. + // This provides access to this medium from other mediums, when protocol + // expects they should communicate. + void RegisterAwdlMedium(api::AwdlMedium& medium); + // Updates advertising info to indicate the current medium is exposing // advertising event. void UpdateWifiLanMediumForAdvertising(api::WifiLanMedium& medium, const NsdServiceInfo& nsd_service_info, bool enabled); + // Updates advertising info to indicate the current medium is exposing + // advertising event. + void UpdateAwdlMediumForAdvertising(api::AwdlMedium& medium, + const NsdServiceInfo& nsd_service_info, + bool enabled); // Updates discovery callback info to allow for dispatch of discovery events. // @@ -290,6 +303,16 @@ class MediumEnvironment { api::WifiLanMedium& medium, WifiLanDiscoveredServiceCallback callback, const std::string& service_type, bool enabled); + // Updates discovery callback info to allow for dispatch of discovery events. + // + // This should be called when discoverable state changes. + // A valid callback should be assigned when discovery `enabled` as true; or + // an empty callback is assigned with discovery `enabled` as false. + void UpdateAwdlMediumForDiscovery(api::AwdlMedium& medium, + AwdlDiscoveredServiceCallback callback, + const std::string& service_type, + bool enabled); + // Gets Fake IP address for WifiLan medium. std::string GetFakeIPAddress() const; @@ -299,10 +322,17 @@ class MediumEnvironment { // Removes medium-related info. This should correspond to device power off. void UnregisterWifiLanMedium(api::WifiLanMedium& medium); + // Removes medium-related info. This should correspond to device power off. + void UnregisterAwdlMedium(api::AwdlMedium& medium); + // Returns WifiLan medium whose advertising service matching IP address and // port, or nullptr. api::WifiLanMedium* GetWifiLanMedium(const std::string& ip_address, int port); + // Returns Awdl medium whose advertising service matching IP address and + // port, or nullptr. + api::AwdlMedium* GetAwdlMedium(const std::string& ip_address, int port); + // Adds medium-related info to allow for start/connect WifiDirect to work. void RegisterWifiDirectMedium(api::WifiDirectMedium& medium); @@ -434,6 +464,16 @@ class MediumEnvironment { absl::flat_hash_map discovered_services; }; + struct AwdlMediumContext { + // advertising service type vs NsdServiceInfo map. + absl::flat_hash_map advertising_services; + // discovered service type vs callback map. + absl::flat_hash_map + discovered_callbacks; + // discovered service vs service type map. + absl::flat_hash_map discovered_services; + }; + struct WifiDirectMediumContext { // Set to "true" for Medium act as WifiDirect GO role; "false" for GC role bool is_go = false; @@ -484,6 +524,10 @@ class MediumEnvironment { const NsdServiceInfo& service_info, bool enabled); + void OnAwdlServiceStateChanged(AwdlMediumContext& info, + const NsdServiceInfo& service_info, + bool enabled); + void RunOnMediumEnvironmentThread(Runnable runnable); std::atomic_bool enabled_ = false; @@ -517,6 +561,8 @@ class MediumEnvironment { absl::flat_hash_map wifi_lan_mediums_; + absl::flat_hash_map awdl_mediums_; + Mutex mutex_; absl::flat_hash_map wifi_direct_mediums_ ABSL_GUARDED_BY(mutex_);