nearbyconnections : Implement WifiLanV2 Connection functions for /medium, /public(wrapper), /g3.

PiperOrigin-RevId: 405820452

cpp/platform/impl/g3/BUILD

@@ -74,6 +74,7 @@ cc_library(
         "//absl/container:flat_hash_map",
         "//absl/container:flat_hash_set",
         "//absl/strings",
+        "//absl/strings:str_format",
         "//absl/synchronization",
         "//platform/api:comm",
         "//platform/base",

cpp/platform/impl/g3/wifi_lan_v2.cc

@@ -19,6 +19,8 @@
 #include <string>
 #include <utility>
 
+#include "absl/strings/escaping.h"
+#include "absl/strings/str_format.h"
 #include "absl/synchronization/mutex.h"
 #include "platform/api/wifi_lan_v2.h"
 #include "platform/base/cancellation_flag_listener.h"
@@ -77,10 +79,8 @@ void WifiLanSocketV2::DoClose() {
     remote_socket_ = nullptr;
     output_->GetOutputStream().Close();
     output_->GetInputStream().Close();
-    if (IsConnectedLocked()) {
-      input_->GetOutputStream().Close();
-      input_->GetInputStream().Close();
-    }
+    input_->GetOutputStream().Close();
+    input_->GetInputStream().Close();
     closed_ = true;
   }
 }
@@ -97,28 +97,59 @@ OutputStream& WifiLanSocketV2::GetLocalOutputStream() {
   return output_->GetOutputStream();
 }
 
