diff --git a/cpp/core/internal/mediums/webrtc.cc b/cpp/core/internal/mediums/webrtc.cc index 2ba4ce0e..96cdec3d 100644 --- a/cpp/core/internal/mediums/webrtc.cc +++ b/cpp/core/internal/mediums/webrtc.cc @@ -48,12 +48,14 @@ const std::string WebRtc::GetDefaultCountryCode() { bool WebRtc::IsAvailable() { return medium_.IsValid(); } -bool WebRtc::IsAcceptingConnections() { +bool WebRtc::IsAcceptingConnections(const std::string& service_id) { MutexLock lock(&mutex_); + // TODO(hais): refractor the implementation with maps. return role_ == Role::kOfferer; } bool WebRtc::StartAcceptingConnections(const PeerId& self_id, + const std::string& service_id, const LocationHint& location_hint, AcceptedConnectionCallback callback) { if (!IsAvailable()) { @@ -64,7 +66,7 @@ bool WebRtc::StartAcceptingConnections(const PeerId& self_id, return false; } - if (IsAcceptingConnections()) { + if (IsAcceptingConnections(service_id)) { NEARBY_LOG(WARNING, "Already accepting WebRTC connections."); return false; } @@ -82,7 +84,8 @@ bool WebRtc::StartAcceptingConnections(const PeerId& self_id, restart_receive_messages_alarm_ = CancelableAlarm( "restart_receiving_messages_webrtc", - std::bind(&WebRtc::RestartReceiveMessages, this, location_hint), + std::bind(&WebRtc::RestartReceiveMessages, this, location_hint, + service_id), kRestartReceiveMessagesDuration, &restart_receive_messages_executor_); SessionDescriptionWrapper offer = connection_flow_->CreateOffer(); @@ -154,11 +157,12 @@ bool WebRtc::SetLocalSessionDescription(SessionDescriptionWrapper sdp) { return true; } -void WebRtc::StopAcceptingConnections() { - if (!IsAcceptingConnections()) { +void WebRtc::StopAcceptingConnections(const std::string& service_id) { + if (!IsAcceptingConnections(service_id)) { NEARBY_LOG(INFO, "Skipped StopAcceptingConnections since we are not currently " - "accepting WebRTC connections"); + "accepting WebRTC connections for %s", + service_id.c_str()); return; } @@ -256,8 +260,7 @@ bool WebRtc::InitWebRtcFlow(Role role, const PeerId& self_id, connection_flow_ = ConnectionFlow::Create(GetLocalIceCandidateListener(), GetDataChannelListener(), medium_); - if (!connection_flow_) - return false; + if (!connection_flow_) return false; return true; } @@ -477,8 +480,9 @@ void WebRtc::OffloadFromSignalingThread(Runnable runnable) { single_thread_executor_.Execute(std::move(runnable)); } -void WebRtc::RestartReceiveMessages(const LocationHint& location_hint) { - if (!IsAcceptingConnections()) { +void WebRtc::RestartReceiveMessages(const LocationHint& location_hint, + const std::string& service_id) { + if (!IsAcceptingConnections(service_id)) { NEARBY_LOG(INFO, "Skipping restart since we are not accepting connections."); return; diff --git a/cpp/core/internal/mediums/webrtc.h b/cpp/core/internal/mediums/webrtc.h index b2831fae..33f39008 100644 --- a/cpp/core/internal/mediums/webrtc.h +++ b/cpp/core/internal/mediums/webrtc.h @@ -23,6 +23,7 @@ #include "platform/public/single_thread_executor.h" #include "platform/public/webrtc.h" #include "location/nearby/mediums/proto/web_rtc_signaling_frames.pb.h" +#include "absl/container/flat_hash_set.h" #include "webrtc/api/data_channel_interface.h" #include "webrtc/api/jsep.h" #include "webrtc/api/scoped_refptr.h" @@ -52,22 +53,24 @@ class WebRtc { // Runs on @MainThread. bool IsAvailable(); - // Returns if the device is ready to accept connections from remote devices. + // Returns if the device is accepting connection with specific service id. // Runs on @MainThread. - bool IsAcceptingConnections() ABSL_LOCKS_EXCLUDED(mutex_); + bool IsAcceptingConnections(const std::string& service_id) + 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 std::string& service_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_); + // Try to stop (accepting) the specific connection with provided service id. + // Runs on @MainThread + void StopAcceptingConnections(const std::string& service_id) + ABSL_LOCKS_EXCLUDED(mutex_); // Initiates a WebRtc connection with peer device identified by |peer_id|. // Runs on @MainThread. @@ -145,7 +148,8 @@ class WebRtc { void OffloadFromSignalingThread(Runnable runnable); // Runs on |restart_receive_messages_executor_|. - void RestartReceiveMessages(const LocationHint& location_hint) + void RestartReceiveMessages(const LocationHint& location_hint, + const std::string& service_id) ABSL_LOCKS_EXCLUDED(mutex_); Mutex mutex_; diff --git a/cpp/core/internal/mediums/webrtc/BUILD b/cpp/core/internal/mediums/webrtc/BUILD index cb614815..9cd3593d 100644 --- a/cpp/core/internal/mediums/webrtc/BUILD +++ b/cpp/core/internal/mediums/webrtc/BUILD @@ -46,6 +46,7 @@ cc_test( "signaling_frames_test.cc", "webrtc_socket_test.cc", ], + tags = ["notsan"], # NOTE(b/139734036): known data race in usrsctplib. deps = [ ":webrtc", "//platform/base", diff --git a/cpp/core/internal/mediums/webrtc_test.cc b/cpp/core/internal/mediums/webrtc_test.cc index 4874e105..ae5c599e 100644 --- a/cpp/core/internal/mediums/webrtc_test.cc +++ b/cpp/core/internal/mediums/webrtc_test.cc @@ -26,7 +26,7 @@ class WebRtcTest : public ::testing::Test { TEST_F(WebRtcTest, NotAcceptingConnections) { WebRtc webrtc; ASSERT_TRUE(webrtc.IsAvailable()); - EXPECT_FALSE(webrtc.IsAcceptingConnections()); + EXPECT_FALSE(webrtc.IsAcceptingConnections(std::string{})); } // Tests the flow when the device tries to accept connections twice. In this @@ -38,14 +38,17 @@ TEST_F(WebRtcTest, StartAcceptingConnectionTwice) { WebRtc webrtc; PeerId self_id("peer_id"); + const std::string service_id("NearbySharing"); LocationHint location_hint{}; ASSERT_TRUE(webrtc.IsAvailable()); ASSERT_TRUE(webrtc.StartAcceptingConnections( - self_id, location_hint, {mock_accepted_callback_.AsStdFunction()})); + self_id, service_id, location_hint, + {mock_accepted_callback_.AsStdFunction()})); EXPECT_FALSE(webrtc.StartAcceptingConnections( - self_id, location_hint, {mock_accepted_callback_.AsStdFunction()})); - EXPECT_TRUE(webrtc.IsAcceptingConnections()); + self_id, service_id, location_hint, + {mock_accepted_callback_.AsStdFunction()})); + EXPECT_TRUE(webrtc.IsAcceptingConnections(std::string{})); } // Tests the flow when the device tries to connect but the data channel times @@ -53,14 +56,15 @@ TEST_F(WebRtcTest, StartAcceptingConnectionTwice) { TEST_F(WebRtcTest, Connect_DataChannelTimeOut) { WebRtc webrtc; PeerId peer_id("peer_id"); + const std::string service_id("NearbySharing"); LocationHint location_hint; ASSERT_TRUE(webrtc.IsAvailable()); WebRtcSocketWrapper wrapper_1 = webrtc.Connect(peer_id, location_hint); EXPECT_FALSE(wrapper_1.IsValid()); - EXPECT_TRUE(webrtc.StartAcceptingConnections(peer_id, location_hint, - AcceptedConnectionCallback())); + EXPECT_TRUE(webrtc.StartAcceptingConnections( + peer_id, service_id, location_hint, AcceptedConnectionCallback())); } // Tests the flow when the device calls Connect() after calling @@ -72,17 +76,20 @@ TEST_F(WebRtcTest, StartAcceptingConnection_ThenConnect) { WebRtc webrtc; PeerId self_id("peer_id"); + const std::string service_id("NearbySharing"); LocationHint location_hint; ASSERT_TRUE(webrtc.IsAvailable()); ASSERT_TRUE(webrtc.StartAcceptingConnections( - self_id, location_hint, {mock_accepted_callback_.AsStdFunction()})); + self_id, service_id, location_hint, + {mock_accepted_callback_.AsStdFunction()})); WebRtcSocketWrapper wrapper = webrtc.Connect(PeerId("random_peer_id"), location_hint); - EXPECT_TRUE(webrtc.IsAcceptingConnections()); + EXPECT_TRUE(webrtc.IsAcceptingConnections(std::string{})); EXPECT_FALSE(wrapper.IsValid()); EXPECT_FALSE(webrtc.StartAcceptingConnections( - self_id, location_hint, {mock_accepted_callback_.AsStdFunction()})); + self_id, service_id, location_hint, + {mock_accepted_callback_.AsStdFunction()})); } // Tests the flow when the device calls StartAcceptingConnections but the medium @@ -94,13 +101,15 @@ TEST_F(WebRtcTest, StartAndStopAcceptingConnections) { WebRtc webrtc; PeerId self_id("peer_id"); + const std::string service_id("NearbySharing"); LocationHint location_hint; ASSERT_TRUE(webrtc.IsAvailable()); ASSERT_TRUE(webrtc.StartAcceptingConnections( - self_id, location_hint, {mock_accepted_callback_.AsStdFunction()})); - webrtc.StopAcceptingConnections(); - EXPECT_FALSE(webrtc.IsAcceptingConnections()); + self_id, service_id, location_hint, + {mock_accepted_callback_.AsStdFunction()})); + webrtc.StopAcceptingConnections(service_id); + EXPECT_FALSE(webrtc.IsAcceptingConnections(std::string{})); } // Tests the flow when the device tries to connect to two different peers @@ -109,12 +118,13 @@ TEST_F(WebRtcTest, ConnectTwice) { WebRtc receiver, sender, device_c; WebRtcSocketWrapper receiver_socket, sender_socket; const PeerId self_id("self_id"), other_id("other_id"); + const std::string service_id("NearbySharing"); LocationHint location_hint; Future connected; ByteArray message("message xyz"); receiver.StartAcceptingConnections( - self_id, location_hint, + self_id, service_id, location_hint, {[&receiver_socket, connected](WebRtcSocketWrapper wrapper) mutable { receiver_socket = wrapper; connected.Set(receiver_socket.IsValid()); @@ -123,7 +133,7 @@ TEST_F(WebRtcTest, ConnectTwice) { using MockAcceptedCallback = testing::MockFunction; testing::StrictMock mock_accepted_callback_; - device_c.StartAcceptingConnections(other_id, location_hint, + device_c.StartAcceptingConnections(other_id, service_id, location_hint, {mock_accepted_callback_.AsStdFunction()}); sender_socket = sender.Connect(self_id, location_hint); @@ -154,12 +164,13 @@ TEST_F(WebRtcTest, ConnectBothDevicesAndAbort) { WebRtc receiver, sender; WebRtcSocketWrapper receiver_socket, sender_socket; const PeerId self_id("self_id"); + const std::string service_id("NearbySharing"); LocationHint location_hint; Future connected; ByteArray message("message xyz"); receiver.StartAcceptingConnections( - self_id, location_hint, + self_id, service_id, location_hint, {[&receiver_socket, connected](WebRtcSocketWrapper wrapper) mutable { receiver_socket = wrapper; connected.Set(receiver_socket.IsValid()); @@ -181,12 +192,13 @@ TEST_F(WebRtcTest, ConnectBothDevicesAndSendData) { WebRtc receiver, sender; WebRtcSocketWrapper receiver_socket, sender_socket; const PeerId self_id("self_id"); + const std::string service_id("NearbySharing"); LocationHint location_hint; Future connected; ByteArray message("message"); receiver.StartAcceptingConnections( - self_id, location_hint, + self_id, service_id, location_hint, {[&receiver_socket, connected](WebRtcSocketWrapper wrapper) mutable { receiver_socket = wrapper; connected.Set(receiver_socket.IsValid()); @@ -214,12 +226,13 @@ TEST_F(WebRtcTest, ConnectBothDevices_ShutdownSignaling_SendData) { WebRtc receiver, sender; WebRtcSocketWrapper receiver_socket, sender_socket; const PeerId self_id("self_id"); + const std::string service_id("NearbySharing"); LocationHint location_hint; Future connected; ByteArray message("message xyz"); receiver.StartAcceptingConnections( - self_id, location_hint, + self_id, service_id, location_hint, {[&receiver_socket, connected](WebRtcSocketWrapper wrapper) mutable { receiver_socket = wrapper; connected.Set(receiver_socket.IsValid()); @@ -233,7 +246,7 @@ TEST_F(WebRtcTest, ConnectBothDevices_ShutdownSignaling_SendData) { EXPECT_TRUE(devices_connected.result()); // Only shuts down signaling channel. - receiver.StopAcceptingConnections(); + receiver.StopAcceptingConnections(service_id); sender_socket.GetOutputStream().Write(message); ExceptionOr received_msg = @@ -252,11 +265,13 @@ TEST_F(WebRtcTest, StartAcceptingConnections_NullPeerConnection) { WebRtc webrtc; PeerId self_id("peer_id"); + const std::string service_id("NearbySharing"); LocationHint location_hint; ASSERT_TRUE(webrtc.IsAvailable()); EXPECT_FALSE(webrtc.StartAcceptingConnections( - self_id, location_hint, {mock_accepted_callback_.AsStdFunction()})); + self_id, service_id, location_hint, + {mock_accepted_callback_.AsStdFunction()})); } TEST_F(WebRtcTest, Connect_NullPeerConnection) { diff --git a/cpp/core/internal/p2p_cluster_pcp_handler.cc b/cpp/core/internal/p2p_cluster_pcp_handler.cc index ba022bd3..17ba3bcd 100644 --- a/cpp/core/internal/p2p_cluster_pcp_handler.cc +++ b/cpp/core/internal/p2p_cluster_pcp_handler.cc @@ -78,17 +78,6 @@ BasePcpHandler::StartOperationResult P2pClusterPcpHandler::StartAdvertisingImpl( std::vector mediums_started_successfully; WebRtcState web_rtc_state{WebRtcState::kUnconnectable}; - if (options.allowed.web_rtc) { - proto::connections::Medium webrtc_medium = - StartListeningForWebRtcConnections( - client, service_id, local_endpoint_id, local_endpoint_info); - if (webrtc_medium != proto::connections::UNKNOWN_MEDIUM) { - NEARBY_LOG(INFO, - "P2pClusterPcpHandler::StartAdvertisingImpl: WebRtc added"); - mediums_started_successfully.push_back(webrtc_medium); - web_rtc_state = WebRtcState::kConnectable; - } - } if (options.allowed.wifi_lan) { const ByteArray wifi_lan_hash = @@ -142,10 +131,6 @@ BasePcpHandler::StartOperationResult P2pClusterPcpHandler::StartAdvertisingImpl( }; } -// StopAcceptingConnections invokes for webrtc is suppressed for now to -// unblock CrOS dogfood integration. Disconnect will invoke ShutdownSignaling -// to release resources. -// TODO (hais): add corresponding logic back (b/172518506). Status P2pClusterPcpHandler::StopAdvertisingImpl(ClientProxy* client) { bluetooth_medium_.TurnOffDiscoverability(); bluetooth_medium_.StopAcceptingConnections(client->GetAdvertisingServiceId()); @@ -194,8 +179,8 @@ bool P2pClusterPcpHandler::IsRecognizedBluetoothEndpoint( void P2pClusterPcpHandler::BluetoothDeviceDiscoveredHandler( ClientProxy* client, const std::string& service_id, - BluetoothDevice& device) { - RunOnPcpHandlerThread([this, client, service_id, &device]() { + BluetoothDevice device) { + RunOnPcpHandlerThread([this, client, service_id, device]() { // Make sure we are still discovering before proceeding. if (!client->IsDiscovering()) { NEARBY_LOG(INFO, @@ -206,7 +191,7 @@ void P2pClusterPcpHandler::BluetoothDeviceDiscoveredHandler( } // Parse the Bluetooth device name. - const std::string& device_name_string = device.GetName(); + const std::string device_name_string = device.GetName(); BluetoothDeviceName device_name(device_name_string); // Make sure the Bluetooth device name points to a valid @@ -691,10 +676,6 @@ BasePcpHandler::ConnectImplResult P2pClusterPcpHandler::ConnectImpl( break; } case proto::connections::Medium::WEB_RTC: { - auto* webrtc_endpoint = down_cast(endpoint); - if (webrtc_endpoint) { - return WebRtcConnectImpl(client, webrtc_endpoint); - } break; } default: @@ -1156,12 +1137,12 @@ P2pClusterPcpHandler::StartListeningForWebRtcConnections( return proto::connections::UNKNOWN_MEDIUM; } - if (!webrtc_medium_.IsAcceptingConnections()) { + if (!webrtc_medium_.IsAcceptingConnections(service_id)) { mediums::PeerId self_id = CreatePeerIdFromAdvertisement( service_id, local_endpoint_id, local_endpoint_info); std::string empty_country_code; if (!webrtc_medium_.StartAcceptingConnections( - self_id, Utils::BuildLocationHint(empty_country_code), + self_id, service_id, Utils::BuildLocationHint(empty_country_code), {[this, client, local_endpoint_info](mediums::WebRtcSocketWrapper socket) { if (!socket.IsValid()) { diff --git a/cpp/core/internal/p2p_cluster_pcp_handler.h b/cpp/core/internal/p2p_cluster_pcp_handler.h index 4922d6d0..58fe7d23 100644 --- a/cpp/core/internal/p2p_cluster_pcp_handler.h +++ b/cpp/core/internal/p2p_cluster_pcp_handler.h @@ -134,7 +134,7 @@ class P2pClusterPcpHandler : public BasePcpHandler { const BluetoothDeviceName& name) const; void BluetoothDeviceDiscoveredHandler(ClientProxy* client, const std::string& service_id, - BluetoothDevice& device); + BluetoothDevice device); void BluetoothDeviceLostHandler(ClientProxy* client, const std::string& service_id, BluetoothDevice& device); diff --git a/cpp/core/internal/webrtc_bwu_handler.cc b/cpp/core/internal/webrtc_bwu_handler.cc index 4d217013..4bd3e269 100644 --- a/cpp/core/internal/webrtc_bwu_handler.cc +++ b/cpp/core/internal/webrtc_bwu_handler.cc @@ -20,10 +20,10 @@ WebrtcBwuHandler::WebrtcBwuHandler(Mediums& mediums, mediums_(mediums) {} void WebrtcBwuHandler::Revert() { - if (!active_service_ids_.empty()) { - webrtc_.StopAcceptingConnections(); - active_service_ids_.clear(); + for (const auto& service_id : active_service_ids_) { + webrtc_.StopAcceptingConnections(service_id); } + active_service_ids_.clear(); NEARBY_LOG(INFO, "WebrtcBwuHandler successfully reverted state."); } @@ -61,9 +61,9 @@ ByteArray WebrtcBwuHandler::InitializeUpgradedMediumForEndpoint( Utils::BuildLocationHint(webrtc_.GetDefaultCountryCode()); mediums::PeerId self_id{mediums::PeerId::FromRandom()}; - if (!webrtc_.IsAcceptingConnections()) { + if (!webrtc_.IsAcceptingConnections(service_id)) { if (!webrtc_.StartAcceptingConnections( - self_id, location_hint, + self_id, upgrade_service_id, location_hint, { .accepted_cb = absl::bind_front( &WebrtcBwuHandler::OnIncomingWebrtcConnection, this, client, diff --git a/cpp/platform/base/medium_environment.cc b/cpp/platform/base/medium_environment.cc index 068dcd1a..7d70af45 100644 --- a/cpp/platform/base/medium_environment.cc +++ b/cpp/platform/base/medium_environment.cc @@ -290,9 +290,7 @@ void MediumEnvironment::UnregisterBluetoothMedium( RunOnMediumEnvironmentThread([this, &medium]() { auto item = bluetooth_mediums_.extract(&medium); if (item.empty()) return; - auto& context = item.mapped(); - NEARBY_LOG(INFO, "Unregistered medium for device=%s", - context.adapter->GetName().c_str()); + NEARBY_LOGS(INFO) << "Unregistered Bluetooth medium:" << &medium; }); }