Roll forward to cl/314549634

Change-Id: I4d1e7eacd5fa4078a094571ad6d5f51e422535ff

cpp/platform_v2/base/BUILD

@@ -24,6 +24,7 @@ cc_library(
         "//platform_v2/api:__subpackages__",
     ],
     deps = [
+        "//absl/meta:type_traits",
         "//absl/strings",
         "//absl/time",
     ],
@@ -32,9 +33,11 @@ cc_library(
 cc_library(
     name = "util",
     srcs = [
+        "base_input_stream.cc",
         "base_pipe.cc",
     ],
     hdrs = [
+        "base_input_stream.h",
         "base_mutex_lock.h",
         "base_pipe.h",
     ],
@@ -44,11 +47,47 @@ cc_library(
     ],
     deps = [
         ":base",
-        "//platform_v2/api",
+        "//platform_v2/api:types",
         "//absl/base:core_headers",
     ],
 )
 
+cc_library(
+    name = "logging",
+    hdrs = [
+        "logging.h",
+    ],
+    visibility = [
+        "//platform_v2:__subpackages__",
+    ],
+    deps = [
+        "//platform:logging",
+    ],
+)
+
+cc_library(
+    name = "test_util",
+    testonly = True,
+    srcs = [
+        "medium_environment.cc",
+    ],
+    hdrs = [
+        "medium_environment.h",
+    ],
+    visibility = [
+        "//core_v2:__subpackages__",
+        "//platform_v2/impl:__subpackages__",
+        "//platform_v2/public:__pkg__",
+    ],
+    deps = [
+        ":base",
+        ":logging",
+        "//platform_v2/api:comm",
+        "//platform_v2/public:types",
+        "//absl/container:flat_hash_map",
+    ],
+)
+
 cc_test(
     name = "platform_base_test",
     srcs = [

cpp/platform_v2/base/base_input_stream.cc

@@ -0,0 +1,85 @@
+#include "platform_v2/base/base_input_stream.h"
+
+namespace location {
+namespace nearby {
+
+ExceptionOr<ByteArray> BaseInputStream::Read(std::int64_t size) {
+  if (!IsAvailable(size)) {
+    return ExceptionOr<ByteArray>{Exception::kIo};
+  }
+
+  ByteArray read_bytes{static_cast<size_t>(size)};
+  if (read_bytes.CopyAt(/*offset=*/0, buffer_,
+                        /*source_offset=*/position_)) {
+    position_ += size;
+    return ExceptionOr<ByteArray>{read_bytes};
+  } else {
+    return ExceptionOr<ByteArray>{Exception::kIo};
+  }
+}
+
+std::uint8_t BaseInputStream::ReadUint8() {
+  constexpr int byte_size = sizeof(std::uint8_t);
+  ByteArray read_bytes = ReadBytes(byte_size);
+  if (read_bytes.Empty() || read_bytes.size() != byte_size) {
+    return -1;
+  }
+
+  return read_bytes.data()[0];
+}
+
+std::uint16_t BaseInputStream::ReadUint16() {
+  constexpr int byte_size = sizeof(std::uint16_t);
+  ByteArray read_bytes = ReadBytes(byte_size);
+  if (read_bytes.Empty() || read_bytes.size() != byte_size) {
+    return -1;
+  }
+
+  // Convert from network order.
+  const char *data = read_bytes.data();
+  return static_cast<uint16_t>(data[0]) << 8 | static_cast<uint16_t>(data[1]);
+}
+
+std::uint32_t BaseInputStream::ReadUint32() {
+  constexpr int byte_size = sizeof(std::uint32_t);
+  ByteArray read_bytes = ReadBytes(byte_size);
+  if (read_bytes.Empty() || read_bytes.size() != byte_size) {
+    return -1;
+  }
+
+  // Convert from network order.
+  const char *data = read_bytes.data();
+  return static_cast<uint32_t>(data[0]) << 24 |
+         static_cast<uint32_t>(data[1]) << 16 |
+         static_cast<uint32_t>(data[2]) << 8 | static_cast<uint32_t>(data[3]);
+}
+
+std::uint64_t BaseInputStream::ReadUint64() {
+  constexpr int byte_size = sizeof(std::uint64_t);
+  ByteArray read_bytes = ReadBytes(byte_size);
+  if (read_bytes.Empty() || read_bytes.size() != byte_size) {
+    return -1;
+  }
+
+  // Convert from network order.
+  const char *data = read_bytes.data();
+  return static_cast<uint64_t>(data[0]) << 56 |
+         static_cast<uint64_t>(data[1]) << 48 |
+         static_cast<uint64_t>(data[2]) << 40 |
+         static_cast<uint64_t>(data[3]) << 32 |
+         static_cast<uint64_t>(data[4]) << 24 |
+         static_cast<uint64_t>(data[5]) << 16 |
+         static_cast<uint64_t>(data[6]) << 8 | static_cast<uint64_t>(data[7]);
+}
+
+ByteArray BaseInputStream::ReadBytes(int size) {
+  ExceptionOr<ByteArray> read_bytes_result = Read(size);
+  if (!read_bytes_result.ok()) {
+    return ByteArray{};
+  }
+
+  return read_bytes_result.GetResult();
+}
+
+}  // namespace nearby
+}  // namespace location

cpp/platform_v2/base/base_input_stream.h

@@ -0,0 +1,44 @@
+#ifndef PLATFORM_V2_BASE_BASE_INPUT_STREAM_H_
+#define PLATFORM_V2_BASE_BASE_INPUT_STREAM_H_
+
+#include "platform_v2/base/byte_array.h"
+#include "platform_v2/base/exception.h"
+#include "platform_v2/base/input_stream.h"
+
+namespace location {
+namespace nearby {
+
+// A base {@link InputStream } for reading the contents of a byte array.
+class BaseInputStream : public InputStream {
+ public:
+  explicit BaseInputStream(ByteArray &buffer) : buffer_{buffer} {}
+  BaseInputStream(const BaseInputStream &) = delete;
+  BaseInputStream &operator=(const BaseInputStream &) = delete;
+  ~BaseInputStream() override { Close(); }
+
+  ExceptionOr<ByteArray> Read(std::int64_t size) override;
+
+  Exception Close() override {
+    // Do nothing.
+    return {Exception::kSuccess};
+  }
+
+  std::uint8_t ReadUint8();
+  std::uint16_t ReadUint16();
+  std::uint32_t ReadUint32();
+  std::uint64_t ReadUint64();
+  bool IsAvailable(int size) const {
+    return buffer_.size() - position_ >= size;
+  }
+
+ private:
+  ByteArray ReadBytes(int size);
+
+  ByteArray &buffer_;
+  int position_{0};
+};
+
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_BASE_BASE_INPUT_STREAM_H_

cpp/platform_v2/base/base_pipe.cc

@@ -1,6 +1,5 @@
 #include "platform_v2/base/base_pipe.h"
 
-#include "platform_v2/api/platform.h"
 #include "platform_v2/base/base_mutex_lock.h"
 #include "platform_v2/base/input_stream.h"
 #include "platform_v2/base/output_stream.h"

cpp/platform_v2/base/byte_array.h

@@ -19,7 +19,9 @@ class ByteArray {
   ByteArray& operator=(ByteArray&&) = default;
 
   // Create ByteArray from string.
-  explicit ByteArray(absl::string_view source) { data_ = source; }
+  explicit ByteArray(absl::string_view source) {
+    SetData(source.data(), source.size());
+  }
 
   // Create default-initialized ByteArray of a given size.
   explicit ByteArray(size_t size) { SetData(size); }

cpp/platform_v2/base/exception.h

@@ -1,9 +1,10 @@
 #ifndef PLATFORM_V2_BASE_EXCEPTION_H_
 #define PLATFORM_V2_BASE_EXCEPTION_H_
 
-#include <type_traits>
 #include <utility>
 
+#include "absl/meta/type_traits.h"
+
 namespace location {
 namespace nearby {
 
@@ -64,7 +65,7 @@ class ExceptionOr {
   ExceptionOr(Exception exception) : exception_{exception} {}         // NOLINT
   // If there exists explicit conversion from from U to T,
   // then allow explicit conversion from ExceptionOr<U> to ExceptionOr<T>.
-  template <typename U, typename = std::void_t<decltype(T{std::declval<U>()})>>
+  template <typename U, typename = absl::void_t<decltype(T{std::declval<U>()})>>
   explicit ExceptionOr<T>(ExceptionOr<U> value) {
     if (!value.ok()) {
       exception_ = value.GetException();

cpp/platform_v2/base/logging.h

@@ -0,0 +1,6 @@
+#ifndef PLATFORM_V2_BASE_LOGGING_H_
+#define PLATFORM_V2_BASE_LOGGING_H_
+
+#include "platform/logging.h"
+
+#endif  // PLATFORM_V2_BASE_LOGGING_H_

cpp/platform_v2/base/medium_environment.cc

@@ -0,0 +1,191 @@
+#include "platform_v2/base/medium_environment.h"
+
+#include <atomic>
+#include <cinttypes>
+#include <new>
+#include <type_traits>
+
+#include "platform_v2/api/bluetooth_adapter.h"
+#include "platform_v2/api/bluetooth_classic.h"
+#include "platform_v2/base/logging.h"
+#include "platform_v2/public/count_down_latch.h"
+
+namespace location {
+namespace nearby {
+
+MediumEnvironment& MediumEnvironment::Instance() {
+  static std::aligned_storage_t<sizeof(MediumEnvironment),
+                                alignof(MediumEnvironment)>
+      storage;
+  static MediumEnvironment* env = new (&storage) MediumEnvironment();
+  return *env;
+}
+
+void MediumEnvironment::Reset() {
+  RunOnMediumEnvironmentThread([this]() {
+    bluetooth_adapters_.clear();
+    bluetooth_mediums_.clear();
+  });
+  Sync();
+}
+
+void MediumEnvironment::Sync(bool enable_notifications) {
+  enable_notifications_ = enable_notifications;
+  int count = 0;
+  do {
+    CountDownLatch latch(1);
+    count = job_count_ + 1;
+    // We are about to schedule one last job.
+    // When it is done, counter must be equal to count.
+    // However, if pending jobs schedule anything else,
+    // it will be pending after us.
+    // If we want to ensure we are completely idle, then we have to
+    // repeat sync, until this becomes true.
+    RunOnMediumEnvironmentThread([&latch]() { latch.CountDown(); });
+    latch.Await();
+  } while (count < job_count_);
+  NEARBY_LOG(INFO, "MediumEnvironment::Sync(): done [count=%d]", count);
+}
+
+void MediumEnvironment::OnBluetoothAdapterChangedState(
+    api::BluetoothAdapter& adapter, api::BluetoothDevice& adapter_device,
+    std::string name, bool enabled, api::BluetoothAdapter::ScanMode mode) {
+  RunOnMediumEnvironmentThread([this, &adapter, &adapter_device,
+                                name = std::move(name), enabled, mode]() {
+    NEARBY_LOG(INFO,
+               "[adapter=%p, device=%p] update: name=%s, enabled=%d, mode=%d",
+               &adapter, &adapter_device, name.c_str(), enabled, mode);
+    for (auto& [medium, info] : bluetooth_mediums_) {
+      // Do not send notification to medium that owns this adapter.
+      if (info.adapter == &adapter) continue;
+      NEARBY_LOG(INFO, "[adapter=%p, device=%p] notify: adapter=%p", &adapter,
+                 &adapter_device, info.adapter);
+      OnDeviceStateChanged(info, adapter_device, name, mode, enabled);
+    }
+    // We don't care if there is an adapter already since all we store is a
+    // pointer. Pointer must remain valid for the duration of a Core session
+    // (since it is owned by the correspoinding Medium, and mediums lifetime
+    // matches Core lifetime).
+    bluetooth_adapters_.emplace(&adapter, &adapter_device);
+  });
+}
+
+void MediumEnvironment::OnDeviceStateChanged(
+    BluetoothMediumContext& info, api::BluetoothDevice& device,
+    const std::string& name, api::BluetoothAdapter::ScanMode mode,
+    bool enabled) {
+  auto item = info.devices.find(&device);
+  if (item == info.devices.end()) {
+    NEARBY_LOG(
+        INFO, "G3 OnDeviceStateChanged [device impl=%p]: new device; notify=%d",
+        &device, enable_notifications_.load());
+    if (mode == api::BluetoothAdapter::ScanMode::kConnectableDiscoverable &&
+        enabled) {
+      // New device is turned on, and is in discoverable state.
+      // Store device name, and report it as discovered.
+      info.devices.emplace(&device, name);
+      if (enable_notifications_) {
+        RunOnMediumEnvironmentThread(
+            [&info, &device]() { info.callback.device_discovered_cb(device); });
+      }
+    }
+  } else {
+    NEARBY_LOG(
+        INFO,
+        "G3 OnDeviceStateChanged [device impl=%p]: exisitng device; notify=%d",
+        &device, enable_notifications_.load());
+    auto& discovered_name = item->second;
+    if (mode == api::BluetoothAdapter::ScanMode::kConnectableDiscoverable &&
+        enabled) {
+      if (name != discovered_name) {
+        // Known device is turned on, and is in discoverable state.
+        // Store device name, and report it as renamed.
+        item->second = name;
+        if (enable_notifications_) {
+          RunOnMediumEnvironmentThread([&info, &device]() {
+            info.callback.device_name_changed_cb(device);
+          });
+        }
+      } else {
+        // Device is in discovery mode, so we are reporting it anyway.
+        if (enable_notifications_) {
+          RunOnMediumEnvironmentThread([&info, &device]() {
+            info.callback.device_discovered_cb(device);
+          });
+        }
+      }
+    }
+    if (!enabled) {
+      // Known device is turned off.
+      // Erase it from the map, and report as lost.
+      if (enable_notifications_) {
+        RunOnMediumEnvironmentThread(
+            [&info, &device]() { info.callback.device_lost_cb(device); });
+      }
+      info.devices.erase(item);
+    }
+  }
+}
+
+void MediumEnvironment::RunOnMediumEnvironmentThread(
+    std::function<void()> runnable) {
+  job_count_++;
+  executor_.Execute(std::move(runnable));
+}
+
+void MediumEnvironment::RegisterBluetoothMedium(
+    api::BluetoothClassicMedium& medium,
+    api::BluetoothAdapter& medium_adapter) {
+  RunOnMediumEnvironmentThread([this, &medium, &medium_adapter]() {
+    auto& context = bluetooth_mediums_
+                        .insert({&medium,
+                                 BluetoothMediumContext{
+                                     .adapter = &medium_adapter,
+                                 }})
+                        .first->second;
+    auto* owned_adapter = context.adapter;
+    NEARBY_LOG(INFO, "Registered: medium=%p; adapter=%p", &medium,
+               owned_adapter);
+    for (auto& [adapter, device] : bluetooth_adapters_) {
+      if (adapter == nullptr) continue;
+      OnDeviceStateChanged(context, *device, adapter->GetName(),
+                           adapter->GetScanMode(), adapter->IsEnabled());
+    }
+  });
+}
+
+void MediumEnvironment::UpdateBluetoothMedium(
+    api::BluetoothClassicMedium& medium, BluetoothDiscoveryCallback callback) {
+  RunOnMediumEnvironmentThread([this, &medium,
+                                callback = std::move(callback)]() {
+    auto item = bluetooth_mediums_.find(&medium);
+    if (item == bluetooth_mediums_.end()) return;
+    auto& context = item->second;
+    context.callback = std::move(callback);
+    auto* owned_adapter = context.adapter;
+    NEARBY_LOG(
+        INFO,
+        "Updated: this=%p; medium=%p; adapter=%p; name=%s; enabled=%d; mode=%d",
+        this, &medium, owned_adapter, owned_adapter->GetName().c_str(),
+        owned_adapter->IsEnabled(), owned_adapter->GetScanMode());
+    for (auto& [adapter, device] : bluetooth_adapters_) {
+      if (adapter == nullptr) continue;
+      OnDeviceStateChanged(context, *device, adapter->GetName(),
+                           adapter->GetScanMode(), adapter->IsEnabled());
+    }
+  });
+}
+
+void MediumEnvironment::UnregisterBluetoothMedium(
+    api::BluetoothClassicMedium& medium) {
+  RunOnMediumEnvironmentThread([this, &medium]() {
+    auto item = bluetooth_mediums_.extract(&medium);
+    if (item.empty()) return;
+    auto& context = item.mapped();
+    NEARBY_LOG(INFO, "Unregistered medium for device=%s",
+               context.adapter->GetName().c_str());
+  });
+}
+
+}  // namespace nearby
+}  // namespace location

cpp/platform_v2/base/medium_environment.h

@@ -0,0 +1,113 @@
+#ifndef PLATFORM_V2_BASE_MEDIUM_ENVIRONMENT_H_
+#define PLATFORM_V2_BASE_MEDIUM_ENVIRONMENT_H_
+
+#include <atomic>
+
+#include "platform_v2/api/bluetooth_adapter.h"
+#include "platform_v2/api/bluetooth_classic.h"
+#include "platform_v2/base/listeners.h"
+#include "platform_v2/public/single_thread_executor.h"
+#include "absl/container/flat_hash_map.h"
+
+namespace location {
+namespace nearby {
+
+// MediumEnvironment is a simulated environment which allows multiple instances
+// of simulated HW devices to "work" together as if they are physical.
+// For each medium type it provides necessary methods to implement
+// advertising, discovery and establishment of a data link.
+// NOTE: this code depends on public:types target.
+class MediumEnvironment {
+ public:
+  using BluetoothDiscoveryCallback =
+      api::BluetoothClassicMedium::DiscoveryCallback;
+  MediumEnvironment(const MediumEnvironment&) = delete;
+  MediumEnvironment& operator=(const MediumEnvironment&) = delete;
+
+  // Creates and returns a reference to the global test environment instance.
+  static MediumEnvironment& Instance();
+
+  // Clears state. No notifications are sent.
+  void Reset();
+
+  // Waits for all previously scheduled jobs to finish.
+  // This method works as a barrier that guarantees that after it returns, all
+  // the activities that started before it was called, or while it was running
+  // are ended. This means that system is at the state of relaxation when this
+  // code returns. It requires external stimulus to get out of relaxation state.
+  //
+  // If enable_notifications is true (default), simulation environment
+  // will send all future notification events to all registered objects,
+  // whenever protocol requires that. This is expected behavior.
+  // If enabled_notifications is false, future event notifications will not be
+  // sent to registered instances. This is useful for protocol shutdown,
+  // where we no longer care about notifications, and where notifications may
+  // otherwise be delivered after the notification source or target lifeteme has
+  // ended, and cause undefined behavior.
+  void Sync(bool enable_notifications = true);
+
+  // Adds an adapter to internal container.
+  // Notify BluetoothClassicMediums if any that adapter state has changed.
+  void OnBluetoothAdapterChangedState(api::BluetoothAdapter& adapter,
+                                      api::BluetoothDevice& adapter_device,
+                                      std::string name, bool enabled,
+                                      api::BluetoothAdapter::ScanMode mode);
+
+  // Adds medium-related info to allow for adapter discovery to work.
+  // This provides acccess to this medium from other mediums, when protocol
+  // expects they should communicate.
+  void RegisterBluetoothMedium(api::BluetoothClassicMedium& medium,
+                               api::BluetoothAdapter& medium_adapter);
+
+  // Updates callback info to allow for dispatch of discovery events.
+  //
+  // Invokes callback asynchronously when any changes happen to discoverable
+  // devices, or if the defice is turned off, whether or not it is discoverable,
+  // if it was ever reported as discoverable.
+  //
+  // This should be called when discoverable state changes.
+  // with user-specified callback when discovery is enabled, and with default
+  // (empty) callback otherwise.
+  void UpdateBluetoothMedium(api::BluetoothClassicMedium& medium,
+                             BluetoothDiscoveryCallback callback);
+
+  // Removes medium-related info. This should correspond to device power off.
+  void UnregisterBluetoothMedium(api::BluetoothClassicMedium& medium);
+
+ private:
+  struct BluetoothMediumContext {
+    BluetoothDiscoveryCallback callback;
+    api::BluetoothAdapter* adapter = nullptr;
+    // discovered device vs device name map.
+    absl::flat_hash_map<api::BluetoothDevice*, std::string> devices;
+  };
+
+  // This is a singleton object, for which destructor will never be called.
+  // Constructor will be invoked once from Instance() static method.
+  // Object is create in-place (with a placement new) to guarantee that
+  // destructor is not scheduled for execution at exit.
+  MediumEnvironment() = default;
+  ~MediumEnvironment() = default;
+
+  void OnDeviceStateChanged(BluetoothMediumContext& info,
+                            api::BluetoothDevice& device,
+                            const std::string& name,
+                            api::BluetoothAdapter::ScanMode mode, bool enabled);
+  void RunOnMediumEnvironmentThread(std::function<void()> runnable);
+
+  std::atomic_int job_count_ = 0;
+  std::atomic_bool enable_notifications_ = false;
+  SingleThreadExecutor executor_;
+
+  // The following data members are accessed in the context of a private
+  // executor_ thread.
+  absl::flat_hash_map<api::BluetoothAdapter*, api::BluetoothDevice*>
+      bluetooth_adapters_;
+  absl::flat_hash_map<api::BluetoothClassicMedium*, BluetoothMediumContext>
+      bluetooth_mediums_;
+};
+
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_BASE_MEDIUM_ENVIRONMENT_H_