Roll forward to cl/338725629.

cpp/core/core.h

@@ -229,6 +229,9 @@ class Core {
   void InitiateBandwidthUpgrade(absl::string_view endpoint_id,
                                 ResultCallback callback);
 
+  // Gets the local endpoint generated by Nearby Connections.
+  std::string GetLocalEndpointId() { return client_.GetLocalEndpointId(); }
+
  private:
   static constexpr absl::Duration kWaitForDisconnect = absl::Milliseconds(5000);
 

cpp/core/internal/BUILD

@@ -19,6 +19,7 @@ cc_library(
         "base_pcp_handler.cc",
         "ble_advertisement.cc",
         "ble_endpoint_channel.cc",
+        "bluetooth_bwu_handler.cc",
         "bluetooth_device_name.cc",
         "bluetooth_endpoint_channel.cc",
         "bwu_manager.cc",
@@ -38,6 +39,7 @@ cc_library(
         "service_controller_router.cc",
         "webrtc_bwu_handler.cc",
         "webrtc_endpoint_channel.cc",
+        "wifi_lan_bwu_handler.cc",
         "wifi_lan_endpoint_channel.cc",
         "wifi_lan_service_info.cc",
     ],
@@ -47,6 +49,7 @@ cc_library(
         "base_pcp_handler.h",
         "ble_advertisement.h",
         "ble_endpoint_channel.h",
+        "bluetooth_bwu_handler.h",
         "bluetooth_device_name.h",
         "bluetooth_endpoint_channel.h",
         "bwu_handler.h",
@@ -71,6 +74,7 @@ cc_library(
         "service_controller_router.h",
         "webrtc_bwu_handler.h",
         "webrtc_endpoint_channel.h",
+        "wifi_lan_bwu_handler.h",
         "wifi_lan_endpoint_channel.h",
         "wifi_lan_service_info.h",
     ],

cpp/core/internal/base_bwu_handler.h

@@ -44,8 +44,6 @@ class BaseBwuHandler : public BwuHandler {
       : channel_manager_(&channel_manager),
         bwu_notifications_(std::move(bwu_notifications)) {}
   ~BaseBwuHandler() override = default;
-  void OnIncomingConnection(ClientProxy* client,
-                            IncomingSocketConnection* connection);
 
  protected:
   // Represents the incoming Socket the Initiator has gotten after initializing

cpp/core/internal/base_pcp_handler.cc

@@ -687,12 +687,12 @@ void BasePcpHandler::OnIncomingFrame(OfflineFrame& frame,
 
 void BasePcpHandler::OnEndpointDisconnect(ClientProxy* client,
                                           const std::string& endpoint_id,
-                                          CountDownLatch* barrier) {
+                                          CountDownLatch barrier) {
   if (stop_.Get()) {
-    if (barrier) barrier->CountDown();
+    barrier.CountDown();
     return;
   }
-  RunOnPcpHandlerThread([this, client, endpoint_id, barrier]() {
+  RunOnPcpHandlerThread([this, client, endpoint_id, barrier]() mutable {
     auto item = pending_alarms_.find(endpoint_id);
     if (item != pending_alarms_.end()) {
       auto& alarm = item->second;
@@ -700,7 +700,7 @@ void BasePcpHandler::OnEndpointDisconnect(ClientProxy* client,
       pending_alarms_.erase(item);
     }
     ProcessPreConnectionResultFailure(client, endpoint_id);
-    barrier->CountDown();
+    barrier.CountDown();
   });
 }
 

cpp/core/internal/base_pcp_handler.h

@@ -163,7 +163,7 @@ class BasePcpHandler : public PcpHandler,
   // approve/reject the connection.
   // @EndpointManagerThread
   void OnEndpointDisconnect(ClientProxy* client, const std::string& endpoint_id,
-                            CountDownLatch* barrier) override;
+                            CountDownLatch barrier) override;
 
   Pcp GetPcp() const override { return pcp_; }
   Strategy GetStrategy() const override { return strategy_; }

cpp/core/internal/base_pcp_handler_test.cc

@@ -393,6 +393,7 @@ TEST_P(BasePcpHandlerTest, ConstructorDestructorWorks) {
   BwuManager bwu(m, em, ecm, {}, {});
   MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
   SUCCEED();
+  bwu.Shutdown();
 }
 
 TEST_P(BasePcpHandlerTest, StartAdvertisingChangesState) {
@@ -403,6 +404,7 @@ TEST_P(BasePcpHandlerTest, StartAdvertisingChangesState) {
   BwuManager bwu(m, em, ecm, {}, {});
   MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
   StartAdvertising(&client, &pcp_handler);
+  bwu.Shutdown();
 }
 
 TEST_P(BasePcpHandlerTest, StopAdvertisingChangesState) {
@@ -417,6 +419,7 @@ TEST_P(BasePcpHandlerTest, StopAdvertisingChangesState) {
   EXPECT_TRUE(client.IsAdvertising());
   pcp_handler.StopAdvertising(&client);
   EXPECT_FALSE(client.IsAdvertising());
+  bwu.Shutdown();
 }
 
 TEST_P(BasePcpHandlerTest, StartDiscoveryChangesState) {
@@ -427,6 +430,7 @@ TEST_P(BasePcpHandlerTest, StartDiscoveryChangesState) {
   BwuManager bwu(m, em, ecm, {}, {});
   MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
   StartDiscovery(&client, &pcp_handler);
+  bwu.Shutdown();
 }
 
 TEST_P(BasePcpHandlerTest, StopDiscoveryChangesState) {
@@ -441,6 +445,7 @@ TEST_P(BasePcpHandlerTest, StopDiscoveryChangesState) {
   EXPECT_TRUE(client.IsDiscovering());
   pcp_handler.StopDiscovery(&client);
   EXPECT_FALSE(client.IsDiscovering());
+  bwu.Shutdown();
 }
 
 TEST_P(BasePcpHandlerTest, RequestConnectionChangesState) {
@@ -464,6 +469,7 @@ TEST_P(BasePcpHandlerTest, RequestConnectionChangesState) {
                     &pcp_handler, connect_medium);
   NEARBY_LOG(INFO, "RequestConnection complete");
   channel_b->Close();
+  bwu.Shutdown();
   pcp_handler.DisconnectFromEndpointManager();
 }
 
@@ -492,6 +498,7 @@ TEST_P(BasePcpHandlerTest, AcceptConnectionChangesState) {
   EXPECT_CALL(mock_connection_listener_.rejected_cb, Call).Times(AtLeast(0));
   NEARBY_LOGS(INFO) << "Closing connection: id=" << endpoint_id;
   channel_b->Close();
+  bwu.Shutdown();
   pcp_handler.DisconnectFromEndpointManager();
 }
 
@@ -516,6 +523,7 @@ TEST_P(BasePcpHandlerTest, RejectConnectionChangesState) {
             Status{Status::kSuccess});
   NEARBY_LOGS(INFO) << "Closing connection: id=" << endpoint_id;
   channel_b->Close();
+  bwu.Shutdown();
   pcp_handler.DisconnectFromEndpointManager();
 }
 
@@ -550,6 +558,7 @@ TEST_P(BasePcpHandlerTest, OnIncomingFrameChangesState) {
                               connect_medium);
   NEARBY_LOGS(INFO) << "Closing connection: id=" << endpoint_id;
   channel_b->Close();
+  bwu.Shutdown();
   pcp_handler.DisconnectFromEndpointManager();
 }
 
@@ -582,6 +591,7 @@ TEST_P(BasePcpHandlerTest, DestructorIsCalledOnProtocolEndpoint) {
     EXPECT_CALL(mock_connection_listener_.rejected_cb, Call).Times(AtLeast(0));
     NEARBY_LOG(INFO, "Closing connection: id=%s", endpoint_id.c_str());
     channel_b->Close();
+    bwu.Shutdown();
     pcp_handler.DisconnectFromEndpointManager();
     bwu.Shutdown();
   }
@@ -629,6 +639,7 @@ TEST_P(BasePcpHandlerTest, MultipleMediumsProduceSingleEndpointLostEvent) {
     }
     NEARBY_LOG(INFO, "Closing connection: id=%s", endpoint_id.c_str());
     channel_b->Close();
+    bwu.Shutdown();
     pcp_handler.DisconnectFromEndpointManager();
   }
   EXPECT_EQ(destroyed_flag.load(), mediums_count);
@@ -686,6 +697,7 @@ TEST_F(BasePcpHandlerTest, InjectEndpoint) {
             .medium = Medium::BLUETOOTH,
             .remote_bluetooth_mac_address = ByteArray(kFakeMacAddress),
         });
+  bwu.Shutdown();
 }
 
 }  // namespace

cpp/core/internal/bluetooth_bwu_handler.cc

@@ -0,0 +1,130 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "core/internal/bluetooth_bwu_handler.h"
+
+#include "core/internal/bluetooth_endpoint_channel.h"
+#include "core/internal/client_proxy.h"
+#include "core/internal/offline_frames.h"
+#include "absl/functional/bind_front.h"
+
+// Manages the Bluetooth-specific methods needed to upgrade an {@link
+// EndpointChannel}.
+
+namespace location {
+namespace nearby {
+namespace connections {
+
+BluetoothBwuHandler::BluetoothBwuHandler(
+    Mediums& mediums, EndpointChannelManager& channel_manager,
+    BwuNotifications notifications)
+    : BaseBwuHandler(channel_manager, std::move(notifications)),
+      mediums_(mediums) {}
+
+void BluetoothBwuHandler::Revert() {
+  for (const std::string& service_id : active_service_ids_) {
+    bluetooth_medium_.StopAcceptingConnections(service_id);
+  }
+  active_service_ids_.clear();
+  NEARBY_LOG(INFO,
+             "BluetoothBwuHandler successfully reverted all Bluetooth state.");
+}
+
+// Accept Connection Callback.
+// Notifies that the remote party called BluetoothClassic::Connect()
+// for this socket.
+void BluetoothBwuHandler::OnIncomingBluetoothConnection(
+    ClientProxy* client, const std::string& service_id,
+    BluetoothSocket socket) {
+  auto channel =
+      absl::make_unique<BluetoothEndpointChannel>(service_id, socket);
+  std::unique_ptr<IncomingSocketConnection> connection{
+      new IncomingSocketConnection{
+          .socket =
+              std::make_unique<BluetoothIncomingSocket>(service_id, socket),
+          .channel = std::move(channel),
+      }};
+  bwu_notifications_.incoming_connection_cb(client, std::move(connection));
+}
+
+// Called by BWU initiator. BT Medium is set up, and BWU request is prepared,
+// with necessary info (service_id, MAC address) for remote party to perform
+// discovery.
+ByteArray BluetoothBwuHandler::InitializeUpgradedMediumForEndpoint(
+    ClientProxy* client, const std::string& service_id,
+    const std::string& endpoint_id) {
+  std::string upgrade_service_id = Utils::WrapUpgradeServiceId(service_id);
+
+  std::string mac_address = bluetooth_medium_.GetMacAddress();
+  if (mac_address.empty()) {
+    return {};
+  }
+
+  if (!bluetooth_medium_.IsAcceptingConnections(upgrade_service_id)) {
+    if (!bluetooth_medium_.StartAcceptingConnections(
+            upgrade_service_id,
+            {
+                .accepted_cb = absl::bind_front(
+                    &BluetoothBwuHandler::OnIncomingBluetoothConnection, this,
+                    client, service_id),
+            })) {
+      return {};
+    }
+  }
+  // cache service ID to revert
+  active_service_ids_.emplace(upgrade_service_id);
+
+  return parser::ForBwuBluetoothPathAvailable(upgrade_service_id, mac_address);
+}
+
+// Called by BWU target. Retrieves a new medium info from incoming message,
+// and establishes connection over BT using this info.
+// Returns a channel ready to exchange data or nullptr on error.
+std::unique_ptr<EndpointChannel>
+BluetoothBwuHandler::CreateUpgradedEndpointChannel(
+    ClientProxy* client, const std::string& service_id,
+    const std::string& endpoint_id, const UpgradePathInfo& upgrade_path_info) {
+  const UpgradePathInfo::BluetoothCredentials& bluetooth_credentials =
+      upgrade_path_info.bluetooth_credentials();
+  if (!bluetooth_credentials.has_service_name() ||
+      !bluetooth_credentials.has_mac_address()) {
+    return nullptr;
+  }
+
+  const std::string& service_name = bluetooth_credentials.service_name();
+  const std::string& mac_address = bluetooth_credentials.mac_address();
+
+  BluetoothDevice device = bluetooth_medium_.GetRemoteDevice(mac_address);
+  if (!device.IsValid()) {
+    return nullptr;
+  }
+
+  BluetoothSocket socket = bluetooth_medium_.Connect(device, service_name);
+  if (!socket.IsValid()) {
+    return nullptr;
+  }
+
+  auto channel =
+      std::make_unique<BluetoothEndpointChannel>(service_name, socket);
+  if (channel == nullptr) {
+    socket.Close();
+    return nullptr;
+  }
+
+  return channel;
+}
+
+}  // namespace connections
+}  // namespace nearby
+}  // namespace location

cpp/core/internal/bluetooth_bwu_handler.h

@@ -0,0 +1,97 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef CORE_INTERNAL_BLUETOOTH_BWU_HANDLER_H_
+#define CORE_INTERNAL_BLUETOOTH_BWU_HANDLER_H_
+
+#include <string>
+
+#include "core/internal/base_bwu_handler.h"
+#include "core/internal/client_proxy.h"
+#include "core/internal/mediums/mediums.h"
+#include "core/internal/mediums/utils.h"
+#include "proto/connections/offline_wire_formats.pb.h"
+#include "platform/public/bluetooth_classic.h"
+#include "platform/public/count_down_latch.h"
+#include "proto/connections_enums.pb.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+
+// Defines the set of methods that need to be implemented to handle the
+// per-Medium-specific operations needed to upgrade an EndpointChannel.
+class BluetoothBwuHandler : public BaseBwuHandler {
+ public:
+  BluetoothBwuHandler(Mediums& mediums, EndpointChannelManager& channel_manager,
+                      BwuNotifications notifications);
+  ~BluetoothBwuHandler() override = default;
+
+ private:
+  constexpr static const int kServiceIdLength = 10;
+
+  // Implements BaseBwuHandler:
+  // Reverts any changes made to the device in the process of upgrading
+  // endpoints.
+  void Revert() override;
+
+  // Cleans up in-progress upgrades after endpoint disconnection.
+  void OnEndpointDisconnect(ClientProxy* client,
+                            const std::string& endpoint_id) override {}
+
+  void OnIncomingBluetoothConnection(ClientProxy* client,
+                                     const std::string& service_id,
+                                     BluetoothSocket socket);
+
+  class BluetoothIncomingSocket : public IncomingSocket {
+   public:
+    explicit BluetoothIncomingSocket(const std::string& name,
+                                     BluetoothSocket socket)
+        : name_(name), socket_(socket) {}
+    ~BluetoothIncomingSocket() override = default;
+    std::string ToString() override { return name_; }
+    void Close() override { socket_.Close(); }
+
+   private:
+    std::string name_;
+    BluetoothSocket socket_;
+  };
+
+  // First part of InitiateBwuForEndpoint implementation;
+  // returns a BWU request to remote party as byte array.
+  ByteArray InitializeUpgradedMediumForEndpoint(
+      ClientProxy* client, const std::string& service_id,
+      const std::string& endpoint_id) override;
+
+  // Invoked from OnBwuNegotiationFrame.
+  std::unique_ptr<EndpointChannel> CreateUpgradedEndpointChannel(
+      ClientProxy* client, const std::string& service_id,
+      const std::string& endpoint_id,
+      const UpgradePathInfo& upgrade_path_info) override;
+
+  // Returns the upgrade medium of the BwuHandler.
+  // @BwuHandlerThread
+  Medium GetUpgradeMedium() const override { return Medium::BLUETOOTH; }
+
+  Mediums& mediums_;
+  absl::flat_hash_set<std::string> active_service_ids_;
+  BluetoothRadio& bluetooth_radio_{mediums_.GetBluetoothRadio()};
+  BluetoothClassic& bluetooth_medium_{mediums_.GetBluetoothClassic()};
+};
+
+}  // namespace connections
+}  // namespace nearby
+}  // namespace location
+
+#endif  // CORE_INTERNAL_BLUETOOTH_BWU_HANDLER_H_

cpp/core/internal/bwu_handler.h

@@ -42,6 +42,7 @@ class BwuHandler {
   virtual ByteArray InitializeUpgradedMediumForEndpoint(
       ClientProxy* client, const std::string& service_id,
       const std::string& endpoint_id) = 0;
+
   // Called to revert any state changed by the Initiator to setup the upgraded
   // medium for an endpoint.
   // @BwuHandlerThread
@@ -55,6 +56,7 @@ class BwuHandler {
       ClientProxy* client, const std::string& service_id,
       const std::string& endpoint_id,
       const UpgradePathInfo& upgrade_path_info) = 0;
+
   // Returns the upgrade medium of the BwuHandler.
   // @BwuHandlerThread
   virtual Medium GetUpgradeMedium() const = 0;

cpp/core/internal/bwu_manager.cc

@@ -17,9 +17,11 @@
 #include <algorithm>
 #include <memory>
 
+#include "core/internal/bluetooth_bwu_handler.h"
 #include "core/internal/bwu_handler.h"
 #include "core/internal/offline_frames.h"
 #include "core/internal/webrtc_bwu_handler.h"
+#include "core/internal/wifi_lan_bwu_handler.h"
 #include "platform/base/byte_array.h"
 #include "platform/public/count_down_latch.h"
 #include "proto/connections_enums.pb.h"
@@ -68,11 +70,21 @@ void BwuManager::InitBwuHandlers() {
       .incoming_connection_cb =
           absl::bind_front(&BwuManager::OnIncomingConnection, this),
   };
+  if (config_.allow_upgrade_to.wifi_lan) {
+    handlers_.emplace(Medium::WIFI_LAN,
+                      std::make_unique<WifiLanBwuHandler>(
+                          *mediums_, *channel_manager_, notifications));
+  }
   if (config_.allow_upgrade_to.web_rtc) {
     handlers_.emplace(Medium::WEB_RTC,
                       std::make_unique<WebrtcBwuHandler>(
                           *mediums_, *channel_manager_, notifications));
   }
+  if (config_.allow_upgrade_to.bluetooth) {
+    handlers_.emplace(Medium::BLUETOOTH,
+                      std::make_unique<BluetoothBwuHandler>(
+                          *mediums_, *channel_manager_, notifications));
+  }
 }
 
 void BwuManager::Shutdown() {
@@ -81,31 +93,26 @@ void BwuManager::Shutdown() {
   endpoint_manager_->UnregisterFrameProcessor(
       V1Frame::BANDWIDTH_UPGRADE_NEGOTIATION, this);
 
-  CountDownLatch latch(1);
-
-  RunOnBwuManagerThread([this, &latch]() {
-    for (auto& item : previous_endpoint_channels_) {
-      EndpointChannel* channel = item.second.get();
-      if (!channel) continue;
-      channel->Close(DisconnectionReason::SHUTDOWN);
-    }
-
-    CancelAllRetryUpgradeAlarms();
-    medium_ = Medium::UNKNOWN_MEDIUM;
-    for (auto& item : handlers_) {
-      BwuHandler& handler = *item.second;
-      handler.Revert();
-    }
-    handlers_.clear();
-    latch.CountDown();
-  });
-
-  latch.Await();
-
   // Stop all the ongoing Runnables (as gracefully as possible).
   alarm_executor_.Shutdown();
   serial_executor_.Shutdown();
 
+  // After worker threads are down we became exclusive owners of data and
+  // may access it from current thread.
+  for (auto& item : previous_endpoint_channels_) {
+    EndpointChannel* channel = item.second.get();
+    if (!channel) continue;
+    channel->Close(DisconnectionReason::SHUTDOWN);
+  }
+
+  CancelAllRetryUpgradeAlarms();
+  medium_ = Medium::UNKNOWN_MEDIUM;
+  for (auto& item : handlers_) {
+    BwuHandler& handler = *item.second;
+    handler.Revert();
+  }
+  handlers_.clear();
+
   NEARBY_LOG(INFO, "BwuHandler has shut down.");
 }
 
@@ -196,10 +203,10 @@ void BwuManager::OnIncomingFrame(OfflineFrame& frame,
 
 void BwuManager::OnEndpointDisconnect(ClientProxy* client,
                                       const std::string& endpoint_id,
-                                      CountDownLatch* barrier) {
-  RunOnBwuManagerThread([this, client, endpoint_id, barrier]() {
+                                      CountDownLatch barrier) {
+  RunOnBwuManagerThread([this, client, endpoint_id, barrier]() mutable {
     if (medium_ == Medium::UNKNOWN_MEDIUM) {
-      barrier->CountDown();
+      barrier.CountDown();
       return;
     }
 
@@ -227,7 +234,7 @@ void BwuManager::OnEndpointDisconnect(ClientProxy* client,
     if (channel_manager_->GetConnectedEndpointsCount() <= 1) {
       Revert();
     }
-    barrier->CountDown();
+    barrier.CountDown();
   });
 }
 
@@ -555,6 +562,8 @@ void BwuManager::ProcessSafeToClosePriorChannelEvent(
              "trying to upgrade endpoint %s.",
              endpoint_id.c_str());
 
+  previous_endpoint_channel->Write(parser::ForDisconnection());
+
   // Wait for in-flight messages to reach their peers.
   SystemClock::Sleep(absl::Seconds(1));
   previous_endpoint_channel->Close(DisconnectionReason::UPGRADED);

cpp/core/internal/bwu_manager.h

@@ -95,7 +95,7 @@ class BwuManager : public EndpointManager::FrameProcessor {
   // @EndpointManagerReaderThread
   void OnEndpointDisconnect(ClientProxy* client_proxy,
                             const std::string& endpoint_id,
-                            CountDownLatch* barrier) override;
+                            CountDownLatch barrier) override;
   void Shutdown();
 
  private:

cpp/core/internal/bwu_manager_test.cc

@@ -20,8 +20,10 @@
 #include "core/internal/endpoint_channel_manager.h"
 #include "core/internal/endpoint_manager.h"
 #include "core/internal/mediums/mediums.h"
+#include "platform/public/system_clock.h"
 #include "gmock/gmock.h"
 #include "gtest/gtest.h"
+#include "absl/time/time.h"
 
 namespace location {
 namespace nearby {
@@ -33,6 +35,10 @@ TEST(BwuManagerTest, CanCreateInstance) {
   EndpointChannelManager ecm;
   EndpointManager em{&ecm};
   BwuManager bwu_manager{mediums, em, ecm, {}, {}};
+
+  SystemClock::Sleep(absl::Seconds(3));
+
+  bwu_manager.Shutdown();
 }
 
 TEST(BwuManagerTest, CanInitiateBwu) {
@@ -45,6 +51,7 @@ TEST(BwuManagerTest, CanInitiateBwu) {
 
   // Method returns void, so we just verify we did not SEGFAULT while calling.
   bwu_manager.InitiateBwuForEndpoint(&client, endpoint_id);
+  SystemClock::Sleep(absl::Seconds(3));
 
   bwu_manager.Shutdown();
 }

cpp/core/internal/endpoint_channel_manager.cc

@@ -16,14 +16,22 @@
 
 #include <memory>
 
+#include "core/internal/offline_frames.h"
+#include "proto/connections/offline_wire_formats.pb.h"
 #include "platform/public/logging.h"
 #include "platform/public/mutex.h"
 #include "platform/public/mutex_lock.h"
+#include "platform/public/system_clock.h"
+#include "absl/time/time.h"
 
 namespace location {
 namespace nearby {
 namespace connections {
 
+namespace {
+const absl::Duration kDataTransferDelay = absl::Milliseconds(500);
+}
+
 EndpointChannelManager::~EndpointChannelManager() {
   MutexLock lock(&mutex_);
   channel_state_.DestroyAll();
@@ -132,6 +140,11 @@ bool EndpointChannelManager::ChannelState::RemoveEndpoint(
   auto item = endpoints_.find(endpoint_id);
   if (item == endpoints_.end()) return false;
   item->second.disconnect_reason = reason;
+  auto channel = item->second.channel;
+  if (channel) {
+    channel->Write(parser::ForDisconnection());
+    SystemClock::Sleep(kDataTransferDelay);
+  }
   endpoints_.erase(item);
   return true;
 }

cpp/core/internal/endpoint_manager.cc

@@ -487,7 +487,7 @@ void EndpointManager::WaitForEndpointDisconnectionProcessing(
     NEARBY_LOGS(INFO) << "processor=" << processor << "; type=" << item.first;
     if (processor) {
       valid++;
-      processor->OnEndpointDisconnect(client, endpoint_id, &barrier);
+      processor->OnEndpointDisconnect(client, endpoint_id, barrier);
     } else {
       barrier.CountDown();
     }

cpp/core/internal/endpoint_manager.h

@@ -84,7 +84,7 @@ class EndpointManager {
     // @EndpointManagerThread
     virtual void OnEndpointDisconnect(ClientProxy* client,
                                       const std::string& endpoint_id,
-                                      CountDownLatch* barrier) = 0;
+                                      CountDownLatch barrier) = 0;
   };
 
   explicit EndpointManager(EndpointChannelManager* manager);

cpp/core/internal/endpoint_manager_test.cc

@@ -85,7 +85,7 @@ class MockFrameProcessor : public EndpointManager::FrameProcessor {
 
   MOCK_METHOD(void, OnEndpointDisconnect,
               (ClientProxy * client, const std::string& endpoint_id,
-               CountDownLatch* barrier),
+               CountDownLatch barrier),
               (override));
 };
 

cpp/core/internal/offline_frames.h

@@ -73,6 +73,7 @@ ByteArray ForBwuLastWrite();
 ByteArray ForBwuSafeToClose();
 
 ByteArray ForKeepAlive();
+ByteArray ForDisconnection();
 
 UpgradePathInfo::Medium MediumToUpgradePathInfoMedium(Medium medium);
 Medium UpgradePathInfoMediumToMedium(UpgradePathInfo::Medium medium);

cpp/core/internal/payload_manager.cc

@@ -402,12 +402,12 @@ void PayloadManager::OnIncomingFrame(
 
 void PayloadManager::OnEndpointDisconnect(ClientProxy* client,
                                           const std::string& endpoint_id,
-                                          CountDownLatch* barrier) {
+                                          CountDownLatch barrier) {
   if (shutdown_.Get()) {
-    if (barrier) barrier->CountDown();
+    barrier.CountDown();
     return;
   }
-  RunOnStatusUpdateThread([this, client, endpoint_id, barrier]() {
+  RunOnStatusUpdateThread([this, client, endpoint_id, barrier]() mutable {
     // Iterate through all our payloads and look for payloads associated
     // with this endpoint.
     MutexLock lock(&mutex_);
@@ -437,7 +437,7 @@ void PayloadManager::OnEndpointDisconnect(ClientProxy* client,
       client->OnPayloadProgress(endpoint_id, update);
     }
 
-    barrier->CountDown();
+    barrier.CountDown();
   });
 }
 

cpp/core/internal/payload_manager.h

@@ -60,7 +60,7 @@ class PayloadManager : public EndpointManager::FrameProcessor {
 
   // @EndpointManagerThread
   void OnEndpointDisconnect(ClientProxy* client, const std::string& endpoint_id,
-                            CountDownLatch* barrier) override;
+                            CountDownLatch barrier) override;
 
   void DisconnectFromEndpointManager();
 

cpp/core/internal/simulation_user.h

@@ -62,6 +62,7 @@ class SimulationUser {
   void Stop() {
     pm_.DisconnectFromEndpointManager();
     mgr_.DisconnectFromEndpointManager();
+    bwu_.Shutdown();
   }
 
   // Calls PcpManager::StartAdvertising.

cpp/core/internal/wifi_lan_bwu_handler.cc

@@ -0,0 +1,129 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "core/internal/wifi_lan_bwu_handler.h"
+
+#include <locale>
+#include <string>
+
+#include "core/internal/client_proxy.h"
+#include "core/internal/mediums/utils.h"
+#include "core/internal/offline_frames.h"
+#include "core/internal/wifi_lan_endpoint_channel.h"
+#include "platform/public/wifi_lan.h"
+#include "absl/functional/bind_front.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+
+WifiLanBwuHandler::WifiLanBwuHandler(Mediums& mediums,
+                                     EndpointChannelManager& channel_manager,
+                                     BwuNotifications notifications)
+    : BaseBwuHandler(channel_manager, std::move(notifications)),
+      mediums_(mediums) {}
+
+// Called by BWU initiator. Set up WifiLan upgraded medium for this endpoint,
+// and returns a upgrade path info (ip address, port) for remote party to
+// perform discovery.
+ByteArray WifiLanBwuHandler::InitializeUpgradedMediumForEndpoint(
+    ClientProxy* client, const std::string& service_id,
+    const std::string& endpoint_id) {
+  // Use wrapped service ID to avoid have the same ID with the one for
+  // startAdvertising. Otherwise, the listening request would be ignored because
+  // the medium already start accepting the connection because the client not
+  // stop the advertising yet.
+  std::string upgrade_service_id = Utils::WrapUpgradeServiceId(service_id);
+
+  if (!wifi_lan_medium_.IsAcceptingConnections(upgrade_service_id)) {
+    if (!wifi_lan_medium_.StartAcceptingConnections(
+            upgrade_service_id,
+            {
+                .accepted_cb = absl::bind_front(
+                    &WifiLanBwuHandler::OnIncomingWifiLanConnection, this,
+                    client),
+            })) {
+      NEARBY_LOG(ERROR,
+                 "WifiLanBwuHandler couldn't initiate the WifiLan upgrade for "
+                 "endpoint %s because it failed to start listening for "
+                 "incoming WifiLan connections.",
+                 endpoint_id.c_str());
+      return {};
+    }
+    NEARBY_LOG(INFO,
+               "WifiLanBwuHandler successfully started listening for incoming "
+               "WifiLan connections while upgrading endpoint %s",
+               endpoint_id.c_str());
+  }
+
+  // cache service ID to revert
+  active_service_ids_.emplace(upgrade_service_id);
+
+  // TODO(b/169303360): Implements wifiLanCredntials for wif_lan_medium to
+  // get ip_address and port.
+  std::string ip_addresss;
+  std::int32_t port = 0;
+  return parser::ForBwuWifiLanPathAvailable(ip_addresss, port);
+}
+
+void WifiLanBwuHandler::Revert() {
+  for (const std::string& service_id : active_service_ids_) {
+    wifi_lan_medium_.StopAcceptingConnections(service_id);
+  }
+  active_service_ids_.clear();
+
+  NEARBY_LOG(INFO, "WifiLanBwuHandler successfully reverted all states.");
+}
+
+// Called by BWU target. Retrieves a new medium info from incoming message,
+// and establishes connection over WifiLan using this info.
+std::unique_ptr<EndpointChannel>
+WifiLanBwuHandler::CreateUpgradedEndpointChannel(
+    ClientProxy* client, const std::string& service_id,
+    const std::string& endpoint_id, const UpgradePathInfo& upgrade_path_info) {
+  // TODO(b/169303360): Implements connect WifiLan over ip address and port.
+  WifiLanSocket socket;
+
+  // Create a new WifiLanEndpointChannel.
+  auto channel = std::make_unique<WifiLanEndpointChannel>(service_id, socket);
+  if (channel == nullptr) {
+    socket.Close();
+    NEARBY_LOG(ERROR,
+               "WifiLanBwuHandler failed to create new EndpointChannel for "
+               "outgoing socket %p, aborting upgrade.",
+               &socket.GetImpl());
+  }
+
+  return channel;
+}
+
+// Accept Connection Callback.
+void WifiLanBwuHandler::OnIncomingWifiLanConnection(
+    ClientProxy* client, WifiLanSocket socket,
+    const std::string& upgrade_service_id) {
+  std::string service_id = Utils::UnwrapUpgradeServiceId(upgrade_service_id);
+  auto channel = std::make_unique<WifiLanEndpointChannel>(service_id, socket);
+  auto wifi_lan_socket =
+      std::make_unique<WifiLanIncomingSocket>(service_id, socket);
+  std::unique_ptr<IncomingSocketConnection> connection(
+      new IncomingSocketConnection{std::move(wifi_lan_socket),
+                                   std::move(channel)});
+
+  bwu_notifications_.incoming_connection_cb(client, std::move(connection));
+}
+
+}  // namespace connections
+}  // namespace nearby
+}  // namespace location
+

cpp/core/internal/wifi_lan_bwu_handler.h

@@ -0,0 +1,79 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef CORE_INTERNAL_WIFI_LAN_BWU_HANDLER_H_
+#define CORE_INTERNAL_WIFI_LAN_BWU_HANDLER_H_
+
+#include "core/internal/base_bwu_handler.h"
+#include "core/internal/client_proxy.h"
+#include "core/internal/endpoint_channel_manager.h"
+#include "core/internal/mediums/mediums.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+
+// Defines the set of methods that need to be implemented to handle the
+// per-Medium-specific operations needed to upgrade an EndpointChannel.
+class WifiLanBwuHandler : public BaseBwuHandler {
+ public:
+  WifiLanBwuHandler(Mediums& mediums, EndpointChannelManager& channel_manager,
+                    BwuNotifications notifications);
+  ~WifiLanBwuHandler() override = default;
+
+ private:
+  ByteArray InitializeUpgradedMediumForEndpoint(
+      ClientProxy* client, const std::string& service_id,
+      const std::string& endpoint_id) override;
+
+  void Revert() override;
+
+  std::unique_ptr<EndpointChannel> CreateUpgradedEndpointChannel(
+      ClientProxy* client, const std::string& service_id,
+      const std::string& endpoint_id,
+      const UpgradePathInfo& upgrade_path_info) override;
+
+  Medium GetUpgradeMedium() const override { return Medium::WIFI_LAN; }
+
+  void OnEndpointDisconnect(ClientProxy* client,
+                            const std::string& endpoint_id) override {}
+
+  void OnIncomingWifiLanConnection(ClientProxy* client, WifiLanSocket socket,
+                                   const std::string& upgrade_service_id);
+
+  class WifiLanIncomingSocket : public BwuHandler::IncomingSocket {
+   public:
+    explicit WifiLanIncomingSocket(const std::string& name,
+                                   WifiLanSocket socket)
+        : name_(name), socket_(socket) {}
+    ~WifiLanIncomingSocket() override = default;
+
+    std::string ToString() override { return name_; }
+    void Close() override { socket_.Close(); }
+
+   private:
+    std::string name_;
+    WifiLanSocket socket_;
+  };
+
+  Mediums& mediums_;
+  WifiLan& wifi_lan_medium_{mediums_.GetWifiLan()};
+  absl::flat_hash_set<std::string> active_service_ids_;
+};
+
+}  // namespace connections
+}  // namespace nearby
+}  // namespace location
+
+#endif  // CORE_INTERNAL_WIFI_LAN_BWU_HANDLER_H_

cpp/core/internal/wifi_lan_service_info.cc

@@ -28,6 +28,11 @@ namespace location {
 namespace nearby {
 namespace connections {
 
+// These definitions are necessary before C++17.
+constexpr absl::string_view WifiLanServiceInfo::kKeyEndpointInfo;
+constexpr std::uint32_t WifiLanServiceInfo::kServiceIdHashLength;
+constexpr int WifiLanServiceInfo::kMaxEndpointInfoLength;
+
 WifiLanServiceInfo::WifiLanServiceInfo(Version version, Pcp pcp,
                                        absl::string_view endpoint_id,
                                        const ByteArray& service_id_hash,

cpp/platform/public/count_down_latch.h

@@ -34,8 +34,8 @@ class CountDownLatch final {
   using Platform = api::ImplementationPlatform;
   explicit CountDownLatch(int count)
       : impl_(Platform::CreateCountDownLatch(count)) {}
-  CountDownLatch(CountDownLatch&&) = default;
-  CountDownLatch& operator=(CountDownLatch&&) = default;
+  CountDownLatch(const CountDownLatch&) = default;
+  CountDownLatch& operator=(const CountDownLatch&) = default;
   ~CountDownLatch() = default;
 
   Exception Await() { return impl_->Await(); }
@@ -45,7 +45,7 @@ class CountDownLatch final {
   void CountDown() { impl_->CountDown(); }
 
  private:
-  std::unique_ptr<api::CountDownLatch> impl_;
+  std::shared_ptr<api::CountDownLatch> impl_;
 };
 
 }  // namespace nearby