From 599a015760ed07ccd9040af4ce6f548f138cf00e Mon Sep 17 00:00:00 2001 From: hai007 Date: Fri, 11 Dec 2020 11:56:14 -0800 Subject: [PATCH] Roll forward to cl/347045028 Signed-off-by: hai007 --- cpp/core/internal/base_pcp_handler.h | 7 ++ cpp/core/internal/mediums/webrtc.cc | 116 ++++++++++++++++--- cpp/core/internal/mediums/webrtc.h | 8 ++ cpp/core/internal/mediums/webrtc_test.cc | 64 +++++++++- cpp/core/internal/p2p_cluster_pcp_handler.h | 21 ---- proto/connections/offline_wire_formats.proto | 2 + proto/discovery_enums.proto | 41 ++++++- 7 files changed, 217 insertions(+), 42 deletions(-) diff --git a/cpp/core/internal/base_pcp_handler.h b/cpp/core/internal/base_pcp_handler.h index 3b3cf4b1..c30271b6 100644 --- a/cpp/core/internal/base_pcp_handler.h +++ b/cpp/core/internal/base_pcp_handler.h @@ -206,6 +206,13 @@ class BasePcpHandler : public PcpHandler, BluetoothDevice bluetooth_device; }; + struct BleEndpoint : public BasePcpHandler::DiscoveredEndpoint { + BleEndpoint(DiscoveredEndpoint endpoint, BlePeripheral peripheral) + : DiscoveredEndpoint(std::move(endpoint)), + ble_peripheral(std::move(peripheral)) {} + BlePeripheral ble_peripheral; + }; + struct WifiLanEndpoint : public DiscoveredEndpoint { WifiLanEndpoint(DiscoveredEndpoint endpoint, WifiLanService service) : DiscoveredEndpoint(std::move(endpoint)), diff --git a/cpp/core/internal/mediums/webrtc.cc b/cpp/core/internal/mediums/webrtc.cc index bbf0c02f..dddef4cf 100644 --- a/cpp/core/internal/mediums/webrtc.cc +++ b/cpp/core/internal/mediums/webrtc.cc @@ -2,6 +2,7 @@ #include #include +#include #include "core/internal/mediums/webrtc/session_description_wrapper.h" #include "core/internal/mediums/webrtc/signaling_frames.h" @@ -35,6 +36,10 @@ constexpr absl::Duration kRestartReceiveMessagesDuration = absl::Seconds(60); WebRtc::WebRtc() = default; WebRtc::~WebRtc() { + { + MutexLock lock(&mutex_); + NEARBY_LOGS(WARNING) << "Destructing Webrtc: " << InternalStatesToString(); + } // This ensures that all pending callbacks are run before we reset the medium // and we are not accepting new runnables. restart_receive_messages_executor_.Shutdown(); @@ -89,6 +94,8 @@ bool WebRtc::StartAcceptingConnections(const std::string& service_id, } { MutexLock lock(&mutex_); + NEARBY_LOGS(WARNING) << "StartAcceptingConnections: " + << InternalStatesToString(); accepting_map_.emplace(service_id, ConnectionInfo{.socket = WebRtcSocketWrapper()}); ConnectionInfo* connection_info = &accepting_map_[service_id]; @@ -107,6 +114,7 @@ bool WebRtc::StartAcceptingConnections(const std::string& service_id, webrtc_frames::EncodeOffer(self_id, offer.GetSdp()); if (!SetLocalSessionDescription(std::move(offer), Role::kOfferer, service_id)) { + NEARBY_LOG(WARNING, "Failed to set local session description."); return false; } @@ -117,7 +125,7 @@ bool WebRtc::StartAcceptingConnections(const std::string& service_id, Role::kOfferer, service_id, connection_info->connection_flow->GetDataChannel(), std::move(callback)); - NEARBY_LOG(INFO, "Started listening for WebRtc connections as %s", + NEARBY_LOG(WARNING, "Started listening for WebRtc connections as %s", self_id.GetId().c_str()); } @@ -127,12 +135,16 @@ bool WebRtc::StartAcceptingConnections(const std::string& service_id, WebRtcSocketWrapper WebRtc::Connect(const PeerId& peer_id, const LocationHint& location_hint) { if (!IsAvailable()) { - Disconnect(Role::kAnswerer, peer_id.GetId()); + MutexLock lock(&mutex_); + LogAndDisconnect(Role::kAnswerer, peer_id.GetId(), + "WebRTC is not available for data transfer."); return WebRtcSocketWrapper(); } { MutexLock lock(&mutex_); + NEARBY_LOGS(WARNING) << "Start Connecting to " << peer_id.GetId() << ":\n" + << InternalStatesToString(); if (connecting_map_.contains(peer_id.GetId())) { NEARBY_LOG( ERROR, @@ -149,7 +161,7 @@ WebRtcSocketWrapper WebRtc::Connect(const PeerId& peer_id, } } - NEARBY_LOG(ERROR, "Attempting to make a WebRTC connection to %s.", + NEARBY_LOG(WARNING, "Attempting to make a WebRTC connection to %s.", peer_id.GetId().c_str()); Future socket_future; { @@ -167,8 +179,14 @@ WebRtcSocketWrapper WebRtc::Connect(const PeerId& peer_id, // in a timeout in creating the socket. ExceptionOr result = socket_future.Get(kDataChannelTimeout); - if (result.ok()) return result.result(); + if (result.ok()) { + NEARBY_LOGS(WARNING) << "Succeeded to make WebRTC connection to " + << peer_id.GetId(); + return result.result(); + } + NEARBY_LOGS(WARNING) << "Failed to make WebRTC connection to " + << peer_id.GetId(); Disconnect(Role::kAnswerer, peer_id.GetId()); return WebRtcSocketWrapper(); } @@ -190,7 +208,7 @@ bool WebRtc::SetLocalSessionDescription(SessionDescriptionWrapper sdp, void WebRtc::StopAcceptingConnections(const std::string& service_id) { if (!IsAcceptingConnections(service_id)) { - NEARBY_LOG(INFO, + NEARBY_LOG(WARNING, "Skipped StopAcceptingConnections since we are not currently " "accepting WebRTC connections for %s", service_id.c_str()); @@ -199,9 +217,11 @@ void WebRtc::StopAcceptingConnections(const std::string& service_id) { { MutexLock lock(&mutex_); - ShutdownSignaling(Role::kOfferer, service_id); + LogAndShutdownSignaling(Role::kOfferer, service_id, + "Invoked by StopAcceptingConnections."); } - NEARBY_LOG(INFO, "Stopped accepting WebRTC connections"); + NEARBY_LOG(WARNING, "Stopped accepting WebRTC connections for %s", + service_id.c_str()); } Future WebRtc::ListenForWebRtcSocketFuture( @@ -261,7 +281,10 @@ bool WebRtc::InitWebRtcFlow(const Role& role, const PeerId& self_id, const LocationHint& location_hint, const std::string& connection_id) { ConnectionInfo* connection_info = GetConnectionInfo(role, connection_id); - if (!connection_info) return false; + if (!connection_info) { + NEARBY_LOGS(WARNING) << "Can not find matching connection info."; + return false; + } connection_info->self_id = self_id; if (connection_info->connection_flow) { @@ -316,6 +339,9 @@ bool WebRtc::InitWebRtcFlow(const Role& role, const PeerId& self_id, return false; } + NEARBY_LOGS(WARNING) << "Succeeded to create connection flow for role:" + << role_names_[role] + << ", connection_id: " << connection_id; return true; } @@ -331,6 +357,8 @@ void WebRtc::OnLocalIceCandidate( ConnectionInfo* connection_info = GetConnectionInfo(role, connection_id); if (IsSignaling(role, connection_id)) { if (connection_info && connection_info->signaling_messenger) { + NEARBY_LOG(WARNING, "Sending local ice candidates to %s", + connection_info->peer_id.GetId().c_str()); connection_info->signaling_messenger->SendMessage( connection_info->peer_id.GetId(), webrtc_frames::EncodeIceCandidates(connection_info->self_id, @@ -421,7 +449,12 @@ void WebRtc::ProcessSignalingMessage(const Role& role, const ByteArray& message) { MutexLock lock(&mutex_); ConnectionInfo* connection_info = GetConnectionInfo(role, connection_id); - if (!connection_info) return; + if (!connection_info) { + NEARBY_LOG(ERROR, + "Could not find connection info for role: %s, connection_id: %s", + role_names_[role].c_str(), connection_id.c_str()); + return; + } if (!connection_info->connection_flow) { LogAndDisconnect(role, connection_id, @@ -444,12 +477,12 @@ void WebRtc::ProcessSignalingMessage(const Role& role, if (frame.has_ready_for_signaling_poke() && !connection_info->peer_id.IsValid()) { connection_info->peer_id = PeerId(frame.sender_id().id()); - NEARBY_LOG(INFO, "Peer %s is ready for signaling", + NEARBY_LOG(WARNING, "Peer %s is ready for signaling", connection_info->peer_id.GetId().c_str()); } if (!IsSignaling(role, connection_id)) { - NEARBY_LOG(INFO, + NEARBY_LOG(WARNING, "Ignoring WebRTC frame: we are not currently listening for " "signaling messages"); return; @@ -457,28 +490,43 @@ void WebRtc::ProcessSignalingMessage(const Role& role, if (frame.sender_id().id() != connection_info->peer_id.GetId()) { NEARBY_LOG( - INFO, "Ignoring WebRTC frame: we are only listening for another peer."); + WARNING, + "Ignoring WebRTC frame: we are only listening for another peer."); return; } if (frame.has_ready_for_signaling_poke()) { + NEARBY_LOG(WARNING, + "Received ready-for-poke for role: %s connection_id: %s", + role_names_[role].c_str(), connection_id.c_str()); SendOfferAndIceCandidatesToPeer(connection_id); } else if (frame.has_offer()) { + NEARBY_LOG(WARNING, "Received offer for role: %s connection_id: %s", + role_names_[role].c_str(), connection_id.c_str()); DCHECK(role == Role::kAnswerer); connection_info->connection_flow->OnOfferReceived( SessionDescriptionWrapper(webrtc_frames::DecodeOffer(frame).release())); SendAnswerToPeer(connection_id); } else if (frame.has_answer()) { + NEARBY_LOG(WARNING, "Received answer for role: %s connection_id: %s", + role_names_[role].c_str(), connection_id.c_str()); DCHECK(role == Role::kOfferer); connection_info->connection_flow->OnAnswerReceived( SessionDescriptionWrapper( webrtc_frames::DecodeAnswer(frame).release())); } else if (frame.has_ice_candidates()) { + NEARBY_LOG(WARNING, + "Received ice candidates for role: %s connection_id: %s", + role_names_[role].c_str(), connection_id.c_str()); if (!connection_info->connection_flow->OnRemoteIceCandidatesReceived( webrtc_frames::DecodeIceCandidates(frame))) { LogAndDisconnect(role, connection_id, "Could not add remote ice candidates."); } + } else { + NEARBY_LOG(WARNING, + "Received unknown frame type for role: %s connection_id: %s", + role_names_[role].c_str(), connection_id.c_str()); } } @@ -493,6 +541,7 @@ void WebRtc::SendOfferAndIceCandidatesToPeer(const std::string& service_id) { return; } + NEARBY_LOG(WARNING, "Sending offer from %s", service_id.c_str()); if (!connection_info->signaling_messenger->SendMessage( connection_info->peer_id.GetId(), connection_info->pending_local_offer)) { @@ -503,6 +552,8 @@ void WebRtc::SendOfferAndIceCandidatesToPeer(const std::string& service_id) { connection_info->pending_local_offer = ByteArray(); if (!connection_info->pending_local_ice_candidates.empty()) { + NEARBY_LOG(WARNING, "Sending local pending ice candidates from %s", + service_id.c_str()); connection_info->signaling_messenger->SendMessage( connection_info->peer_id.GetId(), webrtc_frames::EncodeIceCandidates( @@ -522,6 +573,7 @@ void WebRtc::SendAnswerToPeer(const std::string& peer_id) { if (!SetLocalSessionDescription(std::move(answer), Role::kAnswerer, peer_id)) return; + NEARBY_LOGS(WARNING) << "Sending answer to peer " << peer_id; if (!connection_info->signaling_messenger->SendMessage( connection_info->peer_id.GetId(), answer_message)) { LogAndDisconnect(Role::kAnswerer, peer_id, @@ -542,8 +594,10 @@ void WebRtc::LogAndDisconnect(const Role& role, void WebRtc::LogAndShutdownSignaling(const Role& role, const std::string& connection_id, const std::string& error_message) { - NEARBY_LOG(WARNING, "Stopping WebRTC role: %d, connection id: %s, msg: %s", - role, connection_id.c_str(), error_message.c_str()); + NEARBY_LOG(WARNING, + "Stopping WebRTC role: %s, connection id: %s, msg: %s:\n%s", + role_names_[role].c_str(), connection_id.c_str(), + error_message.c_str(), InternalStatesToString().c_str()); ShutdownSignaling(role, connection_id); } @@ -580,6 +634,9 @@ void WebRtc::Disconnect(const Role& role, const std::string& connection_id) { void WebRtc::DisconnectLocked(const Role& role, const std::string& connection_id) { + NEARBY_LOGS(WARNING) << "Disconnecting role: " << role_names_[role] + << " connection_id: " << connection_id << ":\n" + << InternalStatesToString(); ShutdownSignaling(role, connection_id); ShutdownWebRtcSocket(role, connection_id); ShutdownIceCandidateCollection(role, connection_id); @@ -617,12 +674,12 @@ void WebRtc::OffloadFromSignalingThread(Runnable runnable) { void WebRtc::RestartReceiveMessages(const LocationHint& location_hint, const std::string& service_id) { if (!IsAcceptingConnections(service_id)) { - NEARBY_LOG(INFO, + NEARBY_LOG(WARNING, "Skipping restart since we are not accepting connections."); return; } - NEARBY_LOG(INFO, "Restarting listening for receiving signaling messages."); + NEARBY_LOG(WARNING, "Restarting listening for receiving signaling messages."); { MutexLock lock(&mutex_); ConnectionInfo* connection_info = @@ -668,6 +725,33 @@ WebRtc::ConnectionInfo* WebRtc::GetConnectionInfo( return nullptr; } +std::string WebRtc::ConnectionInfo::ToString() const { + std::ostringstream result; + result << (connection_flow == nullptr ? "connection_flow is null, " + : "connection_flow is valid, "); + result << (signaling_messenger == nullptr ? "signaling_messenger is null, " + : "signaling_messenger is valid, "); + result << (socket.IsValid() ? "socket is valid, " : "socket is not valid, "); + result << "remote peer_id: " << peer_id.GetId() << ", "; + result << "self peer_id: " << self_id.GetId(); + return result.str(); +} + +std::string WebRtc::InternalStatesToString() { + std::ostringstream map_values; + map_values << "connecting map size: " << connecting_map_.size() + << ", accepting map size: " << accepting_map_.size() << "\n"; + for (auto& item : connecting_map_) { + map_values << "connecting " << item.first << ": " << item.second.ToString() + << "\n"; + } + for (auto& item : accepting_map_) { + map_values << "accepting " << item.first << ": " << item.second.ToString() + << "\n"; + } + return map_values.str(); +} + } // namespace mediums } // namespace connections } // namespace nearby diff --git a/cpp/core/internal/mediums/webrtc.h b/cpp/core/internal/mediums/webrtc.h index cd757775..3f098b5b 100644 --- a/cpp/core/internal/mediums/webrtc.h +++ b/cpp/core/internal/mediums/webrtc.h @@ -86,6 +86,11 @@ class WebRtc { kAnswerer = 2, }; + absl::flat_hash_map role_names_{ + {Role::kNone, "None"}, + {Role::kOfferer, "Offerer"}, + {Role::kAnswerer, "Answerer"}}; + struct ConnectionInfo { std::unique_ptr connection_flow; std::unique_ptr signaling_messenger; @@ -97,6 +102,7 @@ class WebRtc { ByteArray pending_local_offer; std::vector<::location::nearby::mediums::IceCandidate> pending_local_ice_candidates; + std::string ToString() const; }; bool InitWebRtcFlow(const Role& role, const PeerId& self_id, @@ -196,6 +202,8 @@ class WebRtc { const std::string& connection_id) ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + std::string InternalStatesToString() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + Mutex mutex_; WebRtcMedium medium_; diff --git a/cpp/core/internal/mediums/webrtc_test.cc b/cpp/core/internal/mediums/webrtc_test.cc index 9b9b4409..117ffad7 100644 --- a/cpp/core/internal/mediums/webrtc_test.cc +++ b/cpp/core/internal/mediums/webrtc_test.cc @@ -13,7 +13,7 @@ namespace connections { namespace mediums { namespace { - +const int kTwoMBSize = 2000000; class WebRtcTest : public ::testing::Test { protected: WebRtcTest() { @@ -142,12 +142,10 @@ TEST_F(WebRtcTest, ConnectTwice) { ASSERT_TRUE(devices_connected.ok()); EXPECT_TRUE(devices_connected.result()); - WebRtcSocketWrapper socket = - sender.Connect(other_id, location_hint); + WebRtcSocketWrapper socket = sender.Connect(other_id, location_hint); EXPECT_TRUE(socket.IsValid()); socket.Close(); - EXPECT_TRUE(receiver_socket.IsValid()); EXPECT_TRUE(sender_socket.IsValid()); @@ -257,6 +255,64 @@ TEST_F(WebRtcTest, ConnectBothDevices_ShutdownSignaling_SendData) { EXPECT_EQ(message, received_msg.result()); } +// Tests the flow when the two devices created two data channel and transfer +// data in the same time. +TEST_F(WebRtcTest, TwoChannels_SendData) { + WebRtc receiver, sender; + WebRtcSocketWrapper receiver_socket1, receiver_socket2, sender_socket1, + sender_socket2; + const PeerId self_id1("self_id1"), self_id2("self_id2"); + const std::string service_id1("service1"), service_id2("service2"); + LocationHint location_hint; + Future connected1, connected2; + ByteArray message; + message.SetData(kTwoMBSize / 10, 'c'); + + receiver.StartAcceptingConnections( + service_id1, self_id1, location_hint, + {[&receiver_socket1, connected1](WebRtcSocketWrapper wrapper) mutable { + receiver_socket1 = wrapper; + connected1.Set(receiver_socket1.IsValid()); + }}); + + receiver.StartAcceptingConnections( + service_id2, self_id2, location_hint, + {[&receiver_socket2, connected2](WebRtcSocketWrapper wrapper) mutable { + receiver_socket2 = wrapper; + connected2.Set(receiver_socket2.IsValid()); + }}); + + sender_socket1 = sender.Connect(self_id1, location_hint); + EXPECT_TRUE(sender_socket1.IsValid()); + + sender_socket2 = sender.Connect(self_id2, location_hint); + EXPECT_TRUE(sender_socket2.IsValid()); + + ExceptionOr devices_connected1 = connected1.Get(); + ASSERT_TRUE(devices_connected1.ok()); + EXPECT_TRUE(devices_connected1.result()); + + ExceptionOr devices_connected2 = connected1.Get(); + ASSERT_TRUE(devices_connected2.ok()); + EXPECT_TRUE(devices_connected2.result()); + + // Only shuts down signaling channel. + receiver.StopAcceptingConnections(service_id1); + receiver.StopAcceptingConnections(service_id2); + + for (int i = 0; i < 10; i++) { + sender_socket1.GetOutputStream().Write(message); + sender_socket2.GetOutputStream().Write(message); + ExceptionOr received_msg1 = + receiver_socket1.GetInputStream().Read(kTwoMBSize / 10); + ASSERT_TRUE(received_msg1.ok()); + ExceptionOr received_msg2 = + receiver_socket2.GetInputStream().Read(kTwoMBSize / 10); + EXPECT_EQ(message, received_msg1.result()); + EXPECT_EQ(message, received_msg2.result()); + } +} + TEST_F(WebRtcTest, StartAcceptingConnections_NullPeerConnection) { using MockAcceptedCallback = testing::MockFunction; diff --git a/cpp/core/internal/p2p_cluster_pcp_handler.h b/cpp/core/internal/p2p_cluster_pcp_handler.h index 731fc9bb..39d0a165 100644 --- a/cpp/core/internal/p2p_cluster_pcp_handler.h +++ b/cpp/core/internal/p2p_cluster_pcp_handler.h @@ -79,20 +79,6 @@ class P2pClusterPcpHandler : public BasePcpHandler { BasePcpHandler::DiscoveredEndpoint* endpoint) override; private: - struct BluetoothEndpoint : public BasePcpHandler::DiscoveredEndpoint { - BluetoothEndpoint(DiscoveredEndpoint endpoint, BluetoothDevice device) - : DiscoveredEndpoint(std::move(endpoint)), - bluetooth_device(std::move(device)) {} - - BluetoothDevice bluetooth_device; - }; - struct BleEndpoint : public BasePcpHandler::DiscoveredEndpoint { - BleEndpoint(DiscoveredEndpoint endpoint, BlePeripheral peripheral) - : DiscoveredEndpoint(std::move(endpoint)), - ble_peripheral(std::move(peripheral)) {} - BlePeripheral ble_peripheral; - }; - // Holds the state required to re-create a BleEndpoint we see on a // BlePeripheral, so BlePeripheralLostHandler can call // BasePcpHandler::OnEndpointLost() with the same information as was passed @@ -105,13 +91,6 @@ class P2pClusterPcpHandler : public BasePcpHandler { std::string endpoint_id; ByteArray endpoint_info; }; - struct WifiLanEndpoint : public BasePcpHandler::DiscoveredEndpoint { - WifiLanEndpoint(DiscoveredEndpoint endpoint, WifiLanService service) - : DiscoveredEndpoint(std::move(endpoint)), - wifi_lan_service(std::move(service)) {} - - WifiLanService wifi_lan_service; - }; using BluetoothDiscoveredDeviceCallback = BluetoothClassic::DiscoveredDeviceCallback; diff --git a/proto/connections/offline_wire_formats.proto b/proto/connections/offline_wire_formats.proto index ef8350d5..20a21c9d 100644 --- a/proto/connections/offline_wire_formats.proto +++ b/proto/connections/offline_wire_formats.proto @@ -264,6 +264,8 @@ message MediumMetadata { optional string bssid = 2; // IP address optional bytes ip_address = 3; + // True if local device supports 6GHz. + optional bool supports_6_ghz = 4; } // LocationHint is used to specify a location as well as format. diff --git a/proto/discovery_enums.proto b/proto/discovery_enums.proto index 426393cf..ff06a300 100644 --- a/proto/discovery_enums.proto +++ b/proto/discovery_enums.proto @@ -12,7 +12,7 @@ option java_package = "com.google.location.nearby.proto"; option java_outer_classname = "DiscoveryEnums"; option go_api_flag = "OPEN_TO_OPAQUE_HYBRID"; // See http://go/go-api-flag. -// NEXT ID: 133 +// NEXT ID: 147 enum DiscoveryEvent { UNKNOWN_DISCOVERY_EVENT = 0; @@ -400,6 +400,45 @@ enum DiscoveryEvent { // companion app. FAST_PAIR_CONNECTION_TRACKER_RECOVER_COMPANION_APP = 132; + // Half sheet is shown for initial Fast Pair. + HALF_SHEET_PAIR_SHOWN = 133; + + // Half sheet is clicked by users. + HALF_SHEET_PAIR_CLICKED = 134; + + // Half sheet is gone without user interactions. + HALF_SHEET_PAIR_AUTO_DISMISSED = 135; + + // Half sheet is dismissed by users. + HALF_SHEET_PAIR_DISMISSED = 136; + + // Half sheet is banned by users. + HALF_SHEET_PAIR_BANNED = 137; + + // Half sheet is shown for available secondary Fast Pair device. + HALF_SHEET_PAIR_SECONDARY_SHOWN = 138; + + // Half sheet is clicked by users. + HALF_SHEET_PAIR_SECONDARY_CLICKED = 139; + + // Half sheet is gone without user interactions. + HALF_SHEET_PAIR_SECONDARY_AUTO_DISMISSED = 140; + + // Half sheet is dismissed by users. + HALF_SHEET_PAIR_SECONDARY_DISMISSED = 141; + + // Half sheet is banned by users. + HALF_SHEET_PAIR_SECONDARY_BANNED = 142; + + // Half sheet is shown for retroactive pairing of a Fast Pair device. + HALF_SHEET_PAIR_RETROACTIVE_SHOWN = 143; + + // Half sheet is clicked by users. + HALF_SHEET_PAIR_RETROACTIVE_CLICKED = 144; + + // Half sheet is dismissed by users. + HALF_SHEET_PAIR_RETROACTIVE_DISMISSED = 145; + // Deprecated. reserved 65, 67 to 72; }