diff --git a/cpp/core/core.h b/cpp/core/core.h index ee6f8486..3b94a1fd 100644 --- a/cpp/core/core.h +++ b/cpp/core/core.h @@ -215,6 +215,9 @@ class Core { void InitiateBandwidthUpgrade(absl::string_view endpoint_id, ResultCallback callback); + // Gets the local endpoint generated by Nearby Connections. + std::string GetLocalEndpointId() { return client_.GetLocalEndpointId(); } + private: static constexpr absl::Duration kWaitForDisconnect = absl::Milliseconds(5000); diff --git a/cpp/core/internal/BUILD b/cpp/core/internal/BUILD index 6baa01da..a2e9dee8 100644 --- a/cpp/core/internal/BUILD +++ b/cpp/core/internal/BUILD @@ -5,6 +5,7 @@ cc_library( "base_pcp_handler.cc", "ble_advertisement.cc", "ble_endpoint_channel.cc", + "bluetooth_bwu_handler.cc", "bluetooth_device_name.cc", "bluetooth_endpoint_channel.cc", "bwu_manager.cc", @@ -24,6 +25,7 @@ cc_library( "service_controller_router.cc", "webrtc_bwu_handler.cc", "webrtc_endpoint_channel.cc", + "wifi_lan_bwu_handler.cc", "wifi_lan_endpoint_channel.cc", "wifi_lan_service_info.cc", ], @@ -33,6 +35,7 @@ cc_library( "base_pcp_handler.h", "ble_advertisement.h", "ble_endpoint_channel.h", + "bluetooth_bwu_handler.h", "bluetooth_device_name.h", "bluetooth_endpoint_channel.h", "bwu_handler.h", @@ -57,6 +60,7 @@ cc_library( "service_controller_router.h", "webrtc_bwu_handler.h", "webrtc_endpoint_channel.h", + "wifi_lan_bwu_handler.h", "wifi_lan_endpoint_channel.h", "wifi_lan_service_info.h", ], diff --git a/cpp/core/internal/base_bwu_handler.h b/cpp/core/internal/base_bwu_handler.h index 33703d46..ab3d7161 100644 --- a/cpp/core/internal/base_bwu_handler.h +++ b/cpp/core/internal/base_bwu_handler.h @@ -30,8 +30,6 @@ class BaseBwuHandler : public BwuHandler { : channel_manager_(&channel_manager), bwu_notifications_(std::move(bwu_notifications)) {} ~BaseBwuHandler() override = default; - void OnIncomingConnection(ClientProxy* client, - IncomingSocketConnection* connection); protected: // Represents the incoming Socket the Initiator has gotten after initializing diff --git a/cpp/core/internal/base_pcp_handler.cc b/cpp/core/internal/base_pcp_handler.cc index 6d74be82..d8e413ce 100644 --- a/cpp/core/internal/base_pcp_handler.cc +++ b/cpp/core/internal/base_pcp_handler.cc @@ -673,12 +673,12 @@ void BasePcpHandler::OnIncomingFrame(OfflineFrame& frame, void BasePcpHandler::OnEndpointDisconnect(ClientProxy* client, const std::string& endpoint_id, - CountDownLatch* barrier) { + CountDownLatch barrier) { if (stop_.Get()) { - if (barrier) barrier->CountDown(); + barrier.CountDown(); return; } - RunOnPcpHandlerThread([this, client, endpoint_id, barrier]() { + RunOnPcpHandlerThread([this, client, endpoint_id, barrier]() mutable { auto item = pending_alarms_.find(endpoint_id); if (item != pending_alarms_.end()) { auto& alarm = item->second; @@ -686,7 +686,7 @@ void BasePcpHandler::OnEndpointDisconnect(ClientProxy* client, pending_alarms_.erase(item); } ProcessPreConnectionResultFailure(client, endpoint_id); - barrier->CountDown(); + barrier.CountDown(); }); } diff --git a/cpp/core/internal/base_pcp_handler.h b/cpp/core/internal/base_pcp_handler.h index c7a6ff0b..3b3cf4b1 100644 --- a/cpp/core/internal/base_pcp_handler.h +++ b/cpp/core/internal/base_pcp_handler.h @@ -149,7 +149,7 @@ class BasePcpHandler : public PcpHandler, // approve/reject the connection. // @EndpointManagerThread void OnEndpointDisconnect(ClientProxy* client, const std::string& endpoint_id, - CountDownLatch* barrier) override; + CountDownLatch barrier) override; Pcp GetPcp() const override { return pcp_; } Strategy GetStrategy() const override { return strategy_; } diff --git a/cpp/core/internal/base_pcp_handler_test.cc b/cpp/core/internal/base_pcp_handler_test.cc index 29246f9f..5e0a879f 100644 --- a/cpp/core/internal/base_pcp_handler_test.cc +++ b/cpp/core/internal/base_pcp_handler_test.cc @@ -379,6 +379,7 @@ TEST_P(BasePcpHandlerTest, ConstructorDestructorWorks) { BwuManager bwu(m, em, ecm, {}, {}); MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu); SUCCEED(); + bwu.Shutdown(); } TEST_P(BasePcpHandlerTest, StartAdvertisingChangesState) { @@ -389,6 +390,7 @@ TEST_P(BasePcpHandlerTest, StartAdvertisingChangesState) { BwuManager bwu(m, em, ecm, {}, {}); MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu); StartAdvertising(&client, &pcp_handler); + bwu.Shutdown(); } TEST_P(BasePcpHandlerTest, StopAdvertisingChangesState) { @@ -403,6 +405,7 @@ TEST_P(BasePcpHandlerTest, StopAdvertisingChangesState) { EXPECT_TRUE(client.IsAdvertising()); pcp_handler.StopAdvertising(&client); EXPECT_FALSE(client.IsAdvertising()); + bwu.Shutdown(); } TEST_P(BasePcpHandlerTest, StartDiscoveryChangesState) { @@ -413,6 +416,7 @@ TEST_P(BasePcpHandlerTest, StartDiscoveryChangesState) { BwuManager bwu(m, em, ecm, {}, {}); MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu); StartDiscovery(&client, &pcp_handler); + bwu.Shutdown(); } TEST_P(BasePcpHandlerTest, StopDiscoveryChangesState) { @@ -427,6 +431,7 @@ TEST_P(BasePcpHandlerTest, StopDiscoveryChangesState) { EXPECT_TRUE(client.IsDiscovering()); pcp_handler.StopDiscovery(&client); EXPECT_FALSE(client.IsDiscovering()); + bwu.Shutdown(); } TEST_P(BasePcpHandlerTest, RequestConnectionChangesState) { @@ -450,6 +455,7 @@ TEST_P(BasePcpHandlerTest, RequestConnectionChangesState) { &pcp_handler, connect_medium); NEARBY_LOG(INFO, "RequestConnection complete"); channel_b->Close(); + bwu.Shutdown(); pcp_handler.DisconnectFromEndpointManager(); } @@ -478,6 +484,7 @@ TEST_P(BasePcpHandlerTest, AcceptConnectionChangesState) { EXPECT_CALL(mock_connection_listener_.rejected_cb, Call).Times(AtLeast(0)); NEARBY_LOGS(INFO) << "Closing connection: id=" << endpoint_id; channel_b->Close(); + bwu.Shutdown(); pcp_handler.DisconnectFromEndpointManager(); } @@ -502,6 +509,7 @@ TEST_P(BasePcpHandlerTest, RejectConnectionChangesState) { Status{Status::kSuccess}); NEARBY_LOGS(INFO) << "Closing connection: id=" << endpoint_id; channel_b->Close(); + bwu.Shutdown(); pcp_handler.DisconnectFromEndpointManager(); } @@ -536,6 +544,7 @@ TEST_P(BasePcpHandlerTest, OnIncomingFrameChangesState) { connect_medium); NEARBY_LOGS(INFO) << "Closing connection: id=" << endpoint_id; channel_b->Close(); + bwu.Shutdown(); pcp_handler.DisconnectFromEndpointManager(); } @@ -568,8 +577,8 @@ TEST_P(BasePcpHandlerTest, DestructorIsCalledOnProtocolEndpoint) { EXPECT_CALL(mock_connection_listener_.rejected_cb, Call).Times(AtLeast(0)); NEARBY_LOG(INFO, "Closing connection: id=%s", endpoint_id.c_str()); channel_b->Close(); - pcp_handler.DisconnectFromEndpointManager(); bwu.Shutdown(); + pcp_handler.DisconnectFromEndpointManager(); } EXPECT_EQ(destroyed_flag.load(), mediums_count); } @@ -615,6 +624,7 @@ TEST_P(BasePcpHandlerTest, MultipleMediumsProduceSingleEndpointLostEvent) { } NEARBY_LOG(INFO, "Closing connection: id=%s", endpoint_id.c_str()); channel_b->Close(); + bwu.Shutdown(); pcp_handler.DisconnectFromEndpointManager(); } EXPECT_EQ(destroyed_flag.load(), mediums_count); @@ -672,6 +682,7 @@ TEST_F(BasePcpHandlerTest, InjectEndpoint) { .medium = Medium::BLUETOOTH, .remote_bluetooth_mac_address = ByteArray(kFakeMacAddress), }); + bwu.Shutdown(); } } // namespace diff --git a/cpp/core/internal/bluetooth_bwu_handler.cc b/cpp/core/internal/bluetooth_bwu_handler.cc new file mode 100644 index 00000000..57b03311 --- /dev/null +++ b/cpp/core/internal/bluetooth_bwu_handler.cc @@ -0,0 +1,116 @@ +#include "core/internal/bluetooth_bwu_handler.h" + +#include "core/internal/bluetooth_endpoint_channel.h" +#include "core/internal/client_proxy.h" +#include "core/internal/offline_frames.h" +#include "absl/functional/bind_front.h" + +// Manages the Bluetooth-specific methods needed to upgrade an {@link +// EndpointChannel}. + +namespace location { +namespace nearby { +namespace connections { + +BluetoothBwuHandler::BluetoothBwuHandler( + Mediums& mediums, EndpointChannelManager& channel_manager, + BwuNotifications notifications) + : BaseBwuHandler(channel_manager, std::move(notifications)), + mediums_(mediums) {} + +void BluetoothBwuHandler::Revert() { + for (const std::string& service_id : active_service_ids_) { + bluetooth_medium_.StopAcceptingConnections(service_id); + } + active_service_ids_.clear(); + NEARBY_LOG(INFO, + "BluetoothBwuHandler successfully reverted all Bluetooth state."); +} + +// Accept Connection Callback. +// Notifies that the remote party called BluetoothClassic::Connect() +// for this socket. +void BluetoothBwuHandler::OnIncomingBluetoothConnection( + ClientProxy* client, const std::string& service_id, + BluetoothSocket socket) { + auto channel = + absl::make_unique(service_id, socket); + std::unique_ptr connection{ + new IncomingSocketConnection{ + .socket = + std::make_unique(service_id, socket), + .channel = std::move(channel), + }}; + bwu_notifications_.incoming_connection_cb(client, std::move(connection)); +} + +// Called by BWU initiator. BT Medium is set up, and BWU request is prepared, +// with necessary info (service_id, MAC address) for remote party to perform +// discovery. +ByteArray BluetoothBwuHandler::InitializeUpgradedMediumForEndpoint( + ClientProxy* client, const std::string& service_id, + const std::string& endpoint_id) { + std::string upgrade_service_id = Utils::WrapUpgradeServiceId(service_id); + + std::string mac_address = bluetooth_medium_.GetMacAddress(); + if (mac_address.empty()) { + return {}; + } + + if (!bluetooth_medium_.IsAcceptingConnections(upgrade_service_id)) { + if (!bluetooth_medium_.StartAcceptingConnections( + upgrade_service_id, + { + .accepted_cb = absl::bind_front( + &BluetoothBwuHandler::OnIncomingBluetoothConnection, this, + client, service_id), + })) { + return {}; + } + } + // cache service ID to revert + active_service_ids_.emplace(upgrade_service_id); + + return parser::ForBwuBluetoothPathAvailable(upgrade_service_id, mac_address); +} + +// Called by BWU target. Retrieves a new medium info from incoming message, +// and establishes connection over BT using this info. +// Returns a channel ready to exchange data or nullptr on error. +std::unique_ptr +BluetoothBwuHandler::CreateUpgradedEndpointChannel( + ClientProxy* client, const std::string& service_id, + const std::string& endpoint_id, const UpgradePathInfo& upgrade_path_info) { + const UpgradePathInfo::BluetoothCredentials& bluetooth_credentials = + upgrade_path_info.bluetooth_credentials(); + if (!bluetooth_credentials.has_service_name() || + !bluetooth_credentials.has_mac_address()) { + return nullptr; + } + + const std::string& service_name = bluetooth_credentials.service_name(); + const std::string& mac_address = bluetooth_credentials.mac_address(); + + BluetoothDevice device = bluetooth_medium_.GetRemoteDevice(mac_address); + if (!device.IsValid()) { + return nullptr; + } + + BluetoothSocket socket = bluetooth_medium_.Connect(device, service_name); + if (!socket.IsValid()) { + return nullptr; + } + + auto channel = + std::make_unique(service_name, socket); + if (channel == nullptr) { + socket.Close(); + return nullptr; + } + + return channel; +} + +} // namespace connections +} // namespace nearby +} // namespace location diff --git a/cpp/core/internal/bluetooth_bwu_handler.h b/cpp/core/internal/bluetooth_bwu_handler.h new file mode 100644 index 00000000..016e33b5 --- /dev/null +++ b/cpp/core/internal/bluetooth_bwu_handler.h @@ -0,0 +1,83 @@ +#ifndef CORE_INTERNAL_BLUETOOTH_BWU_HANDLER_H_ +#define CORE_INTERNAL_BLUETOOTH_BWU_HANDLER_H_ + +#include + +#include "core/internal/base_bwu_handler.h" +#include "core/internal/client_proxy.h" +#include "core/internal/mediums/mediums.h" +#include "core/internal/mediums/utils.h" +#include "proto/connections/offline_wire_formats.pb.h" +#include "platform/public/bluetooth_classic.h" +#include "platform/public/count_down_latch.h" +#include "proto/connections_enums.pb.h" + +namespace location { +namespace nearby { +namespace connections { + +// Defines the set of methods that need to be implemented to handle the +// per-Medium-specific operations needed to upgrade an EndpointChannel. +class BluetoothBwuHandler : public BaseBwuHandler { + public: + BluetoothBwuHandler(Mediums& mediums, EndpointChannelManager& channel_manager, + BwuNotifications notifications); + ~BluetoothBwuHandler() override = default; + + private: + constexpr static const int kServiceIdLength = 10; + + // Implements BaseBwuHandler: + // Reverts any changes made to the device in the process of upgrading + // endpoints. + void Revert() override; + + // Cleans up in-progress upgrades after endpoint disconnection. + void OnEndpointDisconnect(ClientProxy* client, + const std::string& endpoint_id) override {} + + void OnIncomingBluetoothConnection(ClientProxy* client, + const std::string& service_id, + BluetoothSocket socket); + + class BluetoothIncomingSocket : public IncomingSocket { + public: + explicit BluetoothIncomingSocket(const std::string& name, + BluetoothSocket socket) + : name_(name), socket_(socket) {} + ~BluetoothIncomingSocket() override = default; + std::string ToString() override { return name_; } + void Close() override { socket_.Close(); } + + private: + std::string name_; + BluetoothSocket socket_; + }; + + // First part of InitiateBwuForEndpoint implementation; + // returns a BWU request to remote party as byte array. + ByteArray InitializeUpgradedMediumForEndpoint( + ClientProxy* client, const std::string& service_id, + const std::string& endpoint_id) override; + + // Invoked from OnBwuNegotiationFrame. + std::unique_ptr CreateUpgradedEndpointChannel( + ClientProxy* client, const std::string& service_id, + const std::string& endpoint_id, + const UpgradePathInfo& upgrade_path_info) override; + + // Returns the upgrade medium of the BwuHandler. + // @BwuHandlerThread + Medium GetUpgradeMedium() const override { return Medium::BLUETOOTH; } + + Mediums& mediums_; + absl::flat_hash_set active_service_ids_; + BluetoothRadio& bluetooth_radio_{mediums_.GetBluetoothRadio()}; + BluetoothClassic& bluetooth_medium_{mediums_.GetBluetoothClassic()}; +}; + +} // namespace connections +} // namespace nearby +} // namespace location + +#endif // CORE_INTERNAL_BLUETOOTH_BWU_HANDLER_H_ diff --git a/cpp/core/internal/bwu_handler.h b/cpp/core/internal/bwu_handler.h index a3a4587b..3cd76981 100644 --- a/cpp/core/internal/bwu_handler.h +++ b/cpp/core/internal/bwu_handler.h @@ -28,6 +28,7 @@ class BwuHandler { virtual ByteArray InitializeUpgradedMediumForEndpoint( ClientProxy* client, const std::string& service_id, const std::string& endpoint_id) = 0; + // Called to revert any state changed by the Initiator to setup the upgraded // medium for an endpoint. // @BwuHandlerThread @@ -41,6 +42,7 @@ class BwuHandler { ClientProxy* client, const std::string& service_id, const std::string& endpoint_id, const UpgradePathInfo& upgrade_path_info) = 0; + // Returns the upgrade medium of the BwuHandler. // @BwuHandlerThread virtual Medium GetUpgradeMedium() const = 0; diff --git a/cpp/core/internal/bwu_manager.cc b/cpp/core/internal/bwu_manager.cc index 30bdb773..be7e625d 100644 --- a/cpp/core/internal/bwu_manager.cc +++ b/cpp/core/internal/bwu_manager.cc @@ -3,9 +3,11 @@ #include #include +#include "core/internal/bluetooth_bwu_handler.h" #include "core/internal/bwu_handler.h" #include "core/internal/offline_frames.h" #include "core/internal/webrtc_bwu_handler.h" +#include "core/internal/wifi_lan_bwu_handler.h" #include "platform/base/byte_array.h" #include "platform/public/count_down_latch.h" #include "proto/connections_enums.pb.h" @@ -54,11 +56,21 @@ void BwuManager::InitBwuHandlers() { .incoming_connection_cb = absl::bind_front(&BwuManager::OnIncomingConnection, this), }; + if (config_.allow_upgrade_to.wifi_lan) { + handlers_.emplace(Medium::WIFI_LAN, + std::make_unique( + *mediums_, *channel_manager_, notifications)); + } if (config_.allow_upgrade_to.web_rtc) { handlers_.emplace(Medium::WEB_RTC, std::make_unique( *mediums_, *channel_manager_, notifications)); } + if (config_.allow_upgrade_to.bluetooth) { + handlers_.emplace(Medium::BLUETOOTH, + std::make_unique( + *mediums_, *channel_manager_, notifications)); + } } void BwuManager::Shutdown() { @@ -67,31 +79,26 @@ void BwuManager::Shutdown() { endpoint_manager_->UnregisterFrameProcessor( V1Frame::BANDWIDTH_UPGRADE_NEGOTIATION, this); - CountDownLatch latch(1); - - RunOnBwuManagerThread([this, &latch]() { - for (auto& item : previous_endpoint_channels_) { - EndpointChannel* channel = item.second.get(); - if (!channel) continue; - channel->Close(DisconnectionReason::SHUTDOWN); - } - - CancelAllRetryUpgradeAlarms(); - medium_ = Medium::UNKNOWN_MEDIUM; - for (auto& item : handlers_) { - BwuHandler& handler = *item.second; - handler.Revert(); - } - handlers_.clear(); - latch.CountDown(); - }); - - latch.Await(); - // Stop all the ongoing Runnables (as gracefully as possible). alarm_executor_.Shutdown(); serial_executor_.Shutdown(); + // After worker threads are down we became exclusive owners of data and + // may access it from current thread. + for (auto& item : previous_endpoint_channels_) { + EndpointChannel* channel = item.second.get(); + if (!channel) continue; + channel->Close(DisconnectionReason::SHUTDOWN); + } + + CancelAllRetryUpgradeAlarms(); + medium_ = Medium::UNKNOWN_MEDIUM; + for (auto& item : handlers_) { + BwuHandler& handler = *item.second; + handler.Revert(); + } + handlers_.clear(); + NEARBY_LOG(INFO, "BwuHandler has shut down."); } @@ -182,10 +189,10 @@ void BwuManager::OnIncomingFrame(OfflineFrame& frame, void BwuManager::OnEndpointDisconnect(ClientProxy* client, const std::string& endpoint_id, - CountDownLatch* barrier) { - RunOnBwuManagerThread([this, client, endpoint_id, barrier]() { + CountDownLatch barrier) { + RunOnBwuManagerThread([this, client, endpoint_id, barrier]() mutable { if (medium_ == Medium::UNKNOWN_MEDIUM) { - barrier->CountDown(); + barrier.CountDown(); return; } @@ -213,7 +220,7 @@ void BwuManager::OnEndpointDisconnect(ClientProxy* client, if (channel_manager_->GetConnectedEndpointsCount() <= 1) { Revert(); } - barrier->CountDown(); + barrier.CountDown(); }); } @@ -541,6 +548,8 @@ void BwuManager::ProcessSafeToClosePriorChannelEvent( "trying to upgrade endpoint %s.", endpoint_id.c_str()); + previous_endpoint_channel->Write(parser::ForDisconnection()); + // Wait for in-flight messages to reach their peers. SystemClock::Sleep(absl::Seconds(1)); previous_endpoint_channel->Close(DisconnectionReason::UPGRADED); diff --git a/cpp/core/internal/bwu_manager.h b/cpp/core/internal/bwu_manager.h index 0cc73c84..8c743e5e 100644 --- a/cpp/core/internal/bwu_manager.h +++ b/cpp/core/internal/bwu_manager.h @@ -81,7 +81,7 @@ class BwuManager : public EndpointManager::FrameProcessor { // @EndpointManagerReaderThread void OnEndpointDisconnect(ClientProxy* client_proxy, const std::string& endpoint_id, - CountDownLatch* barrier) override; + CountDownLatch barrier) override; void Shutdown(); private: diff --git a/cpp/core/internal/bwu_manager_test.cc b/cpp/core/internal/bwu_manager_test.cc index 00556c08..e3fb271a 100644 --- a/cpp/core/internal/bwu_manager_test.cc +++ b/cpp/core/internal/bwu_manager_test.cc @@ -6,8 +6,10 @@ #include "core/internal/endpoint_channel_manager.h" #include "core/internal/endpoint_manager.h" #include "core/internal/mediums/mediums.h" +#include "platform/public/system_clock.h" #include "gmock/gmock.h" #include "gtest/gtest.h" +#include "absl/time/time.h" namespace location { namespace nearby { @@ -19,6 +21,10 @@ TEST(BwuManagerTest, CanCreateInstance) { EndpointChannelManager ecm; EndpointManager em{&ecm}; BwuManager bwu_manager{mediums, em, ecm, {}, {}}; + + SystemClock::Sleep(absl::Seconds(3)); + + bwu_manager.Shutdown(); } TEST(BwuManagerTest, CanInitiateBwu) { @@ -31,6 +37,7 @@ TEST(BwuManagerTest, CanInitiateBwu) { // Method returns void, so we just verify we did not SEGFAULT while calling. bwu_manager.InitiateBwuForEndpoint(&client, endpoint_id); + SystemClock::Sleep(absl::Seconds(3)); bwu_manager.Shutdown(); } diff --git a/cpp/core/internal/endpoint_channel_manager.cc b/cpp/core/internal/endpoint_channel_manager.cc index a13e5f97..fa4bd69d 100644 --- a/cpp/core/internal/endpoint_channel_manager.cc +++ b/cpp/core/internal/endpoint_channel_manager.cc @@ -2,14 +2,22 @@ #include +#include "core/internal/offline_frames.h" +#include "proto/connections/offline_wire_formats.pb.h" #include "platform/public/logging.h" #include "platform/public/mutex.h" #include "platform/public/mutex_lock.h" +#include "platform/public/system_clock.h" +#include "absl/time/time.h" namespace location { namespace nearby { namespace connections { +namespace { +const absl::Duration kDataTransferDelay = absl::Milliseconds(500); +} + EndpointChannelManager::~EndpointChannelManager() { MutexLock lock(&mutex_); channel_state_.DestroyAll(); @@ -118,6 +126,11 @@ bool EndpointChannelManager::ChannelState::RemoveEndpoint( auto item = endpoints_.find(endpoint_id); if (item == endpoints_.end()) return false; item->second.disconnect_reason = reason; + auto channel = item->second.channel; + if (channel) { + channel->Write(parser::ForDisconnection()); + SystemClock::Sleep(kDataTransferDelay); + } endpoints_.erase(item); return true; } diff --git a/cpp/core/internal/endpoint_manager.cc b/cpp/core/internal/endpoint_manager.cc index 08779491..207fe930 100644 --- a/cpp/core/internal/endpoint_manager.cc +++ b/cpp/core/internal/endpoint_manager.cc @@ -473,7 +473,7 @@ void EndpointManager::WaitForEndpointDisconnectionProcessing( NEARBY_LOGS(INFO) << "processor=" << processor << "; type=" << item.first; if (processor) { valid++; - processor->OnEndpointDisconnect(client, endpoint_id, &barrier); + processor->OnEndpointDisconnect(client, endpoint_id, barrier); } else { barrier.CountDown(); } diff --git a/cpp/core/internal/endpoint_manager.h b/cpp/core/internal/endpoint_manager.h index bd40159d..298ac431 100644 --- a/cpp/core/internal/endpoint_manager.h +++ b/cpp/core/internal/endpoint_manager.h @@ -70,7 +70,7 @@ class EndpointManager { // @EndpointManagerThread virtual void OnEndpointDisconnect(ClientProxy* client, const std::string& endpoint_id, - CountDownLatch* barrier) = 0; + CountDownLatch barrier) = 0; }; explicit EndpointManager(EndpointChannelManager* manager); diff --git a/cpp/core/internal/endpoint_manager_test.cc b/cpp/core/internal/endpoint_manager_test.cc index f29f4e99..deac4931 100644 --- a/cpp/core/internal/endpoint_manager_test.cc +++ b/cpp/core/internal/endpoint_manager_test.cc @@ -71,7 +71,7 @@ class MockFrameProcessor : public EndpointManager::FrameProcessor { MOCK_METHOD(void, OnEndpointDisconnect, (ClientProxy * client, const std::string& endpoint_id, - CountDownLatch* barrier), + CountDownLatch barrier), (override)); }; diff --git a/cpp/core/internal/offline_frames.h b/cpp/core/internal/offline_frames.h index f5bdc01c..447dd2bc 100644 --- a/cpp/core/internal/offline_frames.h +++ b/cpp/core/internal/offline_frames.h @@ -59,6 +59,7 @@ ByteArray ForBwuLastWrite(); ByteArray ForBwuSafeToClose(); ByteArray ForKeepAlive(); +ByteArray ForDisconnection(); UpgradePathInfo::Medium MediumToUpgradePathInfoMedium(Medium medium); Medium UpgradePathInfoMediumToMedium(UpgradePathInfo::Medium medium); diff --git a/cpp/core/internal/payload_manager.cc b/cpp/core/internal/payload_manager.cc index 1e5c578c..64b35f5b 100644 --- a/cpp/core/internal/payload_manager.cc +++ b/cpp/core/internal/payload_manager.cc @@ -388,12 +388,12 @@ void PayloadManager::OnIncomingFrame( void PayloadManager::OnEndpointDisconnect(ClientProxy* client, const std::string& endpoint_id, - CountDownLatch* barrier) { + CountDownLatch barrier) { if (shutdown_.Get()) { - if (barrier) barrier->CountDown(); + barrier.CountDown(); return; } - RunOnStatusUpdateThread([this, client, endpoint_id, barrier]() { + RunOnStatusUpdateThread([this, client, endpoint_id, barrier]() mutable { // Iterate through all our payloads and look for payloads associated // with this endpoint. MutexLock lock(&mutex_); @@ -423,7 +423,7 @@ void PayloadManager::OnEndpointDisconnect(ClientProxy* client, client->OnPayloadProgress(endpoint_id, update); } - barrier->CountDown(); + barrier.CountDown(); }); } diff --git a/cpp/core/internal/payload_manager.h b/cpp/core/internal/payload_manager.h index d7033c60..161cb288 100644 --- a/cpp/core/internal/payload_manager.h +++ b/cpp/core/internal/payload_manager.h @@ -46,7 +46,7 @@ class PayloadManager : public EndpointManager::FrameProcessor { // @EndpointManagerThread void OnEndpointDisconnect(ClientProxy* client, const std::string& endpoint_id, - CountDownLatch* barrier) override; + CountDownLatch barrier) override; void DisconnectFromEndpointManager(); diff --git a/cpp/core/internal/simulation_user.h b/cpp/core/internal/simulation_user.h index 1f04789e..65c8f5b2 100644 --- a/cpp/core/internal/simulation_user.h +++ b/cpp/core/internal/simulation_user.h @@ -48,6 +48,7 @@ class SimulationUser { void Stop() { pm_.DisconnectFromEndpointManager(); mgr_.DisconnectFromEndpointManager(); + bwu_.Shutdown(); } // Calls PcpManager::StartAdvertising. diff --git a/cpp/core/internal/wifi_lan_bwu_handler.cc b/cpp/core/internal/wifi_lan_bwu_handler.cc new file mode 100644 index 00000000..3688fa9e --- /dev/null +++ b/cpp/core/internal/wifi_lan_bwu_handler.cc @@ -0,0 +1,115 @@ +#include "core/internal/wifi_lan_bwu_handler.h" + +#include +#include + +#include "core/internal/client_proxy.h" +#include "core/internal/mediums/utils.h" +#include "core/internal/offline_frames.h" +#include "core/internal/wifi_lan_endpoint_channel.h" +#include "platform/public/wifi_lan.h" +#include "absl/functional/bind_front.h" + +namespace location { +namespace nearby { +namespace connections { + +WifiLanBwuHandler::WifiLanBwuHandler(Mediums& mediums, + EndpointChannelManager& channel_manager, + BwuNotifications notifications) + : BaseBwuHandler(channel_manager, std::move(notifications)), + mediums_(mediums) {} + +// Called by BWU initiator. Set up WifiLan upgraded medium for this endpoint, +// and returns a upgrade path info (ip address, port) for remote party to +// perform discovery. +ByteArray WifiLanBwuHandler::InitializeUpgradedMediumForEndpoint( + ClientProxy* client, const std::string& service_id, + const std::string& endpoint_id) { + // Use wrapped service ID to avoid have the same ID with the one for + // startAdvertising. Otherwise, the listening request would be ignored because + // the medium already start accepting the connection because the client not + // stop the advertising yet. + std::string upgrade_service_id = Utils::WrapUpgradeServiceId(service_id); + + if (!wifi_lan_medium_.IsAcceptingConnections(upgrade_service_id)) { + if (!wifi_lan_medium_.StartAcceptingConnections( + upgrade_service_id, + { + .accepted_cb = absl::bind_front( + &WifiLanBwuHandler::OnIncomingWifiLanConnection, this, + client), + })) { + NEARBY_LOG(ERROR, + "WifiLanBwuHandler couldn't initiate the WifiLan upgrade for " + "endpoint %s because it failed to start listening for " + "incoming WifiLan connections.", + endpoint_id.c_str()); + return {}; + } + NEARBY_LOG(INFO, + "WifiLanBwuHandler successfully started listening for incoming " + "WifiLan connections while upgrading endpoint %s", + endpoint_id.c_str()); + } + + // cache service ID to revert + active_service_ids_.emplace(upgrade_service_id); + + // TODO(b/169303360): Implements wifiLanCredntials for wif_lan_medium to + // get ip_address and port. + std::string ip_addresss; + std::int32_t port = 0; + return parser::ForBwuWifiLanPathAvailable(ip_addresss, port); +} + +void WifiLanBwuHandler::Revert() { + for (const std::string& service_id : active_service_ids_) { + wifi_lan_medium_.StopAcceptingConnections(service_id); + } + active_service_ids_.clear(); + + NEARBY_LOG(INFO, "WifiLanBwuHandler successfully reverted all states."); +} + +// Called by BWU target. Retrieves a new medium info from incoming message, +// and establishes connection over WifiLan using this info. +std::unique_ptr +WifiLanBwuHandler::CreateUpgradedEndpointChannel( + ClientProxy* client, const std::string& service_id, + const std::string& endpoint_id, const UpgradePathInfo& upgrade_path_info) { + // TODO(b/169303360): Implements connect WifiLan over ip address and port. + WifiLanSocket socket; + + // Create a new WifiLanEndpointChannel. + auto channel = std::make_unique(service_id, socket); + if (channel == nullptr) { + socket.Close(); + NEARBY_LOG(ERROR, + "WifiLanBwuHandler failed to create new EndpointChannel for " + "outgoing socket %p, aborting upgrade.", + &socket.GetImpl()); + } + + return channel; +} + +// Accept Connection Callback. +void WifiLanBwuHandler::OnIncomingWifiLanConnection( + ClientProxy* client, WifiLanSocket socket, + const std::string& upgrade_service_id) { + std::string service_id = Utils::UnwrapUpgradeServiceId(upgrade_service_id); + auto channel = std::make_unique(service_id, socket); + auto wifi_lan_socket = + std::make_unique(service_id, socket); + std::unique_ptr connection( + new IncomingSocketConnection{std::move(wifi_lan_socket), + std::move(channel)}); + + bwu_notifications_.incoming_connection_cb(client, std::move(connection)); +} + +} // namespace connections +} // namespace nearby +} // namespace location + diff --git a/cpp/core/internal/wifi_lan_bwu_handler.h b/cpp/core/internal/wifi_lan_bwu_handler.h new file mode 100644 index 00000000..a724e939 --- /dev/null +++ b/cpp/core/internal/wifi_lan_bwu_handler.h @@ -0,0 +1,65 @@ +#ifndef CORE_INTERNAL_WIFI_LAN_BWU_HANDLER_H_ +#define CORE_INTERNAL_WIFI_LAN_BWU_HANDLER_H_ + +#include "core/internal/base_bwu_handler.h" +#include "core/internal/client_proxy.h" +#include "core/internal/endpoint_channel_manager.h" +#include "core/internal/mediums/mediums.h" + +namespace location { +namespace nearby { +namespace connections { + +// Defines the set of methods that need to be implemented to handle the +// per-Medium-specific operations needed to upgrade an EndpointChannel. +class WifiLanBwuHandler : public BaseBwuHandler { + public: + WifiLanBwuHandler(Mediums& mediums, EndpointChannelManager& channel_manager, + BwuNotifications notifications); + ~WifiLanBwuHandler() override = default; + + private: + ByteArray InitializeUpgradedMediumForEndpoint( + ClientProxy* client, const std::string& service_id, + const std::string& endpoint_id) override; + + void Revert() override; + + std::unique_ptr CreateUpgradedEndpointChannel( + ClientProxy* client, const std::string& service_id, + const std::string& endpoint_id, + const UpgradePathInfo& upgrade_path_info) override; + + Medium GetUpgradeMedium() const override { return Medium::WIFI_LAN; } + + void OnEndpointDisconnect(ClientProxy* client, + const std::string& endpoint_id) override {} + + void OnIncomingWifiLanConnection(ClientProxy* client, WifiLanSocket socket, + const std::string& upgrade_service_id); + + class WifiLanIncomingSocket : public BwuHandler::IncomingSocket { + public: + explicit WifiLanIncomingSocket(const std::string& name, + WifiLanSocket socket) + : name_(name), socket_(socket) {} + ~WifiLanIncomingSocket() override = default; + + std::string ToString() override { return name_; } + void Close() override { socket_.Close(); } + + private: + std::string name_; + WifiLanSocket socket_; + }; + + Mediums& mediums_; + WifiLan& wifi_lan_medium_{mediums_.GetWifiLan()}; + absl::flat_hash_set active_service_ids_; +}; + +} // namespace connections +} // namespace nearby +} // namespace location + +#endif // CORE_INTERNAL_WIFI_LAN_BWU_HANDLER_H_ diff --git a/cpp/core/internal/wifi_lan_service_info.cc b/cpp/core/internal/wifi_lan_service_info.cc index 7cda3436..4a796eb2 100644 --- a/cpp/core/internal/wifi_lan_service_info.cc +++ b/cpp/core/internal/wifi_lan_service_info.cc @@ -14,6 +14,11 @@ namespace location { namespace nearby { namespace connections { +// These definitions are necessary before C++17. +constexpr absl::string_view WifiLanServiceInfo::kKeyEndpointInfo; +constexpr std::uint32_t WifiLanServiceInfo::kServiceIdHashLength; +constexpr int WifiLanServiceInfo::kMaxEndpointInfoLength; + WifiLanServiceInfo::WifiLanServiceInfo(Version version, Pcp pcp, absl::string_view endpoint_id, const ByteArray& service_id_hash, diff --git a/cpp/platform/public/count_down_latch.h b/cpp/platform/public/count_down_latch.h index 921691e9..7550d15e 100644 --- a/cpp/platform/public/count_down_latch.h +++ b/cpp/platform/public/count_down_latch.h @@ -20,8 +20,8 @@ class CountDownLatch final { using Platform = api::ImplementationPlatform; explicit CountDownLatch(int count) : impl_(Platform::CreateCountDownLatch(count)) {} - CountDownLatch(CountDownLatch&&) = default; - CountDownLatch& operator=(CountDownLatch&&) = default; + CountDownLatch(const CountDownLatch&) = default; + CountDownLatch& operator=(const CountDownLatch&) = default; ~CountDownLatch() = default; Exception Await() { return impl_->Await(); } @@ -31,7 +31,7 @@ class CountDownLatch final { void CountDown() { impl_->CountDown(); } private: - std::unique_ptr impl_; + std::shared_ptr impl_; }; } // namespace nearby