From d3b3d07805e4626dbdbd82ec7d2d7c5010a7b902 Mon Sep 17 00:00:00 2001 From: hai007 Date: Fri, 23 Oct 2020 12:43:22 -0700 Subject: [PATCH] Roll forward to cl/338725629 Signed-off-by: hai007 --- cpp/core/core.h | 3 + cpp/core/internal/BUILD | 4 + cpp/core/internal/base_bwu_handler.h | 2 - cpp/core/internal/base_pcp_handler.cc | 8 +- cpp/core/internal/base_pcp_handler.h | 2 +- cpp/core/internal/base_pcp_handler_test.cc | 13 +- cpp/core/internal/bluetooth_bwu_handler.cc | 116 +++ cpp/core/internal/bluetooth_bwu_handler.h | 83 ++ cpp/core/internal/bwu_handler.h | 2 + cpp/core/internal/bwu_manager.cc | 59 +- cpp/core/internal/bwu_manager.cc.orig | 780 ------------------ cpp/core/internal/bwu_manager.h | 2 +- cpp/core/internal/bwu_manager_test.cc | 7 + cpp/core/internal/endpoint_channel_manager.cc | 13 + cpp/core/internal/endpoint_manager.cc | 2 +- cpp/core/internal/endpoint_manager.h | 2 +- cpp/core/internal/endpoint_manager_test.cc | 2 +- cpp/core/internal/mediums/webrtc.h.orig | 174 ---- cpp/core/internal/offline_frames.h | 1 + .../offline_service_controller_test.cc.orig | 374 --------- cpp/core/internal/payload_manager.cc | 8 +- cpp/core/internal/payload_manager.h | 2 +- cpp/core/internal/simulation_user.h | 1 + cpp/core/internal/wifi_lan_bwu_handler.cc | 115 +++ cpp/core/internal/wifi_lan_bwu_handler.h | 65 ++ cpp/core/internal/wifi_lan_service_info.cc | 5 + cpp/platform/impl/shared/BUILD.orig | 71 -- cpp/platform/public/BUILD.orig | 118 --- cpp/platform/public/count_down_latch.h | 6 +- 29 files changed, 478 insertions(+), 1562 deletions(-) create mode 100644 cpp/core/internal/bluetooth_bwu_handler.cc create mode 100644 cpp/core/internal/bluetooth_bwu_handler.h delete mode 100644 cpp/core/internal/bwu_manager.cc.orig delete mode 100644 cpp/core/internal/mediums/webrtc.h.orig delete mode 100644 cpp/core/internal/offline_service_controller_test.cc.orig create mode 100644 cpp/core/internal/wifi_lan_bwu_handler.cc create mode 100644 cpp/core/internal/wifi_lan_bwu_handler.h delete mode 100644 cpp/platform/impl/shared/BUILD.orig delete mode 100644 cpp/platform/public/BUILD.orig 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.cc.orig b/cpp/core/internal/bwu_manager.cc.orig deleted file mode 100644 index 8e46b3b2..00000000 --- a/cpp/core/internal/bwu_manager.cc.orig +++ /dev/null @@ -1,780 +0,0 @@ -#include "core/internal/bwu_manager.h" - -#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 "platform/base/byte_array.h" -#include "platform/public/count_down_latch.h" -#include "proto/connections_enums.pb.h" -#include "absl/functional/bind_front.h" -#include "absl/time/time.h" - -namespace location { -namespace nearby { -namespace connections { - -using ::location::nearby::proto::connections::ConnectionAttemptResult; -using ::location::nearby::proto::connections::DisconnectionReason; - -BwuManager::BwuManager( - Mediums& mediums, EndpointManager& endpoint_manager, - EndpointChannelManager& channel_manager, - absl::flat_hash_map> handlers, - Config config) - : config_(config), - mediums_(&mediums), - endpoint_manager_(&endpoint_manager), - channel_manager_(&channel_manager) { - if (config_.bandwidth_upgrade_retry_delay == absl::ZeroDuration()) { - config_.bandwidth_upgrade_retry_delay = absl::Seconds(5); - } - if (config_.bandwidth_upgrade_retry_max_delay == absl::ZeroDuration()) { - config_.bandwidth_upgrade_retry_max_delay = absl::Seconds(10); - } - if (config_.allow_upgrade_to.All(false)) { - config_.allow_upgrade_to.web_rtc = true; - } - if (!handlers.empty()) { - handlers_ = std::move(handlers); - } else { - InitBwuHandlers(); - } - - // Register the offline frame processor. - endpoint_manager_->RegisterFrameProcessor( - V1Frame::BANDWIDTH_UPGRADE_NEGOTIATION, this); -} - -void BwuManager::InitBwuHandlers() { - // Register the supported concrete BwuMedium implementations. - BwuHandler::BwuNotifications notifications{ - .incoming_connection_cb = - absl::bind_front(&BwuManager::OnIncomingConnection, this), - }; - 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() { - NEARBY_LOG(INFO, "Initiating shutdown of BwuManager."); - - 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(); - - NEARBY_LOG(INFO, "BwuHandler has shut down."); -} - -// This is the point on the Initiator side where the -// medium_ is set. -void BwuManager::InitiateBwuForEndpoint(ClientProxy* client, - const std::string& endpoint_id, - Medium new_medium) { - RunOnBwuManagerThread([this, client, endpoint_id, new_medium]() { - Medium proposed_medium = ChooseBestUpgradeMedium( - client->GetUpgradeMediums(endpoint_id).GetMediums(true)); - if (new_medium != Medium::UNKNOWN_MEDIUM) { - proposed_medium = new_medium; - } - auto* handler = SetCurrentBwuHandler(proposed_medium); - - if (!handler) return; - - if (in_progress_upgrades_.contains(endpoint_id)) { - return; - } - - auto channel = channel_manager_->GetChannelForEndpoint(endpoint_id); - - if (channel == nullptr) { - return; - } - - // Ignore requests where the medium we're upgrading to is the medium we're - // already connected over. This can happen now that Bluetooth is both an - // advertising medium and a potential bandwidth upgrade, and will continue - // to be possible as we add other new advertising mediums like mDNS (WiFi - // LAN). Very specifically, this happens now when a device uses P2P_CLUSTER, - // connects over Bluetooth, and is not connected to LAN. Bluetooth is the - // best medium, and we attempt to upgrade from Bluetooth to Bluetooth. - if (medium_ == channel->GetMedium()) { - return; - } - - std::string service_id = client->GetServiceId(); - ByteArray bytes = handler->InitializeUpgradedMediumForEndpoint( - client, service_id, endpoint_id); - - // Because we grab the endpointChannel first thing, it is possible the - // endpointChannel is stale by the time we attempt to write over it. - if (bytes.Empty()) { - NEARBY_LOG(ERROR, - "Couldn't complete the upgrade for endpoint " - "%s to %d because it failed to initialize the " - "BWU_NEGOTIATION.UPGRADE_PATH_AVAILABLE OfflineFrame.", - endpoint_id.c_str(), medium_); - UpgradePathInfo info; - info.set_medium(parser::MediumToUpgradePathInfoMedium(medium_)); - - ProcessUpgradeFailureEvent(client, endpoint_id, info); - return; - } - if (!channel->Write(bytes).Ok()) { - NEARBY_LOG(ERROR, - "Couldn't complete the upgrade for endpoint %s to %d because " - "it failed to write the " - "BWU_NEGOTIATION.UPGRADE_PATH_AVAILABLE OfflineFrame.", - endpoint_id.c_str(), medium_); - return; - } - - NEARBY_LOG(INFO, - "Successfully wrote the BWU_NEGOTIATION.UPGRADE_PATH_AVAILABLE " - "OfflineFrame while upgrading endpoint %s to %d.", - endpoint_id.c_str(), medium_); - in_progress_upgrades_.emplace(endpoint_id, client); - }); -} - -void BwuManager::OnIncomingFrame(OfflineFrame& frame, - const std::string& endpoint_id, - ClientProxy* client, Medium medium) { - if (parser::GetFrameType(frame) != V1Frame::BANDWIDTH_UPGRADE_NEGOTIATION) - return; - auto bwu_frame = frame.v1().bandwidth_upgrade_negotiation(); - CountDownLatch latch(1); - RunOnBwuManagerThread([this, client, endpoint_id, &bwu_frame, &latch]() { - OnBwuNegotiationFrame(client, bwu_frame, endpoint_id); - latch.CountDown(); - }); - latch.Await(); -} - -void BwuManager::OnEndpointDisconnect(ClientProxy* client, - const std::string& endpoint_id, - CountDownLatch* barrier) { - RunOnBwuManagerThread([this, client, endpoint_id, barrier]() { - if (medium_ == Medium::UNKNOWN_MEDIUM) { - barrier->CountDown(); - return; - } - - if (handler_) { - handler_->OnEndpointDisconnect(client, endpoint_id); - } - - auto item = previous_endpoint_channels_.extract(endpoint_id); - - if (!item.empty()) { - auto old_channel = item.mapped(); - if (old_channel != nullptr) { - old_channel->Close(DisconnectionReason::SHUTDOWN); - } - } - in_progress_upgrades_.erase(endpoint_id); - CancelRetryUpgradeAlarm(endpoint_id); - - successfully_upgraded_endpoints_.erase(endpoint_id); - - // If this was our very last endpoint: - // - // a) revert all the changes for currentBwuMedium. - // b) reset currentBwuMedium. - if (channel_manager_->GetConnectedEndpointsCount() <= 1) { - Revert(); - } - barrier->CountDown(); - }); -} - -BwuHandler* BwuManager::SetCurrentBwuHandler(Medium medium) { - handler_ = nullptr; - medium_ = medium; - if (medium != Medium::UNKNOWN_MEDIUM) { - auto item = handlers_.find(medium); - if (item != handlers_.end()) { - handler_ = item->second.get(); - } - } - return handler_; -} - -void BwuManager::Revert() { - if (handler_) { - handler_->Revert(); - medium_ = Medium::UNKNOWN_MEDIUM; - handler_ = nullptr; - } -} - -void BwuManager::OnBwuNegotiationFrame(ClientProxy* client, - const BwuNegotiationFrame& frame, - const string& endpoint_id) { - switch (frame.event_type()) { - case BwuNegotiationFrame::UPGRADE_PATH_AVAILABLE: - ProcessBwuPathAvailableEvent(client, endpoint_id, - frame.upgrade_path_info()); - break; - case BwuNegotiationFrame::UPGRADE_FAILURE: - ProcessUpgradeFailureEvent(client, endpoint_id, - frame.upgrade_path_info()); - break; - case BwuNegotiationFrame::LAST_WRITE_TO_PRIOR_CHANNEL: - ProcessLastWriteToPriorChannelEvent(client, endpoint_id); - break; - case BwuNegotiationFrame::SAFE_TO_CLOSE_PRIOR_CHANNEL: - ProcessSafeToClosePriorChannelEvent(client, endpoint_id); - break; - default: - break; - } -} - -void BwuManager::OnIncomingConnection( - ClientProxy* client, - std::unique_ptr mutable_connection) { - std::shared_ptr connection( - mutable_connection.release()); - RunOnBwuManagerThread([this, client, connection]() { - EndpointChannel* channel = connection->channel.get(); - if (channel == nullptr) { - connection->socket->Close(); - return; - } - - ClientIntroduction introduction; - if (!ReadClientIntroductionFrame(channel, introduction)) { - // This was never a fully EstablishedConnection, no need to provide a - // closure reason. - channel->Close(); - return; - } - - const std::string& endpoint_id = introduction.endpoint_id(); - auto item = in_progress_upgrades_.extract(endpoint_id); - if (item.empty()) return; - ClientProxy* mapped_client = item.mapped(); - CancelRetryUpgradeAlarm(endpoint_id); - if (mapped_client == nullptr) { - // This was never a fully EstablishedConnection, no need to provide a - // closure reason. - channel->Close(); - return; - } - - CHECK(client == mapped_client); - - // Use the introductory client information sent over to run the upgrade - // protocol. - RunUpgradeProtocol(mapped_client, endpoint_id, - std::move(connection->channel)); - }); -} - -void BwuManager::RunOnBwuManagerThread(Runnable runnable) { - serial_executor_.Execute(std::move(runnable)); -} - -void BwuManager::RunUpgradeProtocol( - ClientProxy* client, const std::string& endpoint_id, - std::unique_ptr new_channel) { - // First, register this new EndpointChannel as *the* EndpointChannel to use - // for this endpoint here onwards. NOTE: We pause this new EndpointChannel - // until we've completely drained the old EndpointChannel to avoid out of - // order reads on the other side. This is a consequence of using the same - // UKEY2 context for both the previous and new EndpointChannels. UKEY2 uses - // sequence numbers for writes and reads, and simultaneously sending Payloads - // on the new channel and control messages on the old channel cause the other - // side to read messages out of sequence - new_channel->Pause(); - auto old_channel = channel_manager_->GetChannelForEndpoint(endpoint_id); - if (!old_channel) return; - channel_manager_->ReplaceChannelForEndpoint(client, endpoint_id, - std::move(new_channel)); - - // Next, initiate a clean shutdown for the previous EndpointChannel used for - // this endpoint by telling the remote device that it will not receive any - // more writes over that EndpointChannel. - if (!old_channel->Write(parser::ForBwuLastWrite()).Ok()) { - return; - } - - // The remainder of this clean shutdown for the previous EndpointChannel will - // continue when we receive a corresponding - // BANDWIDTH_UPGRADE_NEGOTIATION.LAST_WRITE_TO_PRIOR_CHANNEL OfflineFrame from - // the remote device, so for now, just store that previous EndpointChannel. - previous_endpoint_channels_.emplace(endpoint_id, old_channel); - - // If we already read LAST_WRITE on the old endpoint channel, then we can - // safely close it now. - auto item = successfully_upgraded_endpoints_.extract(endpoint_id); - if (!item.empty()) { - ProcessLastWriteToPriorChannelEvent(client, endpoint_id); - } -} - -// Outgoing BWU session. -void BwuManager::ProcessBwuPathAvailableEvent( - ClientProxy* client, const string& endpoint_id, - const UpgradePathInfo& upgrade_path_info) { - Medium medium = - parser::UpgradePathInfoMediumToMedium(upgrade_path_info.medium()); - if (medium_ == Medium::UNKNOWN_MEDIUM) { - SetCurrentBwuHandler(medium); - } - // Check for the correct medium so we don't process an incorrect OfflineFrame. - if (medium != medium_) { - RunUpgradeFailedProtocol(client, endpoint_id, upgrade_path_info); - return; - } - - auto channel = ProcessBwuPathAvailableEventInternal(client, endpoint_id, - upgrade_path_info); - ConnectionAttemptResult connectionAttemptResult; - if (channel != nullptr) { - connectionAttemptResult = ConnectionAttemptResult::RESULT_SUCCESS; - } else { - connectionAttemptResult = ConnectionAttemptResult::RESULT_ERROR; - } - - if (channel == nullptr) { - RunUpgradeFailedProtocol(client, endpoint_id, upgrade_path_info); - return; - } - - RunUpgradeProtocol(client, endpoint_id, std::move(channel)); -} - -std::unique_ptr -BwuManager::ProcessBwuPathAvailableEventInternal( - ClientProxy* client, const string& endpoint_id, - const UpgradePathInfo& upgrade_path_info) { - std::unique_ptr channel = - handler_->CreateUpgradedEndpointChannel(client, client->GetServiceId(), - endpoint_id, upgrade_path_info); - if (!channel) { - return nullptr; - } - - // Write the requisite BANDWIDTH_UPGRADE_NEGOTIATION.CLIENT_INTRODUCTION as - // the first OfflineFrame on this new EndpointChannel. - if (!channel->Write(parser::ForBwuIntroduction(client->GetLocalEndpointId())) - .Ok()) { - // This was never a fully EstablishedConnection, no need to provide a - // closure reason. - channel->Close(); - - NEARBY_LOG( - ERROR, - "Failed to write BWU_NEGOTIATION.CLIENT_INTRODUCTION OfflineFrame to " - "newly-created EndpointChannel %s, aborting upgrade.", - channel->GetName().c_str()); - - return {}; - } - - NEARBY_LOG( - INFO, - "Successfully wrote BWU_NEGOTIATION.CLIENT_INTRODUCTION OfflineFrame to " - "newly-created EndpointChannel %s while upgrading endpoint %s.", - channel->GetName().c_str(), endpoint_id.c_str()); - - // Set the AnalyticsRecorder so that the future closure of this - // EndpointChannel will be recorded. - return channel; -} - -void BwuManager::RunUpgradeFailedProtocol( - ClientProxy* client, const std::string& endpoint_id, - const UpgradePathInfo& upgrade_path_info) { - // We attempted to connect to the new medium that the remote device has set up - // for us but we failed. We need to let the remote device know so that they - // can pick another medium for us to try. - std::shared_ptr channel = - channel_manager_->GetChannelForEndpoint(endpoint_id); - if (!channel) { - NEARBY_LOG(ERROR, - "Couldn't find a previous EndpointChannel for %s " - "when sending an upgrade failure frame, short-circuiting the " - "upgrade protocol.", - endpoint_id.c_str()); - return; - } - - // Report UPGRADE_FAILURE to the remote device. - if (!channel->Write(parser::ForBwuFailure(upgrade_path_info)).Ok()) { - channel->Close(DisconnectionReason::IO_ERROR); - - NEARBY_LOG( - ERROR, - "Failed to write BANDWIDTH_UPGRADE_NEGOTIATION.UPGRADE_FAILURE " - "OfflineFrame to endpoint %s, short-circuiting the upgrade protocol.", - endpoint_id.c_str()); - return; - } - - // And lastly, clean up our currentBwuMedium since we failed to - // utilize it anyways. - if (medium_ != Medium::UNKNOWN_MEDIUM) { - Revert(); - } -} - -bool BwuManager::ReadClientIntroductionFrame(EndpointChannel* channel, - ClientIntroduction& introduction) { - auto data = channel->Read(); - if (!data.ok()) return false; - auto transfer(parser::FromBytes(data.result())); - if (!transfer.ok()) return false; - OfflineFrame frame = transfer.result(); - if (!frame.has_v1() || !frame.v1().has_bandwidth_upgrade_negotiation()) - return false; - const auto& frame_intro = - frame.v1().bandwidth_upgrade_negotiation().client_introduction(); - introduction = frame_intro; - return true; -} - -void BwuManager::ProcessLastWriteToPriorChannelEvent( - ClientProxy* client, const std::string& endpoint_id) { - // By this point in the upgrade protocol, there is the guarantee that both - // involved endpoints have registered a new EndpointChannel with the - // EndpointChannelManager as the official channel for communication; given - // the way communication is structured in the EndpointManager, this means - // that all new writes are happening over that new EndpointChannel, but - // reads are still happening over this prior EndpointChannel (to avoid data - // loss). But now that we've received this definitive final write over that - // prior EndpointChannel, we can let the remote device that they can safely - // close their end of this now-dormant EndpointChannel. - EndpointChannel* previous_endpoint_channel = - previous_endpoint_channels_[endpoint_id].get(); - if (!previous_endpoint_channel) { - NEARBY_LOG( - ERROR, - "Received a BWU_NEGOTIATION.LAST_WRITE_TO_PRIOR_CHANNEL OfflineFrame " - "for unknown endpoint %s, can't complete the upgrade protocol.", - endpoint_id.c_str()); - - successfully_upgraded_endpoints_.emplace(endpoint_id); - return; - } - - if (!previous_endpoint_channel->Write(parser::ForBwuSafeToClose()).Ok()) { - previous_endpoint_channel->Close(DisconnectionReason::IO_ERROR); - // Remove this prior EndpointChannel from previous_endpoint_channels to - // avoid leaks. - previous_endpoint_channels_.erase(endpoint_id); - - NEARBY_LOG( - ERROR, - "Failed to write BWU_NEGOTIATION.SAFE_TO_CLOSE_PRIOR_CHANNEL " - "OfflineFrame to endpoint %s, short-circuiting the upgrade protocol.", - endpoint_id.c_str()); - return; - } - - // The upgrade protocol's clean shutdown of the prior EndpointChannel will - // conclude when we receive a corresponding - // BANDWIDTH_UPGRADE_NEGOTIATION.SAFE_TO_CLOSE_PRIOR_CHANNEL OfflineFrame - // from the remote device. -} - -void BwuManager::ProcessSafeToClosePriorChannelEvent( - ClientProxy* client, const std::string& endpoint_id) { - // By this point in the upgrade protocol, there's no more writes happening - // over the prior EndpointChannel, and the remote device has given us the - // go-ahead to close this EndpointChannel [1], so we can safely close it - // (and depend on the EndpointManager querying the EndpointChannelManager to - // start reading from the new EndpointChannel). - // - // [1] Which also implies that they've received our - // BANDWIDTH_UPGRADE_NEGOTIATION.LAST_WRITE_TO_PRIOR_CHANNEL OfflineFrame), - // so there can be no data loss, regardless of whether the EndpointChannel - // allows reads of queued, unread data after the EndpointChannel has been - // closed from the other end (as is the case with conventional TCP sockets) - // or not (as is the case with Android's Bluetooth sockets, where closing - // instantly throws an IOException on the remote device). - auto item = previous_endpoint_channels_.extract(endpoint_id); - auto& previous_endpoint_channel = item.mapped(); - if (previous_endpoint_channel == nullptr) { - NEARBY_LOG( - ERROR, - "Received a BWU_NEGOTIATION.SAFE_TO_CLOSE_PRIOR_CHANNEL OfflineFrame " - "for unknown endpoint %s, can't complete the upgrade protocol.", - endpoint_id.c_str()); - return; - } - - NEARBY_LOG(INFO, - "BwuManager successfully received a " - "BWU_NEGOTIATION.SAFE_TO_CLOSE_PRIOR_CHANNEL OfflineFrame while " - "trying to upgrade endpoint %s.", - endpoint_id.c_str()); - - // Wait for in-flight messages to reach their peers. - SystemClock::Sleep(absl::Seconds(1)); - previous_endpoint_channel->Close(DisconnectionReason::UPGRADED); - - // Now that the old channel has been drained, we can unpause the new channel - std::shared_ptr channel = - channel_manager_->GetChannelForEndpoint(endpoint_id); - - if (!channel) { - NEARBY_LOG(ERROR, - "Attempted to resume the current EndpointChannel with endpoint " - "%s, but none was found", - endpoint_id.c_str()); - return; - } - - channel->Resume(); - - // Report the success to the client - client->OnBandwidthChanged(endpoint_id, channel->GetMedium()); -} - -void BwuManager::ProcessUpgradeFailureEvent( - ClientProxy* client, const std::string& endpoint_id, - const UpgradePathInfo& upgrade_info) { - // The remote device failed to upgrade to the new medium we set up for them. - // That's alright! We'll just try the next available medium (if there is - // one). - in_progress_upgrades_.erase(endpoint_id); - - // The first thing we have to do is to replace our - // currentBwuMedium with the next best upgrade medium we share - // with the remote device. The catch is that we can only do this if we only - // have one connected endpoint. Otherwise, we'll end up disrupting our other - // connected peers. - if (channel_manager_->GetConnectedEndpointsCount() > 1) { - // We can't change the currentBwuMedium, so there are no more - // upgrade attempts for this endpoint. Sorry. - NEARBY_LOG( - ERROR, - "Failed to attempt a new bandwidth upgrade for endpoint %s because we " - "have other connected endpoints and can't try a new upgrade medium.", - endpoint_id.c_str()); - return; - } - - // Revert the existing upgrade medium for now. - if (medium_ != Medium::UNKNOWN_MEDIUM) { - Revert(); - } - - // Loop through the ordered list of upgrade mediums. One by one, remove the - // top element until we get to the medium we last attempted to upgrade to. - // The remainder of the list will contain the mediums we haven't attempted - // yet. - Medium last = parser::UpgradePathInfoMediumToMedium(upgrade_info.medium()); - std::vector all_possible_mediums = - client->GetUpgradeMediums(endpoint_id).GetMediums(true); - std::vector untried_mediums(all_possible_mediums); - for (Medium medium : all_possible_mediums) { - untried_mediums.erase(untried_mediums.begin()); - if (medium == last) { - break; - } - } - - RetryUpgradeMediums(client, endpoint_id, untried_mediums); -} - -void BwuManager::RetryUpgradeMediums(ClientProxy* client, - const std::string& endpoint_id, - std::vector upgrade_mediums) { - Medium next_medium = ChooseBestUpgradeMedium(upgrade_mediums); - - // If current medium is not WiFi and we have not succeeded with upgrading - // yet, retry upgrade. - Medium current_medium = GetEndpointMedium(endpoint_id); - if (current_medium != Medium::WIFI_LAN && - (next_medium == current_medium || next_medium == Medium::UNKNOWN_MEDIUM || - upgrade_mediums.empty())) { - RetryUpgradesAfterDelay(client, endpoint_id); - return; - } - - // Attempt to set the new upgrade medium. - if (!SetCurrentBwuHandler(next_medium)) { - NEARBY_LOG( - INFO, - "BwuManager failed to attempt a new bandwidth upgrade for endpoint %s " - "because we couldn't set a new bandwidth upgrade medium.", - endpoint_id.c_str()); - return; - } - - // Now that we've successfully picked a new upgrade medium to try, - // re-initiate the bandwidth upgrade. - NEARBY_LOG(INFO, - "BwuManager is attempting to upgrade endpoint %s again with a new " - " bandwidth upgrade medium.", - endpoint_id.c_str()); - InitiateBwuForEndpoint(client, endpoint_id); -} - -std::vector BwuManager::StripOutUnavailableMediums( - const std::vector& mediums) { - std::vector available_mediums; - for (Medium m : mediums) { - bool available = false; - switch (m) { - case Medium::WIFI_LAN: - available = mediums_->GetWifiLan().IsAvailable(); - break; - case Medium::WEB_RTC: - available = mediums_->GetWebRtc().IsAvailable(); - break; - case Medium::BLUETOOTH: - available = mediums_->GetBluetoothClassic().IsAvailable(); - break; - default: - break; - } - if (available) { - available_mediums.push_back(m); - } - } - return available_mediums; -} - -// Returns the optimal medium supported by both devices. -// Each medium in the passed in list is checked for its availability with the -// medium_manager_ to ensure that the chosen upgrade medium is supported and -// available locally before continuing the upgrade. Once we pick a medium, all -// future connections will use it too. eg. If we chose Wifi LAN, we'll attempt -// to upgrade the 2nd, 3rd, etc remote endpoints with Wifi LAN even if they're -// on a different network (or had a better medium). This is a quick and easy -// way to prevent mediums, like Wifi Hotspot, from interfering with active -// connections (although it's suboptimal for bandwidth throughput). When all -// endpoints disconnect, we reset the bandwidth upgrade medium. -Medium BwuManager::ChooseBestUpgradeMedium(const std::vector& mediums) { - auto available_mediums = StripOutUnavailableMediums(mediums); - if (medium_ == Medium::UNKNOWN_MEDIUM) { - if (!available_mediums.empty()) { - // Case 1: This is our first time upgrading, and we have at least one - // supported medium to choose from. Return the first medium in the list, - // since they are ordered by preference. - return available_mediums[0]; - } - // Case 2: This is our first time upgrading, but there are no available - // upgrade mediums. Fall through to returning UNKNOWN_MEDIUM at the - // bottom. - NEARBY_LOG( - INFO, - "Current upgrade medium is unset, but there are no common supported " - "upgrade mediums."); - } else { - // Case 3: We have already upgraded, and there is a list of supported - // mediums to check against. Return the current upgrade medium if it's in - // the supported list. - if (std::find(available_mediums.begin(), available_mediums.end(), - medium_) != available_mediums.end()) { - return medium_; - } - // Case 4: We have already upgraded, but the current medium is not - // supported by the remote endpoint (it's not in the list, or the list is - // empty). Fall through and return Medium.UNKNOWN_MEDIUM because we cannot - // continue with the current upgrade medium, and we are not allowed to - // switch. - NEARBY_LOG( - INFO, - "Current upgrade medium %d is not supported by the remote endpoint", - medium_); - } - - return Medium::UNKNOWN_MEDIUM; -} - -void BwuManager::RetryUpgradesAfterDelay(ClientProxy* client, - const std::string& endpoint_id) { - absl::Duration delay = CalculateNextRetryDelay(endpoint_id); - CancelRetryUpgradeAlarm(endpoint_id); - CancelableAlarm alarm( - "BWU alarm", - [this, client, endpoint_id]() { - RunOnBwuManagerThread([this, client, endpoint_id]() { - if (!client->IsConnectedToEndpoint(endpoint_id)) { - return; - } - RetryUpgradeMediums( - client, endpoint_id, - client->GetUpgradeMediums(endpoint_id).GetMediums(true)); - }); - }, - delay, &alarm_executor_); - - retry_upgrade_alarms_.emplace(endpoint_id, - std::make_pair(std::move(alarm), delay)); - NEARBY_LOGS(INFO) << "Retry bandwidth upgrade after " << delay; -} - -absl::Duration BwuManager::CalculateNextRetryDelay( - const std::string& endpoint_id) { - auto item = retry_upgrade_alarms_.find(endpoint_id); - auto initial_delay = config_.bandwidth_upgrade_retry_delay; - auto delay = item == retry_upgrade_alarms_.end() - ? initial_delay - : item->second.second + initial_delay; - return std::min(delay, config_.bandwidth_upgrade_retry_max_delay); -} - -void BwuManager::CancelRetryUpgradeAlarm(const std::string& endpoint_id) { - auto item = retry_upgrade_alarms_.extract(endpoint_id); - if (item.empty()) return; - auto& pair = item.mapped(); - pair.first.Cancel(); -} - -void BwuManager::CancelAllRetryUpgradeAlarms() { - for (const auto& item : retry_upgrade_alarms_) { - const std::string& endpoint_id = item.first; - CancelRetryUpgradeAlarm(endpoint_id); - } -} - -Medium BwuManager::GetEndpointMedium(const std::string& endpoint_id) { - auto channel = channel_manager_->GetChannelForEndpoint(endpoint_id); - return channel == nullptr ? Medium::UNKNOWN_MEDIUM : channel->GetMedium(); -} - -} // namespace connections -} // namespace nearby -} // namespace location 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/mediums/webrtc.h.orig b/cpp/core/internal/mediums/webrtc.h.orig deleted file mode 100644 index a2456020..00000000 --- a/cpp/core/internal/mediums/webrtc.h.orig +++ /dev/null @@ -1,174 +0,0 @@ -#ifndef CORE_V2_INTERNAL_MEDIUMS_WEBRTC_H_ -#define CORE_V2_INTERNAL_MEDIUMS_WEBRTC_H_ - -#include -#include - -#include "core_v2/internal/mediums/webrtc/connection_flow.h" -#include "core_v2/internal/mediums/webrtc/data_channel_listener.h" -#include "core_v2/internal/mediums/webrtc/local_ice_candidate_listener.h" -#include "core_v2/internal/mediums/webrtc/peer_id.h" -#include "core_v2/internal/mediums/webrtc/webrtc_socket.h" -#include "core_v2/internal/mediums/webrtc/webrtc_socket_wrapper.h" -#include "proto/connections/offline_wire_formats.pb.h" -#include "proto/connections/offline_wire_formats.pb.h" -#include "platform_v2/base/byte_array.h" -#include "platform_v2/base/listeners.h" -#include "platform_v2/base/runnable.h" -#include "platform_v2/public/atomic_boolean.h" -#include "platform_v2/public/cancelable_alarm.h" -#include "platform_v2/public/future.h" -#include "platform_v2/public/mutex.h" -#include "platform_v2/public/scheduled_executor.h" -#include "platform_v2/public/single_thread_executor.h" -#include "platform_v2/public/webrtc.h" -#include "location/nearby/mediums/proto/web_rtc_signaling_frames.pb.h" -#include "webrtc/api/data_channel_interface.h" -#include "webrtc/api/jsep.h" -#include "webrtc/api/scoped_refptr.h" - -namespace location { -namespace nearby { -namespace connections { -namespace mediums { - -// Callback that is invoked when a new connection is accepted. -struct AcceptedConnectionCallback { - std::function accepted_cb = - DefaultCallback(); -}; - -// Entry point for connecting a data channel between two devices via WebRtc. -class WebRtc { - public: - WebRtc(); - ~WebRtc(); - - // Returns if WebRtc is available as a medium for nearby to transport data. - // Runs on @MainThread. - bool IsAvailable(); - - // Returns if the device is ready to accept connections from remote devices. - // Runs on @MainThread. - bool IsAcceptingConnections() ABSL_LOCKS_EXCLUDED(mutex_); - - // Prepares the device to accept incoming WebRtc connections. Returns a - // boolean value indicating if the device has started accepting connections. - // Runs on @MainThread. - bool StartAcceptingConnections(const PeerId& self_id, - const LocationHint& location_hint, - AcceptedConnectionCallback callback) - ABSL_LOCKS_EXCLUDED(mutex_); - - // Prevents device from accepting future connections until - // StartAcceptingConnections() is called. - // Runs on @MainThread. - void StopAcceptingConnections() ABSL_LOCKS_EXCLUDED(mutex_); - - // Initiates a WebRtc connection with peer device identified by |peer_id|. - // Runs on @MainThread. - WebRtcSocketWrapper Connect(const PeerId& peer_id, - const LocationHint& location_hint) - ABSL_LOCKS_EXCLUDED(mutex_); - - private: - enum class Role { - kNone = 0, - kOfferer = 1, - kAnswerer = 2, - }; - - bool InitWebRtcFlow(Role role, const PeerId& self_id, - const LocationHint& location_hint) - ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - - Future ListenForWebRtcSocketFuture( - Future> - data_channel_future, - AcceptedConnectionCallback callback); - - WebRtcSocketWrapper CreateWebRtcSocketWrapper( - rtc::scoped_refptr data_channel); - - LocalIceCandidateListener GetLocalIceCandidateListener(); - void OnLocalIceCandidate( - const webrtc::IceCandidateInterface* local_ice_candidate); - - DataChannelListener GetDataChannelListener(); - void OnDataChannelClosed(); - void OnDataChannelMessageReceived(const ByteArray& message); - void OnDataChannelBufferedAmountChanged(); - - // Runs on @MainThread and |single_thread_executor_|. - bool SetLocalSessionDescription(SessionDescriptionWrapper sdp) - ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - - // Runs on |single_thread_executor_|. - bool IsSignaling() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - - // Runs on |single_thread_executor_|. - void ProcessSignalingMessage(const ByteArray& message) - ABSL_LOCKS_EXCLUDED(mutex_); - - // Runs on |single_thread_executor_|. - void SendOfferAndIceCandidatesToPeer() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - - // Runs on |single_thread_executor_|. - void SendAnswerToPeer() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - - // Runs on @MainThread and |single_thread_executor_|. - void LogAndDisconnect(const std::string& error_message) - ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - - // Runs on @MainThread. - void Disconnect() ABSL_LOCKS_EXCLUDED(mutex_); - - // Runs on @MainThread and |single_thread_executor_|. - void DisconnectLocked() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - - void LogAndShutdownSignaling(const std::string& error_message) - ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - - // Runs on @MainThread and |single_thread_executor_|. - void ShutdownSignaling() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - - // Runs on @MainThread and |single_thread_executor_|. - void ShutdownWebRtcSocket() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - - // Runs on @MainThread and |single_thread_executor_|. - void ShutdownIceCandidateCollection(); - - void OffloadFromSignalingThread(Runnable runnable); - - // Runs on |restart_receive_messages_executor_|. - void RestartReceiveMessages(const LocationHint& location_hint) - ABSL_LOCKS_EXCLUDED(mutex_); - - Mutex mutex_; - - Role role_ ABSL_GUARDED_BY(mutex_) = Role::kNone; - PeerId self_id_ ABSL_GUARDED_BY(mutex_); - PeerId peer_id_ ABSL_GUARDED_BY(mutex_); - ByteArray pending_local_offer_ ABSL_GUARDED_BY(mutex_); - std::vector<::location::nearby::mediums::IceCandidate> - pending_local_ice_candidates_ ABSL_GUARDED_BY(mutex_); - - WebRtcMedium medium_; - std::unique_ptr connection_flow_; - std::unique_ptr signaling_messenger_ - ABSL_GUARDED_BY(mutex_); - WebRtcSocketWrapper socket_ ABSL_GUARDED_BY(mutex_); - - SingleThreadExecutor single_thread_executor_; - - // Restarts the signaling messenger for receiving messages. - ScheduledExecutor restart_receive_messages_executor_; - CancelableAlarm restart_receive_messages_alarm_; -}; - -} // namespace mediums -} // namespace connections -} // namespace nearby -} // namespace location - -#endif // CORE_V2_INTERNAL_MEDIUMS_WEBRTC_H_ 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/offline_service_controller_test.cc.orig b/cpp/core/internal/offline_service_controller_test.cc.orig deleted file mode 100644 index da1d51bd..00000000 --- a/cpp/core/internal/offline_service_controller_test.cc.orig +++ /dev/null @@ -1,374 +0,0 @@ -#include "core_v2/internal/offline_service_controller.h" - -#include - -#include "core_v2/internal/offline_simulation_user.h" -#include "platform_v2/base/medium_environment.h" -#include "platform_v2/base/output_stream.h" -#include "platform_v2/public/count_down_latch.h" -#include "platform_v2/public/logging.h" -#include "platform_v2/public/pipe.h" -#include "platform_v2/public/system_clock.h" -#include "gmock/gmock.h" -#include "gtest/gtest.h" - -namespace location { -namespace nearby { -namespace connections { -namespace { - -using ::testing::Eq; - -constexpr std::array kFakeMacAddress = {'a', 'b', 'c', 'd', 'e', 'f'}; -constexpr absl::string_view kServiceId = "service-id"; -constexpr absl::string_view kDeviceA = "device-a"; -constexpr absl::string_view kDeviceB = "device-b"; -constexpr absl::string_view kMessage = "message"; -constexpr absl::Duration kProgressTimeout = absl::Milliseconds(1000); -constexpr absl::Duration kDefaultTimeout = absl::Milliseconds(1000); -constexpr absl::Duration kDisconnectTimeout = absl::Milliseconds(15000); - -constexpr BooleanMediumSelector kTestCases[] = { - BooleanMediumSelector{ - .bluetooth = true, - }, - BooleanMediumSelector{ - .wifi_lan = true, - }, - BooleanMediumSelector{ - .bluetooth = true, - .wifi_lan = true, - }, -}; - -class OfflineServiceControllerTest - : public ::testing::TestWithParam { - protected: - OfflineServiceControllerTest() { env_.Stop(); } - - bool SetupConnection(OfflineSimulationUser& user_a, - OfflineSimulationUser& user_b) { - user_a.StartAdvertising(std::string(kServiceId), &connect_latch_); - user_b.StartDiscovery(std::string(kServiceId), &discover_latch_); - EXPECT_TRUE(discover_latch_.Await(kDefaultTimeout).result()); - EXPECT_EQ(user_b.GetDiscovered().service_id, kServiceId); - EXPECT_EQ(user_b.GetDiscovered().endpoint_info, user_a.GetInfo()); - EXPECT_FALSE(user_b.GetDiscovered().endpoint_id.empty()); - NEARBY_LOG(INFO, "EP-B: [discovered] %s", - user_b.GetDiscovered().endpoint_id.c_str()); - user_b.RequestConnection(&connect_latch_); - EXPECT_TRUE(connect_latch_.Await(kDefaultTimeout).result()); - EXPECT_FALSE(user_a.GetDiscovered().endpoint_id.empty()); - NEARBY_LOG(INFO, "EP-A: [discovered] %s", - user_a.GetDiscovered().endpoint_id.c_str()); - NEARBY_LOG(INFO, "Both users discovered their peers."); - user_a.AcceptConnection(&accept_latch_); - user_b.AcceptConnection(&accept_latch_); - EXPECT_TRUE(accept_latch_.Await(kDefaultTimeout).result()); - NEARBY_LOG(INFO, "Both users reached connected state."); - return user_a.IsConnected() && user_b.IsConnected(); - } - - CountDownLatch discover_latch_{1}; - CountDownLatch lost_latch_{1}; - CountDownLatch connect_latch_{2}; - CountDownLatch accept_latch_{2}; - CountDownLatch payload_latch_{1}; - MediumEnvironment& env_ = MediumEnvironment::Instance(); -}; - -TEST_P(OfflineServiceControllerTest, CanCreateOne) { - env_.Start(); - OfflineSimulationUser user_a(kDeviceA, GetParam()); - env_.Stop(); -} - -TEST_P(OfflineServiceControllerTest, CanCreateMany) { - env_.Start(); - OfflineSimulationUser user_a(kDeviceA, GetParam()); - OfflineSimulationUser user_b(kDeviceB, GetParam()); - env_.Stop(); -} - -TEST_P(OfflineServiceControllerTest, CanStartAdvertising) { - env_.Start(); - OfflineSimulationUser user_a(kDeviceA, GetParam()); - OfflineSimulationUser user_b(kDeviceB, GetParam()); - EXPECT_FALSE(user_a.IsAdvertising()); - EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), nullptr), - Eq(Status{Status::kSuccess})); - EXPECT_TRUE(user_a.IsAdvertising()); - env_.Stop(); -} - -TEST_P(OfflineServiceControllerTest, CanStartDiscoveryBeforeAdvertising) { - env_.Start(); - OfflineSimulationUser user_a(kDeviceA, GetParam()); - OfflineSimulationUser user_b(kDeviceB, GetParam()); - EXPECT_FALSE(user_b.IsDiscovering()); - EXPECT_THAT(user_b.StartDiscovery(std::string(kServiceId), &discover_latch_), - Eq(Status{Status::kSuccess})); - EXPECT_TRUE(user_b.IsDiscovering()); - EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), nullptr), - Eq(Status{Status::kSuccess})); - EXPECT_TRUE(discover_latch_.Await(kDefaultTimeout).result()); - user_a.Stop(); - user_b.Stop(); - env_.Stop(); -} - -TEST_P(OfflineServiceControllerTest, CanStartDiscoveryAfterAdvertising) { - env_.Start(); - OfflineSimulationUser user_a(kDeviceA, GetParam()); - OfflineSimulationUser user_b(kDeviceB, GetParam()); - EXPECT_FALSE(user_b.IsDiscovering()); - EXPECT_FALSE(user_b.IsAdvertising()); - EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), nullptr), - Eq(Status{Status::kSuccess})); - EXPECT_THAT(user_b.StartDiscovery(std::string(kServiceId), &discover_latch_), - Eq(Status{Status::kSuccess})); - EXPECT_TRUE(user_a.IsAdvertising()); - EXPECT_TRUE(user_b.IsDiscovering()); - EXPECT_TRUE(discover_latch_.Await(kDefaultTimeout).result()); - user_a.Stop(); - user_b.Stop(); - env_.Stop(); -} - -TEST_P(OfflineServiceControllerTest, CanStopAdvertising) { - env_.Start(); - OfflineSimulationUser user_a(kDeviceA, GetParam()); - OfflineSimulationUser user_b(kDeviceB, GetParam()); - EXPECT_FALSE(user_a.IsAdvertising()); - EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), nullptr), - Eq(Status{Status::kSuccess})); - EXPECT_TRUE(user_a.IsAdvertising()); - user_a.StopAdvertising(); - EXPECT_FALSE(user_a.IsAdvertising()); - EXPECT_THAT(user_b.StartDiscovery(std::string(kServiceId), &discover_latch_, - &lost_latch_), - Eq(Status{Status::kSuccess})); - EXPECT_TRUE(user_b.IsDiscovering()); - auto discover_none = discover_latch_.Await(kDefaultTimeout).GetResult(); - if (!discover_none) { - EXPECT_TRUE(true); - } else { - // There are rare cases (1/1000) that advertisment data has been captured by - // discovery device before advertising is stopped. So we need to check if - // lost_cb has grabbed the event in the end to prove the advertising service - // is stopped. - EXPECT_TRUE(lost_latch_.Await(kDefaultTimeout).result()); - } - user_a.Stop(); - user_b.Stop(); - env_.Stop(); -} - -TEST_P(OfflineServiceControllerTest, CanStopDiscovery) { - env_.Start(); - OfflineSimulationUser user_a(kDeviceA, GetParam()); - OfflineSimulationUser user_b(kDeviceB, GetParam()); - EXPECT_FALSE(user_b.IsDiscovering()); - EXPECT_THAT(user_b.StartDiscovery(std::string(kServiceId), &discover_latch_), - Eq(Status{Status::kSuccess})); - EXPECT_TRUE(user_b.IsDiscovering()); - user_b.StopDiscovery(); - EXPECT_FALSE(user_b.IsDiscovering()); - EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), nullptr), - Eq(Status{Status::kSuccess})); - EXPECT_FALSE(discover_latch_.Await(kDefaultTimeout).result()); - user_a.Stop(); - user_b.Stop(); - env_.Stop(); -} - -TEST_P(OfflineServiceControllerTest, CanConnect) { - env_.Start(); - OfflineSimulationUser user_a(kDeviceA, GetParam()); - OfflineSimulationUser user_b(kDeviceB, GetParam()); - EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), &connect_latch_), - Eq(Status{Status::kSuccess})); - EXPECT_THAT(user_b.StartDiscovery(std::string(kServiceId), &discover_latch_), - Eq(Status{Status::kSuccess})); - EXPECT_TRUE(discover_latch_.Await(kDefaultTimeout).result()); - EXPECT_THAT(user_b.RequestConnection(&connect_latch_), - Eq(Status{Status::kSuccess})); - EXPECT_TRUE(connect_latch_.Await(kDefaultTimeout).result()); - user_a.Stop(); - user_b.Stop(); - env_.Stop(); -} - -TEST_P(OfflineServiceControllerTest, CanAcceptConnection) { - env_.Start(); - OfflineSimulationUser user_a(kDeviceA, GetParam()); - OfflineSimulationUser user_b(kDeviceB, GetParam()); - EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), &connect_latch_), - Eq(Status{Status::kSuccess})); - EXPECT_THAT(user_b.StartDiscovery(std::string(kServiceId), &discover_latch_), - Eq(Status{Status::kSuccess})); - EXPECT_TRUE(discover_latch_.Await(kDefaultTimeout).result()); - EXPECT_THAT(user_b.RequestConnection(&connect_latch_), - Eq(Status{Status::kSuccess})); - EXPECT_TRUE(connect_latch_.Await(kDefaultTimeout).result()); - EXPECT_THAT(user_a.AcceptConnection(&accept_latch_), - Eq(Status{Status::kSuccess})); - EXPECT_THAT(user_b.AcceptConnection(&accept_latch_), - Eq(Status{Status::kSuccess})); - EXPECT_TRUE(accept_latch_.Await(kDefaultTimeout).result()); - EXPECT_TRUE(user_a.IsConnected()); - EXPECT_TRUE(user_b.IsConnected()); - user_a.Stop(); - user_b.Stop(); - env_.Stop(); -} - -TEST_P(OfflineServiceControllerTest, CanRejectConnection) { - env_.Start(); - OfflineSimulationUser user_a(kDeviceA, GetParam()); - OfflineSimulationUser user_b(kDeviceB, GetParam()); - CountDownLatch reject_latch(1); - EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), &connect_latch_), - Eq(Status{Status::kSuccess})); - EXPECT_THAT(user_b.StartDiscovery(std::string(kServiceId), &discover_latch_), - Eq(Status{Status::kSuccess})); - EXPECT_TRUE(discover_latch_.Await(kDefaultTimeout).result()); - EXPECT_THAT(user_b.RequestConnection(&connect_latch_), - Eq(Status{Status::kSuccess})); - EXPECT_TRUE(connect_latch_.Await(kDefaultTimeout).result()); - user_a.ExpectRejectedConnection(reject_latch); - EXPECT_THAT(user_b.RejectConnection(nullptr), Eq(Status{Status::kSuccess})); - EXPECT_TRUE(reject_latch.Await(kDefaultTimeout).result()); - user_a.Stop(); - user_b.Stop(); - env_.Stop(); -} - -TEST_P(OfflineServiceControllerTest, CanSendBytePayload) { - env_.Start(); - OfflineSimulationUser user_a(kDeviceA, GetParam()); - OfflineSimulationUser user_b(kDeviceB, GetParam()); - ASSERT_TRUE(SetupConnection(user_a, user_b)); - ByteArray message(std::string{kMessage}); - user_a.SendPayload(Payload(message)); - user_b.ExpectPayload(payload_latch_); - EXPECT_TRUE(payload_latch_.Await(kDefaultTimeout).result()); - EXPECT_EQ(user_b.GetPayload().AsBytes(), message); - user_a.Stop(); - user_b.Stop(); - env_.Stop(); -} - -TEST_P(OfflineServiceControllerTest, CanSendStreamPayload) { - env_.Start(); - OfflineSimulationUser user_a(kDeviceA, GetParam()); - OfflineSimulationUser user_b(kDeviceB, GetParam()); - ASSERT_TRUE(SetupConnection(user_a, user_b)); - ByteArray message(std::string{kMessage}); - auto pipe = std::make_shared(); - OutputStream& tx = pipe->GetOutputStream(); - user_a.SendPayload(Payload([pipe]() -> InputStream& { - return pipe->GetInputStream(); // NOLINT - })); - user_b.ExpectPayload(payload_latch_); - tx.Write(message); - EXPECT_TRUE(payload_latch_.Await(kDefaultTimeout).result()); - EXPECT_NE(user_b.GetPayload().AsStream(), nullptr); - InputStream& rx = *user_b.GetPayload().AsStream(); - ASSERT_TRUE(user_b.WaitForProgress( - [size = message.size()](const PayloadProgressInfo& info) -> bool { - return info.bytes_transferred >= size; - }, - kProgressTimeout)); - EXPECT_EQ(rx.Read(Pipe::kChunkSize).result(), message); - user_a.Stop(); - user_b.Stop(); - env_.Stop(); -} - -TEST_P(OfflineServiceControllerTest, CanCancelStreamPayload) { - env_.Start(); - OfflineSimulationUser user_a(kDeviceA, GetParam()); - OfflineSimulationUser user_b(kDeviceB, GetParam()); - ASSERT_TRUE(SetupConnection(user_a, user_b)); - ByteArray message(std::string{kMessage}); - auto pipe = std::make_shared(); - OutputStream& tx = pipe->GetOutputStream(); - user_a.SendPayload(Payload([pipe]() -> InputStream& { - return pipe->GetInputStream(); // NOLINT - })); - user_b.ExpectPayload(payload_latch_); - tx.Write(message); - EXPECT_TRUE(payload_latch_.Await(kDefaultTimeout).result()); - EXPECT_NE(user_b.GetPayload().AsStream(), nullptr); - InputStream& rx = *user_b.GetPayload().AsStream(); - ASSERT_TRUE(user_b.WaitForProgress( - [size = message.size()](const PayloadProgressInfo& info) -> bool { - return info.bytes_transferred >= size; - }, - kProgressTimeout)); - EXPECT_EQ(rx.Read(Pipe::kChunkSize).result(), message); - user_b.CancelPayload(); - int count = 0; - while (true) { - count++; - if (!tx.Write(message).Ok()) break; - SystemClock::Sleep(kDefaultTimeout); - } - EXPECT_TRUE(user_a.WaitForProgress( - [](const PayloadProgressInfo& info) -> bool { - return info.status == PayloadProgressInfo::Status::kCanceled; - }, - kProgressTimeout)); - EXPECT_LT(count, 10); - user_a.Stop(); - user_b.Stop(); - env_.Stop(); -} - -TEST_P(OfflineServiceControllerTest, CanDisconnect) { - env_.Start(); - CountDownLatch disconnect_latch(1); - OfflineSimulationUser user_a(kDeviceA, GetParam()); - OfflineSimulationUser user_b(kDeviceB, GetParam()); - ASSERT_TRUE(SetupConnection(user_a, user_b)); - NEARBY_LOGS(INFO) << "Disconnecting"; - user_b.ExpectDisconnect(disconnect_latch); - user_b.Disconnect(); - EXPECT_TRUE(disconnect_latch.Await(kDisconnectTimeout).result()); - NEARBY_LOGS(INFO) << "Disconnected"; - EXPECT_FALSE(user_b.IsConnected()); - user_a.Stop(); - user_b.Stop(); - env_.Stop(); -} - -INSTANTIATE_TEST_SUITE_P(ParametrisedOfflineServiceControllerTest, - OfflineServiceControllerTest, - ::testing::ValuesIn(kTestCases)); - -// Verifies that InjectEndpoint() can be run successfully; does not test the -// full connection flow given that normal discovery/advertisement is skipped. -// Note: Not parameterized because InjectEndpoint only works over Bluetooth. -TEST_F(OfflineServiceControllerTest, InjectEndpoint) { - env_.Start(); - OfflineSimulationUser user_a(kDeviceA, - BooleanMediumSelector{.bluetooth = true}); - EXPECT_THAT(user_a.StartDiscovery(std::string(kServiceId), - /*found_latch=*/nullptr), - Eq(Status{Status::kSuccess})); - EXPECT_TRUE(user_a.IsDiscovering()); - user_a.InjectEndpoint( - std::string(kServiceId), - OutOfBandConnectionMetadata{ - .medium = Medium::BLUETOOTH, - .remote_bluetooth_mac_address = ByteArray(kFakeMacAddress), - }); - user_a.Stop(); - env_.Stop(); -} - -} // namespace -} // namespace connections -} // namespace nearby -} // namespace location 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/impl/shared/BUILD.orig b/cpp/platform/impl/shared/BUILD.orig deleted file mode 100644 index 05e5ff1b..00000000 --- a/cpp/platform/impl/shared/BUILD.orig +++ /dev/null @@ -1,71 +0,0 @@ -cc_library( - name = "posix_lock", - srcs = [ - "posix_lock.cc", - ], - hdrs = [ - "posix_lock.h", - ], - visibility = [ - "//googlemac/iPhone/Shared/Nearby/Connections:__subpackages__", - "//platform/impl:__subpackages__", - ], - deps = [ - "//platform/api", - ], -) - -cc_library( - name = "posix_condition_variable", - srcs = [ - "posix_condition_variable.cc", - ], - hdrs = [ - "posix_condition_variable.h", - ], - visibility = [ - "//googlemac/iPhone/Shared/Nearby/Connections:__subpackages__", - "//platform/impl:__subpackages__", - ], - deps = [ - ":posix_lock", - "//platform:types", - "//platform/api:condition_variable", - ], -) - -cc_library( - name = "atomic_boolean", - hdrs = ["atomic_boolean_impl.h"], - visibility = ["//platform/impl:__subpackages__"], - deps = ["//platform/api"], -) - -cc_library( - name = "file", - srcs = ["file_impl.cc"], - hdrs = ["file_impl.h"], - visibility = [ - "//googlemac/iPhone/Shared/Nearby/Connections:__subpackages__", - "//core:__subpackages__", - "//platform/impl:__subpackages__", - ], - deps = [ - "//platform:types", - "//platform/api", - ], -) - -cc_test( - name = "file_test", - timeout = "short", - srcs = [ - "file_impl_test.cc", - ], - deps = [ - ":file", - "//file/util:temp_path", - "//testing/base/public:gunit_main", - "//absl/strings", - ], -) diff --git a/cpp/platform/public/BUILD.orig b/cpp/platform/public/BUILD.orig deleted file mode 100644 index 31c87b6d..00000000 --- a/cpp/platform/public/BUILD.orig +++ /dev/null @@ -1,118 +0,0 @@ -cc_library( - name = "types", - srcs = [ - "pipe.cc", - ], - hdrs = [ - "atomic_boolean.h", - "atomic_reference.h", - "cancelable.h", - "cancelable_alarm.h", - "condition_variable.h", - "count_down_latch.h", - "crypto.h", - "file.h", - "future.h", - "logging.h", - "multi_thread_executor.h", - "mutex.h", - "mutex_lock.h", - "pipe.h", - "scheduled_executor.h", - "settable_future.h", - "single_thread_executor.h", - "submittable_executor.h", - "system_clock.h", - ], - visibility = [ - "//core_v2:__subpackages__", - "//platform_v2/base:__pkg__", - ], - deps = [ - ":logging", - "//platform_v2/api:platform", - "//platform_v2/api:types", - "//platform_v2/base", - "//platform_v2/base:logging", - "//platform_v2/base:util", - "//absl/base:core_headers", - "//absl/container:flat_hash_map", - "//absl/time", - ], -) - -cc_library( - name = "comm", - srcs = [ - "ble.cc", - "bluetooth_classic.cc", - "wifi_lan.cc", - ], - hdrs = [ - "ble.h", - "bluetooth_adapter.h", - "bluetooth_classic.h", - "webrtc.h", - "wifi_lan.h", - ], - visibility = ["//core_v2:__subpackages__"], - deps = [ - ":logging", - ":types", - "//proto/connections:offline_wire_formats_portable_proto", - "//platform_v2/api:comm", - "//platform_v2/api:platform", - "//platform_v2/base", - "//absl/container:flat_hash_map", - "//absl/strings", - "//webrtc/api:libjingle_peerconnection_api", - ], -) - -cc_library( - name = "logging", - hdrs = [ - "logging.h", - ], - visibility = ["//core_v2:__subpackages__"], - deps = [ - "//platform_v2/base:logging", - ], -) - -cc_test( - name = "public_test", - size = "small", - srcs = [ - "atomic_boolean_test.cc", - "atomic_reference_test.cc", - "ble_test.cc", - "bluetooth_adapter_test.cc", - "bluetooth_classic_test.cc", - "cancelable_alarm_test.cc", - "condition_variable_test.cc", - "count_down_latch_test.cc", - "crypto_test.cc", - "future_test.cc", - "logging_test.cc", - "multi_thread_executor_test.cc", - "mutex_test.cc", - "pipe_test.cc", - "scheduled_executor_test.cc", - "single_thread_executor_test.cc", - "wifi_lan_test.cc", - ], - shard_count = 16, - deps = [ - ":comm", - ":logging", - ":types", - "//platform_v2/base", - "//platform_v2/base:test_util", - "//platform_v2/impl/g3", # build_cleaner: keep - "//testing/base/public:gunit_main", - "//absl/strings", - "//absl/synchronization", - "//absl/time", - ], -) 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