Merge branch 'google3' up to cl/358285146.

cpp/core/internal/BUILD

@@ -185,6 +185,7 @@ cc_test(
         ":internal_test",
         "//core:core_types",
         "//core/internal/mediums",
+        "//core/internal/mediums:utils",
         "//proto/connections:offline_wire_formats_portable_proto",
         "//platform/base",
         "//platform/base:test_util",

cpp/core/internal/bwu_manager.cc

@@ -205,12 +205,18 @@ void BwuManager::OnIncomingFrame(OfflineFrame& frame,
   if (parser::GetFrameType(frame) != V1Frame::BANDWIDTH_UPGRADE_NEGOTIATION)
     return;
   auto bwu_frame = frame.v1().bandwidth_upgrade_negotiation();
-  CountDownLatch latch(1);
-  RunOnBwuManagerThread([this, client, endpoint_id, &bwu_frame, &latch]() {
-    OnBwuNegotiationFrame(client, bwu_frame, endpoint_id);
-    latch.CountDown();
-  });
-  latch.Await();
+  if (FeatureFlags::GetInstance().GetFlags().enable_async_bandwidth_upgrade) {
+    RunOnBwuManagerThread([this, client, endpoint_id, bwu_frame]() {
+      OnBwuNegotiationFrame(client, bwu_frame, endpoint_id);
+    });
+  } else {
+    CountDownLatch latch(1);
+    RunOnBwuManagerThread([this, client, endpoint_id, bwu_frame, &latch]() {
+      OnBwuNegotiationFrame(client, bwu_frame, endpoint_id);
+      latch.CountDown();
+    });
+    latch.Await();
+  }
 }
 
 void BwuManager::OnEndpointDisconnect(ClientProxy* client,
@@ -274,7 +280,7 @@ void BwuManager::Revert() {
 }
 
 void BwuManager::OnBwuNegotiationFrame(ClientProxy* client,
-                                       const BwuNegotiationFrame& frame,
+                                       const BwuNegotiationFrame frame,
                                        const string& endpoint_id) {
   NEARBY_LOG(INFO, "OnBwuNegotiationFrame for endpoint %s",
              endpoint_id.c_str());

cpp/core/internal/bwu_manager.h

@@ -114,7 +114,7 @@ class BwuManager : public EndpointManager::FrameProcessor {
   // Processes the BwuNegotiationFrames that come over the
   // EndpointChannel on both initiator and responder side of the upgrade.
   void OnBwuNegotiationFrame(ClientProxy* client,
-                             const BwuNegotiationFrame& frame,
+                             const BwuNegotiationFrame frame,
                              const string& endpoint_id);
 
   // Called to revert any state changed by the Initiator or Responder in the

cpp/core/internal/bwu_manager_test.cc

@@ -20,6 +20,7 @@
 #include "core/internal/endpoint_channel_manager.h"
 #include "core/internal/endpoint_manager.h"
 #include "core/internal/mediums/mediums.h"
+#include "core/internal/mediums/utils.h"
 #include "platform/public/system_clock.h"
 #include "gmock/gmock.h"
 #include "gtest/gtest.h"
@@ -56,6 +57,23 @@ TEST(BwuManagerTest, CanInitiateBwu) {
   bwu_manager.Shutdown();
 }
 
+TEST(BwuManagerTest, CanProcessPathAvailableFrame) {
+  ClientProxy client;
+  std::string endpoint_id("EP_A");
+  Mediums mediums;
+  EndpointChannelManager ecm;
+  EndpointManager em{&ecm};
+  BwuManager bwu_manager{mediums, em, ecm, {}, {}};
+
+  LocationHint location_hint = Utils::BuildLocationHint("US");
+  ExceptionOr<OfflineFrame> wrapped_frame = parser::FromBytes(
+      parser::ForBwuWebrtcPathAvailable("my_id", location_hint));
+
+  bwu_manager.OnIncomingFrame(wrapped_frame.result(), endpoint_id, &client,
+                              Medium::WEB_RTC);
+  bwu_manager.Shutdown();
+}
+
 }  // namespace
 }  // namespace connections
 }  // namespace nearby

cpp/platform/base/feature_flags.h

@@ -35,6 +35,7 @@ class FeatureFlags {
     // Ignore subsequent BWU Available events when we're still processing the
     // first one.
     bool disallow_out_of_order_bwu_avail_event = true;
+    bool enable_async_bandwidth_upgrade = true;
   };
 
   static const FeatureFlags& GetInstance() {