From 0be1cd03ce7a486e7271d2a237c88261b3044c33 Mon Sep 17 00:00:00 2001 From: hai007 Date: Tue, 12 Jan 2021 11:16:21 -0800 Subject: [PATCH] Roll forward up to cl/351408510. --- cpp/core/internal/base_pcp_handler.cc | 58 ++++++++------- cpp/core/internal/base_pcp_handler.h | 11 +-- cpp/core/internal/bwu_manager.cc | 70 +++++++++++++++++++ cpp/core/internal/bwu_manager.h | 4 ++ .../internal/mediums/bluetooth_classic.cc | 12 ++++ cpp/core/internal/mediums/bluetooth_classic.h | 14 +++- cpp/core/internal/offline_frames.cc | 19 +++++ cpp/core/internal/offline_frames.h | 1 + cpp/core/internal/offline_frames_test.cc | 5 ++ proto/connections/offline_wire_formats.proto | 8 +++ proto/mediums/BUILD | 1 - 11 files changed, 171 insertions(+), 32 deletions(-) diff --git a/cpp/core/internal/base_pcp_handler.cc b/cpp/core/internal/base_pcp_handler.cc index a0bd9f1f..356c0dfe 100644 --- a/cpp/core/internal/base_pcp_handler.cc +++ b/cpp/core/internal/base_pcp_handler.cc @@ -142,8 +142,7 @@ void BasePcpHandler::InjectEndpoint( ClientProxy* client, const std::string& service_id, const OutOfBandConnectionMetadata& metadata) { CountDownLatch latch(1); - RunOnPcpHandlerThread([this, client, service_id, metadata, - &latch]() { + RunOnPcpHandlerThread([this, client, service_id, metadata, &latch]() { InjectEndpointImpl(client, service_id, metadata); latch.CountDown(); }); @@ -330,7 +329,7 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client, BluetoothUtils::ToString(options.remote_bluetooth_mac_address); if (!remote_bluetooth_mac_address.empty()) { if (AppendRemoteBluetoothMacAddressEndpoint(endpoint_id, - remote_bluetooth_mac_address)) + remote_bluetooth_mac_address)) NEARBY_LOGS(INFO) << "Appended remote Bluetooth MAC Address endpoint " << "[" << remote_bluetooth_mac_address << "]"; } @@ -343,6 +342,7 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client, ConnectImplResult connect_impl_result; for (auto connect_endpoint : discovered_endpoints) { + if (!MediumSupported(connect_endpoint->medium, options)) continue; connect_impl_result = ConnectImpl(client, connect_endpoint); if (connect_impl_result.status.Ok()) { channel = std::move(connect_impl_result.endpoint_channel); @@ -366,7 +366,7 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client, // endpoint about ourselves. Exception write_exception = WriteConnectionRequestFrame( channel.get(), client->GetLocalEndpointId(), info.endpoint_info, nonce, - GetConnectionMediumsByPriority()); + GetSupportedConnectionMediumsByPriority(options)); if (!write_exception.Ok()) { NEARBY_LOG(INFO, "Failed to send connection request: id=%s", endpoint_id.c_str()); @@ -416,6 +416,27 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client, return status; } +bool BasePcpHandler::MediumSupported(const proto::connections::Medium& medium, + const ConnectionOptions& options) const { + for (auto supported_medium : + options.allowed.GetMediums(/* is supported = */true)) { + if (medium == supported_medium) return true; + } + return false; +} + +std::vector +BasePcpHandler::GetSupportedConnectionMediumsByPriority( + const ConnectionOptions& options) { + std::vector supported_mediums_by_priority; + for (auto medium_by_priority : GetConnectionMediumsByPriority()) { + if (MediumSupported(medium_by_priority, options)) { + supported_mediums_by_priority.push_back(medium_by_priority); + } + } + return supported_mediums_by_priority; +} + // Get any single discovered endpoint for a given endpoint_id. BasePcpHandler::DiscoveredEndpoint* BasePcpHandler::GetDiscoveredEndpoint( const std::string& endpoint_id) { @@ -1038,13 +1059,8 @@ bool BasePcpHandler::AppendRemoteBluetoothMacAddressEndpoint( auto bluetooth_endpoint = std::make_shared(BluetoothEndpoint{ - { - endpoint_id, - endpoint->endpoint_info, - endpoint->service_id, - proto::connections::Medium::BLUETOOTH, - WebRtcState::kUnconnectable - }, + {endpoint_id, endpoint->endpoint_info, endpoint->service_id, + proto::connections::Medium::BLUETOOTH, WebRtcState::kUnconnectable}, remote_bluetooth_device, }); @@ -1069,20 +1085,12 @@ bool BasePcpHandler::AppendWebRTCEndpoint(const std::string& endpoint_id) { } if (!should_connect_web_rtc) return false; - auto webrtc_endpoint = - std::make_shared(WebRtcEndpoint{ - { - endpoint_id, - endpoint->endpoint_info, - endpoint->service_id, - proto::connections::Medium::WEB_RTC, - WebRtcState::kConnectable - }, - CreatePeerIdFromAdvertisement( - endpoint->service_id, - endpoint->endpoint_id, - endpoint->endpoint_info), - }); + auto webrtc_endpoint = std::make_shared(WebRtcEndpoint{ + {endpoint_id, endpoint->endpoint_info, endpoint->service_id, + proto::connections::Medium::WEB_RTC, WebRtcState::kConnectable}, + CreatePeerIdFromAdvertisement(endpoint->service_id, endpoint->endpoint_id, + endpoint->endpoint_info), + }); discovered_endpoints_.emplace(endpoint_id, std::move(webrtc_endpoint)); return true; diff --git a/cpp/core/internal/base_pcp_handler.h b/cpp/core/internal/base_pcp_handler.h index c30271b6..8dfea1a1 100644 --- a/cpp/core/internal/base_pcp_handler.h +++ b/cpp/core/internal/base_pcp_handler.h @@ -118,8 +118,7 @@ class BasePcpHandler : public PcpHandler, // otherwise does nothing. void StopDiscovery(ClientProxy* client) override; - void InjectEndpoint(ClientProxy* client, - const std::string& service_id, + void InjectEndpoint(ClientProxy* client, const std::string& service_id, const OutOfBandConnectionMetadata& metadata) override; // Requests a newly discovered remote endpoint it to form a connection. @@ -280,8 +279,7 @@ class BasePcpHandler : public PcpHandler, // @PcpHandlerThread virtual Status InjectEndpointImpl( - ClientProxy* client, - const std::string& service_id, + ClientProxy* client, const std::string& service_id, const OutOfBandConnectionMetadata& metadata) = 0; // @PcpHandlerThread @@ -466,6 +464,11 @@ class BasePcpHandler : public PcpHandler, void WaitForLatch(const std::string& method_name, CountDownLatch* latch); Status WaitForResult(const std::string& method_name, std::int64_t client_id, Future* future); + bool MediumSupported(const proto::connections::Medium& medium, + const ConnectionOptions& options) const; + std::vector + GetSupportedConnectionMediumsByPriority( + const ConnectionOptions& options); AtomicReference bwu_medium_{Medium::UNKNOWN_MEDIUM}; ScheduledExecutor alarm_executor_; diff --git a/cpp/core/internal/bwu_manager.cc b/cpp/core/internal/bwu_manager.cc index 1d021924..0f26504c 100644 --- a/cpp/core/internal/bwu_manager.cc +++ b/cpp/core/internal/bwu_manager.cc @@ -287,6 +287,13 @@ void BwuManager::OnIncomingConnection( return; } + if (!WriteClientIntroductionAckFrame(channel)) { + // 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; @@ -410,6 +417,22 @@ BwuManager::ProcessBwuPathAvailableEventInternal( return {}; } + if (upgrade_path_info.supports_client_introduction_ack()) { + if (!ReadClientIntroductionAckFrame(channel.get())) { + // This was never a fully EstablishedConnection, no need to provide a + // closure reason. + channel->Close(); + + NEARBY_LOG( + ERROR, + "Failed to read BWU_NEGOTIATION.CLIENT_INTRODUCTION_ACK OfflineFrame " + "to newly-created EndpointChannel %s, aborting upgrade.", + channel->GetName().c_str()); + + return {}; + } + } + NEARBY_LOG( INFO, "Successfully wrote BWU_NEGOTIATION.CLIENT_INTRODUCTION OfflineFrame to " @@ -459,19 +482,66 @@ void BwuManager::RunUpgradeFailedProtocol( bool BwuManager::ReadClientIntroductionFrame(EndpointChannel* channel, ClientIntroduction& introduction) { + CancelableAlarm timeout_alarm( + "BwuManager::ReadClientIntroductionFrame", + [channel]() { + NEARBY_LOG( + ERROR, + "In BandwidthUpgradeManager, failed to read the " + "ClientIntroductionFrame after %d seconds. Timing out and closing " + "EndpointChannel %s.", + kReadClientIntroductionFrameTimeout, channel->GetType().c_str()); + channel->Close(); + }, + kReadClientIntroductionFrameTimeout, &alarm_executor_); auto data = channel->Read(); + timeout_alarm.Cancel(); 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; + if (frame.v1().bandwidth_upgrade_negotiation().event_type() != + BandwidthUpgradeNegotiationFrame::CLIENT_INTRODUCTION) + return false; const auto& frame_intro = frame.v1().bandwidth_upgrade_negotiation().client_introduction(); introduction = frame_intro; return true; } +bool BwuManager::ReadClientIntroductionAckFrame(EndpointChannel* channel) { + CancelableAlarm timeout_alarm( + "BwuManager::ReadClientIntroductionAckFrame", + [channel]() { + NEARBY_LOG(ERROR, + "In BandwidthUpgradeManager, failed to read the " + "ClientIntroductionAckFrame after %d seconds. Timing out " + "and closing EndpointChannel %s.", + kReadClientIntroductionFrameTimeout, + channel->GetType().c_str()); + channel->Close(); + }, + kReadClientIntroductionFrameTimeout, &alarm_executor_); + auto data = channel->Read(); + timeout_alarm.Cancel(); + 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; + if (frame.v1().bandwidth_upgrade_negotiation().event_type() != + BandwidthUpgradeNegotiationFrame::CLIENT_INTRODUCTION_ACK) + return false; + return true; +} + +bool BwuManager::WriteClientIntroductionAckFrame(EndpointChannel* channel) { + return channel->Write(parser::ForBwuIntroductionAck()).Ok(); +} + void BwuManager::ProcessLastWriteToPriorChannelEvent( ClientProxy* client, const std::string& endpoint_id) { // By this point in the upgrade protocol, there is the guarantee that both diff --git a/cpp/core/internal/bwu_manager.h b/cpp/core/internal/bwu_manager.h index 8c743e5e..a1f6c480 100644 --- a/cpp/core/internal/bwu_manager.h +++ b/cpp/core/internal/bwu_manager.h @@ -85,6 +85,8 @@ class BwuManager : public EndpointManager::FrameProcessor { void Shutdown(); private: + static constexpr absl::Duration kReadClientIntroductionFrameTimeout = + absl::Seconds(5); BwuHandler* SetCurrentBwuHandler(Medium medium); void InitBwuHandlers(); void RunOnBwuManagerThread(std::function runnable); @@ -129,6 +131,8 @@ class BwuManager : public EndpointManager::FrameProcessor { const std::string& endpoint_id); bool ReadClientIntroductionFrame(EndpointChannel* endpoint_channel, ClientIntroduction& introduction); + bool ReadClientIntroductionAckFrame(EndpointChannel* endpoint_channel); + bool WriteClientIntroductionAckFrame(EndpointChannel* endpoint_channel); void ProcessEndpointDisconnection(ClientProxy* client, const std::string& endpoint_id, CountDownLatch* barrier); diff --git a/cpp/core/internal/mediums/bluetooth_classic.cc b/cpp/core/internal/mediums/bluetooth_classic.cc index 63e8df68..06aecac6 100644 --- a/cpp/core/internal/mediums/bluetooth_classic.cc +++ b/cpp/core/internal/mediums/bluetooth_classic.cc @@ -332,6 +332,18 @@ bool BluetoothClassic::StopAcceptingConnections( BluetoothSocket BluetoothClassic::Connect(BluetoothDevice& bluetooth_device, const std::string& service_name) { + for (int attempts_count = 0; attempts_count < kConnectAttemptsLimit; + attempts_count++) { + auto wrapper_result = AttemptToConnect(bluetooth_device, service_name); + if (wrapper_result.IsValid()) { + return wrapper_result; + } + } + return BluetoothSocket(); +} + +BluetoothSocket BluetoothClassic::AttemptToConnect( + BluetoothDevice& bluetooth_device, const std::string& service_name) { MutexLock lock(&mutex_); NEARBY_LOG(INFO, "BluetoothClassic::Connect: device=%p", &bluetooth_device); // Socket to return. To allow for NRVO to work, it has to be a single object. diff --git a/cpp/core/internal/mediums/bluetooth_classic.h b/cpp/core/internal/mediums/bluetooth_classic.h index 18262767..e1e05094 100644 --- a/cpp/core/internal/mediums/bluetooth_classic.h +++ b/cpp/core/internal/mediums/bluetooth_classic.h @@ -91,8 +91,8 @@ class BluetoothClassic { return adapter_.IsValid(); } - // Establishes connection to BT service that was might be started on another - // device with StartAcceptingConnections() using the same service_name. + // Establishes connection to BT service with internal retry for maximum + // attempts of kConnectAttemptsLimit. // Blocks until connection is established, or server-side is terminated. // Returns socket instance. On success, BluetoothSocket.IsValid() return true. // Called by client. @@ -112,6 +112,8 @@ class BluetoothClassic { static constexpr int kMaxConcurrentAcceptLoops = 5; + static constexpr int kConnectAttemptsLimit = 3; + // Constructs UUID object from arbitrary string, using MD5 hash, and then // converts UUID to a readable UUID string and returns it. static std::string GenerateUuidFromString(const std::string& data); @@ -146,6 +148,14 @@ class BluetoothClassic { // Returns true if device is currently in discovery mode. bool IsDiscovering() const ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + // Establishes connection to BT service that was might be started on another + // device with StartAcceptingConnections() using the same service_name. + // Blocks until connection is established, or server-side is terminated. + // Returns socket instance. On success, BluetoothSocket.IsValid() return true. + // Called by client. + BluetoothSocket AttemptToConnect(BluetoothDevice& bluetooth_device, + const std::string& service_name); + mutable Mutex mutex_; BluetoothRadio& radio_ ABSL_GUARDED_BY(mutex_); BluetoothAdapter& adapter_ ABSL_GUARDED_BY(mutex_){ diff --git a/cpp/core/internal/offline_frames.cc b/cpp/core/internal/offline_frames.cc index a8050b6a..fae6baae 100644 --- a/cpp/core/internal/offline_frames.cc +++ b/cpp/core/internal/offline_frames.cc @@ -146,6 +146,7 @@ ByteArray ForBwuWifiHotspotPathAvailable(const std::string& ssid, BandwidthUpgradeNegotiationFrame::UPGRADE_PATH_AVAILABLE); auto* upgrade_path_info = sub_frame->mutable_upgrade_path_info(); upgrade_path_info->set_medium(UpgradePathInfo::WIFI_HOTSPOT); + upgrade_path_info->set_supports_client_introduction_ack(true); upgrade_path_info->set_supports_disabling_encryption( supports_disabling_encryption); auto* wifi_hotspot_credentials = @@ -170,6 +171,7 @@ ByteArray ForBwuWifiLanPathAvailable(const std::string& ip_address, BandwidthUpgradeNegotiationFrame::UPGRADE_PATH_AVAILABLE); auto* upgrade_path_info = sub_frame->mutable_upgrade_path_info(); upgrade_path_info->set_medium(UpgradePathInfo::WIFI_LAN); + upgrade_path_info->set_supports_client_introduction_ack(true); auto* wifi_lan_socket = upgrade_path_info->mutable_wifi_lan_socket(); wifi_lan_socket->set_ip_address(ip_address); wifi_lan_socket->set_wifi_port(port); @@ -191,6 +193,7 @@ ByteArray ForBwuWifiAwarePathAvailable(const std::string& service_id, BandwidthUpgradeNegotiationFrame::UPGRADE_PATH_AVAILABLE); auto* upgrade_path_info = sub_frame->mutable_upgrade_path_info(); upgrade_path_info->set_medium(UpgradePathInfo::WIFI_AWARE); + upgrade_path_info->set_supports_client_introduction_ack(true); upgrade_path_info->set_supports_disabling_encryption( supports_disabling_encryption); auto* wifi_aware_credentials = @@ -217,6 +220,7 @@ ByteArray ForBwuWifiDirectPathAvailable(const std::string& ssid, BandwidthUpgradeNegotiationFrame::UPGRADE_PATH_AVAILABLE); auto* upgrade_path_info = sub_frame->mutable_upgrade_path_info(); upgrade_path_info->set_medium(UpgradePathInfo::WIFI_DIRECT); + upgrade_path_info->set_supports_client_introduction_ack(true); upgrade_path_info->set_supports_disabling_encryption( supports_disabling_encryption); auto* wifi_direct_credentials = @@ -241,6 +245,7 @@ ByteArray ForBwuBluetoothPathAvailable(const std::string& service_id, BandwidthUpgradeNegotiationFrame::UPGRADE_PATH_AVAILABLE); auto* upgrade_path_info = sub_frame->mutable_upgrade_path_info(); upgrade_path_info->set_medium(UpgradePathInfo::BLUETOOTH); + upgrade_path_info->set_supports_client_introduction_ack(true); auto* bluetooth_credentials = upgrade_path_info->mutable_bluetooth_credentials(); bluetooth_credentials->set_mac_address(mac_address); @@ -261,6 +266,7 @@ ByteArray ForBwuWebrtcPathAvailable(const std::string& peer_id, BandwidthUpgradeNegotiationFrame::UPGRADE_PATH_AVAILABLE); auto* upgrade_path_info = sub_frame->mutable_upgrade_path_info(); upgrade_path_info->set_medium(UpgradePathInfo::WEB_RTC); + upgrade_path_info->set_supports_client_introduction_ack(true); auto* webrtc_credentials = upgrade_path_info->mutable_web_rtc_credentials(); webrtc_credentials->set_peer_id(peer_id); auto* local_location_hint = webrtc_credentials->mutable_location_hint(); @@ -310,6 +316,19 @@ ByteArray ForBwuIntroduction(const std::string& endpoint_id) { return ToBytes(std::move(frame)); } +ByteArray ForBwuIntroductionAck() { + OfflineFrame frame; + + frame.set_version(OfflineFrame::V1); + auto* v1_frame = frame.mutable_v1(); + v1_frame->set_type(V1Frame::BANDWIDTH_UPGRADE_NEGOTIATION); + auto* sub_frame = v1_frame->mutable_bandwidth_upgrade_negotiation(); + sub_frame->set_event_type( + BandwidthUpgradeNegotiationFrame::CLIENT_INTRODUCTION_ACK); + + return ToBytes(std::move(frame)); +} + ByteArray ForBwuFailure(const UpgradePathInfo& info) { OfflineFrame frame; diff --git a/cpp/core/internal/offline_frames.h b/cpp/core/internal/offline_frames.h index 47b4565d..62a80f96 100644 --- a/cpp/core/internal/offline_frames.h +++ b/cpp/core/internal/offline_frames.h @@ -46,6 +46,7 @@ ByteArray ForControlPayloadTransfer( // Builds Bandwidth Upgrade [BWU] messages. ByteArray ForBwuIntroduction(const std::string& endpoint_id); +ByteArray ForBwuIntroductionAck(); ByteArray ForBwuWifiHotspotPathAvailable(const std::string& ssid, const std::string& password, std::int32_t port, diff --git a/cpp/core/internal/offline_frames_test.cc b/cpp/core/internal/offline_frames_test.cc index 82cbabe2..96383869 100644 --- a/cpp/core/internal/offline_frames_test.cc +++ b/cpp/core/internal/offline_frames_test.cc @@ -189,6 +189,7 @@ TEST(OfflineFramesTest, CanGenerateBwuWifiHotspotPathAvailable) { gateway: "0.0.0.0" > supports_disabling_encryption: false + supports_client_introduction_ack: true > > >)pb"; @@ -211,6 +212,7 @@ TEST(OfflineFramesTest, CanGenerateBwuWifiLanPathAvailable) { upgrade_path_info: < medium: WIFI_LAN wifi_lan_socket: < ip_address: "\x01\x02\x03\x04" wifi_port: 1234 > + supports_client_introduction_ack: true > > >)pb"; @@ -237,6 +239,7 @@ TEST(OfflineFramesTest, CanGenerateBwuWifiAwarePathAvailable) { password: "password" > supports_disabling_encryption: false + supports_client_introduction_ack: true > > >)pb"; @@ -265,6 +268,7 @@ TEST(OfflineFramesTest, CanGenerateBwuWifiDirectPathAvailable) { frequency: 1000 > supports_disabling_encryption: false + supports_client_introduction_ack: true > > >)pb"; @@ -290,6 +294,7 @@ TEST(OfflineFramesTest, CanGenerateBwuBluetoothPathAvailable) { service_name: "service" mac_address: "\x11\x22\x33\x44\x55\x66" > + supports_client_introduction_ack: true > > >)pb"; diff --git a/proto/connections/offline_wire_formats.proto b/proto/connections/offline_wire_formats.proto index 5225503e..625f6919 100644 --- a/proto/connections/offline_wire_formats.proto +++ b/proto/connections/offline_wire_formats.proto @@ -152,6 +152,7 @@ message BandwidthUpgradeNegotiationFrame { SAFE_TO_CLOSE_PRIOR_CHANNEL = 3; CLIENT_INTRODUCTION = 4; UPGRADE_FAILURE = 5; + CLIENT_INTRODUCTION_ACK = 6; } // Accompanies UPGRADE_PATH_AVAILABLE and UPGRADE_FAILURE events. @@ -223,6 +224,9 @@ message BandwidthUpgradeNegotiationFrame { // Disable Encryption for this upgrade medium to improve throughput. optional bool supports_disabling_encryption = 7; + + // An ack will be sent after the CLIENT_INTRODUCTION frame. + optional bool supports_client_introduction_ack = 9; } // Accompanies CLIENT_INTRODUCTION events. @@ -231,11 +235,15 @@ message BandwidthUpgradeNegotiationFrame { optional bool supports_disabling_encryption = 2; } + // Accompanies CLIENT_INTRODUCTION_ACK events. + message ClientIntroductionAck {} + optional EventType event_type = 1; // Exactly one of the following fields will be set. optional UpgradePathInfo upgrade_path_info = 2; optional ClientIntroduction client_introduction = 3; + optional ClientIntroductionAck client_introduction_ack = 4; } message KeepAliveFrame { diff --git a/proto/mediums/BUILD b/proto/mediums/BUILD index dcf12b59..e28f5866 100644 --- a/proto/mediums/BUILD +++ b/proto/mediums/BUILD @@ -67,7 +67,6 @@ cc_proto_library( java_lite_proto_library( name = "web_rtc_signaling_frames_java_proto_lite", - strict_deps = 0, visibility = [ "//java/com/google/android/gmscore/integ/modules/nearby/src/com/google/android/gms/nearby/mediums:__subpackages__", "//javatests/com/google/android/gmscore/integ/modules/nearby/src/com/google/android/gms/nearby/mediums:__subpackages__",