-std::string WifiLanServerSocketV2::GetIpAddress() {
-  absl::MutexLock lock(&mutex_);
-  return {};
-}
-
-int WifiLanServerSocketV2::GetPort() {
-  absl::MutexLock lock(&mutex_);
-  return 0;
+std::string WifiLanServerSocketV2::GetName(const std::string& ip_address,
+                                           int port) {
+  std::string dot_delimited_string;
+  if (!ip_address.empty()) {
+    for (auto byte : ip_address) {
+      if (!dot_delimited_string.empty())
+        absl::StrAppend(&dot_delimited_string, ".");
+      absl::StrAppend(&dot_delimited_string, absl::StrFormat("%d", byte));
+    }
+  }
+  std::string out = absl::StrCat(dot_delimited_string, ":", port);
+  return out;
 }
 
 std::unique_ptr<api::WifiLanSocketV2> WifiLanServerSocketV2::Accept() {
   absl::MutexLock lock(&mutex_);
-  return nullptr;
+  while (!closed_ && pending_sockets_.empty()) {
+    cond_.Wait(&mutex_);
+  }
+  // whether or not we were running in the wait loop, return early if closed.
+  if (closed_) return {};
+  auto* remote_socket =
+      pending_sockets_.extract(pending_sockets_.begin()).value();
+  CHECK(remote_socket);
+
+  auto local_socket = std::make_unique<WifiLanSocketV2>();
+  local_socket->Connect(*remote_socket);
+  remote_socket->Connect(*local_socket);
+  cond_.SignalAll();
+  return local_socket;
 }
 
 bool WifiLanServerSocketV2::Connect(WifiLanSocketV2& socket) {
   absl::MutexLock lock(&mutex_);
+  if (closed_) return false;
+  if (socket.IsConnected()) {
+    NEARBY_LOGS(ERROR)
+        << "Failed to connect to WifiLan server socket: already connected";
+    return true;  // already connected.
+  }
+  // add client socket to the pending list
+  pending_sockets_.insert(&socket);
+  cond_.SignalAll();
+  while (!socket.IsConnected()) {
+    cond_.Wait(&mutex_);
+    if (closed_) return false;
+  }
   return true;
 }
 
 void WifiLanServerSocketV2::SetCloseNotifier(std::function<void()> notifier) {
   absl::MutexLock lock(&mutex_);
+  close_notifier_ = std::move(notifier);
 }
 
 WifiLanServerSocketV2::~WifiLanServerSocketV2() {
@@ -131,7 +162,22 @@ Exception WifiLanServerSocketV2::Close() {
   return DoClose();
 }
 
-Exception WifiLanServerSocketV2::DoClose() { return {Exception::kSuccess}; }
+Exception WifiLanServerSocketV2::DoClose() {
+  bool should_notify = !closed_;
+  closed_ = true;
+  if (should_notify) {
+    cond_.SignalAll();
+    if (close_notifier_) {
+      auto notifier = std::move(close_notifier_);
+      mutex_.Unlock();
+      // Notifier may contain calls to public API, and may cause deadlock, if
+      // mutex_ is held during the call.
+      notifier();
+      mutex_.Lock();
+    }
+  }
+  return {Exception::kSuccess};
+}
 
 WifiLanMediumV2::WifiLanMediumV2() {
   auto& env = MediumEnvironment::Instance();
@@ -235,19 +281,90 @@ bool WifiLanMediumV2::StopDiscovery(const std::string& service_type) {
 }
 
 std::unique_ptr<api::WifiLanSocketV2> WifiLanMediumV2::ConnectToService(
-    NsdServiceInfo& remote_service_info, CancellationFlag* cancellation_flag) {
-  return nullptr;
+    const NsdServiceInfo& remote_service_info,
+    CancellationFlag* cancellation_flag) {
+  std::string service_type = remote_service_info.GetServiceType();
+  NEARBY_LOGS(INFO) << "G3 WifiLan ConnectToService [self]: medium=" << this
+                    << ", service_type=" << service_type;
+  return ConnectToService(remote_service_info.GetIPAddress(),
+                          remote_service_info.GetPort(), cancellation_flag);
 }
 
 std::unique_ptr<api::WifiLanSocketV2> WifiLanMediumV2::ConnectToService(
     const std::string& ip_address, int port,
     CancellationFlag* cancellation_flag) {
-  return nullptr;
+  NEARBY_LOGS(INFO) << "G3 WifiLan ConnectToService [self]: medium=" << this
+                    << ", ip address + port="
+                    << WifiLanServerSocketV2::GetName(ip_address, port);
+  // First, find an instance of remote medium, that exposed this service.
+  auto& env = MediumEnvironment::Instance();
+  auto* remote_medium =
+      static_cast<WifiLanMediumV2*>(env.GetWifiLanV2Medium(ip_address, port));
+  if (!remote_medium) {
+    return {};
+  }
+
+  WifiLanServerSocketV2* server_socket = nullptr;
+  NEARBY_LOGS(INFO) << "G3 WifiLan ConnectToService [peer]: medium="
+                    << remote_medium << ", remote ip address + port="
+                    << WifiLanServerSocketV2::GetName(ip_address, port);
+  // Then, find our server socket context in this medium.
+  std::string socket_name = WifiLanServerSocketV2::GetName(ip_address, port);
+  {
+    absl::MutexLock medium_lock(&remote_medium->mutex_);
+    auto item = remote_medium->server_sockets_.find(socket_name);
+    server_socket = item != server_sockets_.end() ? item->second : nullptr;
+    if (server_socket == nullptr) {
+      NEARBY_LOGS(ERROR)
+          << "G3 WifiLan Failed to find WifiLan Server socket: socket_name="
+          << socket_name;
+      return {};
+    }
+  }
+
+  if (cancellation_flag->Cancelled()) {
+    NEARBY_LOGS(ERROR) << "G3 WifiLan Connect: Has been cancelled: socket_name="
+                       << socket_name;
+    return {};
+  }
+
+  CancellationFlagListener listener(cancellation_flag, [&server_socket]() {
+    NEARBY_LOGS(INFO) << "G3 WifiLan Cancel Connect.";
+    if (server_socket != nullptr) {
+      server_socket->Close();
+    }
+  });
+
+  auto socket = std::make_unique<WifiLanSocketV2>();
+  // Finally, Request to connect to this socket.
+  if (!server_socket->Connect(*socket)) {
+    NEARBY_LOGS(ERROR) << "G3 WifiLan Failed to connect to existing WifiLan "
+                          "Server socket: name="
+                       << socket_name;
+    return {};
+  }
+  NEARBY_LOGS(INFO) << "G3 WifiLan ConnectToService: connected: socket="
+                    << socket.get();
+  return socket;
 }
 
 std::unique_ptr<api::WifiLanServerSocketV2> WifiLanMediumV2::ListenForService(
     int port) {
-  return {};
+  auto& env = MediumEnvironment::Instance();
+  auto server_socket = std::make_unique<WifiLanServerSocketV2>();
+  server_socket->SetIPAddress(env.GetFakeIPAddress());
+  server_socket->SetPort(port == 0 ? env.GetFakePort() : port);
+  std::string socket_name = WifiLanServerSocketV2::GetName(
+      server_socket->GetIPAddress(), server_socket->GetPort());
+  server_socket->SetCloseNotifier([this, socket_name]() {
+    absl::MutexLock lock(&mutex_);
+    server_sockets_.erase(socket_name);
+  });
+  NEARBY_LOGS(INFO) << "G3 WifiLan Adding server socket: medium=" << this
+                    << ", socket_name=" << socket_name;
+  absl::MutexLock lock(&mutex_);
+  server_sockets_.insert({socket_name, server_socket.get()});
+  return server_socket;
 }
 
 }  // namespace g3

cpp/platform/impl/g3/wifi_lan_v2.h

@@ -92,13 +92,33 @@ class WifiLanSocketV2 : public api::WifiLanSocketV2 {
 
 class WifiLanServerSocketV2 : public api::WifiLanServerSocketV2 {
  public:
+  static std::string GetName(const std::string& ip_address, int port);
+
   ~WifiLanServerSocketV2() override;
 
-  // Returns ip address.
-  std::string GetIpAddress() override ABSL_LOCKS_EXCLUDED(mutex_);
+  // Gets ip address.
+  std::string GetIPAddress() const override ABSL_LOCKS_EXCLUDED(mutex_) {
+    absl::MutexLock lock(&mutex_);
+    return ip_address_;
+  }
 
-  // Returns port.
-  int GetPort() override ABSL_LOCKS_EXCLUDED(mutex_);
+  // Sets the ip address.
+  void SetIPAddress(const std::string& ip_address) ABSL_LOCKS_EXCLUDED(mutex_) {
+    absl::MutexLock lock(&mutex_);
+    ip_address_ = ip_address;
+  }
+
+  // Gets the port.
+  int GetPort() const override ABSL_LOCKS_EXCLUDED(mutex_) {
+    absl::MutexLock lock(&mutex_);
+    return port_;
+  }
+
+  // Sets the port.
+  void SetPort(int port) ABSL_LOCKS_EXCLUDED(mutex_) {
+    absl::MutexLock lock(&mutex_);
+    port_ = port;
+  }
 
   // Blocks until either:
   // - at least one incoming connection request is available, or
@@ -135,7 +155,14 @@ class WifiLanServerSocketV2 : public api::WifiLanServerSocketV2 {
  private:
   Exception DoClose() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
 
-  absl::Mutex mutex_;
+  mutable absl::Mutex mutex_;
+  std::string ip_address_ ABSL_GUARDED_BY(mutex_);
+  int port_ ABSL_GUARDED_BY(mutex_);
+  absl::CondVar cond_;
+  absl::flat_hash_set<WifiLanSocketV2*> pending_sockets_
+      ABSL_GUARDED_BY(mutex_);
+  std::function<void()> close_notifier_ ABSL_GUARDED_BY(mutex_);
+  bool closed_ ABSL_GUARDED_BY(mutex_) = false;
 };
 
 // Container of operations that can be performed over the WifiLan medium.
@@ -187,7 +214,7 @@ class WifiLanMediumV2 : public api::WifiLanMediumV2 {
   // On success, returns a new WifiLanSocket.
   // On error, returns nullptr.
   std::unique_ptr<api::WifiLanSocketV2> ConnectToService(
-      NsdServiceInfo& remote_service_info,
+      const NsdServiceInfo& remote_service_info,
       CancellationFlag* cancellation_flag) override;
 
   // Connects to a WifiLan service by ip address and port.
@@ -242,6 +269,8 @@ class WifiLanMediumV2 : public api::WifiLanMediumV2 {
   absl::Mutex mutex_;
   AdvertisingInfo advertising_info_ ABSL_GUARDED_BY(mutex_);
   DiscoveringInfo discovering_info_ ABSL_GUARDED_BY(mutex_);
+  absl::flat_hash_map<std::string, WifiLanServerSocketV2*> server_sockets_
+      ABSL_GUARDED_BY(mutex_);
 };
 
 }  // namespace g3