Implement AWDL in connection layer (part 1)

PiperOrigin-RevId: 742925127

internal/platform/implementation/g3/BUILD

@@ -62,6 +62,7 @@ cc_library(
     name = "comm",
     testonly = True,
     srcs = [
+        "awdl.cc",
         "ble.cc",
         "ble_v2.cc",
         "bluetooth_adapter.cc",
@@ -72,6 +73,7 @@ cc_library(
         "wifi_lan.cc",
     ],
     hdrs = [
+        "awdl.h",
         "ble.h",
         "ble_v2.h",
         "bluetooth_adapter.h",

internal/platform/implementation/g3/awdl.cc

@@ -0,0 +1,309 @@
+// 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 "internal/platform/implementation/g3/awdl.h"
+
+#include <iostream>
+#include <memory>
+#include <string>
+#include <utility>
+
+#include "absl/log/check.h"
+#include "absl/strings/str_cat.h"
+#include "absl/strings/str_format.h"
+#include "absl/synchronization/mutex.h"
+#include "internal/platform/cancellation_flag.h"
+#include "internal/platform/cancellation_flag_listener.h"
+#include "internal/platform/exception.h"
+#include "internal/platform/implementation/awdl.h"
+#include "internal/platform/logging.h"
+#include "internal/platform/medium_environment.h"
+#include "internal/platform/nsd_service_info.h"
+
+namespace nearby {
+namespace g3 {
+
+std::string AwdlServerSocket::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::AwdlSocket> AwdlServerSocket::Accept() {
+  absl::MutexLock lock(&mutex_);
+  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<AwdlSocket>();
+  local_socket->Connect(*remote_socket);
+  remote_socket->Connect(*local_socket);
+  cond_.SignalAll();
+  return local_socket;
+}
+
+bool AwdlServerSocket::Connect(AwdlSocket& socket) {
+  absl::MutexLock lock(&mutex_);
+  if (closed_) return false;
+  if (socket.IsConnected()) {
+    NEARBY_LOGS(ERROR)
+        << "Failed to connect to Awdl 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 AwdlServerSocket::SetCloseNotifier(
+    absl::AnyInvocable<void()> notifier) {
+  absl::MutexLock lock(&mutex_);
+  close_notifier_ = std::move(notifier);
+}
+
+AwdlServerSocket::~AwdlServerSocket() {
+  absl::MutexLock lock(&mutex_);
+  DoClose();
+}
+
+Exception AwdlServerSocket::Close() {
+  absl::MutexLock lock(&mutex_);
+  return DoClose();
+}
+
+Exception AwdlServerSocket::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};
+}
+
+AwdlMedium::AwdlMedium() {
+  auto& env = MediumEnvironment::Instance();
+  env.RegisterAwdlMedium(*this);
+}
+
+AwdlMedium::~AwdlMedium() {
+  auto& env = MediumEnvironment::Instance();
+  env.UnregisterAwdlMedium(*this);
+}
+
+bool AwdlMedium::StartAdvertising(const NsdServiceInfo& nsd_service_info) {
+  std::string service_type = nsd_service_info.GetServiceType();
+  NEARBY_LOGS(INFO) << "G3 Awdl StartAdvertising: nsd_service_info="
+                    << &nsd_service_info
+                    << ", service_name=" << nsd_service_info.GetServiceName()
+                    << ", service_type=" << service_type;
+  {
+    absl::MutexLock lock(&mutex_);
+    if (advertising_info_.Existed(service_type)) {
+      NEARBY_LOGS(INFO)
+          << "G3 Awdl StartAdvertising: Can't start advertising because "
+             "service_type="
+          << service_type << ", has started already.";
+      return false;
+    }
+  }
+  auto& env = MediumEnvironment::Instance();
+  env.UpdateAwdlMediumForAdvertising(*this, nsd_service_info,
+                                        /*enabled=*/true);
+  {
+    absl::MutexLock lock(&mutex_);
+    advertising_info_.Add(service_type);
+  }
+  return true;
+}
+
+bool AwdlMedium::StopAdvertising(const NsdServiceInfo& nsd_service_info) {
+  std::string service_type = nsd_service_info.GetServiceType();
+  NEARBY_LOGS(INFO) << "G3 Awdl StopAdvertising: nsd_service_info="
+                    << &nsd_service_info
+                    << ", service_name=" << nsd_service_info.GetServiceName()
+                    << ", service_type=" << service_type;
+  {
+    absl::MutexLock lock(&mutex_);
+    if (!advertising_info_.Existed(service_type)) {
+      NEARBY_LOGS(INFO)
+          << "G3 Awdl StopAdvertising: Can't stop advertising because "
+             "we never started advertising for service_type="
+          << service_type;
+      return false;
+    }
+    advertising_info_.Remove(service_type);
+  }
+  auto& env = MediumEnvironment::Instance();
+  env.UpdateAwdlMediumForAdvertising(*this, nsd_service_info,
+                                        /*enabled=*/false);
+  return true;
+}
+
+bool AwdlMedium::StartDiscovery(const std::string& service_type,
+                                   DiscoveredServiceCallback callback) {
+  NEARBY_LOGS(INFO) << "G3 Awdl StartDiscovery: service_type="
+                    << service_type;
+  {
+    absl::MutexLock lock(&mutex_);
+    if (discovering_info_.Existed(service_type)) {
+      NEARBY_LOGS(INFO)
+          << "G3 Awdl StartDiscovery: Can't start discovery because "
+             "service_type="
+          << service_type << " has started already.";
+      return false;
+    }
+  }
+  auto& env = MediumEnvironment::Instance();
+  env.UpdateAwdlMediumForDiscovery(*this, std::move(callback), service_type,
+                                      true);
+  {
+    absl::MutexLock lock(&mutex_);
+    discovering_info_.Add(service_type);
+  }
+  return true;
+}
+
+bool AwdlMedium::StopDiscovery(const std::string& service_type) {
+  NEARBY_LOGS(INFO) << "G3 Awdl StopDiscovery: service_type="
+                    << service_type;
+  {
+    absl::MutexLock lock(&mutex_);
+    if (!discovering_info_.Existed(service_type)) {
+      NEARBY_LOGS(INFO)
+          << "G3 Awdl StopDiscovery: Can't stop discovering because we "
+             "never started discovering.";
+      return false;
+    }
+    discovering_info_.Remove(service_type);
+  }
+  auto& env = MediumEnvironment::Instance();
+  env.UpdateAwdlMediumForDiscovery(*this, {}, service_type, false);
+  return true;
+}
+
+std::unique_ptr<api::AwdlSocket> AwdlMedium::ConnectToService(
+    const NsdServiceInfo& remote_service_info,
+    CancellationFlag* cancellation_flag) {
+  std::string service_type = remote_service_info.GetServiceType();
+  NEARBY_LOGS(INFO) << "G3 Awdl ConnectToService [self]: medium=" << this
+                    << ", service_type=" << service_type;
+  return ConnectToService(remote_service_info.GetIPAddress(),
+                          remote_service_info.GetPort(), cancellation_flag);
+}
+
+std::unique_ptr<api::AwdlSocket> AwdlMedium::ConnectToService(
+    const std::string& ip_address, int port,
+    CancellationFlag* cancellation_flag) {
+  std::string socket_name = AwdlServerSocket::GetName(ip_address, port);
+  NEARBY_LOGS(INFO) << "G3 Awdl ConnectToService [self]: medium=" << this
+                    << ", ip address + port=" << socket_name;
+  // First, find an instance of remote medium, that exposed this service.
+  auto& env = MediumEnvironment::Instance();
+  auto* remote_medium =
+      static_cast<AwdlMedium*>(env.GetAwdlMedium(ip_address, port));
+  if (!remote_medium) {
+    return {};
+  }
+
+  AwdlServerSocket* server_socket = nullptr;
+  NEARBY_LOGS(INFO) << "G3 Awdl ConnectToService [peer]: medium="
+                    << remote_medium
+                    << ", remote ip address + port=" << socket_name;
+  // Then, find our server socket context in this medium.
+  {
+    absl::MutexLock medium_lock(&remote_medium->mutex_);
+    auto item = remote_medium->server_sockets_.find(socket_name);
+    server_socket =
+        item != remote_medium->server_sockets_.end() ? item->second : nullptr;
+    if (server_socket == nullptr) {
+      NEARBY_LOGS(ERROR)
+          << "G3 Awdl Failed to find Awdl Server socket: socket_name="
+          << socket_name;
+      return {};
+    }
+  }
+
+  if (cancellation_flag->Cancelled()) {
+    NEARBY_LOGS(ERROR) << "G3 Awdl Connect: Has been cancelled: socket_name="
+                       << socket_name;
+    return {};
+  }
+
+  CancellationFlagListener listener(cancellation_flag, [&server_socket]() {
+    NEARBY_LOGS(INFO) << "G3 Awdl Cancel Connect.";
+    if (server_socket != nullptr) {
+      server_socket->Close();
+    }
+  });
+
+  auto socket = std::make_unique<AwdlSocket>();
+  // Finally, Request to connect to this socket.
+  if (!server_socket->Connect(*socket)) {
+    NEARBY_LOGS(ERROR) << "G3 Awdl Failed to connect to existing Awdl "
+                          "Server socket: name="
+                       << socket_name;
+    return {};
+  }
+  NEARBY_LOGS(INFO) << "G3 Awdl ConnectToService: connected: socket="
+                    << socket.get();
+  return socket;
+}
+
+std::unique_ptr<api::AwdlServerSocket> AwdlMedium::ListenForService(
+    int port) {
+  auto& env = MediumEnvironment::Instance();
+  auto server_socket = std::make_unique<AwdlServerSocket>();
+  server_socket->SetIPAddress(env.GetFakeIPAddress());
+  server_socket->SetPort(port == 0 ? env.GetFakePort() : port);
+  std::string socket_name = AwdlServerSocket::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 Awdl 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
+}  // namespace nearby

internal/platform/implementation/g3/awdl.h

@@ -0,0 +1,254 @@
+// 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 PLATFORM_IMPL_G3_AWDL_H_
+#define PLATFORM_IMPL_G3_AWDL_H_
+
+#include <memory>
+#include <string>
+#include <utility>
+
+#include "absl/container/flat_hash_map.h"
+#include "absl/container/flat_hash_set.h"
+#include "absl/synchronization/mutex.h"
+#include "internal/platform/byte_array.h"
+#include "internal/platform/implementation/g3/multi_thread_executor.h"
+#include "internal/platform/implementation/g3/socket_base.h"
+#include "internal/platform/implementation/awdl.h"
+#include "internal/platform/input_stream.h"
+#include "internal/platform/nsd_service_info.h"
+#include "internal/platform/output_stream.h"
+
+namespace nearby {
+namespace g3 {
+
+class AwdlMedium;
+
+class AwdlSocket : public api::AwdlSocket, public SocketBase {
+ public:
+  // Returns the InputStream of this connected AwdlSocket.
+  InputStream& GetInputStream() override {
+    return SocketBase::GetInputStream();
+  }
+
+  // Returns the OutputStream of this connected AwdlSocket.
+  // This stream is for local side to write.
+  OutputStream& GetOutputStream() override {
+    return SocketBase::GetOutputStream();
+  }
+
+  // Returns address of a remote AwdlSocket or nullptr.
+  AwdlSocket* GetRemoteSocket() {
+    return static_cast<AwdlSocket*>(SocketBase::GetRemoteSocket());
+  }
+
+  // Returns Exception::kIo on error, Exception::kSuccess otherwise.
+  Exception Close() override { return SocketBase::Close(); }
+};
+
+class AwdlServerSocket : public api::AwdlServerSocket {
+ public:
+  static std::string GetName(const std::string& ip_address, int port);
+
+  ~AwdlServerSocket() override;
+
+  // Gets ip address.
+  std::string GetIPAddress() const override ABSL_LOCKS_EXCLUDED(mutex_) {
+    absl::MutexLock lock(&mutex_);
+    return ip_address_;
+  }
+
+  // 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
+  // - ServerSocket is closed.
+  // On success, returns connected socket, ready to exchange data.
+  // Returns nullptr on error.
+  // Once error is reported, it is permanent, and ServerSocket has to be closed.
+  //
+  // Called by the server side of a connection.
+  // Returns AwdlSocket to the server side.
+  // If not null, returned socket is connected to its remote (client-side) peer.
+  std::unique_ptr<api::AwdlSocket> Accept() override
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Blocks until either:
+  // - connection is available, or
+  // - server socket is closed, or
+  // - error happens.
+  //
+  // Called by the client side of a connection.
+  // Returns true, if socket is successfully connected.
+  bool Connect(AwdlSocket& socket) ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Called by the server side of a connection before passing ownership of
+  // AwdlServerSocker to user, to track validity of a pointer to this
+  // server socket.
+  void SetCloseNotifier(absl::AnyInvocable<void()> notifier)
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns Exception::kIo on error, Exception::kSuccess otherwise.
+  // Calls close_notifier if it was previously set, and marks socket as closed.
+  Exception Close() override ABSL_LOCKS_EXCLUDED(mutex_);
+
+ private:
+  Exception DoClose() ABSL_EXCLUSIVE_LOCKS_REQUIRED(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<AwdlSocket*> pending_sockets_ ABSL_GUARDED_BY(mutex_);
+  absl::AnyInvocable<void()> close_notifier_ ABSL_GUARDED_BY(mutex_);
+  bool closed_ ABSL_GUARDED_BY(mutex_) = false;
+};
+
+// Container of operations that can be performed over the Awdl medium.
+class AwdlMedium : public api::AwdlMedium {
+ public:
+  AwdlMedium();
+  ~AwdlMedium() override;
+
+  // Check if a network connection to a primary router exist.
+  bool IsNetworkConnected() const override { return true; }
+
+  // Starts Awdl advertising.
+  //
+  // nsd_service_info - NsdServiceInfo data that's advertised through mDNS
+  //                    service.
+  // On success if the service is now advertising.
+  // On error if the service cannot start to advertise or the service type in
+  // NsdServiceInfo has been passed previously which StopAdvertising is not
+  // been called.
+  bool StartAdvertising(const NsdServiceInfo& nsd_service_info) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Stops Awdl advertising.
+  //
+  // nsd_service_info - NsdServiceInfo data that's advertised through mDNS
+  //                    service.
+  // On success if the service stops advertising.
+  // On error if the service cannot stop advertising or the service type in
+  // NsdServiceInfo cannot be found.
+  bool StopAdvertising(const NsdServiceInfo& nsd_service_info) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Starts the discovery of nearby Awdl services.
+  //
+  // Returns true once the Awdl discovery has been initiated. The
+  // service_type is associated with callback.
+  bool StartDiscovery(const std::string& service_type,
+                      DiscoveredServiceCallback callback) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Stops the discovery of nearby Awdl services.
+  //
+  // service_type - The one assigend in StartDiscovery.
+  // On success if service_type is matched to the callback and will be removed
+  //            from the list. If list is empty then stops the Awdl discovery
+  //            service.
+  // On error if the service_type is not existed, then return immediately.
+  bool StopDiscovery(const std::string& service_type) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Connects to a Awdl service.
+  // On success, returns a new AwdlSocket.
+  // On error, returns nullptr.
+  std::unique_ptr<api::AwdlSocket> ConnectToService(
+      const NsdServiceInfo& remote_service_info,
+      CancellationFlag* cancellation_flag) override ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Connects to a Awdl service by ip address and port.
+  // On success, returns a new AwdlSocket.
+  // On error, returns nullptr.
+  std::unique_ptr<api::AwdlSocket> ConnectToService(
+      const std::string& ip_address, int port,
+      CancellationFlag* cancellation_flag) override ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Listens for incoming connection.
+  //
+  // port - A port number.
+  //         0 : use a random port.
+  //   1~65536 : open a server socket on that exact port.
+  // On success, returns a new AwdlServerSocket.
+  // On error, returns nullptr.
+  std::unique_ptr<api::AwdlServerSocket> ListenForService(int port) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns the port range as a pair of min and max port.
+  absl::optional<std::pair<std::int32_t, std::int32_t>> GetDynamicPortRange()
+      override {
+    return std::make_pair(49152, 65535);
+  }
+
+ private:
+  struct AdvertisingInfo {
+    bool Empty() const { return service_types.empty(); }
+    void Clear() { service_types.clear(); }
+    void Add(const std::string& service_type) {
+      service_types.insert(service_type);
+    }
+    void Remove(const std::string& service_type) {
+      service_types.erase(service_type);
+    }
+    bool Existed(const std::string& service_type) const {
+      return service_types.contains(service_type);
+    }
+
+    absl::flat_hash_set<std::string> service_types;
+  };
+  struct DiscoveringInfo {
+    bool Empty() const { return service_types.empty(); }
+    void Clear() { service_types.clear(); }
+    void Add(const std::string& service_type) {
+      service_types.insert(service_type);
+    }
+    void Remove(const std::string& service_type) {
+      service_types.erase(service_type);
+    }
+    bool Existed(const std::string& service_type) const {
+      return service_types.contains(service_type);
+    }
+
+    absl::flat_hash_set<std::string> service_types;
+  };
+
+  absl::Mutex mutex_;
+  AdvertisingInfo advertising_info_ ABSL_GUARDED_BY(mutex_);
+  DiscoveringInfo discovering_info_ ABSL_GUARDED_BY(mutex_);
+  absl::flat_hash_map<std::string, AwdlServerSocket*> server_sockets_
+      ABSL_GUARDED_BY(mutex_);
+};
+
+}  // namespace g3
+}  // namespace nearby
+
+#endif  // PLATFORM_IMPL_G3_AWDL_H_

internal/platform/implementation/g3/platform.cc

@@ -28,6 +28,7 @@
 #include "internal/base/files.h"
 #include "internal/platform/implementation/atomic_boolean.h"
 #include "internal/platform/implementation/atomic_reference.h"
+#include "internal/platform/implementation/awdl.h"
 #include "internal/platform/implementation/ble.h"
 #include "internal/platform/implementation/ble_v2.h"
 #include "internal/platform/implementation/bluetooth_adapter.h"
@@ -36,6 +37,7 @@
 #include "internal/platform/implementation/count_down_latch.h"
 #include "internal/platform/implementation/credential_storage.h"
 #include "internal/platform/implementation/device_info.h"
+#include "internal/platform/implementation/g3/awdl.h"
 #include "internal/platform/implementation/http_loader.h"
 #include "internal/platform/implementation/input_file.h"
 #include "internal/platform/implementation/log_message.h"
@@ -213,6 +215,10 @@ std::unique_ptr<WifiLanMedium> ImplementationPlatform::CreateWifiLanMedium() {
   return std::make_unique<g3::WifiLanMedium>();
 }
 
+std::unique_ptr<AwdlMedium> ImplementationPlatform::CreateAwdlMedium() {
+  return std::make_unique<g3::AwdlMedium>();
+}
+
 std::unique_ptr<WifiHotspotMedium>
 ImplementationPlatform::CreateWifiHotspotMedium() {
   return std::make_unique<g3::WifiHotspotMedium>();

internal/platform/implementation/windows/platform.cc

@@ -39,6 +39,7 @@
 #include "internal/base/files.h"
 #include "internal/platform/implementation/atomic_boolean.h"
 #include "internal/platform/implementation/atomic_reference.h"
+#include "internal/platform/implementation/awdl.h"
 #include "internal/platform/implementation/ble.h"
 #include "internal/platform/implementation/ble_v2.h"
 #include "internal/platform/implementation/bluetooth_adapter.h"
@@ -303,6 +304,10 @@ std::unique_ptr<WifiLanMedium> ImplementationPlatform::CreateWifiLanMedium() {
   return std::make_unique<windows::WifiLanMedium>();
 }
 
+std::unique_ptr<AwdlMedium> ImplementationPlatform::CreateAwdlMedium() {
+  return nullptr;
+}
+
 std::unique_ptr<WifiHotspotMedium>
 ImplementationPlatform::CreateWifiHotspotMedium() {
   return std::make_unique<windows::WifiHotspotMedium>();