diff --git a/cpp/core/internal/BUILD b/cpp/core/internal/BUILD
index 9ce76017..ff017980 100644
--- a/cpp/core/internal/BUILD
+++ b/cpp/core/internal/BUILD
@@ -84,6 +84,7 @@ cc_library(
     ],
     visibility = [
         "//core:__pkg__",
+        "//core/internal/fuzzers:__pkg__",
     ],
     deps = [
         ":message_lite",
@@ -94,6 +95,7 @@ cc_library(
         "//proto/connections:offline_wire_formats_portable_proto",
         "//platform/api:comm",
         "//platform/base",
+        "//platform/base:cancellation_flag",
         "//platform/base:util",
         "//platform/public:comm",
         "//platform/public:logging",
diff --git a/cpp/core/internal/base_pcp_handler.cc b/cpp/core/internal/base_pcp_handler.cc
index c568d192..1cddd548 100644
--- a/cpp/core/internal/base_pcp_handler.cc
+++ b/cpp/core/internal/base_pcp_handler.cc
@@ -95,14 +95,12 @@ Status BasePcpHandler::StartAdvertising(ClientProxy* client,
     // Now that we've succeeded, mark the client as advertising.
     // Save the advertising options for local reference in later process like
     // upgrading bandwidth.
-    // TODO(hais): saving advertising_options_ in clientProxy instead of here
-    // as java implementation does.
     std::vector<proto::connections::Medium> supported_mediums =
         advertising_options.GetMediums();
-    advertising_options_ = advertising_options;
     advertising_listener_ = info.listener;
     client->StartedAdvertising(service_id, GetStrategy(), info.listener,
-                               absl::MakeSpan(result.mediums));
+                               absl::MakeSpan(result.mediums),
+                               advertising_options);
     response.Set({Status::kSuccess});
   });
   return WaitForResult(
@@ -115,7 +113,6 @@ void BasePcpHandler::StopAdvertising(ClientProxy* client) {
   RunOnPcpHandlerThread([this, client, &latch]() {
     StopAdvertisingImpl(client);
     client->StoppedAdvertising();
-    // advertising_options_ is purposefully not cleared here.
     latch.CountDown();
   });
   WaitForLatch("StopAdvertising", &latch);
@@ -142,23 +139,22 @@ Status BasePcpHandler::StartDiscovery(ClientProxy* client,
 
   NEARBY_LOG(INFO, "StartDiscovery with supported mediums: %s",
              GetStringValueOfSupportedMediums(options).c_str());
-  RunOnPcpHandlerThread(
-      [this, client, service_id, discovery_options, &listener, &response]() {
-        // Ask the implementation to attempt to start discovery.
-        auto result = StartDiscoveryImpl(client, service_id, discovery_options);
-        if (!result.status.Ok()) {
-          response.Set(result.status);
-          return;
-        }
+  RunOnPcpHandlerThread([this, client, service_id, discovery_options, &listener,
+                         &response]() {
+    // Ask the implementation to attempt to start discovery.
+    auto result = StartDiscoveryImpl(client, service_id, discovery_options);
+    if (!result.status.Ok()) {
+      response.Set(result.status);
+      return;
+    }
 
-        // Now that we've succeeded, mark the client as discovering and clear
-        // out any old endpoints we had discovered.
-        discovery_options_ = discovery_options;
-        discovered_endpoints_.clear();
-        client->StartedDiscovery(service_id, GetStrategy(), listener,
-                                 absl::MakeSpan(result.mediums));
-        response.Set({Status::kSuccess});
-      });
+    // Now that we've succeeded, mark the client as discovering and clear
+    // out any old endpoints we had discovered.
+    discovered_endpoints_.clear();
+    client->StartedDiscovery(service_id, GetStrategy(), listener,
+                             absl::MakeSpan(result.mediums), discovery_options);
+    response.Set({Status::kSuccess});
+  });
   return WaitForResult(absl::StrCat("StartDiscovery(", service_id, ")"),
                        client->GetClientId(), &response);
 }
@@ -168,7 +164,6 @@ void BasePcpHandler::StopDiscovery(ClientProxy* client) {
   RunOnPcpHandlerThread([this, client, &latch]() {
     StopDiscoveryImpl(client);
     client->StoppedDiscovery();
-    // discovery_options_ is purposefully not cleared here.
     latch.CountDown();
   });
 
@@ -346,7 +341,7 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client,
 
     // If our child class says we can't send any more outgoing connections,
     // listen to them.
-    if (ShouldEnforceTopologyConstraints() &&
+    if (ShouldEnforceTopologyConstraints(client->GetAdvertisingOptions()) &&
         !CanSendOutgoingConnection(client)) {
       NEARBY_LOG(INFO, "Outgoing connection not allowed: id=%s",
                  endpoint_id.c_str());
@@ -365,13 +360,14 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client,
     auto remote_bluetooth_mac_address =
         BluetoothUtils::ToString(options.remote_bluetooth_mac_address);
     if (!remote_bluetooth_mac_address.empty()) {
-      if (AppendRemoteBluetoothMacAddressEndpoint(endpoint_id,
-                                                  remote_bluetooth_mac_address))
+      if (AppendRemoteBluetoothMacAddressEndpoint(
+              endpoint_id, remote_bluetooth_mac_address,
+              client->GetDiscoveryOptions()))
         NEARBY_LOGS(INFO) << "Appended remote Bluetooth MAC Address endpoint "
                           << "[" << remote_bluetooth_mac_address << "]";
     }
 
-    if (AppendWebRTCEndpoint(endpoint_id))
+    if (AppendWebRTCEndpoint(endpoint_id, client->GetDiscoveryOptions()))
       NEARBY_LOGS(INFO) << "Appended Web RTC endpoint.";
 
     auto discovered_endpoints = GetDiscoveredEndpoints(endpoint_id);
@@ -380,7 +376,7 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client,
 
     for (auto connect_endpoint : discovered_endpoints) {
       if (!MediumSupportedByClientOptions(connect_endpoint->medium,
-                                          discovery_options_))
+                                          client->GetDiscoveryOptions()))
         continue;
       connect_impl_result = ConnectImpl(client, connect_endpoint);
       if (connect_impl_result.status.Ok()) {
@@ -405,7 +401,7 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client,
     // endpoint about ourselves.
     Exception write_exception = WriteConnectionRequestFrame(
         channel.get(), client->GetLocalEndpointId(), info.endpoint_info, nonce,
-        GetSupportedConnectionMediumsByPriority(discovery_options_));
+        GetSupportedConnectionMediumsByPriority(client->GetDiscoveryOptions()));
     if (!write_exception.Ok()) {
       NEARBY_LOG(INFO, "Failed to send connection request: id=%s",
                  endpoint_id.c_str());
@@ -467,7 +463,7 @@ bool BasePcpHandler::MediumSupportedByClientOptions(
 }
 
 // Get ordered supported connection medium based on local advertising/discovery
-// option. local_option is either advertising_options_ or discovery_options_.
+// option.
 std::vector<proto::connections::Medium>
 BasePcpHandler::GetSupportedConnectionMediumsByPriority(
     const ConnectionOptions& local_option) {
@@ -501,6 +497,19 @@ BasePcpHandler::GetDiscoveredEndpoints(const std::string& endpoint_id) {
             [this](DiscoveredEndpoint* a, DiscoveredEndpoint* b) -> bool {
               return IsPreferred(*a, *b);
             });
+
+  return result;
+}
+
+std::vector<BasePcpHandler::DiscoveredEndpoint*>
+BasePcpHandler::GetDiscoveredEndpoints(
+    const proto::connections::Medium medium) {
+  std::vector<BasePcpHandler::DiscoveredEndpoint*> result;
+  for (const auto& item : discovered_endpoints_) {
+    if (item.second->medium == medium) {
+      result.push_back(item.second.get());
+    }
+  }
   return result;
 }
 
@@ -569,22 +578,24 @@ void BasePcpHandler::ProcessPreConnectionResultFailure(
   client->OnConnectionRejected(endpoint_id, {Status::kError});
 }
 
-bool BasePcpHandler::ShouldEnforceTopologyConstraints() const {
+bool BasePcpHandler::ShouldEnforceTopologyConstraints(
+    const ConnectionOptions& local_advertising_options) const {
   // Topology constraints only matter for the advertiser.
   // For discoverers, we'll always enforce them.
-  if (advertising_options_.strategy.IsNone()) {
+  if (local_advertising_options.strategy.IsNone()) {
     return true;
   }
 
-  return advertising_options_.enforce_topology_constraints;
+  return local_advertising_options.enforce_topology_constraints;
 }
 
-bool BasePcpHandler::AutoUpgradeBandwidth() const {
-  if (advertising_options_.strategy.IsNone()) {
+bool BasePcpHandler::AutoUpgradeBandwidth(
+    const ConnectionOptions& local_advertising_options) const {
+  if (local_advertising_options.strategy.IsNone()) {
     return true;
   }
 
-  return advertising_options_.auto_upgrade_bandwidth;
+  return local_advertising_options.auto_upgrade_bandwidth;
 }
 
 Status BasePcpHandler::AcceptConnection(
@@ -765,14 +776,6 @@ BluetoothDevice BasePcpHandler::GetRemoteBluetoothDevice(
       remote_bluetooth_mac_address);
 }
 
-ConnectionOptions BasePcpHandler::GetConnectionOptions() const {
-  return advertising_options_;
-}
-
-ConnectionOptions BasePcpHandler::GetDiscoveryOptions() const {
-  return discovery_options_;
-}
-
 void BasePcpHandler::OnEndpointFound(
     ClientProxy* client, std::shared_ptr<DiscoveredEndpoint> endpoint) {
   // Check if we've seen this endpoint ID before.
@@ -933,7 +936,7 @@ Exception BasePcpHandler::OnIncomingConnection(
 
   // If our child class says we can't accept any more incoming connections,
   // listen to them.
-  if (ShouldEnforceTopologyConstraints() &&
+  if (ShouldEnforceTopologyConstraints(client->GetAdvertisingOptions()) &&
       !CanReceiveIncomingConnection(client)) {
     return {Exception::kIo};
   }
@@ -1029,15 +1032,18 @@ void BasePcpHandler::InitiateBandwidthUpgrade(
   // sense to dynamically select the proper medium for upgrading.
   // TODO(hais): when we add more mediums like Wifi Hotspot, we need to prevent
   // upgrading interfering with active connections.
-  Medium bwu_medium = ChooseBestUpgradeMedium(their_supported_mediums);
+  Medium bwu_medium = ChooseBestUpgradeMedium(their_supported_mediums,
+                                              client->GetAdvertisingOptions());
 
-  if (AutoUpgradeBandwidth() && bwu_medium != Medium::UNKNOWN_MEDIUM) {
+  if (AutoUpgradeBandwidth(client->GetAdvertisingOptions()) &&
+      bwu_medium != Medium::UNKNOWN_MEDIUM) {
     bwu_manager_->InitiateBwuForEndpoint(client, endpoint_id, bwu_medium);
   }
 }
 
 proto::connections::Medium BasePcpHandler::ChooseBestUpgradeMedium(
-    const std::vector<proto::connections::Medium>& their_supported_mediums) {
+    const std::vector<proto::connections::Medium>& their_supported_mediums,
+    const ConnectionOptions& local_advertising_options) {
   // If the remote side did not report their supported mediums, choose an
   // appropriate default.
   std::vector<proto::connections::Medium> their_mediums =
@@ -1048,7 +1054,7 @@ proto::connections::Medium BasePcpHandler::ChooseBestUpgradeMedium(
 
   // Otherwise, pick the best medium we support.
   std::vector<proto::connections::Medium> my_mediums =
-      GetSupportedConnectionMediumsByPriority(advertising_options_);
+      GetSupportedConnectionMediumsByPriority(local_advertising_options);
   for (const auto& my_medium : my_mediums) {
     for (const auto& their_medium : their_mediums) {
       if (my_medium == their_medium) {
@@ -1062,8 +1068,9 @@ proto::connections::Medium BasePcpHandler::ChooseBestUpgradeMedium(
 
 bool BasePcpHandler::AppendRemoteBluetoothMacAddressEndpoint(
     const std::string& endpoint_id,
-    const std::string& remote_bluetooth_mac_address) {
-  if (!discovery_options_.allowed.bluetooth) {
+    const std::string& remote_bluetooth_mac_address,
+    const ConnectionOptions& local_discovery_options) {
+  if (!local_discovery_options.allowed.bluetooth) {
     return false;
   }
 
@@ -1103,8 +1110,10 @@ bool BasePcpHandler::AppendRemoteBluetoothMacAddressEndpoint(
   return true;
 }
 
-bool BasePcpHandler::AppendWebRTCEndpoint(const std::string& endpoint_id) {
-  if (!discovery_options_.allowed.web_rtc) {
+bool BasePcpHandler::AppendWebRTCEndpoint(
+    const std::string& endpoint_id,
+    const ConnectionOptions& local_discovery_options) {
+  if (!local_discovery_options.allowed.web_rtc) {
     return false;
   }
 
diff --git a/cpp/core/internal/base_pcp_handler.h b/cpp/core/internal/base_pcp_handler.h
index efc7dfcf..a160ad6a 100644
--- a/cpp/core/internal/base_pcp_handler.h
+++ b/cpp/core/internal/base_pcp_handler.h
@@ -311,6 +311,10 @@ class BasePcpHandler : public PcpHandler,
   std::vector<BasePcpHandler::DiscoveredEndpoint*> GetDiscoveredEndpoints(
       const std::string& endpoint_id);
 
+  // Returns a vector of discovered endpoints that share a given Medium.
+  std::vector<BasePcpHandler::DiscoveredEndpoint*> GetDiscoveredEndpoints(
+      const proto::connections::Medium medium);
+
   mediums::PeerId CreatePeerIdFromAdvertisement(const string& service_id,
                                                 const string& endpoint_id,
                                                 const ByteArray& endpoint_info);
@@ -407,11 +411,13 @@ class BasePcpHandler : public PcpHandler,
                    const BasePcpHandler::DiscoveredEndpoint& old_endpoint);
 
   // Returns true, if connection party should respect the specified topology.
-  bool ShouldEnforceTopologyConstraints() const;
+  bool ShouldEnforceTopologyConstraints(
+      const ConnectionOptions& local_advertising_options) const;
 
   // Returns true, if connection party should attempt to upgrade itself to
   // use a higher bandwidth medium, if it is available.
-  bool AutoUpgradeBandwidth() const;
+  bool AutoUpgradeBandwidth(
+      const ConnectionOptions& local_advertising_options) const;
 
   // Returns true if the incoming connection should be killed. This only
   // happens when an incoming connection arrives while we have an outgoing
@@ -438,18 +444,21 @@ class BasePcpHandler : public PcpHandler,
 
   // Returns the optimal medium supported by both devices.
   proto::connections::Medium ChooseBestUpgradeMedium(
-      const std::vector<proto::connections::Medium>& supported_mediums);
+      const std::vector<proto::connections::Medium>& supported_mediums,
+      const ConnectionOptions& local_advertising_options);
 
   // Returns true if the bluetooth endpoint based on remote bluetooth mac
   // address is created and appended into discovered_endpoints_ with key
   // endpoint_id.
   bool AppendRemoteBluetoothMacAddressEndpoint(
       const std::string& endpoint_id,
-      const std::string& remote_bluetooth_mac_address);
+      const std::string& remote_bluetooth_mac_address,
+      const ConnectionOptions& local_discovery_options);
 
   // Returns true if the webrtc endpoint is created and appended into
   // discovered_endpoints_ with key endpoint_id.
-  bool AppendWebRTCEndpoint(const std::string& endpoint_id);
+  bool AppendWebRTCEndpoint(const std::string& endpoint_id,
+                            const ConnectionOptions& local_discovery_options);
 
   void ProcessPreConnectionInitiationFailure(const std::string& endpoint_id,
                                              EndpointChannel* channel,
@@ -506,22 +515,10 @@ class BasePcpHandler : public PcpHandler,
   // doesn't happen.
   absl::flat_hash_map<std::string, CancelableAlarm> pending_alarms_;
 
-  // The active ClientProxy's advertising constraints. Empty()
-  // returns true if the client hasn't started advertising false otherwise.
-  // Note: this is not cleared when the client stops advertising because it
-  // might still be useful downstream of advertising (eg: establishing
-  // connections, performing bandwidth upgrades, etc.)
-  ConnectionOptions advertising_options_;
   // The active ClientProxy's connection lifecycle listener. Non-null while
   // advertising.
   ConnectionListener advertising_listener_;
 
-  // The active ClientProxy's discovery constraints. Null if the client
-  // hasn't started discovering. Note: this is not cleared when the client
-  // stops discovering because it might still be useful downstream of
-  // discovery (eg: connection speed, etc.)
-  ConnectionOptions discovery_options_;
-
   AtomicBoolean stop_{false};
   Pcp pcp_;
   Strategy strategy_{PcpToStrategy(pcp_)};
diff --git a/cpp/core/internal/base_pcp_handler_test.cc b/cpp/core/internal/base_pcp_handler_test.cc
index 1bcc98aa..a38580c7 100644
--- a/cpp/core/internal/base_pcp_handler_test.cc
+++ b/cpp/core/internal/base_pcp_handler_test.cc
@@ -130,7 +130,8 @@ class MockPcpHandler : public BasePcpHandler {
   MOCK_METHOD(Status, StopDiscoveryImpl, (ClientProxy * client), (override));
   MOCK_METHOD(Status, InjectEndpointImpl,
               (ClientProxy * client, const std::string& service_id,
-              const OutOfBandConnectionMetadata& metadata), (override));
+               const OutOfBandConnectionMetadata& metadata),
+              (override));
   MOCK_METHOD(ConnectImplResult, ConnectImpl,
               (ClientProxy * client, DiscoveredEndpoint* endpoint), (override));
   MOCK_METHOD(proto::connections::Medium, GetDefaultUpgradeMedium, (),
@@ -138,7 +139,9 @@ class MockPcpHandler : public BasePcpHandler {
 
   std::vector<proto::connections::Medium> GetConnectionMediumsByPriority()
       override {
-    return GetDiscoveryMediums();
+    return std::vector<proto::connections::Medium>{
+        proto::connections::WIFI_LAN, proto::connections::WEB_RTC,
+        proto::connections::BLUETOOTH, proto::connections::BLE};
   }
 
   // Mock adapters for protected non-virtual methods of a base class.
@@ -154,9 +157,9 @@ class MockPcpHandler : public BasePcpHandler {
     return BasePcpHandler::GetDiscoveredEndpoints(endpoint_id);
   }
 
-  std::vector<proto::connections::Medium> GetDiscoveryMediums() {
-    auto allowed =
-        BasePcpHandler::GetDiscoveryOptions().CompatibleOptions().allowed;
+  std::vector<proto::connections::Medium> GetDiscoveryMediums(
+      ClientProxy* client) {
+    auto allowed = client->GetDiscoveryOptions().CompatibleOptions().allowed;
     return GetMediumsFromSelector(allowed);
   }
 
@@ -322,7 +325,7 @@ class BasePcpHandlerTest
     EXPECT_CALL(mock_connection_listener_.initiated_cb, Call).Times(1);
     // Simulate successful discovery.
     auto encryption_runner = std::make_unique<EncryptionRunner>();
-    auto allowed_mediums = pcp_handler->GetDiscoveryMediums();
+    auto allowed_mediums = pcp_handler->GetDiscoveryMediums(client);
 
     EXPECT_CALL(*pcp_handler, ConnectImpl)
         .WillOnce(Invoke([&channel_a, connect_medium](
@@ -457,7 +460,7 @@ TEST_P(BasePcpHandlerTest, RequestConnectionChangesState) {
   BwuManager bwu(m, em, ecm, {}, {});
   MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
   StartDiscovery(&client, &pcp_handler);
-  auto mediums = pcp_handler.GetDiscoveryMediums();
+  auto mediums = pcp_handler.GetDiscoveryMediums(&client);
   auto connect_medium = mediums[mediums.size() - 1];
   auto channel_pair = SetupConnection(pipe_a_, pipe_b_, connect_medium);
   auto& channel_a = channel_pair.first;
@@ -482,7 +485,7 @@ TEST_P(BasePcpHandlerTest, AcceptConnectionChangesState) {
   BwuManager bwu(m, em, ecm, {}, {});
   MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
   StartDiscovery(&client, &pcp_handler);
-  auto mediums = pcp_handler.GetDiscoveryMediums();
+  auto mediums = pcp_handler.GetDiscoveryMediums(&client);
   auto connect_medium = mediums[mediums.size() - 1];
   auto channel_pair = SetupConnection(pipe_a_, pipe_b_, connect_medium);
   auto& channel_a = channel_pair.first;
@@ -511,7 +514,7 @@ TEST_P(BasePcpHandlerTest, RejectConnectionChangesState) {
   BwuManager bwu(m, em, ecm, {}, {});
   MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
   StartDiscovery(&client, &pcp_handler);
-  auto mediums = pcp_handler.GetDiscoveryMediums();
+  auto mediums = pcp_handler.GetDiscoveryMediums(&client);
   auto connect_medium = mediums[mediums.size() - 1];
   auto channel_pair = SetupConnection(pipe_a_, pipe_b_, connect_medium);
   auto& channel_b = channel_pair.second;
@@ -536,7 +539,7 @@ TEST_P(BasePcpHandlerTest, OnIncomingFrameChangesState) {
   BwuManager bwu(m, em, ecm, {}, {});
   MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
   StartDiscovery(&client, &pcp_handler);
-  auto mediums = pcp_handler.GetDiscoveryMediums();
+  auto mediums = pcp_handler.GetDiscoveryMediums(&client);
   auto connect_medium = mediums[mediums.size() - 1];
   auto channel_pair = SetupConnection(pipe_a_, pipe_b_, connect_medium);
   auto& channel_a = channel_pair.first;
@@ -574,7 +577,7 @@ TEST_P(BasePcpHandlerTest, DestructorIsCalledOnProtocolEndpoint) {
     BwuManager bwu(m, em, ecm, {}, {});
     MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
     StartDiscovery(&client, &pcp_handler);
-    auto mediums = pcp_handler.GetDiscoveryMediums();
+    auto mediums = pcp_handler.GetDiscoveryMediums(&client);
     auto connect_medium = mediums[mediums.size() - 1];
     auto channel_pair = SetupConnection(pipe_a_, pipe_b_, connect_medium);
     auto& channel_a = channel_pair.first;
@@ -616,7 +619,7 @@ TEST_P(BasePcpHandlerTest, MultipleMediumsProduceSingleEndpointLostEvent) {
     BwuManager bwu(m, em, ecm, {}, {});
     MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
     StartDiscovery(&client, &pcp_handler);
-    auto mediums = pcp_handler.GetDiscoveryMediums();
+    auto mediums = pcp_handler.GetDiscoveryMediums(&client);
     auto connect_medium = mediums[mediums.size() - 1];
     auto channel_pair = SetupConnection(pipe_a_, pipe_b_, connect_medium);
     auto& channel_a = channel_pair.first;
@@ -626,7 +629,7 @@ TEST_P(BasePcpHandlerTest, MultipleMediumsProduceSingleEndpointLostEvent) {
     EXPECT_CALL(mock_discovery_listener_.endpoint_lost_cb, Call).Times(1);
     RequestConnection(endpoint_id, std::move(channel_a), channel_b.get(),
                       &client, &pcp_handler, connect_medium, &destroyed_flag);
-    auto allowed_mediums = pcp_handler.GetDiscoveryMediums();
+    auto allowed_mediums = pcp_handler.GetDiscoveryMediums(&client);
     mediums_count = allowed_mediums.size();
     NEARBY_LOG(INFO, "Attempting to accept connection: id=%s",
                endpoint_id.c_str());
@@ -657,9 +660,11 @@ TEST_F(BasePcpHandlerTest, InjectEndpoint) {
   EndpointManager em(&ecm);
   BwuManager bwu(m, em, ecm, {}, {});
   MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
-  BooleanMediumSelector allowed{ .bluetooth = true, };
+  BooleanMediumSelector allowed{
+      .bluetooth = true,
+  };
   ConnectionOptions options{
-      .allowed =  allowed,
+      .allowed = allowed,
       .is_out_of_band_connection = true,
   };
   EXPECT_CALL(mock_discovery_listener_.endpoint_found_cb, Call);
@@ -675,9 +680,8 @@ TEST_F(BasePcpHandlerTest, InjectEndpoint) {
 
   EXPECT_CALL(pcp_handler, InjectEndpointImpl(&client, service_id, _))
       .WillOnce(Invoke([&pcp_handler, &endpoint_id](
-          ClientProxy* client,
-          const std::string& service_id,
-          const OutOfBandConnectionMetadata& metadata) {
+                           ClientProxy* client, const std::string& service_id,
+                           const OutOfBandConnectionMetadata& metadata) {
         pcp_handler.OnEndpointFound(
             client,
             std::make_shared<MockDiscoveredEndpoint>(MockDiscoveredEndpoint{
@@ -692,11 +696,12 @@ TEST_F(BasePcpHandlerTest, InjectEndpoint) {
             }));
         return Status{Status::kSuccess};
       }));
-  pcp_handler.InjectEndpoint(&client, service_id,
-        OutOfBandConnectionMetadata{
-            .medium = Medium::BLUETOOTH,
-            .remote_bluetooth_mac_address = ByteArray(kFakeMacAddress),
-        });
+  pcp_handler.InjectEndpoint(
+      &client, service_id,
+      OutOfBandConnectionMetadata{
+          .medium = Medium::BLUETOOTH,
+          .remote_bluetooth_mac_address = ByteArray(kFakeMacAddress),
+      });
   bwu.Shutdown();
 }
 
diff --git a/cpp/core/internal/bluetooth_bwu_handler.cc b/cpp/core/internal/bluetooth_bwu_handler.cc
index 07eed2e6..efbe1c58 100644
--- a/cpp/core/internal/bluetooth_bwu_handler.cc
+++ b/cpp/core/internal/bluetooth_bwu_handler.cc
@@ -110,7 +110,8 @@ BluetoothBwuHandler::CreateUpgradedEndpointChannel(
     return nullptr;
   }
 
-  BluetoothSocket socket = bluetooth_medium_.Connect(device, service_name);
+  BluetoothSocket socket = bluetooth_medium_.Connect(
+      device, service_name, client->GetCancellationFlag(endpoint_id));
   if (!socket.IsValid()) {
     return nullptr;
   }
diff --git a/cpp/core/internal/bwu_manager.cc b/cpp/core/internal/bwu_manager.cc
index 1df953b5..57e85c95 100644
--- a/cpp/core/internal/bwu_manager.cc
+++ b/cpp/core/internal/bwu_manager.cc
@@ -32,7 +32,6 @@ namespace location {
 namespace nearby {
 namespace connections {
 
-using ::location::nearby::proto::connections::ConnectionAttemptResult;
 using ::location::nearby::proto::connections::DisconnectionReason;
 
 // Required for C++ 14 support in Chrome
@@ -391,13 +390,6 @@ void BwuManager::ProcessBwuPathAvailableEvent(
 
   auto channel = ProcessBwuPathAvailableEventInternal(client, endpoint_id,
                                                       upgrade_path_info);
-  ConnectionAttemptResult connectionAttemptResult;
-  if (channel != nullptr) {
-    connectionAttemptResult = ConnectionAttemptResult::RESULT_SUCCESS;
-  } else {
-    connectionAttemptResult = ConnectionAttemptResult::RESULT_ERROR;
-  }
-
   if (channel == nullptr) {
     RunUpgradeFailedProtocol(client, endpoint_id, upgrade_path_info);
     return;
diff --git a/cpp/core/internal/client_proxy.cc b/cpp/core/internal/client_proxy.cc
index 010d9712..aac4e09c 100644
--- a/cpp/core/internal/client_proxy.cc
+++ b/cpp/core/internal/client_proxy.cc
@@ -68,9 +68,11 @@ void ClientProxy::Reset() {
 void ClientProxy::StartedAdvertising(
     const std::string& service_id, Strategy strategy,
     const ConnectionListener& listener,
-    absl::Span<proto::connections::Medium> mediums) {
+    absl::Span<proto::connections::Medium> mediums,
+    const ConnectionOptions& advertising_options) {
   MutexLock lock(&mutex_);
   advertising_info_ = {service_id, listener};
+  advertising_options_ = advertising_options;
 }
 
 void ClientProxy::StoppedAdvertising() {
@@ -79,6 +81,7 @@ void ClientProxy::StoppedAdvertising() {
   if (IsAdvertising()) {
     advertising_info_.Clear();
   }
+  // advertising_options_ is purposefully not cleared here.
   ResetLocalEndpointIdIfNeeded();
 }
 
@@ -103,9 +106,11 @@ std::string ClientProxy::GetServiceId() const {
 void ClientProxy::StartedDiscovery(
     const std::string& service_id, Strategy strategy,
     const DiscoveryListener& listener,
-    absl::Span<proto::connections::Medium> mediums) {
+    absl::Span<proto::connections::Medium> mediums,
+    const ConnectionOptions& discovery_options) {
   MutexLock lock(&mutex_);
   discovery_info_ = DiscoveryInfo{service_id, listener};
+  discovery_options_ = discovery_options;
 }
 
 void ClientProxy::StoppedDiscovery() {
@@ -115,6 +120,7 @@ void ClientProxy::StoppedDiscovery() {
     discovered_endpoint_ids_.clear();
     discovery_info_.Clear();
   }
+  // discovery_options_ is purposefully not cleared here.
   ResetLocalEndpointIdIfNeeded();
 }
 
@@ -202,6 +208,11 @@ void ClientProxy::OnConnectionInitiated(const std::string& endpoint_id,
   // Note: we allow devices to connect to an advertiser even after it stops
   // advertising, so no need to check IsAdvertising() here.
   item.connection_listener.initiated_cb(endpoint_id, info);
+
+  if (info.is_incoming_connection) {
+    // Add CancellationFlag for advertisers once encryption succeeds.
+    AddCancellationFlag(endpoint_id);
+  }
 }
 
 void ClientProxy::OnConnectionAccepted(const std::string& endpoint_id) {
@@ -262,6 +273,8 @@ void ClientProxy::OnDisconnected(const std::string& endpoint_id, bool notify) {
     connections_.erase(endpoint_id);
     ResetLocalEndpointIdIfNeeded();
   }
+
+  CancelEndpoint(endpoint_id);
 }
 
 bool ClientProxy::ConnectionStatusMatches(const std::string& endpoint_id,
@@ -457,6 +470,42 @@ bool ClientProxy::RemoteConnectionIsAccepted(std::string endpoint_id) const {
       endpoint_id, ClientProxy::Connection::kRemoteEndpointAccepted);
 }
 
+void ClientProxy::AddCancellationFlag(const std::string& endpoint_id) {
+  auto item = cancellation_flags_.find(endpoint_id);
+  if (item != cancellation_flags_.end()) {
+    return;
+  }
+  cancellation_flags_.emplace(endpoint_id,
+                              std::make_unique<CancellationFlag>());
+}
+
+CancellationFlag* ClientProxy::GetCancellationFlag(
+    const std::string& endpoint_id) {
+  const auto item = cancellation_flags_.find(endpoint_id);
+  if (item == cancellation_flags_.end()) {
+    return default_cancellation_flag_.get();
+  }
+  return item->second.get();
+}
+
+void ClientProxy::CancelEndpoint(const std::string& endpoint_id) {
+  const auto item = cancellation_flags_.find(endpoint_id);
+  if (item == cancellation_flags_.end()) return;
+  item->second->Cancel();
+  cancellation_flags_.erase(item);
+}
+
+void ClientProxy::CancelAllEndpoints() {
+  for (const auto& item : cancellation_flags_) {
+    CancellationFlag* cancellation_flag = item.second.get();
+    if (cancellation_flag->Cancelled()) {
+      continue;
+    }
+    cancellation_flag->Cancel();
+  }
+  cancellation_flags_.clear();
+}
+
 void ClientProxy::OnPayload(const std::string& endpoint_id, Payload payload) {
   MutexLock lock(&mutex_);
 
@@ -507,6 +556,7 @@ void ClientProxy::RemoveAllEndpoints() {
   // endpoint, in the case when this is called from stopAllEndpoints(). For now,
   // just remove without notifying.
   connections_.clear();
+  cancellation_flags_.clear();
   local_endpoint_id_.clear();
 }
 
@@ -535,6 +585,14 @@ void ClientProxy::AppendConnectionStatus(const std::string& endpoint_id,
   }
 }
 
+ConnectionOptions ClientProxy::GetAdvertisingOptions() const {
+  return advertising_options_;
+}
+
+ConnectionOptions ClientProxy::GetDiscoveryOptions() const {
+  return discovery_options_;
+}
+
 }  // namespace connections
 }  // namespace nearby
 }  // namespace location
diff --git a/cpp/core/internal/client_proxy.h b/cpp/core/internal/client_proxy.h
index 3aa5b667..58a81de6 100644
--- a/cpp/core/internal/client_proxy.h
+++ b/cpp/core/internal/client_proxy.h
@@ -24,6 +24,7 @@
 #include "core/status.h"
 #include "core/strategy.h"
 #include "platform/base/byte_array.h"
+#include "platform/base/cancellation_flag.h"
 #include "platform/base/prng.h"
 #include "platform/public/mutex.h"
 #include "proto/connections_enums.pb.h"
@@ -59,7 +60,8 @@ class ClientProxy final {
   void StartedAdvertising(
       const std::string& service_id, Strategy strategy,
       const ConnectionListener& connection_lifecycle_listener,
-      absl::Span<proto::connections::Medium> mediums);
+      absl::Span<proto::connections::Medium> mediums,
+      const ConnectionOptions& advertising_options = ConnectionOptions{});
   // Marks this client as not advertising.
   void StoppedAdvertising();
   bool IsAdvertising() const;
@@ -70,9 +72,11 @@ class ClientProxy final {
   std::string GetServiceId() const;
 
   // Marks this client as discovering with the given callback.
-  void StartedDiscovery(const std::string& service_id, Strategy strategy,
-                        const DiscoveryListener& discovery_listener,
-                        absl::Span<proto::connections::Medium> mediums);
+  void StartedDiscovery(
+      const std::string& service_id, Strategy strategy,
+      const DiscoveryListener& discovery_listener,
+      absl::Span<proto::connections::Medium> mediums,
+      const ConnectionOptions& discovery_options = ConnectionOptions{});
   // Marks this client as not discovering at all.
   void StoppedDiscovery();
   bool IsDiscoveringServiceId(const std::string& service_id) const;
@@ -153,6 +157,17 @@ class ClientProxy final {
   bool LocalConnectionIsAccepted(std::string endpoint_id) const;
   bool RemoteConnectionIsAccepted(std::string endpoint_id) const;
 
+  // Adds a CancellationFlag for endpoint id.
+  void AddCancellationFlag(const std::string& endpoint_id);
+  // Returns the CancellationFlag for endpoint id,
+  CancellationFlag* GetCancellationFlag(const std::string& endpoint_id);
+  // Sets the CancellationFlag to true for endpoint id.
+  void CancelEndpoint(const std::string& endpoint_id);
+  // Cancels all CancellationFlags.
+  void CancelAllEndpoints();
+  ConnectionOptions GetAdvertisingOptions() const;
+  ConnectionOptions GetDiscoveryOptions() const;
+
  private:
   struct Connection {
     // Status: may be either:
@@ -221,6 +236,19 @@ class ClientProxy final {
   // If not empty, we are currently discovering for the given service_id.
   DiscoveryInfo discovery_info_;
 
+  // The active ClientProxy's advertising constraints. Empty()
+  // returns true if the client hasn't started advertising false otherwise.
+  // Note: this is not cleared when the client stops advertising because it
+  // might still be useful downstream of advertising (eg: establishing
+  // connections, performing bandwidth upgrades, etc.)
+  ConnectionOptions advertising_options_;
+
+  // The active ClientProxy's discovery constraints. Null if the client
+  // hasn't started discovering. Note: this is not cleared when the client
+  // stops discovering because it might still be useful downstream of
+  // discovery (eg: connection speed, etc.)
+  ConnectionOptions discovery_options_;
+
   // Maps endpoint_id to endpoint connection state.
   absl::flat_hash_map<std::string, Connection> connections_;
 
@@ -230,6 +258,13 @@ class ClientProxy final {
   // happen because some mediums (like Bluetooth) repeatedly give us the same
   // endpoints after each scan.
   absl::flat_hash_set<std::string> discovered_endpoint_ids_;
+
+  // Maps endpoint_id to CancellationFlag.
+  absl::flat_hash_map<std::string, std::unique_ptr<CancellationFlag>>
+      cancellation_flags_;
+  // A default cancellation flag with isCancelled set be true.
+  std::unique_ptr<CancellationFlag> default_cancellation_flag_ =
+      std::make_unique<CancellationFlag>(true);
 };
 
 // Operator overloads when comparing Ptr<ClientProxy>.
diff --git a/cpp/core/internal/client_proxy_test.cc b/cpp/core/internal/client_proxy_test.cc
index f3ee1726..2a1b1735 100644
--- a/cpp/core/internal/client_proxy_test.cc
+++ b/cpp/core/internal/client_proxy_test.cc
@@ -93,8 +93,7 @@ class ClientProxyTest : public testing::Test {
 
   void OnDiscoveryEndpointFound(ClientProxy* client, const Endpoint& endpoint) {
     EXPECT_CALL(mock_discovery_.endpoint_found_cb, Call).Times(1);
-    client->OnEndpointFound(service_id_, endpoint.id, endpoint.info,
-                            medium_);
+    client->OnEndpointFound(service_id_, endpoint.id, endpoint.info, medium_);
   }
 
   void OnDiscoveryEndpointLost(ClientProxy* client, const Endpoint& endpoint) {
@@ -112,6 +111,8 @@ class ClientProxyTest : public testing::Test {
                                   connection_options_,
                                   discovery_connection_listener_);
     EXPECT_TRUE(client->HasPendingConnectionToEndpoint(endpoint.id));
+    // Cancellation flag has been created and added into map.
+    EXPECT_FALSE(client->GetCancellationFlag(endpoint.id)->Cancelled());
   }
 
   void OnDiscoveryConnectionLocalAccepted(ClientProxy* client,
@@ -174,6 +175,8 @@ class ClientProxyTest : public testing::Test {
                                          const Endpoint& endpoint) {
     EXPECT_CALL(mock_discovery_connection_.disconnected_cb, Call).Times(1);
     client->OnDisconnected(endpoint.id, true);
+    // The Cancelled is always true as the default flag being returned.
+    EXPECT_TRUE(client->GetCancellationFlag(endpoint.id)->Cancelled());
   }
 
   void OnPayload(ClientProxy* client, const Endpoint& endpoint) {
@@ -235,8 +238,7 @@ TEST_F(ClientProxyTest, ClientIdIsUnique) {
 }
 
 TEST_F(ClientProxyTest, GeneratedEndpointIdIsUnique) {
-  EXPECT_NE(client1_.GetLocalEndpointId(),
-            client2_.GetLocalEndpointId());
+  EXPECT_NE(client1_.GetLocalEndpointId(), client2_.GetLocalEndpointId());
 }
 
 TEST_F(ClientProxyTest, ResetClearsState) {
@@ -369,6 +371,72 @@ TEST_F(ClientProxyTest, OnPayloadProgressChangesState) {
   OnPayloadProgress(&client2_, advertising_endpoint);
 }
 
+TEST_F(ClientProxyTest, CanCancelEndpoint) {
+  Endpoint advertising_endpoint =
+      StartAdvertising(&client1_, advertising_connection_listener_);
+  StartDiscovery(&client2_, discovery_listener_);
+  OnDiscoveryEndpointFound(&client2_, advertising_endpoint);
+  OnDiscoveryConnectionInitiated(&client2_, advertising_endpoint);
+
+  EXPECT_FALSE(
+      client2_.GetCancellationFlag(advertising_endpoint.id)->Cancelled());
+
+  client2_.CancelEndpoint(advertising_endpoint.id);
+
+  // The Cancelled is always true as the default flag being returned.
+  EXPECT_TRUE(
+      client2_.GetCancellationFlag(advertising_endpoint.id)->Cancelled());
+}
+
+TEST_F(ClientProxyTest, CanCancelAllEndpoints) {
+  Endpoint advertising_endpoint =
+      StartAdvertising(&client1_, advertising_connection_listener_);
+  StartDiscovery(&client2_, discovery_listener_);
+  OnDiscoveryEndpointFound(&client2_, advertising_endpoint);
+  OnDiscoveryConnectionInitiated(&client2_, advertising_endpoint);
+
+  EXPECT_FALSE(
+      client2_.GetCancellationFlag(advertising_endpoint.id)->Cancelled());
+
+  client2_.CancelAllEndpoints();
+
+  // The Cancelled is always true as the default flag being returned.
+  EXPECT_TRUE(
+      client2_.GetCancellationFlag(advertising_endpoint.id)->Cancelled());
+}
+
+TEST_F(ClientProxyTest, CanCancelAllEndpointsWithDifferentEndpoint) {
+  ConnectionListener advertising_connection_listener_2;
+  ConnectionListener advertising_connection_listener_3;
+  ClientProxy client3;
+
+  StartDiscovery(&client1_, discovery_listener_);
+  Endpoint advertising_endpoint_2 =
+      StartAdvertising(&client2_, advertising_connection_listener_2);
+  Endpoint advertising_endpoint_3 =
+      StartAdvertising(&client3, advertising_connection_listener_3);
+  OnDiscoveryEndpointFound(&client1_, advertising_endpoint_2);
+  OnDiscoveryConnectionInitiated(&client1_, advertising_endpoint_2);
+  OnDiscoveryEndpointFound(&client1_, advertising_endpoint_3);
+  OnDiscoveryConnectionInitiated(&client1_, advertising_endpoint_3);
+
+  // The CancellationFlag of endpoint_2 and endpoint_3 have been added. Default
+  // Cancelled is false.
+  EXPECT_FALSE(
+      client1_.GetCancellationFlag(advertising_endpoint_2.id)->Cancelled());
+  EXPECT_FALSE(
+      client1_.GetCancellationFlag(advertising_endpoint_3.id)->Cancelled());
+
+  client1_.CancelAllEndpoints();
+
+  // Expect the CancellationFlag of endpoint_2 and endpoint_3 has been removed.
+  // The Cancelled is always true as the default flag being returned.
+  EXPECT_TRUE(
+      client1_.GetCancellationFlag(advertising_endpoint_2.id)->Cancelled());
+  EXPECT_TRUE(
+      client1_.GetCancellationFlag(advertising_endpoint_3.id)->Cancelled());
+}
+
 }  // namespace
 }  // namespace connections
 }  // namespace nearby
diff --git a/cpp/core/internal/endpoint_manager.cc b/cpp/core/internal/endpoint_manager.cc
index 3c080e3a..95616f3d 100644
--- a/cpp/core/internal/endpoint_manager.cc
+++ b/cpp/core/internal/endpoint_manager.cc
@@ -310,8 +310,6 @@ EndpointManager::FrameProcessor* EndpointManager::GetFrameProcessor(
     latch.CountDown();
   });
   latch.Await();
-  NEARBY_LOG(INFO, "GetFrameProcessor: type=%d; processor=%p", frame_type,
-             processor);
   return processor;
 }
 
diff --git a/cpp/core/internal/fuzzers/BUILD b/cpp/core/internal/fuzzers/BUILD
new file mode 100644
index 00000000..0c113e12
--- /dev/null
+++ b/cpp/core/internal/fuzzers/BUILD
@@ -0,0 +1,12 @@
+load("//security/fuzzing/blaze:cc_fuzz_target.bzl", "cc_fuzz_target")
+
+cc_fuzz_target(
+    name = "offline_frames_fuzzer",
+    srcs = ["offline_frames_fuzzer.cc"],
+    componentid = 148515,
+    deps = [
+        "//core/internal",
+        "//platform/base",
+        "//security/fuzzing/blaze:default_init_google_for_cc_fuzz_target",
+    ],
+)
diff --git a/cpp/core/internal/fuzzers/offline_frames_fuzzer.cc b/cpp/core/internal/fuzzers/offline_frames_fuzzer.cc
new file mode 100644
index 00000000..50a6062d
--- /dev/null
+++ b/cpp/core/internal/fuzzers/offline_frames_fuzzer.cc
@@ -0,0 +1,11 @@
+#include "core/internal/offline_frames.h"
+#include "platform/base/byte_array.h"
+
+extern "C" int LLVMFuzzerTestOneInput(const uint8_t* data, size_t size) {
+  location::nearby::ByteArray byte_array;
+  byte_array.SetData(reinterpret_cast<const char*>(data), size);
+
+  location::nearby::connections::parser::FromBytes(byte_array);
+
+  return 0;
+}
diff --git a/cpp/core/internal/mediums/BUILD b/cpp/core/internal/mediums/BUILD
index d52e2ac9..59c3fd33 100644
--- a/cpp/core/internal/mediums/BUILD
+++ b/cpp/core/internal/mediums/BUILD
@@ -45,6 +45,7 @@ cc_library(
         "//core/internal/mediums/webrtc",
         "//proto/connections:offline_wire_formats_portable_proto",
         "//platform/base",
+        "//platform/base:cancellation_flag",
         "//platform/public:comm",
         "//platform/public:logging",
         "//platform/public:types",
diff --git a/cpp/core/internal/mediums/ble.cc b/cpp/core/internal/mediums/ble.cc
index 3c21d8ce..91191bc7 100644
--- a/cpp/core/internal/mediums/ble.cc
+++ b/cpp/core/internal/mediums/ble.cc
@@ -304,8 +304,9 @@ bool Ble::IsAcceptingConnectionsLocked(const std::string& service_id) {
   return accepting_connections_info_.Existed(service_id);
 }
 
-BleSocket Ble::Connect(BlePeripheral& peripheral,
-                       const std::string& service_id) {
+// TODO(b/169303284): Handles Cancellation and registration.
+BleSocket Ble::Connect(BlePeripheral& peripheral, const std::string& service_id,
+                       CancellationFlag* cancellation_flag) {
   MutexLock lock(&mutex_);
   NEARBY_LOGS(INFO) << "BLE::Connect: service=" << &peripheral;
   // Socket to return. To allow for NRVO to work, it has to be a single object.
diff --git a/cpp/core/internal/mediums/ble.h b/cpp/core/internal/mediums/ble.h
index 8ced26a9..b8263773 100644
--- a/cpp/core/internal/mediums/ble.h
+++ b/cpp/core/internal/mediums/ble.h
@@ -21,6 +21,7 @@
 #include "core/internal/mediums/bluetooth_radio.h"
 #include "core/listeners.h"
 #include "platform/base/byte_array.h"
+#include "platform/base/cancellation_flag.h"
 #include "platform/public/ble.h"
 #include "platform/public/multi_thread_executor.h"
 #include "platform/public/mutex.h"
@@ -98,7 +99,8 @@ class Ble {
   // service_id. Blocks until connection is established, or server-side is
   // terminated. Returns socket instance. On success, BleSocket.IsValid() return
   // true.
-  BleSocket Connect(BlePeripheral& peripheral, const std::string& service_id)
+  BleSocket Connect(BlePeripheral& peripheral, const std::string& service_id,
+                    CancellationFlag* cancellation_flag)
       ABSL_LOCKS_EXCLUDED(mutex_);
 
  private:
diff --git a/cpp/core/internal/mediums/ble_test.cc b/cpp/core/internal/mediums/ble_test.cc
index 95b32d7d..b420ae40 100644
--- a/cpp/core/internal/mediums/ble_test.cc
+++ b/cpp/core/internal/mediums/ble_test.cc
@@ -172,10 +172,8 @@ TEST_F(BleTest, CanStartAcceptingConnectionsAndConnect) {
 
   EXPECT_TRUE(found_latch.Await(kWaitDuration).result());
   ASSERT_TRUE(discovered_peripheral.IsValid());
-
-  BleSocket socket =
-      ble_b.Connect(discovered_peripheral, service_id);
-
+  CancellationFlag flag;
+  BleSocket socket = ble_b.Connect(discovered_peripheral, service_id, &flag);
   EXPECT_TRUE(accept_latch.Await(kWaitDuration).result());
   EXPECT_TRUE(socket.IsValid());
   ble_b.StopScanning(service_id);
diff --git a/cpp/core/internal/mediums/bluetooth_classic.cc b/cpp/core/internal/mediums/bluetooth_classic.cc
index a00fcd6f..beff1280 100644
--- a/cpp/core/internal/mediums/bluetooth_classic.cc
+++ b/cpp/core/internal/mediums/bluetooth_classic.cc
@@ -344,8 +344,10 @@ bool BluetoothClassic::StopAcceptingConnections(
   return true;
 }
 
+// TODO(b/169303284): Handles Cancellation and registration.
 BluetoothSocket BluetoothClassic::Connect(BluetoothDevice& bluetooth_device,
-                                          const std::string& service_name) {
+                                          const std::string& service_name,
+                                          CancellationFlag* cancellation_flag) {
   for (int attempts_count = 0; attempts_count < kConnectAttemptsLimit;
        attempts_count++) {
     auto wrapper_result = AttemptToConnect(bluetooth_device, service_name);
diff --git a/cpp/core/internal/mediums/bluetooth_classic.h b/cpp/core/internal/mediums/bluetooth_classic.h
index b5bda949..15a8d2d7 100644
--- a/cpp/core/internal/mediums/bluetooth_classic.h
+++ b/cpp/core/internal/mediums/bluetooth_classic.h
@@ -21,6 +21,7 @@
 #include "core/internal/mediums/bluetooth_radio.h"
 #include "core/listeners.h"
 #include "platform/base/byte_array.h"
+#include "platform/base/cancellation_flag.h"
 #include "platform/public/bluetooth_adapter.h"
 #include "platform/public/bluetooth_classic.h"
 #include "platform/public/multi_thread_executor.h"
@@ -111,7 +112,8 @@ class BluetoothClassic {
   // Returns socket instance. On success, BluetoothSocket.IsValid() return true.
   // Called by client.
   BluetoothSocket Connect(BluetoothDevice& bluetooth_device,
-                          const std::string& service_name)
+                          const std::string& service_name,
+                          CancellationFlag* cancellation_flag)
       ABSL_LOCKS_EXCLUDED(mutex_);
 
   std::string GetMacAddress() const ABSL_LOCKS_EXCLUDED(mutex_);
diff --git a/cpp/core/internal/mediums/bluetooth_classic_test.cc b/cpp/core/internal/mediums/bluetooth_classic_test.cc
index 49dd9f2b..02945400 100644
--- a/cpp/core/internal/mediums/bluetooth_classic_test.cc
+++ b/cpp/core/internal/mediums/bluetooth_classic_test.cc
@@ -194,8 +194,9 @@ TEST_F(BluetoothClassicTest, CanConnect) {
                 accept_latch.CountDown();
               },
       }));
+  CancellationFlag flag;
   BluetoothSocket socket_for_client =
-      bt_client.Connect(discovered_device, std::string(kServiceName));
+      bt_client.Connect(discovered_device, std::string(kServiceName), &flag);
   EXPECT_TRUE(accept_latch.Await(kWaitDuration).result());
   EXPECT_TRUE(bt_server.StopAcceptingConnections(std::string(kServiceName)));
   EXPECT_TRUE(socket_for_server.IsValid());
diff --git a/cpp/core/internal/mediums/webrtc.cc b/cpp/core/internal/mediums/webrtc.cc
index 719f7ad9..8447f9ea 100644
--- a/cpp/core/internal/mediums/webrtc.cc
+++ b/cpp/core/internal/mediums/webrtc.cc
@@ -208,7 +208,8 @@ void WebRtc::StopAcceptingConnections(const std::string& service_id) {
 
 WebRtcSocketWrapper WebRtc::Connect(const std::string& service_id,
                                     const PeerId& remote_peer_id,
-                                    const LocationHint& location_hint) {
+                                    const LocationHint& location_hint,
+                                    CancellationFlag* cancellation_flag) {
   for (int attempts_count = 0; attempts_count < kConnectAttemptsLimit;
        attempts_count++) {
     auto wrapper_result =
diff --git a/cpp/core/internal/mediums/webrtc.h b/cpp/core/internal/mediums/webrtc.h
index 3e18f50c..494695aa 100644
--- a/cpp/core/internal/mediums/webrtc.h
+++ b/cpp/core/internal/mediums/webrtc.h
@@ -27,6 +27,7 @@
 #include "proto/connections/offline_wire_formats.pb.h"
 #include "proto/connections/offline_wire_formats.pb.h"
 #include "platform/base/byte_array.h"
+#include "platform/base/cancellation_flag.h"
 #include "platform/base/listeners.h"
 #include "platform/base/runnable.h"
 #include "platform/public/atomic_boolean.h"
@@ -92,7 +93,8 @@ class WebRtc {
   // Runs on @MainThread.
   WebRtcSocketWrapper Connect(const std::string& service_id,
                               const PeerId& peer_id,
-                              const LocationHint& location_hint)
+                              const LocationHint& location_hint,
+                              CancellationFlag* cancellation_flag)
       ABSL_LOCKS_EXCLUDED(mutex_);
 
  private:
diff --git a/cpp/core/internal/mediums/webrtc_test.cc b/cpp/core/internal/mediums/webrtc_test.cc
index ecf95996..4eb7509b 100644
--- a/cpp/core/internal/mediums/webrtc_test.cc
+++ b/cpp/core/internal/mediums/webrtc_test.cc
@@ -75,8 +75,9 @@ TEST_F(WebRtcTest, Connect_DataChannelTimeOut) {
   LocationHint location_hint;
 
   ASSERT_TRUE(webrtc.IsAvailable());
+  CancellationFlag flag;
   WebRtcSocketWrapper wrapper_1 =
-      webrtc.Connect(service_id, peer_id, location_hint);
+      webrtc.Connect(service_id, peer_id, location_hint, &flag);
   EXPECT_FALSE(wrapper_1.IsValid());
 
   EXPECT_TRUE(webrtc.StartAcceptingConnections(
@@ -99,8 +100,9 @@ TEST_F(WebRtcTest, StartAcceptingConnection_ThenConnect) {
   ASSERT_TRUE(webrtc.StartAcceptingConnections(
       service_id, self_id, location_hint,
       {mock_accepted_callback_.AsStdFunction()}));
-  WebRtcSocketWrapper wrapper =
-      webrtc.Connect(service_id, PeerId("random_peer_id"), location_hint);
+  CancellationFlag flag;
+  WebRtcSocketWrapper wrapper = webrtc.Connect(
+      service_id, PeerId("random_peer_id"), location_hint, &flag);
   EXPECT_TRUE(webrtc.IsAcceptingConnections(service_id));
   EXPECT_FALSE(wrapper.IsValid());
   EXPECT_FALSE(webrtc.StartAcceptingConnections(
@@ -150,7 +152,8 @@ TEST_F(WebRtcTest, ConnectTwice) {
   device_c.StartAcceptingConnections(service_id, other_id, location_hint,
                                      {[](WebRtcSocketWrapper wrapper) {}});
 
-  sender_socket = sender.Connect(service_id, self_id, location_hint);
+  CancellationFlag flag;
+  sender_socket = sender.Connect(service_id, self_id, location_hint, &flag);
   EXPECT_TRUE(sender_socket.IsValid());
 
   ExceptionOr<bool> devices_connected = connected.Get();
@@ -158,7 +161,7 @@ TEST_F(WebRtcTest, ConnectTwice) {
   EXPECT_TRUE(devices_connected.result());
 
   WebRtcSocketWrapper socket =
-      sender.Connect(service_id, other_id, location_hint);
+      sender.Connect(service_id, other_id, location_hint, &flag);
   EXPECT_TRUE(socket.IsValid());
   socket.Close();
 
@@ -192,7 +195,8 @@ TEST_F(WebRtcTest, ConnectBothDevicesAndAbort) {
         connected.Set(receiver_socket.IsValid());
       }});
 
-  sender_socket = sender.Connect(service_id, self_id, location_hint);
+  CancellationFlag flag;
+  sender_socket = sender.Connect(service_id, self_id, location_hint, &flag);
   EXPECT_TRUE(sender_socket.IsValid());
 
   ExceptionOr<bool> devices_connected = connected.Get();
@@ -220,7 +224,8 @@ TEST_F(WebRtcTest, ConnectBothDevicesAndSendData) {
         connected.Set(receiver_socket.IsValid());
       }});
 
-  sender_socket = sender.Connect(service_id, self_id, location_hint);
+  CancellationFlag flag;
+  sender_socket = sender.Connect(service_id, self_id, location_hint, &flag);
   EXPECT_TRUE(sender_socket.IsValid());
 
   ExceptionOr<bool> devices_connected = connected.Get();
@@ -254,7 +259,8 @@ TEST_F(WebRtcTest, ConnectBothDevices_ShutdownSignaling_SendData) {
         connected.Set(receiver_socket.IsValid());
       }});
 
-  sender_socket = sender.Connect(service_id, self_id, location_hint);
+  CancellationFlag flag;
+  sender_socket = sender.Connect(service_id, self_id, location_hint, &flag);
   EXPECT_TRUE(sender_socket.IsValid());
 
   ExceptionOr<bool> devices_connected = connected.Get();
@@ -285,8 +291,9 @@ TEST_F(WebRtcTest, Connect_NullPeerConnection) {
   LocationHint location_hint;
 
   ASSERT_TRUE(webrtc.IsAvailable());
-  WebRtcSocketWrapper wrapper =
-      webrtc.Connect(service_id, PeerId("random_peer_id"), location_hint);
+  CancellationFlag flag;
+  WebRtcSocketWrapper wrapper = webrtc.Connect(
+      service_id, PeerId("random_peer_id"), location_hint, &flag);
   EXPECT_FALSE(wrapper.IsValid());
 }
 
diff --git a/cpp/core/internal/mediums/wifi_lan.cc b/cpp/core/internal/mediums/wifi_lan.cc
index 5736dc87..9f1e5487 100644
--- a/cpp/core/internal/mediums/wifi_lan.cc
+++ b/cpp/core/internal/mediums/wifi_lan.cc
@@ -222,8 +222,10 @@ bool WifiLan::IsAcceptingConnectionsLocked(const std::string& service_id) {
   return accepting_connections_info_.Existed(service_id);
 }
 
+// TODO(b/169303284): Handles Cancellation and registration.
 WifiLanSocket WifiLan::Connect(WifiLanService& wifi_lan_service,
-                               const std::string& service_id) {
+                               const std::string& service_id,
+                               CancellationFlag* cancellation_flag) {
   MutexLock lock(&mutex_);
   NEARBY_LOGS(INFO) << "WifiLan::Connect: wifi_lan_service="
                     << &wifi_lan_service << ", service_info_name="
diff --git a/cpp/core/internal/mediums/wifi_lan.h b/cpp/core/internal/mediums/wifi_lan.h
index 5c9d1822..53eb4d85 100644
--- a/cpp/core/internal/mediums/wifi_lan.h
+++ b/cpp/core/internal/mediums/wifi_lan.h
@@ -19,6 +19,7 @@
 #include <string>
 
 #include "platform/base/byte_array.h"
+#include "platform/base/cancellation_flag.h"
 #include "platform/public/multi_thread_executor.h"
 #include "platform/public/mutex.h"
 #include "platform/public/wifi_lan.h"
@@ -81,7 +82,8 @@ class WifiLan {
   // Blocks until connection is established, or server-side is terminated.
   // Returns socket instance. On success, WifiLanSocket.IsValid() return true.
   WifiLanSocket Connect(WifiLanService& wifi_lan_service,
-                        const std::string& service_id)
+                        const std::string& service_id,
+                        CancellationFlag* cancellation_flag)
       ABSL_LOCKS_EXCLUDED(mutex_);
 
   WifiLanService GetRemoteWifiLanService(const std::string& ip_address,
diff --git a/cpp/core/internal/mediums/wifi_lan_test.cc b/cpp/core/internal/mediums/wifi_lan_test.cc
index fb86e0aa..9f1bc63e 100644
--- a/cpp/core/internal/mediums/wifi_lan_test.cc
+++ b/cpp/core/internal/mediums/wifi_lan_test.cc
@@ -159,10 +159,9 @@ TEST_F(WifiLanTest, CanStartAcceptingConnectionsAndConnect) {
 
   EXPECT_TRUE(found_latch.Await(kWaitDuration).result());
   ASSERT_TRUE(discovered_service.IsValid());
-
+  CancellationFlag flag;
   WifiLanSocket socket =
-      wifi_lan_b.Connect(discovered_service, service_id);
-
+      wifi_lan_b.Connect(discovered_service, service_id, &flag);
   EXPECT_TRUE(accept_latch.Await(kWaitDuration).result());
   EXPECT_TRUE(socket.IsValid());
   wifi_lan_b.StopDiscovery(service_id);
diff --git a/cpp/core/internal/p2p_cluster_pcp_handler.cc b/cpp/core/internal/p2p_cluster_pcp_handler.cc
index cc89c391..d5e84caa 100644
--- a/cpp/core/internal/p2p_cluster_pcp_handler.cc
+++ b/cpp/core/internal/p2p_cluster_pcp_handler.cc
@@ -241,6 +241,83 @@ void P2pClusterPcpHandler::BluetoothDeviceDiscoveredHandler(
   });
 }
 
+void P2pClusterPcpHandler::BluetoothNameChangedHandler(
+    ClientProxy* client, const std::string& service_id,
+    BluetoothDevice device) {
+  RunOnPcpHandlerThread([this, client, service_id, device]() {
+    // Make sure we are still discovering before proceeding.
+    if (!client->IsDiscovering()) {
+      NEARBY_LOG(INFO,
+                 "BT discovery handler (CHANGED) [client=%p, service=%s]: not "
+                 "in discovery mode",
+                 client, service_id.c_str());
+      return;
+    }
+
+    // Parse the Bluetooth device name.
+    const std::string device_name_string = device.GetName();
+    BluetoothDeviceName device_name(device_name_string);
+    NEARBY_LOG(INFO,
+               "BT discovery handler (CHANGED) [client=%p, service=%s]: "
+               "processing new name %s",
+               client, service_id.c_str(), device_name_string.c_str());
+
+    // By this point, the BluetoothDevice passed to us has a different name than
+    // what we may have discovered before. We need to iterate over the found
+    // BluetoothEndpoints and compare their addresses to see the devices are the
+    // same. We are not guaranteed to discover a match, since the old name may
+    // not have been formatted for Nearby Connections.
+    for (auto endpoint :
+         GetDiscoveredEndpoints(proto::connections::Medium::BLUETOOTH)) {
+      BluetoothEndpoint* bluetoothEndpoint =
+          static_cast<BluetoothEndpoint*>(endpoint);
+      NEARBY_LOG(INFO,
+                 "BT discovery handler (CHANGED) [client=%p, service=%s]: "
+                 "comparing MAC addresses with existing endpoint %s. They have "
+                 "MAC address %s and the new endpoint has MAC address %s.",
+                 client, service_id.c_str(),
+                 bluetoothEndpoint->bluetooth_device.GetName().c_str(),
+                 bluetoothEndpoint->bluetooth_device.GetMacAddress().c_str(),
+                 device.GetMacAddress().c_str());
+      if (bluetoothEndpoint->bluetooth_device.GetMacAddress() ==
+          device.GetMacAddress()) {
+        // Report the BluetoothEndpoint as lost to the client.
+        NEARBY_LOG(
+            INFO,
+            "BT discovery handler (LOST) [client=%p, service=%s]: report "
+            "to client",
+            client, service_id.c_str());
+        OnEndpointLost(client, *endpoint);
+        break;
+      }
+    }
+
+    // Make sure the Bluetooth device name points to a valid
+    // endpoint we're discovering.
+    if (!IsRecognizedBluetoothEndpoint(device_name_string, service_id,
+                                       device_name)) {
+      NEARBY_LOG(INFO,
+                 "BT discovery handler (CHANGED) [client=%p, service=%s]: The "
+                 "new name is not recognized. Ignoring.",
+                 client, service_id.c_str());
+      return;
+    }
+
+    // Report the discovered endpoint to the client.
+    NEARBY_LOGS(INFO)
+        << "Invoking BasePcpHandler::OnEndpointFound() for BT service="
+        << service_id << "; id=" << device_name.GetEndpointId() << "; name="
+        << absl::BytesToHexString(device_name.GetEndpointInfo().data());
+    OnEndpointFound(
+        client, std::make_shared<BluetoothEndpoint>(BluetoothEndpoint{
+                    {device_name.GetEndpointId(), device_name.GetEndpointInfo(),
+                     service_id, proto::connections::Medium::BLUETOOTH,
+                     device_name.GetWebRtcState()},
+                    device,
+                }));
+  });
+}
+
 void P2pClusterPcpHandler::BluetoothDeviceLostHandler(
     ClientProxy* client, const std::string& service_id,
     BluetoothDevice& device) {
@@ -264,7 +341,7 @@ void P2pClusterPcpHandler::BluetoothDeviceLostHandler(
                                        device_name))
       return;
 
-    // Report the discovered endpoint to the client.
+    // Report the BluetoothEndpoint as lost to the client.
     NEARBY_LOG(INFO,
                "BT discovery handler (LOST) [client=%p, service=%s]: report "
                "to client",
@@ -590,7 +667,7 @@ BasePcpHandler::StartOperationResult P2pClusterPcpHandler::StartDiscoveryImpl(
                 &P2pClusterPcpHandler::BluetoothDeviceDiscoveredHandler, this,
                 client, service_id),
             .device_name_changed_cb = absl::bind_front(
-                &P2pClusterPcpHandler::BluetoothDeviceDiscoveredHandler, this,
+                &P2pClusterPcpHandler::BluetoothNameChangedHandler, this,
                 client, service_id),
             .device_lost_cb = absl::bind_front(
                 &P2pClusterPcpHandler::BluetoothDeviceLostHandler, this, client,
@@ -828,8 +905,9 @@ BasePcpHandler::ConnectImplResult P2pClusterPcpHandler::BluetoothConnectImpl(
     ClientProxy* client, BluetoothEndpoint* endpoint) {
   BluetoothDevice& device = endpoint->bluetooth_device;
 
-  BluetoothSocket bluetooth_socket =
-      bluetooth_medium_.Connect(device, endpoint->service_id);
+  BluetoothSocket bluetooth_socket = bluetooth_medium_.Connect(
+      device, endpoint->service_id,
+      client->GetCancellationFlag(endpoint->endpoint_id));
   if (!bluetooth_socket.IsValid()) {
     return BasePcpHandler::ConnectImplResult{
         .status = {Status::kBluetoothError},
@@ -1009,7 +1087,9 @@ BasePcpHandler::ConnectImplResult P2pClusterPcpHandler::BleConnectImpl(
     ClientProxy* client, BleEndpoint* endpoint) {
   BlePeripheral& peripheral = endpoint->ble_peripheral;
 
-  BleSocket ble_socket = ble_medium_.Connect(peripheral, endpoint->service_id);
+  BleSocket ble_socket =
+      ble_medium_.Connect(peripheral, endpoint->service_id,
+                          client->GetCancellationFlag(endpoint->endpoint_id));
   if (!ble_socket.IsValid()) {
     return BasePcpHandler::ConnectImplResult{
         .status = {Status::kBleError},
@@ -1139,8 +1219,9 @@ BasePcpHandler::ConnectImplResult P2pClusterPcpHandler::WifiLanConnectImpl(
     ClientProxy* client, WifiLanEndpoint* endpoint) {
   WifiLanService& wifi_lan_service = endpoint->wifi_lan_service;
 
-  WifiLanSocket wifi_lan_socket =
-      wifi_lan_medium_.Connect(wifi_lan_service, endpoint->service_id);
+  WifiLanSocket wifi_lan_socket = wifi_lan_medium_.Connect(
+      wifi_lan_service, endpoint->service_id,
+      client->GetCancellationFlag(endpoint->endpoint_id));
   if (!wifi_lan_socket.IsValid()) {
     return BasePcpHandler::ConnectImplResult{
         .status = {Status::kWifiLanError},
@@ -1204,7 +1285,8 @@ BasePcpHandler::ConnectImplResult P2pClusterPcpHandler::WebRtcConnectImpl(
   std::string empty_country_code;
   mediums::WebRtcSocketWrapper socket_wrapper = webrtc_medium_.Connect(
       webrtc_endpoint->service_id, webrtc_endpoint->peer_id,
-      Utils::BuildLocationHint(empty_country_code));
+      Utils::BuildLocationHint(empty_country_code),
+      client->GetCancellationFlag(webrtc_endpoint->endpoint_id));
   if (!socket_wrapper.IsValid()) {
     return BasePcpHandler::ConnectImplResult{.status = {Status::kError}};
   }
diff --git a/cpp/core/internal/p2p_cluster_pcp_handler.h b/cpp/core/internal/p2p_cluster_pcp_handler.h
index 4c25b6d6..43b84f3e 100644
--- a/cpp/core/internal/p2p_cluster_pcp_handler.h
+++ b/cpp/core/internal/p2p_cluster_pcp_handler.h
@@ -130,6 +130,9 @@ class P2pClusterPcpHandler : public BasePcpHandler {
   void BluetoothDeviceDiscoveredHandler(ClientProxy* client,
                                         const std::string& service_id,
                                         BluetoothDevice device);
+  void BluetoothNameChangedHandler(ClientProxy* client,
+                                   const std::string& service_id,
+                                   BluetoothDevice device);
   void BluetoothDeviceLostHandler(ClientProxy* client,
                                   const std::string& service_id,
                                   BluetoothDevice& device);
diff --git a/cpp/core/internal/service_controller_router.cc b/cpp/core/internal/service_controller_router.cc
index a1a5c103..61a96f2c 100644
--- a/cpp/core/internal/service_controller_router.cc
+++ b/cpp/core/internal/service_controller_router.cc
@@ -160,6 +160,10 @@ void ServiceControllerRouter::RequestConnection(
     ClientProxy* client, absl::string_view endpoint_id,
     const ConnectionRequestInfo& info, const ConnectionOptions& options,
     const ResultCallback& callback) {
+  // Cancellations can be fired from clients anytime, need to add the
+  // CancellationListener as soon as possible.
+  client->AddCancellationFlag(std::string(endpoint_id));
+
   RouteToServiceController([this, client,
                             endpoint_id = std::string(endpoint_id), info,
                             options, callback]() {
@@ -174,8 +178,12 @@ void ServiceControllerRouter::RequestConnection(
       return;
     }
 
-    callback.result_cb(service_controller_->RequestConnection(
-        client, endpoint_id, info, options));
+    Status status = service_controller_->RequestConnection(client, endpoint_id,
+                                                           info, options);
+    if (!status.Ok()) {
+      client->CancelEndpoint(endpoint_id);
+    }
+    callback.result_cb(status);
   });
 }
 
@@ -213,6 +221,8 @@ void ServiceControllerRouter::AcceptConnection(ClientProxy* client,
 void ServiceControllerRouter::RejectConnection(ClientProxy* client,
                                                absl::string_view endpoint_id,
                                                const ResultCallback& callback) {
+  client->CancelEndpoint(std::string(endpoint_id));
+
   RouteToServiceController(
       [this, client, endpoint_id = std::string(endpoint_id), callback]() {
         if (!ClientHasAcquiredServiceController(client)) {
@@ -310,6 +320,10 @@ void ServiceControllerRouter::CancelPayload(ClientProxy* client,
 void ServiceControllerRouter::DisconnectFromEndpoint(
     ClientProxy* client, absl::string_view endpoint_id,
     const ResultCallback& callback) {
+  // Client can emit the cancellation at anytime, we need to execute the request
+  // without further posting it.
+  client->CancelEndpoint(std::string(endpoint_id));
+
   RouteToServiceController(
       [this, client, endpoint_id = std::string(endpoint_id), callback]() {
         if (ClientHasAcquiredServiceController(client)) {
@@ -326,6 +340,10 @@ void ServiceControllerRouter::DisconnectFromEndpoint(
 
 void ServiceControllerRouter::StopAllEndpoints(ClientProxy* client,
                                                const ResultCallback& callback) {
+  // Client can emit the cancellation at anytime, we need to execute the request
+  // without further posting it.
+  client->CancelAllEndpoints();
+
   RouteToServiceController([this, client, callback]() {
     if (ClientHasAcquiredServiceController(client)) {
       DoneWithStrategySessionForClient(client);
@@ -336,6 +354,10 @@ void ServiceControllerRouter::StopAllEndpoints(ClientProxy* client,
 
 void ServiceControllerRouter::ClientDisconnecting(
     ClientProxy* client, const ResultCallback& callback) {
+  // Client can emit the cancellation at anytime, we need to execute the request
+  // without further posting it.
+  client->CancelAllEndpoints();
+
   RouteToServiceController([this, client, callback]() {
     if (ClientHasAcquiredServiceController(client)) {
       DoneWithStrategySessionForClient(client);
diff --git a/cpp/core/internal/webrtc_bwu_handler.cc b/cpp/core/internal/webrtc_bwu_handler.cc
index 354b6624..d4ca9e39 100644
--- a/cpp/core/internal/webrtc_bwu_handler.cc
+++ b/cpp/core/internal/webrtc_bwu_handler.cc
@@ -123,7 +123,8 @@ WebrtcBwuHandler::CreateUpgradedEndpointChannel(
              peer_id.GetId().c_str(), location_hint.DebugString().c_str());
 
   mediums::WebRtcSocketWrapper socket =
-      webrtc_.Connect(service_id, peer_id, location_hint);
+      webrtc_.Connect(service_id, peer_id, location_hint,
+                      client->GetCancellationFlag(endpoint_id));
   if (!socket.IsValid()) {
     NEARBY_LOG(ERROR,
                "WebRtcBwuHandler failed to connect to remote peer (%s) on "
diff --git a/cpp/core/internal/wifi_lan_bwu_handler.cc b/cpp/core/internal/wifi_lan_bwu_handler.cc
index 93d0dd40..43976360 100644
--- a/cpp/core/internal/wifi_lan_bwu_handler.cc
+++ b/cpp/core/internal/wifi_lan_bwu_handler.cc
@@ -117,8 +117,8 @@ WifiLanBwuHandler::CreateUpgradedEndpointChannel(
   if (!wifi_lan_service.IsValid()) {
     return nullptr;
   }
-  WifiLanSocket socket =
-      wifi_lan_medium_.Connect(wifi_lan_service, service_id);
+  WifiLanSocket socket = wifi_lan_medium_.Connect(
+      wifi_lan_service, service_id, client->GetCancellationFlag(endpoint_id));
   if (!socket.IsValid()) {
     return nullptr;
   }
diff --git a/cpp/platform/base/BUILD b/cpp/platform/base/BUILD
index 765cca6b..48225e1a 100644
--- a/cpp/platform/base/BUILD
+++ b/cpp/platform/base/BUILD
@@ -94,6 +94,28 @@ cc_library(
     ],
 )
 
+cc_library(
+    name = "cancellation_flag",
+    srcs = [
+        "cancellation_flag.cc",
+    ],
+    hdrs = [
+        "cancellation_flag.h",
+    ],
+    visibility = [
+        "//core/internal:__subpackages__",
+        "//platform/api:__subpackages__",
+        "//platform/impl:__subpackages__",
+        "//platform/public:__pkg__",
+    ],
+    deps = [
+        ":base",
+        ":util",
+        "//absl/container:flat_hash_set",
+        "//absl/synchronization",
+    ],
+)
+
 cc_library(
     name = "test_util",
     testonly = True,
@@ -143,6 +165,17 @@ cc_test(
     ],
 )
 
+cc_test(
+    name = "cancellation_flag_test",
+    srcs = [
+        "cancellation_flag_test.cc",
+    ],
+    deps = [
+        ":cancellation_flag",
+        "//testing/base/public:gunit_main",
+    ],
+)
+
 cc_with_non_compile_test(
     name = "exception_test",
     srcs = [
diff --git a/cpp/platform/base/cancellation_flag.cc b/cpp/platform/base/cancellation_flag.cc
new file mode 100644
index 00000000..1bb6e062
--- /dev/null
+++ b/cpp/platform/base/cancellation_flag.cc
@@ -0,0 +1,32 @@
+#include "platform/base/cancellation_flag.h"
+
+namespace location {
+namespace nearby {
+
+CancellationFlag::CancellationFlag() {
+  mutex_ = std::make_unique<absl::Mutex>();
+}
+
+CancellationFlag::CancellationFlag(bool cancelled) {
+  mutex_ = std::make_unique<absl::Mutex>();
+  cancelled_ = cancelled;
+}
+
+void CancellationFlag::Cancel() {
+  absl::MutexLock lock(mutex_.get());
+
+  if (cancelled_) {
+    // Someone already cancelled. Return immediately.
+    return;
+  }
+  cancelled_ = true;
+}
+
+bool CancellationFlag::Cancelled() const {
+  absl::MutexLock lock(mutex_.get());
+
+  return cancelled_;
+}
+
+}  // namespace nearby
+}  // namespace location
diff --git a/cpp/platform/base/cancellation_flag.h b/cpp/platform/base/cancellation_flag.h
new file mode 100644
index 00000000..4ef3f1ba
--- /dev/null
+++ b/cpp/platform/base/cancellation_flag.h
@@ -0,0 +1,37 @@
+#ifndef PLATFORM_BASE_CANCELLATION_FLAG_H_
+#define PLATFORM_BASE_CANCELLATION_FLAG_H_
+
+#include <memory>
+
+#include "absl/synchronization/mutex.h"
+
+namespace location {
+namespace nearby {
+
+// A cancellation flag to mark an operation has been cancelled and should be
+// cleaned up as soon as possible.
+class CancellationFlag {
+ public:
+  CancellationFlag();
+  explicit CancellationFlag(bool cancelled);
+  CancellationFlag(const CancellationFlag &) = delete;
+  CancellationFlag &operator=(const CancellationFlag &) = delete;
+  CancellationFlag(CancellationFlag &&) = default;
+  CancellationFlag &operator=(CancellationFlag &&) = default;
+  virtual ~CancellationFlag() = default;
+
+  // Set the flag as cancelled.
+  void Cancel() ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns true if the flag has been set to cancelled.
+  bool Cancelled() const ABSL_LOCKS_EXCLUDED(mutex_);
+
+ private:
+  std::unique_ptr<absl::Mutex> mutex_;
+  bool cancelled_ ABSL_GUARDED_BY(mutex_) = false;
+};
+
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_BASE_CANCELLATION_FLAG_H_
diff --git a/cpp/platform/base/cancellation_flag_test.cc b/cpp/platform/base/cancellation_flag_test.cc
new file mode 100644
index 00000000..32d79e0d
--- /dev/null
+++ b/cpp/platform/base/cancellation_flag_test.cc
@@ -0,0 +1,20 @@
+#include "platform/base/cancellation_flag.h"
+
+#include "gtest/gtest.h"
+
+namespace location {
+namespace nearby {
+
+TEST(CancellationFlagTest, InitialValueIsFalse) {
+  CancellationFlag flag;
+  EXPECT_FALSE(flag.Cancelled());
+}
+
+TEST(CancellationFlagTest, CanCancel) {
+  CancellationFlag flag;
+  flag.Cancel();
+  EXPECT_TRUE(flag.Cancelled());
+}
+
+}  // namespace nearby
+}  // namespace location
diff --git a/cpp/platform/impl/g3/webrtc.h b/cpp/platform/impl/g3/webrtc.h
index b362238f..20f310c8 100644
--- a/cpp/platform/impl/g3/webrtc.h
+++ b/cpp/platform/impl/g3/webrtc.h
@@ -41,7 +41,7 @@ class WebRtcSignalingMessenger : public api::WebRtcSignalingMessenger {
   void StopReceivingMessages() override;
 
  private:
-  absl::string_view self_id_;
+  std::string self_id_;
   connections::LocationHint location_hint_;
 };
 
diff --git a/cpp/platform/public/BUILD b/cpp/platform/public/BUILD
index bc64154d..0d7d89a3 100644
--- a/cpp/platform/public/BUILD
+++ b/cpp/platform/public/BUILD
@@ -52,7 +52,6 @@ cc_library(
         "//platform/base:logging",
         "//platform/base:util",
         "//absl/base:core_headers",
-        "//absl/container:flat_hash_map",
         "//absl/time",
     ],
 )