diff --git a/cpp/core/internal/base_pcp_handler.h b/cpp/core/internal/base_pcp_handler.h index def460a3..08f4874e 100644 --- a/cpp/core/internal/base_pcp_handler.h +++ b/cpp/core/internal/base_pcp_handler.h @@ -220,6 +220,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 cfa7034d..a729e095 100644 --- a/cpp/core/internal/mediums/webrtc.cc +++ b/cpp/core/internal/mediums/webrtc.cc @@ -16,6 +16,7 @@ #include #include +#include #include "core/internal/mediums/webrtc/session_description_wrapper.h" #include "core/internal/mediums/webrtc/signaling_frames.h" @@ -49,6 +50,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(); @@ -103,6 +108,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]; @@ -121,6 +128,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; } @@ -131,7 +139,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()); } @@ -141,12 +149,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, @@ -163,7 +175,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; { @@ -181,8 +193,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(); } @@ -204,7 +222,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()); @@ -213,9 +231,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( @@ -275,7 +295,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) { @@ -330,6 +353,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; } @@ -345,6 +371,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, @@ -435,7 +463,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, @@ -458,12 +491,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; @@ -471,28 +504,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()); } } @@ -507,6 +555,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)) { @@ -517,6 +566,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( @@ -536,6 +587,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, @@ -556,8 +608,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); } @@ -594,6 +648,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); @@ -631,12 +688,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 = @@ -682,6 +739,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 4252efcb..0be2ecfc 100644 --- a/cpp/core/internal/mediums/webrtc.h +++ b/cpp/core/internal/mediums/webrtc.h @@ -100,6 +100,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; @@ -111,6 +116,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, @@ -210,6 +216,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 0572c7f4..567a46dd 100644 --- a/cpp/core/internal/mediums/webrtc_test.cc +++ b/cpp/core/internal/mediums/webrtc_test.cc @@ -27,7 +27,7 @@ namespace connections { namespace mediums { namespace { - +const int kTwoMBSize = 2000000; class WebRtcTest : public ::testing::Test { protected: WebRtcTest() { @@ -156,12 +156,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()); @@ -271,6 +269,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 c9c3060b..43c58b61 100644 --- a/cpp/core/internal/p2p_cluster_pcp_handler.h +++ b/cpp/core/internal/p2p_cluster_pcp_handler.h @@ -93,20 +93,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 @@ -119,13 +105,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 39d522c2..0a3de54d 100644 --- a/proto/connections/offline_wire_formats.proto +++ b/proto/connections/offline_wire_formats.proto @@ -278,6 +278,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 bccc5737..d63ce004 100644 --- a/proto/discovery_enums.proto +++ b/proto/discovery_enums.proto @@ -20,7 +20,7 @@ option optimize_for = LITE_RUNTIME; option java_package = "com.google.location.nearby.proto"; option java_outer_classname = "DiscoveryEnums"; -// NEXT ID: 133 +// NEXT ID: 147 enum DiscoveryEvent { UNKNOWN_DISCOVERY_EVENT = 0; @@ -408,6 +408,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; }