Roll forward to cl/334770381

Signed-off-by: Alexey Polyudov <apolyudov@google.com>
Change-Id: I22b685b17c37357d6281cedfa7c228e304de8835

cpp/core_v2/internal/bwu_manager.cc

@@ -30,11 +30,11 @@ BwuManager::BwuManager(
   if (config_.bandwidth_upgrade_retry_delay == absl::ZeroDuration()) {
     config_.bandwidth_upgrade_retry_delay = absl::Seconds(5);
   }
-  if (config_.bandwidth_upgrade_retry_delay == absl::ZeroDuration()) {
-    config_.bandwidth_upgrade_retry_delay = absl::Seconds(10);
+  if (config_.bandwidth_upgrade_retry_max_delay == absl::ZeroDuration()) {
+    config_.bandwidth_upgrade_retry_max_delay = absl::Seconds(10);
   }
   if (config_.allow_upgrade_to.All(false)) {
-    config.allow_upgrade_to.web_rtc = true;
+    config_.allow_upgrade_to.web_rtc = true;
   }
   if (!handlers.empty()) {
     handlers_ = std::move(handlers);
@@ -43,7 +43,7 @@ BwuManager::BwuManager(
   }
 
   // Register the offline frame processor.
-  endpoint_manager.RegisterFrameProcessor(
+  endpoint_manager_->RegisterFrameProcessor(
       V1Frame::BANDWIDTH_UPGRADE_NEGOTIATION, this);
 }
 

cpp/core_v2/internal/endpoint_manager.cc

@@ -136,6 +136,9 @@ ExceptionOr<bool> EndpointManager::HandleData(
       // no explicit handler.
       if (frame_type == V1Frame::KEEP_ALIVE) {
         NEARBY_LOG(INFO, "KeepAlive message for: id=%s", endpoint_id.c_str());
+      } else if (frame_type == V1Frame::DISCONNECTION) {
+        NEARBY_LOG(INFO, "Disconnect message for: id=%s", endpoint_id.c_str());
+        endpoint_channel->Close();
       } else {
         NEARBY_LOG(ERROR, "Unhandled message: id=%s, type=%d",
                    endpoint_id.c_str(), frame_type);

cpp/core_v2/internal/offline_frames.cc

@@ -254,6 +254,17 @@ ByteArray ForKeepAlive() {
   return ToBytes(std::move(frame));
 }
 
+ByteArray ForDisconnection() {
+  OfflineFrame frame;
+
+  frame.set_version(OfflineFrame::V1);
+  auto* v1_frame = frame.mutable_v1();
+  v1_frame->set_type(V1Frame::DISCONNECTION);
+  v1_frame->mutable_disconnection();
+
+  return ToBytes(std::move(frame));
+}
+
 UpgradePathInfo::Medium MediumToUpgradePathInfoMedium(Medium medium) {
   switch (medium) {
     case Medium::MDNS:

cpp/core_v2/internal/p2p_point_to_point_pcp_handler.cc

@@ -13,9 +13,18 @@ P2pPointToPointPcpHandler::P2pPointToPointPcpHandler(
 std::vector<proto::connections::Medium>
 P2pPointToPointPcpHandler::GetConnectionMediumsByPriority() {
   std::vector<proto::connections::Medium> mediums;
+  if (mediums_->GetWifiLan().IsAvailable()) {
+    mediums.push_back(proto::connections::WIFI_LAN);
+  }
+  if (mediums_->GetWebRtc().IsAvailable()) {
+    mediums.push_back(proto::connections::WEB_RTC);
+  }
   if (mediums_->GetBluetoothClassic().IsAvailable()) {
     mediums.push_back(proto::connections::BLUETOOTH);
   }
+  if (mediums_->GetBle().IsAvailable()) {
+    mediums.push_back(proto::connections::BLE);
+  }
   return mediums;
 }
 

cpp/core_v2/internal/p2p_star_pcp_handler.cc

@@ -16,9 +16,18 @@ P2pStarPcpHandler::P2pStarPcpHandler(Mediums& mediums,
 std::vector<proto::connections::Medium>
 P2pStarPcpHandler::GetConnectionMediumsByPriority() {
   std::vector<proto::connections::Medium> mediums;
+  if (mediums_->GetWifiLan().IsAvailable()) {
+    mediums.push_back(proto::connections::WIFI_LAN);
+  }
+  if (mediums_->GetWebRtc().IsAvailable()) {
+    mediums.push_back(proto::connections::WEB_RTC);
+  }
   if (mediums_->GetBluetoothClassic().IsAvailable()) {
     mediums.push_back(proto::connections::BLUETOOTH);
   }
+  if (mediums_->GetBle().IsAvailable()) {
+    mediums.push_back(proto::connections::BLE);
+  }
   return mediums;
 }
 

cpp/platform_v2/impl/ios/BUILD

@@ -0,0 +1,56 @@
+objc_library(
+    name = "types",
+    srcs = [
+        "log_message.mm",
+        "scheduled_executor.mm",
+    ],
+    hdrs = [
+        "atomic_boolean.h",
+        "atomic_reference.h",
+        "condition_variable.h",
+        "count_down_latch.h",
+        "log_message.h",
+        "multi_thread_executor.h",
+        "mutex.h",
+        "scheduled_executor.h",
+        "single_thread_executor.h",
+    ],
+    visibility = [
+        "//platform_v2/impl/ios:__pkg__",
+    ],
+    deps = [
+        "//base",
+        "//platform_v2/api:platform",
+        "//platform_v2/api:types",
+        "//platform_v2/base",
+        "//platform_v2/base:util",
+        "//platform_v2/impl/shared:posix_mutex",
+        "//absl/base:core_headers",
+        "//absl/synchronization",
+        "//absl/time",
+        "//thread",
+    ],
+)
+
+objc_library(
+    name = "ios",
+    srcs = [
+        "platform.mm",
+    ],
+    visibility = [
+        "//googlemac/iPhone/Shared/Nearby/Connections:__subpackages__",
+        "//core_v2:__subpackages__",
+        "//platform_v2:__subpackages__",
+    ],
+    deps = [
+        ":types",
+        "//platform_v2/api:comm",
+        "//platform_v2/api:platform",
+        "//platform_v2/api:types",
+        "//platform_v2/impl/shared:file",
+        "//absl/base:core_headers",
+        "//absl/memory",
+        "//absl/strings",
+        "//absl/time",
+    ],
+)

cpp/platform_v2/impl/ios/atomic_boolean.h

@@ -0,0 +1,28 @@
+#ifndef PLATFORM_V2_IMPL_IOS_ATOMIC_BOOLEAN_H_
+#define PLATFORM_V2_IMPL_IOS_ATOMIC_BOOLEAN_H_
+
+#include <atomic>
+
+#include "platform_v2/api/atomic_boolean.h"
+
+namespace location {
+namespace nearby {
+namespace ios {
+
+class AtomicBoolean : public api::AtomicBoolean {
+ public:
+  explicit AtomicBoolean(bool initial_value) : value_(initial_value) {}
+  ~AtomicBoolean() override = default;
+
+  bool Get() const override { return value_.load(); }
+  bool Set(bool value) override { return value_.exchange(value); }
+
+ private:
+  std::atomic_bool value_;
+};
+
+}  // namespace ios
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_IMPL_IOS_ATOMIC_BOOLEAN_H_

cpp/platform_v2/impl/ios/atomic_reference.h

@@ -0,0 +1,33 @@
+#ifndef PLATFORM_V2_IMPL_IOS_ATOMIC_REFERENCE_H_
+#define PLATFORM_V2_IMPL_IOS_ATOMIC_REFERENCE_H_
+
+#include <atomic>
+#include <cstdint>
+
+#include "platform_v2/api/atomic_reference.h"
+
+namespace location {
+namespace nearby {
+namespace ios {
+
+class AtomicUint32 : public api::AtomicUint32 {
+ public:
+  explicit AtomicUint32(std::int32_t value) : value_(value) {}
+  ~AtomicUint32() override = default;
+
+  std::uint32_t Get() const override {
+    return value_;
+  }
+  void Set(std::uint32_t value) override {
+    value_ = value;
+  }
+
+ private:
+  std::atomic<std::uint32_t> value_;
+};
+
+}  // namespace ios
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_IMPL_IOS_ATOMIC_REFERENCE_H_

cpp/platform_v2/impl/ios/condition_variable.h

@@ -0,0 +1,37 @@
+#ifndef PLATFORM_V2_IMPL_IOS_CONDITION_VARIABLE_H_
+#define PLATFORM_V2_IMPL_IOS_CONDITION_VARIABLE_H_
+
+#include "platform_v2/api/condition_variable.h"
+#include "platform_v2/base/exception.h"
+#include "platform_v2/impl/ios/mutex.h"
+#include "absl/synchronization/mutex.h"
+
+namespace location {
+namespace nearby {
+namespace ios {
+
+class ConditionVariable : public api::ConditionVariable {
+ public:
+  explicit ConditionVariable(ios::Mutex* mutex) : mutex_(&mutex->mutex_) {}
+  ~ConditionVariable() override = default;
+
+  Exception Wait() override {
+    cond_var_.Wait(mutex_);
+    return {Exception::kSuccess};
+  }
+  Exception Wait(absl::Duration timeout) override {
+    cond_var_.WaitWithTimeout(mutex_, timeout);
+    return {Exception::kSuccess};
+  }
+  void Notify() override { cond_var_.SignalAll(); }
+
+ private:
+  absl::Mutex* mutex_;
+  absl::CondVar cond_var_;
+};
+
+}  // namespace ios
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_IMPL_IOS_CONDITION_VARIABLE_H_

cpp/platform_v2/impl/ios/count_down_latch.h

@@ -0,0 +1,55 @@
+#ifndef PLATFORM_V2_IMPL_IOS_COUNT_DOWN_LATCH_H_
+#define PLATFORM_V2_IMPL_IOS_COUNT_DOWN_LATCH_H_
+
+#include "platform_v2/api/count_down_latch.h"
+#include "absl/base/thread_annotations.h"
+#include "absl/synchronization/mutex.h"
+#include "absl/time/clock.h"
+
+namespace location {
+namespace nearby {
+namespace ios {
+
+class CountDownLatch final : public api::CountDownLatch {
+ public:
+  explicit CountDownLatch(int count) : count_(count) {}
+  CountDownLatch(const CountDownLatch&) = delete;
+  CountDownLatch& operator=(const CountDownLatch&) = delete;
+  CountDownLatch(CountDownLatch&&) = delete;
+  CountDownLatch& operator=(CountDownLatch&&) = delete;
+  ExceptionOr<bool> Await(absl::Duration timeout) override {
+    absl::MutexLock lock(&mutex_);
+    absl::Time deadline = absl::Now() + timeout;
+    while (count_ > 0) {
+      if (cond_.WaitWithDeadline(&mutex_, deadline)) {
+        return ExceptionOr<bool>(false);
+      }
+    }
+    return ExceptionOr<bool>(true);
+  }
+  Exception Await() override {
+    absl::MutexLock lock(&mutex_);
+    while (count_ > 0) {
+      cond_.Wait(&mutex_);
+    }
+    return {Exception::kSuccess};
+  }
+  void CountDown() override {
+    absl::MutexLock lock(&mutex_);
+    if (count_ > 0 && --count_ == 0) {
+      cond_.SignalAll();
+    }
+  }
+
+ private:
+  absl::Mutex mutex_;   // Mutex to be used with cond_.Wait...() method family.
+  absl::CondVar cond_;  // Condition to synchronize up to N waiting threads.
+  int count_
+      ABSL_GUARDED_BY(mutex_);  // When zero, latch should release all waiters.
+};
+
+}  // namespace ios
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_IMPL_IOS_COUNT_DOWN_LATCH_H_

cpp/platform_v2/impl/ios/log_message.h

@@ -0,0 +1,28 @@
+#ifndef PLATFORM_V2_IMPL_IOS_LOG_MESSAGE_H_
+#define PLATFORM_V2_IMPL_IOS_LOG_MESSAGE_H_
+
+#include "base/logging.h"
+#include "platform_v2/api/log_message.h"
+
+namespace location {
+namespace nearby {
+namespace ios {
+
+class LogMessage : public api::LogMessage {
+ public:
+  LogMessage(const char* file, int line, Severity severity);
+  ~LogMessage() override;
+
+  void Print(const char* format, ...) override;
+
+  std::ostream& Stream() override;
+
+ private:
+  absl::LogStreamer log_streamer_;
+};
+
+}  // namespace ios
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_IMPL_IOS_LOG_MESSAGE_H_

cpp/platform_v2/impl/ios/log_message.mm

@@ -0,0 +1,56 @@
+#include "platform_v2/impl/ios/log_message.h"
+
+#include <algorithm>
+
+#include "base/stringprintf.h"
+
+namespace location {
+namespace nearby {
+namespace ios {
+
+api::LogMessage::Severity kMinLogSeverity = api::LogMessage::Severity::kInfo;
+
+inline absl::LogSeverity ConvertSeverity(api::LogMessage::Severity severity) {
+  switch (severity) {
+    case api::LogMessage::Severity::kInfo:
+      return absl::LogSeverity::kInfo;
+    case api::LogMessage::Severity::kWarning:
+      return absl::LogSeverity::kWarning;
+    case api::LogMessage::Severity::kError:
+      return absl::LogSeverity::kError;
+    case api::LogMessage::Severity::kFatal:
+      return absl::LogSeverity::kFatal;
+  }
+}
+
+LogMessage::LogMessage(const char* file, int line, Severity severity)
+    : log_streamer_(ConvertSeverity(severity), file, line) {}
+
+LogMessage::~LogMessage() = default;
+
+void LogMessage::Print(const char* format, ...) {
+  va_list ap;
+  va_start(ap, format);
+  std::string result;
+  StringAppendV(&result, format, ap);
+  log_streamer_.stream() << result;
+  va_end(ap);
+}
+
+std::ostream& LogMessage::Stream() { return log_streamer_.stream(); }
+
+}  // namespace ios
+
+namespace api {
+
+void LogMessage::SetMinLogSeverity(Severity severity) {
+  ios::kMinLogSeverity = severity;
+}
+
+bool LogMessage::ShouldCreateLogMessage(Severity severity) {
+  return severity >= ios::kMinLogSeverity;
+}
+
+}  // namespace api
+}  // namespace nearby
+}  // namespace location

cpp/platform_v2/impl/ios/multi_thread_executor.h

@@ -0,0 +1,57 @@
+#ifndef PLATFORM_V2_IMPL_IOS_MULTI_THREAD_EXECUTOR_H_
+#define PLATFORM_V2_IMPL_IOS_MULTI_THREAD_EXECUTOR_H_
+
+#include <atomic>
+
+#include "platform_v2/api/submittable_executor.h"
+#include "platform_v2/impl/ios/count_down_latch.h"
+#include "absl/time/clock.h"
+#include "thread/threadpool.h"
+
+namespace location {
+namespace nearby {
+namespace ios {
+
+class MultiThreadExecutor : public api::SubmittableExecutor {
+ public:
+  explicit MultiThreadExecutor(int max_parallelism)
+      : thread_pool_(max_parallelism) {
+    thread_pool_.StartWorkers();
+  }
+  void Execute(Runnable&& runnable) override {
+    if (!shutdown_) {
+      thread_pool_.Schedule(std::move(runnable));
+    }
+  }
+  bool DoSubmit(Runnable&& runnable) override {
+    if (shutdown_) return false;
+    thread_pool_.Schedule(std::move(runnable));
+    return true;
+  }
+  void Shutdown() override { DoShutdown(); }
+  ~MultiThreadExecutor() override { DoShutdown(); }
+
+  int GetTid(int index) const override {
+    const auto* thread = thread_pool_.thread(index);
+    return thread ? *(int*)(thread->tid()) : 0;
+  }
+
+  void ScheduleAfter(absl::Duration delay, Runnable&& runnable) {
+    if (shutdown_) return;
+    thread_pool_.ScheduleAt(absl::Now() + delay, std::move(runnable));
+  }
+  bool InShutdown() const { return shutdown_; }
+
+ private:
+  void DoShutdown() {
+    shutdown_ = true;
+  }
+  std::atomic_bool shutdown_ = false;
+  ThreadPool thread_pool_;
+};
+
+}  // namespace ios
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_IMPL_IOS_MULTI_THREAD_EXECUTOR_H_

cpp/platform_v2/impl/ios/mutex.h

@@ -0,0 +1,47 @@
+#ifndef PLATFORM_V2_IMPL_IOS_MUTEX_H_
+#define PLATFORM_V2_IMPL_IOS_MUTEX_H_
+
+#include "platform_v2/api/mutex.h"
+#include "platform_v2/impl/shared/posix_mutex.h"
+#include "absl/synchronization/mutex.h"
+
+namespace location {
+namespace nearby {
+namespace ios {
+
+class ABSL_LOCKABLE Mutex : public api::Mutex {
+ public:
+  explicit Mutex(bool check) : check_(check) {}
+  ~Mutex() override = default;
+  Mutex(Mutex&&) = delete;
+  Mutex& operator=(Mutex&&) = delete;
+  Mutex(const Mutex&) = delete;
+  Mutex& operator=(const Mutex&) = delete;
+
+  void Lock() ABSL_EXCLUSIVE_LOCK_FUNCTION() override {
+    mutex_.Lock();
+    if (!check_) mutex_.ForgetDeadlockInfo();
+  }
+  void Unlock() ABSL_UNLOCK_FUNCTION() override { mutex_.Unlock(); }
+
+ private:
+  friend class ConditionVariable;
+  absl::Mutex mutex_;
+  bool check_;
+};
+
+class ABSL_LOCKABLE RecursiveMutex : public posix::Mutex {
+ public:
+  ~RecursiveMutex() override = default;
+  RecursiveMutex() = default;
+  RecursiveMutex(RecursiveMutex&&) = delete;
+  RecursiveMutex& operator=(RecursiveMutex&&) = delete;
+  RecursiveMutex(const RecursiveMutex&) = delete;
+  RecursiveMutex& operator=(const RecursiveMutex&) = delete;
+};
+
+}  // namespace ios
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_IMPL_IOS_MUTEX_H_

cpp/platform_v2/impl/ios/platform.mm

@@ -0,0 +1,124 @@
+#include "platform_v2/api/platform.h"
+
+#include <atomic>
+#include <memory>
+
+#include "platform_v2/api/atomic_boolean.h"
+#include "platform_v2/api/atomic_reference.h"
+#include "platform_v2/api/condition_variable.h"
+#include "platform_v2/api/count_down_latch.h"
+#include "platform_v2/api/log_message.h"
+#include "platform_v2/api/mutex.h"
+#include "platform_v2/api/scheduled_executor.h"
+#include "platform_v2/api/submittable_executor.h"
+#include "platform_v2/impl/ios/atomic_boolean.h"
+#include "platform_v2/impl/ios/atomic_reference.h"
+#include "platform_v2/impl/ios/condition_variable.h"
+#include "platform_v2/impl/ios/count_down_latch.h"
+#include "platform_v2/impl/ios/log_message.h"
+#include "platform_v2/impl/ios/multi_thread_executor.h"
+#include "platform_v2/impl/ios/mutex.h"
+#include "platform_v2/impl/ios/scheduled_executor.h"
+#include "platform_v2/impl/ios/single_thread_executor.h"
+#include "platform_v2/impl/shared/file.h"
+#include "absl/memory/memory.h"
+
+namespace location {
+namespace nearby {
+namespace api {
+
+namespace {
+std::string GetPayloadPath(PayloadId payload_id) {
+  return absl::StrCat("/tmp/", payload_id);
+}
+}  // namespace
+
+std::unique_ptr<AtomicBoolean> ImplementationPlatform::CreateAtomicBoolean(bool initial_value) {
+  return absl::make_unique<ios::AtomicBoolean>(initial_value);
+}
+
+std::unique_ptr<AtomicUint32> ImplementationPlatform::CreateAtomicUint32(std::uint32_t value) {
+  return absl::make_unique<ios::AtomicUint32>(value);
+}
+
+std::unique_ptr<CountDownLatch> ImplementationPlatform::CreateCountDownLatch(
+    std::int32_t count) {
+  return absl::make_unique<ios::CountDownLatch>(count);
+}
+
+std::unique_ptr<Mutex> ImplementationPlatform::CreateMutex(Mutex::Mode mode) {
+  if (mode == Mutex::Mode::kRecursive)
+    return absl::make_unique<ios::RecursiveMutex>();
+  else
+    return absl::make_unique<ios::Mutex>(mode == Mutex::Mode::kRegular);
+}
+
+std::unique_ptr<ConditionVariable> ImplementationPlatform::CreateConditionVariable(Mutex* mutex) {
+  return std::unique_ptr<ConditionVariable>(
+      new ios::ConditionVariable(static_cast<ios::Mutex*>(mutex)));
+}
+
+std::unique_ptr<InputFile> ImplementationPlatform::CreateInputFile(PayloadId payload_id,
+                                                                   std::int64_t total_size) {
+  return absl::make_unique<shared::InputFile>(GetPayloadPath(payload_id), total_size);
+}
+
+std::unique_ptr<OutputFile> ImplementationPlatform::CreateOutputFile(PayloadId payload_id) {
+  return absl::make_unique<shared::OutputFile>(GetPayloadPath(payload_id));
+}
+
+std::unique_ptr<LogMessage> ImplementationPlatform::CreateLogMessage(
+    const char* file, int line, LogMessage::Severity severity) {
+  return absl::make_unique<ios::LogMessage>(file, line, severity);
+}
+
+std::unique_ptr<SubmittableExecutor> ImplementationPlatform::CreateSingleThreadExecutor() {
+  return absl::make_unique<ios::SingleThreadExecutor>();
+}
+
+std::unique_ptr<SubmittableExecutor> ImplementationPlatform::CreateMultiThreadExecutor(
+    int max_concurrency) {
+  return absl::make_unique<ios::MultiThreadExecutor>(max_concurrency);
+}
+
+std::unique_ptr<ScheduledExecutor> ImplementationPlatform::CreateScheduledExecutor() {
+  return absl::make_unique<ios::ScheduledExecutor>();
+}
+
+std::unique_ptr<BluetoothAdapter> ImplementationPlatform::CreateBluetoothAdapter() {
+  return std::unique_ptr<BluetoothAdapter>();
+}
+
+std::unique_ptr<BluetoothClassicMedium> ImplementationPlatform::CreateBluetoothClassicMedium(
+    api::BluetoothAdapter& adapter) {
+  return std::unique_ptr<BluetoothClassicMedium>();
+}
+
+std::unique_ptr<BleMedium> ImplementationPlatform::CreateBleMedium(api::BluetoothAdapter& adapter) {
+  return std::unique_ptr<BleMedium>();
+}
+
+std::unique_ptr<ble_v2::BleMedium> ImplementationPlatform::CreateBleV2Medium(
+    api::BluetoothAdapter& adapter) {
+  return std::unique_ptr<ble_v2::BleMedium>();
+}
+
+std::unique_ptr<ServerSyncMedium> ImplementationPlatform::CreateServerSyncMedium() {
+  return std::unique_ptr<ServerSyncMedium>();
+}
+
+std::unique_ptr<WifiMedium> ImplementationPlatform::CreateWifiMedium() {
+  return std::unique_ptr<WifiMedium>();
+}
+
+std::unique_ptr<WifiLanMedium> ImplementationPlatform::CreateWifiLanMedium() {
+   return std::unique_ptr<WifiLanMedium>();
+}
+
+std::unique_ptr<WebRtcMedium> ImplementationPlatform::CreateWebRtcMedium() {
+  return std::unique_ptr<WebRtcMedium>();
+}
+
+}  // namespace api
+}  // namespace nearby
+}  // namespace location

cpp/platform_v2/impl/ios/scheduled_executor.h

@@ -0,0 +1,43 @@
+#ifndef PLATFORM_V2_IMPL_IOS_SCHEDULED_EXECUTOR_H_
+#define PLATFORM_V2_IMPL_IOS_SCHEDULED_EXECUTOR_H_
+
+#include <atomic>
+#include <memory>
+
+#include "platform_v2/api/cancelable.h"
+#include "platform_v2/api/scheduled_executor.h"
+#include "platform_v2/base/runnable.h"
+#include "platform_v2/impl/ios/single_thread_executor.h"
+#include "absl/time/clock.h"
+#include "thread/threadpool.h"
+
+namespace location {
+namespace nearby {
+namespace ios {
+
+class ScheduledExecutor final : public api::ScheduledExecutor {
+ public:
+  ScheduledExecutor() = default;
+  ~ScheduledExecutor() override {
+    executor_.Shutdown();
+  }
+
+  void Execute(Runnable&& runnable) override {
+    executor_.Execute(std::move(runnable));
+  }
+  std::shared_ptr<api::Cancelable> Schedule(Runnable&& runnable,
+                                            absl::Duration delay) override;
+  void Shutdown() override { executor_.Shutdown(); }
+
+  int GetTid(int index) const override {
+    return executor_.GetTid(index);
+  }
+ private:
+  SingleThreadExecutor executor_;
+};
+
+}  // namespace ios
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_IMPL_IOS_SCHEDULED_EXECUTOR_H_

cpp/platform_v2/impl/ios/scheduled_executor.mm

@@ -0,0 +1,65 @@
+#include "platform_v2/impl/ios/scheduled_executor.h"
+
+#include <atomic>
+#include <memory>
+
+#include "platform_v2/api/cancelable.h"
+#include "platform_v2/base/runnable.h"
+#include "absl/time/clock.h"
+
+namespace location {
+namespace nearby {
+namespace ios {
+
+namespace {
+
+class ScheduledCancelable : public api::Cancelable {
+ public:
+  bool Cancel() override {
+    Status expected = kNotRun;
+    while (expected == kNotRun) {
+      if (status_.compare_exchange_strong(expected, kCanceled)) {
+        return true;
+      }
+    }
+    return false;
+  }
+  bool MarkExecuted() {
+    Status expected = kNotRun;
+    while (expected == kNotRun) {
+      if (status_.compare_exchange_strong(expected, kExecuted)) {
+        return true;
+      }
+    }
+    return false;
+  }
+
+ private:
+  enum Status {
+    kNotRun,
+    kExecuted,
+    kCanceled,
+  };
+  std::atomic<Status> status_ = kNotRun;
+};
+
+}  // namespace
+
+std::shared_ptr<api::Cancelable> ScheduledExecutor::Schedule(
+    Runnable&& runnable, absl::Duration delay) {
+  auto scheduled_cancelable = std::make_shared<ScheduledCancelable>();
+  if (executor_.InShutdown()) {
+    return scheduled_cancelable;
+  }
+  executor_.ScheduleAfter(
+      delay, [this, scheduled_cancelable, runnable(std::move(runnable))]() {
+        if (!executor_.InShutdown() && scheduled_cancelable->MarkExecuted()) {
+          runnable();
+        }
+      });
+  return scheduled_cancelable;
+}
+
+}  // namespace ios
+}  // namespace nearby
+}  // namespace location

cpp/platform_v2/impl/ios/single_thread_executor.h

@@ -0,0 +1,20 @@
+#ifndef PLATFORM_V2_IMPL_IOS_SINGLE_THREAD_EXECUTOR_H_
+#define PLATFORM_V2_IMPL_IOS_SINGLE_THREAD_EXECUTOR_H_
+
+#include "platform_v2/impl/ios/multi_thread_executor.h"
+
+namespace location {
+namespace nearby {
+namespace ios {
+
+class SingleThreadExecutor final : public MultiThreadExecutor {
+ public:
+  SingleThreadExecutor() : MultiThreadExecutor(1) {}
+  ~SingleThreadExecutor() override = default;
+};
+
+}  // namespace ios
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_IMPL_IOS_SINGLE_THREAD_EXECUTOR_H_