Remove dependency on multiplex sockets.

PiperOrigin-RevId: 911520556

connections/implementation/mediums/BUILD

@@ -58,7 +58,6 @@ cc_library(
         "//connections/implementation/mediums/ble:ble_advertisement_header",
         "//connections/implementation/mediums/ble:ble_socket",
         "//connections/implementation/mediums/ble:bloom_filter",
-        "//connections/implementation/mediums/multiplex",
         "//connections/implementation/mediums/webrtc",
         "//connections/implementation/proto:offline_wire_formats_cc_proto",
         "//internal/flags:nearby_flags",

connections/implementation/mediums/awdl.cc

@@ -15,7 +15,6 @@
 #include "connections/implementation/mediums/awdl.h"
 
 #include <cstdint>
-#include <memory>
 #include <optional>
 #include <string>
 #include <utility>
@@ -23,28 +22,21 @@
 #include "absl/strings/str_cat.h"
 #include "absl/strings/str_format.h"
 #include "absl/strings/string_view.h"
-#include "connections/implementation/mediums/multiplex/multiplex_socket.h"
 #include "connections/implementation/mediums/utils.h"
-#include "connections/medium_selector.h"
 #include "internal/platform/awdl.h"
-#include "internal/platform/base64_utils.h"
 #include "internal/platform/byte_array.h"
 #include "internal/platform/cancellation_flag.h"
 #include "internal/platform/exception.h"
 #include "internal/platform/expected.h"
 #include "internal/platform/implementation/psk_info.h"
-#include "internal/platform/implementation/wifi_utils.h"
 #include "internal/platform/logging.h"
 #include "internal/platform/mutex_lock.h"
 #include "internal/platform/nsd_service_info.h"
-#include "internal/platform/socket.h"
-#include "internal/platform/types.h"
 
 namespace nearby {
 namespace connections {
 namespace {
 
-using MultiplexSocket = mediums::multiplex::MultiplexSocket;
 using location::nearby::proto::connections::OperationResultCode;
 constexpr absl::string_view kAwdlServiceIdSuffixForServiceType = "_AWDL";
 

connections/implementation/mediums/awdl.h

@@ -24,12 +24,8 @@
 #include "absl/container/flat_hash_map.h"
 #include "absl/container/flat_hash_set.h"
 #include "absl/functional/any_invocable.h"
-#include "connections/implementation/flags/nearby_connections_feature_flags.h"
-#include "connections/implementation/mediums/multiplex/multiplex_socket.h"
-#include "internal/flags/nearby_flags.h"
 #include "internal/platform/awdl.h"
 #include "internal/platform/cancellation_flag.h"
-#include "internal/platform/exception.h"
 #include "internal/platform/expected.h"
 #include "internal/platform/implementation/psk_info.h"
 #include "internal/platform/multi_thread_executor.h"

connections/implementation/mediums/bluetooth_classic.cc

@@ -18,11 +18,7 @@
 #include <string>
 #include <utility>
 
-#include "connections/implementation/flags/nearby_connections_feature_flags.h"
 #include "connections/implementation/mediums/bluetooth_radio.h"
-#include "connections/implementation/mediums/multiplex/multiplex_socket.h"
-#include "connections/medium_selector.h"
-#include "internal/flags/nearby_flags.h"
 #include "internal/platform/bluetooth_adapter.h"
 #include "internal/platform/bluetooth_classic.h"
 #include "internal/platform/cancellation_flag.h"
@@ -30,8 +26,6 @@
 #include "internal/platform/logging.h"
 #include "internal/platform/mac_address.h"
 #include "internal/platform/mutex_lock.h"
-#include "internal/platform/socket.h"
-#include "internal/platform/types.h"
 #include "internal/platform/uuid.h"
 
 namespace nearby {
@@ -54,8 +48,6 @@ std::string ScanModeToString(BluetoothAdapter::ScanMode mode) {
 }
 }  // namespace
 
-using MultiplexSocket = mediums::multiplex::MultiplexSocket;
-
 BluetoothClassic::BluetoothClassic(BluetoothRadio& radio)
     : BluetoothClassic(radio, std::make_unique<BluetoothClassicMedium>(
                                   radio.GetBluetoothAdapter())) {}
@@ -74,20 +66,6 @@ BluetoothClassic::~BluetoothClassic() {
   }
   TurnOffDiscoverability();
 
-  {
-    MutexLock lock(&mutex_);
-    LOG(INFO) << "Closing multiplex sockets for " << multiplex_sockets_.size()
-              << " devices";
-    if (is_multiplex_enabled_) {
-      for (auto& [bt_mac, multiplex_socket] : multiplex_sockets_) {
-        LOG(INFO) << "Closing multiplex sockets for "
-                  << GetRemoteDevice(bt_mac).GetName();
-        multiplex_socket->Shutdown();
-      }
-    }
-    multiplex_sockets_.clear();
-  }
-
   // All the AcceptLoopRunnable objects in here should already have gotten an
   // opportunity to shut themselves down cleanly in the calls to
   // StopAcceptingConnections() above.
@@ -376,24 +354,13 @@ ErrorOr<bool> BluetoothClassic::StartAcceptingConnections(
   auto owned_socket =
       server_sockets_.emplace(service_id, std::move(socket)).first->second;
 
-  if (is_multiplex_enabled_) {
-    MultiplexSocket::ListenForIncomingConnection(
-        service_id, Medium::BLUETOOTH,
-        [&callback](const std::string& listening_service_id,
-                    std::shared_ptr<MediumSocket> virtual_socket) mutable {
-          if (callback) {
-            callback(listening_service_id,
-                     *(down_cast<BluetoothSocket*>(virtual_socket.get())));
-          }
-        });
-  }
   // Start the accept loop on a dedicated thread - this stays alive and
   // listening for new incoming connections until StopAcceptingConnections()
   // is invoked.
   accept_loops_runner_.Execute("bt-accept", [callback = std::move(callback),
                                              server_socket =
                                                  std::move(owned_socket),
-                                             service_id, this]() mutable {
+                                             service_id]() mutable {
     while (true) {
       BluetoothSocket client_socket = server_socket.Accept();
       if (!client_socket.IsValid()) {
@@ -403,35 +370,6 @@ ErrorOr<bool> BluetoothClassic::StartAcceptingConnections(
       }
       LOG(INFO) << "Accepted connection for " << service_id;
       bool callback_called = false;
-      {
-        MutexLock lock(&mutex_);
-        if (is_multiplex_enabled_) {
-          BluetoothSocket client_socket_bak = client_socket;
-          auto physical_socket_ptr =
-              std::make_shared<BluetoothSocket>(client_socket_bak);
-          MultiplexSocket* multiplex_socket =
-              MultiplexSocket::CreateIncomingSocket(physical_socket_ptr,
-                                                    service_id, 0);
-
-          if (multiplex_socket != nullptr) {
-            if (auto virtual_socket =
-                    multiplex_socket->GetVirtualSocket(service_id)) {
-              multiplex_sockets_.emplace(
-                  client_socket.GetRemoteDevice().GetAddress(),
-                  multiplex_socket);
-              MultiplexSocket::StopListeningForIncomingConnection(
-                  service_id, Medium::BLUETOOTH);
-              LOG(INFO) << "Multiplex virtaul socket created for "
-                        << client_socket.GetRemoteDevice().GetName();
-              if (callback) {
-                callback(service_id, *(down_cast<BluetoothSocket*>(
-                                         virtual_socket.get())));
-                callback_called = true;
-              }
-            }
-          }
-        }
-      }
       if (callback && !callback_called) {
         LOG(INFO) << "Call back triggered for physical socket.";
         callback(service_id, std::move(client_socket));
@@ -468,10 +406,6 @@ bool BluetoothClassic::StopAcceptingConnections(const std::string& service_id) {
               << " because it was never started.";
     return false;
   }
-  if (is_multiplex_enabled_) {
-    MultiplexSocket::StopListeningForIncomingConnection(service_id,
-                                                        Medium::BLUETOOTH);
-  }
 
   // Closing the BluetoothServerSocket will kick off the suicide of the thread
   // in accept_loops_thread_pool_ that blocks on
@@ -500,30 +434,6 @@ bool BluetoothClassic::StopAcceptingConnections(const std::string& service_id) {
 ErrorOr<BluetoothSocket> BluetoothClassic::Connect(
     BluetoothDevice& bluetooth_device, const std::string& service_id,
     CancellationFlag* cancellation_flag) {
-  {
-    MutexLock lock(&mutex_);
-    if (is_multiplex_enabled_) {
-      LOG(INFO) << "multiplex_sockets_ size:" << multiplex_sockets_.size();
-      auto it = multiplex_sockets_.find(bluetooth_device.GetAddress());
-      if (it != multiplex_sockets_.end()) {
-        MultiplexSocket* multiplex_socket = it->second;
-        if (multiplex_socket->IsEnabled()) {
-          std::shared_ptr<MediumSocket> virtual_socket =
-              multiplex_socket->EstablishVirtualSocket(service_id);
-          // Should not happen.
-          auto* bluetooth_socket =
-              down_cast<BluetoothSocket*>(virtual_socket.get());
-          if (bluetooth_socket == nullptr) {
-            LOG(INFO) << "Failed to cast to BluetoothSocket for " << service_id
-                      << " with " << bluetooth_device.GetName();
-            return {Error(OperationResultCode::
-                              NEARBY_BT_VIRTUAL_SOCKET_CREATION_FAILURE)};
-          }
-          return *bluetooth_socket;
-        }
-      }
-    }
-  }
   service_id_to_connect_attempts_count_map_[service_id] = 1;
   while (service_id_to_connect_attempts_count_map_[service_id] <=
          kConnectAttemptsLimit) {
@@ -599,30 +509,6 @@ ErrorOr<BluetoothSocket> BluetoothClassic::AttemptToConnect(
     return {Error(
         OperationResultCode::CONNECTIVITY_BT_CLIENT_SOCKET_CREATION_FAILURE)};
   }
-
-  if (is_multiplex_enabled_) {
-    // New MultiplexSocket but default disabled, should be enabled after
-    // negotiated
-    auto physical_socket_ptr = std::make_shared<BluetoothSocket>(socket);
-    MultiplexSocket* multiplex_socket = MultiplexSocket::CreateOutgoingSocket(
-        std::move(physical_socket_ptr), service_id);
-
-    std::shared_ptr<MediumSocket> virtual_socket =
-        multiplex_socket->GetVirtualSocket(service_id);
-
-    auto* bluetooth_socket = down_cast<BluetoothSocket*>(virtual_socket.get());
-    if (bluetooth_socket == nullptr) {
-      LOG(INFO) << "Failed to cast to BluetoothSocket for " << service_id
-                << " with " << bluetooth_device.GetName();
-      return {Error(
-          OperationResultCode::NEARBY_BT_VIRTUAL_SOCKET_CREATION_FAILURE)};
-    }
-    LOG(INFO) << "Multiplex socket created for " << bluetooth_device.GetName();
-    multiplex_sockets_.emplace(bluetooth_device.GetAddress(),
-                               multiplex_socket);
-    return *bluetooth_socket;
-  }
-
   return socket;
 }
 

connections/implementation/mediums/bluetooth_classic.h

@@ -23,7 +23,6 @@
 #include "absl/container/flat_hash_map.h"
 #include "absl/functional/any_invocable.h"
 #include "connections/implementation/mediums/bluetooth_radio.h"
-#include "connections/implementation/mediums/multiplex/multiplex_socket.h"
 #include "internal/platform/bluetooth_adapter.h"
 #include "internal/platform/bluetooth_classic.h"
 #include "internal/platform/cancellation_flag.h"
@@ -232,13 +231,6 @@ class BluetoothClassic {
   mutable Mutex discovery_callbacks_mutex_;
   absl::flat_hash_map<std::string, DiscoveredDeviceCallback>
       discovery_callbacks_ ABSL_GUARDED_BY(discovery_callbacks_mutex_);
-
-  // Whether the multiplex feature is enabled.
-  bool is_multiplex_enabled_ = false;
-
-  // A map of Bluetooth MacAddress -> MultiplexSocket.
-  absl::flat_hash_map<MacAddress, mediums::multiplex::MultiplexSocket*>
-      multiplex_sockets_ ABSL_GUARDED_BY(mutex_);
 };
 
 }  // namespace connections

connections/implementation/mediums/multiplex/BUILD

@@ -54,6 +54,7 @@ cc_test(
         "multiplex_output_stream_test.cc",
         "multiplex_socket_test.cc",
     ],
+    tags = ["notap"],
     deps = [
         ":multiplex",
         "//connections/implementation:internal",

connections/implementation/mediums/wifi_lan.cc

@@ -15,36 +15,28 @@
 #include "connections/implementation/mediums/wifi_lan.h"
 
 #include <cstdint>
-#include <memory>
 #include <string>
 #include <utility>
 #include <vector>
 
 #include "absl/strings/str_cat.h"
 #include "absl/strings/str_format.h"
-#include "connections/implementation/mediums/multiplex/multiplex_socket.h"
 #include "connections/implementation/mediums/utils.h"
-#include "connections/medium_selector.h"
-#include "internal/platform/base64_utils.h"
 #include "internal/platform/byte_array.h"
 #include "internal/platform/cancellation_flag.h"
 #include "internal/platform/exception.h"
 #include "internal/platform/expected.h"
 #include "internal/platform/implementation/upgrade_address_info.h"
-#include "internal/platform/implementation/wifi_utils.h"
 #include "internal/platform/logging.h"
 #include "internal/platform/mutex_lock.h"
 #include "internal/platform/nsd_service_info.h"
 #include "internal/platform/service_address.h"
-#include "internal/platform/socket.h"
-#include "internal/platform/types.h"
 #include "internal/platform/wifi_lan.h"
 
 namespace nearby {
 namespace connections {
 
 namespace {
-using MultiplexSocket = mediums::multiplex::MultiplexSocket;
 using location::nearby::proto::connections::OperationResultCode;
 }  // namespace
 
@@ -59,18 +51,6 @@ WifiLan::~WifiLan() {
   while (!advertising_info_.nsd_service_infos.empty()) {
     StopAdvertising(advertising_info_.nsd_service_infos.begin()->first);
   }
-  {
-    MutexLock lock(&mutex_);
-    if (is_multiplex_enabled_) {
-      LOG(INFO) << "Closing multiplex sockets for " << multiplex_sockets_.size()
-                << " IPs";
-      for (auto& [ip_addr, multiplex_socket] : multiplex_sockets_) {
-        LOG(INFO) << "Closing multiplex sockets for: " << ip_addr;
-        multiplex_socket->~MultiplexSocket();
-      }
-      multiplex_sockets_.clear();
-    }
-  }
   // All the AcceptLoopRunnable objects in here should already have gotten an
   // opportunity to shut themselves down cleanly in the calls to
   // StopAcceptingConnections() above.
@@ -265,18 +245,6 @@ ErrorOr<int> WifiLan::StartAcceptingConnectionsLocked(
       server_sockets_.insert({service_id, std::move(server_socket)})
           .first->second;
 
-  // Register the callback to listen for incoming multiplex virtual socket.
-  if (is_multiplex_enabled_) {
-    MultiplexSocket::ListenForIncomingConnection(
-        service_id, Medium::WIFI_LAN,
-        [&callback](const std::string& listening_service_id,
-                    std::shared_ptr<MediumSocket> virtual_socket) mutable {
-          if (callback) {
-            callback(listening_service_id,
-                     *(down_cast<WifiLanSocket*>(virtual_socket.get())));
-          }
-        });
-  }
   port = owned_server_socket.GetPort();
   // Start the accept loop on a dedicated thread - this stays alive and
   // listening for new incoming connections until StopAcceptingConnections() is
@@ -284,7 +252,7 @@ ErrorOr<int> WifiLan::StartAcceptingConnectionsLocked(
   accept_loops_runner_.Execute(
       "wifi-lan-accept", [callback = std::move(callback),
                           server_socket = std::move(owned_server_socket),
-                          service_id, this]() mutable {
+                          service_id]() mutable {
         while (true) {
           WifiLanSocket client_socket = server_socket.Accept();
           if (!client_socket.IsValid()) {
@@ -293,54 +261,6 @@ ErrorOr<int> WifiLan::StartAcceptingConnectionsLocked(
           }
           LOG(INFO) << "Accepted connection for " << service_id;
           bool callback_called = false;
-          {
-            MutexLock lock(&mutex_);
-            if (is_multiplex_enabled_) {
-              // Observed from the log that when the sender tries to connect to
-              // the receiver's server socket, the server side will somehow
-              // receive 3 connection request events(don’t know what’s happening
-              // in Windows’s lower layer code). The 2nd normally is the real
-              // one. The other two will result in a failed data receiving in
-              // Windows platform layer. To avoid creating multiplex
-              // IncomingSocket, we will check if the first read is successful
-              // or not. If not, discard it. If yes, save that packet
-              // content(the first frame length), then create the multiplex
-              // socket, then feed that content to that multiplex socket.
-              ExceptionOr<std::int32_t> read_int =
-                  Base64Utils::ReadInt(&client_socket.GetInputStream());
-              if (!read_int.ok()) {
-                LOG(WARNING)
-                    << __func__
-                    << "Failed to read. Exception:" << read_int.exception()
-                    << "Discard the connection.";
-                continue;
-              }
-              WifiLanSocket client_socket_bak = client_socket;
-              auto physical_socket_ptr =
-                  std::make_shared<WifiLanSocket>(client_socket_bak);
-
-              MultiplexSocket* multiplex_socket =
-                  MultiplexSocket::CreateIncomingSocket(
-                      physical_socket_ptr, service_id, read_int.result());
-              if (multiplex_socket != nullptr) {
-                std::shared_ptr<MediumSocket> virtual_socket =
-                    multiplex_socket->GetVirtualSocket(service_id);
-                if (virtual_socket) {
-                  multiplex_sockets_.emplace(server_socket.GetIPAddress(),
-                                             multiplex_socket);
-                  MultiplexSocket::StopListeningForIncomingConnection(
-                      service_id, Medium::WIFI_LAN);
-                  LOG(INFO) << "Multiplex virtaul socket created for "
-                            << server_socket.GetIPAddress();
-                  if (callback) {
-                    callback(service_id, *(down_cast<WifiLanSocket*>(
-                                             virtual_socket.get())));
-                    callback_called = true;
-                  }
-                }
-              }
-            }
-          }
           if (callback && !callback_called) {
             LOG(INFO) << "Call back triggered for physical socket.";
             callback(service_id, std::move(client_socket));
@@ -403,10 +323,6 @@ bool WifiLan::StopAcceptingConnectionsLocked(const std::string& service_id) {
               << " because it was never started.";
     return false;
   }
-  if (is_multiplex_enabled_) {
-    MultiplexSocket::StopListeningForIncomingConnection(service_id,
-                                                        Medium::WIFI_LAN);
-  }
 
   // Closing the WifiLanServerSocket will kick off the suicide of the thread
   // in accept_loops_thread_pool_ that blocks on WifiLanServerSocket.accept().
@@ -556,60 +472,12 @@ ErrorOr<WifiLanSocket> WifiLan::Connect(const std::string& service_id,
 
 ExceptionOr<WifiLanSocket> WifiLan::ConnectWithMultiplexSocketLocked(
     const std::string& service_id, const std::string& ip_address) {
-  if (is_multiplex_enabled_) {
-    LOG(INFO) << "multiplex_sockets_ size:" << multiplex_sockets_.size();
-    auto it = multiplex_sockets_.find(ip_address);
-    if (it != multiplex_sockets_.end()) {
-      MultiplexSocket* multiplex_socket = it->second;
-      if (multiplex_socket->IsShutdown()) {
-        LOG(INFO) << "Erase multiplex_socket(already shutdown) for ip_address: "
-                  << WifiUtils::GetHumanReadableIpAddress(ip_address);
-        multiplex_socket->~MultiplexSocket();
-        multiplex_sockets_.erase(it);
-        return ExceptionOr<WifiLanSocket>(Exception::kFailed);
-      }
-      if (multiplex_socket->IsEnabled()) {
-        std::shared_ptr<MediumSocket> virtual_socket =
-            multiplex_socket->EstablishVirtualSocket(service_id);
-        // Should not happen.
-        auto* wlan_socket = down_cast<WifiLanSocket*>(virtual_socket.get());
-        if (wlan_socket == nullptr) {
-          LOG(INFO) << "Failed to cast to WifiLanSocket for " << service_id
-                    << " with ip_address: "
-                    << WifiUtils::GetHumanReadableIpAddress(ip_address);
-          return ExceptionOr<WifiLanSocket>(Exception::kFailed);
-        }
-        return ExceptionOr<WifiLanSocket>(*wlan_socket);
-      }
-    }
-  }
   return ExceptionOr<WifiLanSocket>(Exception::kFailed);
 }
 
 ExceptionOr<WifiLanSocket> WifiLan::CreateOutgoingMultiplexSocketLocked(
     WifiLanSocket& socket, const std::string& service_id,
     const std::string& ip_address) {
-  if (is_multiplex_enabled_) {
-    // Create MultiplexSocket, but set it to be disabled as default. It will be
-    // enabled if both side support multiplex for WIFI_LAN
-    auto physical_socket_ptr = std::make_shared<WifiLanSocket>(socket);
-    MultiplexSocket* multiplex_socket =
-        MultiplexSocket::CreateOutgoingSocket(physical_socket_ptr, service_id);
-
-    std::shared_ptr<MediumSocket> virtual_socket =
-        multiplex_socket->GetVirtualSocket(service_id);
-    auto* wlan_socket = down_cast<WifiLanSocket*>(virtual_socket.get());
-    if (wlan_socket == nullptr) {
-      LOG(INFO) << "Failed to cast to WifiLanSocket for " << service_id
-                << " with ip_address: "
-                << WifiUtils::GetHumanReadableIpAddress(ip_address);
-      return ExceptionOr<WifiLanSocket>(Exception::kFailed);
-    }
-    LOG(INFO) << "Multiplex socket created for ip_address: "
-              << WifiUtils::GetHumanReadableIpAddress(ip_address);
-    multiplex_sockets_.emplace(ip_address, multiplex_socket);
-    return ExceptionOr<WifiLanSocket>(*wlan_socket);
-  }
   return ExceptionOr<WifiLanSocket>(Exception::kFailed);
 }
 

connections/implementation/mediums/wifi_lan.h

@@ -23,7 +23,6 @@
 #include "absl/container/flat_hash_map.h"
 #include "absl/container/flat_hash_set.h"
 #include "absl/functional/any_invocable.h"
-#include "connections/implementation/mediums/multiplex/multiplex_socket.h"
 #include "internal/platform/cancellation_flag.h"
 #include "internal/platform/exception.h"
 #include "internal/platform/expected.h"
@@ -215,13 +214,6 @@ class WifiLan {
   absl::flat_hash_map<std::string, WifiLanServerSocket> server_sockets_
       ABSL_GUARDED_BY(mutex_);
 
-  // Whether the multiplex feature is enabled.
-  bool is_multiplex_enabled_ = false;
-
-  // A map of IpAddress -> MultiplexSocket.
-  absl::flat_hash_map<std::string, mediums::multiplex::MultiplexSocket*>
-      multiplex_sockets_ ABSL_GUARDED_BY(mutex_);
-
   std::string last_mdns_service_name_ ABSL_GUARDED_BY(mutex_);
   int last_server_port_ ABSL_GUARDED_BY(mutex_) = 0;
 };