Merge branch 'google3' to roll forward to cl/341966255.

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;

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_;

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",

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<bool> 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<void(WebRtcSocketWrapper socket)>;
   testing::StrictMock<MockAcceptedCallback> 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<bool> 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<bool> 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<bool> 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<ByteArray> 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) {

cpp/core/internal/p2p_cluster_pcp_handler.cc

@@ -78,17 +78,6 @@ BasePcpHandler::StartOperationResult P2pClusterPcpHandler::StartAdvertisingImpl(
   std::vector<proto::connections::Medium> 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<WebRtcEndpoint*>(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()) {

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);

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,

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;
   });
 }