roll forward to cl/317978613

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

cpp/platform_v2/public/BUILD

@@ -27,11 +27,13 @@ cc_library(
         "crypto.h",
         "file.h",
         "future.h",
+        "logging.h",
         "multi_thread_executor.h",
         "mutex.h",
         "mutex_lock.h",
         "pipe.h",
         "scheduled_executor.h",
+        "settable_future.h",
         "single_thread_executor.h",
         "submittable_executor.h",
         "system_clock.h",
@@ -46,6 +48,7 @@ cc_library(
         "//platform_v2/api:platform",
         "//platform_v2/api:types",
         "//platform_v2/base",
+        "//platform_v2/base:logging",
         "//platform_v2/base:util",
         "//absl/base:core_headers",
         "//absl/container:flat_hash_map",
@@ -58,11 +61,13 @@ cc_library(
     name = "comm",
     srcs = [
         "bluetooth_classic.cc",
+        "wifi_lan.cc",
     ],
     hdrs = [
         "bluetooth_adapter.h",
         "bluetooth_classic.h",
         "webrtc.h",
+        "wifi_lan.h",
     ],
     visibility = [
         "//core_v2:__subpackages__",
@@ -103,6 +108,7 @@ cc_test(
         "atomic_reference_test.cc",
         "bluetooth_adapter_test.cc",
         "bluetooth_classic_test.cc",
+        "condition_variable_test.cc",
         "count_down_latch_test.cc",
         "crypto_test.cc",
         "future_test.cc",
@@ -112,6 +118,7 @@ cc_test(
         "pipe_test.cc",
         "scheduled_executor_test.cc",
         "single_thread_executor_test.cc",
+        "wifi_lan_test.cc",
     ],
     shard_count = 16,
     deps = [

cpp/platform_v2/public/atomic_reference.h

@@ -16,36 +16,71 @@
 #define PLATFORM_V2_PUBLIC_ATOMIC_REFERENCE_H_
 
 #include <memory>
+#include <type_traits>
 
 #include "platform_v2/api/atomic_reference.h"
 #include "platform_v2/api/platform.h"
+#include "platform_v2/public/mutex.h"
+#include "platform_v2/public/mutex_lock.h"
 #include "absl/types/any.h"
 
 namespace location {
 namespace nearby {
 
 // An object reference that may be updated atomically.
-//
-// https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/atomic/AtomicReference.html
+template <typename, typename = void>
+class AtomicReference;
+
+// Platform-based atomic type, for something convertible to std::uint32_t.
 template <typename T>
-class AtomicReference final : public api::AtomicReference<T> {
+class AtomicReference<T, std::enable_if_t<sizeof(T) <= sizeof(std::uint32_t) &&
+                                              std::is_trivially_copyable_v<T>,
+                                          void>>
+    final {
  public:
   using Platform = api::ImplementationPlatform;
-  explicit AtomicReference(const T& value)
-      : impl_(Platform::CreateAtomicReferenceAny(value)) {}
-  explicit AtomicReference(T&& value)
-      : impl_(Platform::CreateAtomicReferenceAny(std::move(value))) {}
-  ~AtomicReference() override = default;
+  explicit AtomicReference(T value)
+      : impl_(Platform::CreateAtomicUint32(static_cast<std::uint32_t>(value))) {
+  }
+  ~AtomicReference() = default;
   AtomicReference(AtomicReference&&) = default;
   AtomicReference& operator=(AtomicReference&&) = default;
 
-  T Get() const& override { return absl::any_cast<T>(impl_->Get()); }
-  T Get() && override { return absl::any_cast<T>(std::move(impl_->Get())); }
-  void Set(const T& value) override { impl_->Set(absl::any(value)); }
-  void Set(T&& value) override { impl_->Set(absl::any(value)); }
+  T Get() const { return static_cast<T>(impl_->Get()); }
+  void Set(T value) { impl_->Set(static_cast<std::uint32_t>(value)); }
 
  private:
-  std::unique_ptr<api::AtomicReference<absl::any>> impl_;
+  std::unique_ptr<api::AtomicUint32> impl_;
+};
+
+// Atomic type that is using Platform mutex to provide atomicity.
+// Supports any copyable type.
+template <typename T>
+class AtomicReference<T, std::enable_if_t<(sizeof(T) > sizeof(std::uint32_t) ||
+                                           !std::is_trivially_copyable_v<T>),
+                                          void>>
+    final {
+ public:
+  explicit AtomicReference(T value) {
+    MutexLock lock(&mutex_);
+    value_ = std::move(value);
+  }
+  void Set(T value) {
+    MutexLock lock(&mutex_);
+    value_ = std::move(value);
+  }
+  T Get() const& {
+    MutexLock lock(&mutex_);
+    return value_;
+  }
+  T&& Get() const&& {
+    MutexLock lock(&mutex_);
+    return std::move(value_);
+  }
+
+ private:
+  mutable Mutex mutex_;
+  T value_;
 };
 
 }  // namespace nearby

cpp/platform_v2/public/bluetooth_classic_test.cc

@@ -33,6 +33,7 @@ class BluetoothClassicMediumTest : public ::testing::Test {
  protected:
   using DiscoveryCallback = BluetoothClassicMedium::DiscoveryCallback;
   BluetoothClassicMediumTest() {
+    env_.Start();
     env_.Reset();
     adapter_a_ = std::make_unique<BluetoothAdapter>();
     adapter_b_ = std::make_unique<BluetoothAdapter>();
@@ -54,6 +55,7 @@ class BluetoothClassicMediumTest : public ::testing::Test {
     adapter_a_.reset();
     adapter_b_.reset();
     env_.Reset();
+    env_.Stop();
   }
 
   MediumEnvironment& env_{MediumEnvironment::Instance()};

cpp/platform_v2/public/condition_variable.h

@@ -35,10 +35,9 @@ class ConditionVariable final {
   ConditionVariable(ConditionVariable&&) = default;
   ConditionVariable& operator=(ConditionVariable&&) = default;
 
-  // https://docs.oracle.com/javase/8/docs/api/java/lang/Object.html#notify--
   void Notify() { impl_->Notify(); }
-  // https://docs.oracle.com/javase/8/docs/api/java/lang/Object.html#wait--
   Exception Wait() { return impl_->Wait(); }
+  Exception Wait(absl::Duration timeout) { return impl_->Wait(timeout); }
 
  private:
   std::unique_ptr<api::ConditionVariable> impl_;

cpp/platform_v2/public/condition_variable_test.cc

@@ -0,0 +1,76 @@
+// 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 "platform_v2/public/condition_variable.h"
+
+#include "platform_v2/public/logging.h"
+#include "platform_v2/public/mutex.h"
+#include "platform_v2/public/single_thread_executor.h"
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+#include "absl/time/time.h"
+
+namespace location {
+namespace nearby {
+namespace {
+
+TEST(ConditionVariableTest, CanCreate) {
+  Mutex mutex;
+  ConditionVariable cond{&mutex};
+}
+
+TEST(ConditionVariableTest, CanWakeupWaiter) {
+  Mutex mutex;
+  ConditionVariable cond{&mutex};
+  bool done = false;
+  bool waiting = false;
+  NEARBY_LOG(INFO, "At start; done=%d", done);
+  {
+    SingleThreadExecutor executor;
+    executor.Execute([&cond, &mutex, &done, &waiting]() {
+      MutexLock lock(&mutex);
+      NEARBY_LOG(INFO, "Before cond.Wait(); done=%d", done);
+      waiting = true;
+      cond.Wait();
+      waiting = false;
+      done = true;
+      NEARBY_LOG(INFO, "After cond.Wait(); done=%d", done);
+    });
+    while (true) {
+      {
+        MutexLock lock(&mutex);
+        if (waiting) break;
+      }
+      SystemClock::Sleep(absl::Milliseconds(100));
+    }
+    {
+      MutexLock lock(&mutex);
+      cond.Notify();
+      EXPECT_FALSE(done);
+    }
+  }
+  NEARBY_LOG(INFO, "After executor shutdown: done=%d", done);
+  EXPECT_TRUE(done);
+}
+
+TEST(ConditionVariableTest, WaitTerminatesOnTimeoutWithoutNotify) {
+  Mutex mutex;
+  ConditionVariable cond{&mutex};
+  MutexLock lock(&mutex);
+  EXPECT_EQ(cond.Wait(absl::Milliseconds(100)), Exception{Exception::kTimeout});
+}
+
+}  // namespace
+}  // namespace nearby
+}  // namespace location

cpp/platform_v2/public/file.h

@@ -24,45 +24,89 @@
 #include "platform_v2/api/platform.h"
 #include "platform_v2/base/byte_array.h"
 #include "platform_v2/base/exception.h"
+#include "platform_v2/base/input_stream.h"
+#include "platform_v2/base/output_stream.h"
 
 namespace location {
 namespace nearby {
 
-class InputFile final : public api::InputFile {
+class InputFile final {
  public:
   using Platform = api::ImplementationPlatform;
-  InputFile(std::int64_t payload_id, std::int64_t size)
-      : impl_(Platform::CreateInputFile(payload_id, size)) {}
-  ~InputFile() override = default;
+  InputFile(PayloadId payload_id, std::int64_t size)
+      : impl_(Platform::CreateInputFile(payload_id, size)), id_(payload_id) {}
+  ~InputFile() = default;
   InputFile(InputFile&&) = default;
   InputFile& operator=(InputFile&&) = default;
 
-  ExceptionOr<ByteArray> Read(std::int64_t size) override {
-    return impl_->Read(size);
-  }
-  std::string GetFilePath() const override { return impl_->GetFilePath(); }
-  std::int64_t GetTotalSize() const override { return impl_->GetTotalSize(); }
-  Exception Close() override { return impl_->Close(); }
+  // Reads up to size bytes and returns as a ByteArray object wrapped by
+  // ExceptionOr.
+  // Returns Exception::kIo on error, or end of file.
+  ExceptionOr<ByteArray> Read(std::int64_t size) { return impl_->Read(size); }
+
+  // Returns a string that uniqely identifies this file.
+  std::string GetFilePath() const { return impl_->GetFilePath(); }
+
+  // Returns total size of this file in bytes.
+  std::int64_t GetTotalSize() const { return impl_->GetTotalSize(); }
+
+  // Disallows further reads from the file and frees system resources,
+  // associated with it.
+  Exception Close() { return impl_->Close(); }
+
+  // Returns a handle to the underlying input stream.
+  //
+  // Returned handle will remain valid even if InputFile is moved, for as long
+  // as original InputFile lifetime continues.
+  // Side effects of any non-const operation invoked for InputFile (such as
+  // Read, or Close will be observable through InputStream& handle, and vice
+  // versa.
+  InputStream& GetInputStream() { return *impl_; }
+
+  // Returns payload id of this file. The closest "file" equivalent is inode.
+  PayloadId GetPayloadId() const { return id_; }
 
  private:
   std::unique_ptr<api::InputFile> impl_;
+  PayloadId id_;
 };
 
-class OutputFile final : public api::OutputFile {
+class OutputFile final {
  public:
   using Platform = api::ImplementationPlatform;
-  explicit OutputFile(std::int64_t payload_id)
-      : impl_(Platform::CreateOutputFile(payload_id)) {}
-  ~OutputFile() override = default;
+  explicit OutputFile(PayloadId payload_id)
+      : impl_(Platform::CreateOutputFile(payload_id)), id_(payload_id) {}
+  ~OutputFile() = default;
   OutputFile(OutputFile&&) = default;
   OutputFile& operator=(OutputFile&&) = default;
 
-  Exception Write(const ByteArray& data) override { return impl_->Write(data); }
-  Exception Flush() override { return impl_->Flush(); }
-  Exception Close() override { return impl_->Close(); }
+  // Writes all data from ByteArray object to the underlying stream.
+  // Returns Exception::kIo on error, Exception::kSuccess otherwise.
+  Exception Write(const ByteArray& data) { return impl_->Write(data); }
+
+  // Ensures that all data written by previous calls to Write() is passed
+  // down to the applicable transport layer.
+  Exception Flush() { return impl_->Flush(); }
+
+  // Disallows further writes to the file and frees system resources,
+  // associated with it.
+  Exception Close() { return impl_->Close(); }
+
+  // Returns a handle to the underlying  output stream.
+  //
+  // Returned handle will remain valid even if OutputFile is moved, for as long
+  // as original OutputFile lifetime continues.
+  // Side effects of any non-const operation invoked for OutputFile (such as
+  // Write, or Close will be observable through OutputStream& handle, and vice
+  // versa.
+  OutputStream& GetOutputStream() { return *impl_; }
+
+  // Returns payload id of this file. The closest "file" equivalent is inode.
+  PayloadId GetPayloadId() const { return id_; }
 
  private:
   std::unique_ptr<api::OutputFile> impl_;
+  PayloadId id_;
 };
 
 }  // namespace nearby

cpp/platform_v2/public/future.h

@@ -15,60 +15,38 @@
 #ifndef PLATFORM_V2_PUBLIC_FUTURE_H_
 #define PLATFORM_V2_PUBLIC_FUTURE_H_
 
-#include "platform_v2/api/executor.h"
-#include "platform_v2/api/platform.h"
-#include "platform_v2/api/settable_future.h"
-#include "platform_v2/base/exception.h"
-#include "platform_v2/base/runnable.h"
-#include "absl/time/time.h"
-#include "absl/types/any.h"
+#include "platform_v2/public/settable_future.h"
 
 namespace location {
 namespace nearby {
 
 template <typename T>
-class Future final : public api::SettableFuture<T> {
+class Future final {
  public:
-  using Platform = api::ImplementationPlatform;
-  ~Future() override = default;
-  Future() : impl_(Platform::CreateSettableFutureAny().release()) {}
-  Future(Future&& other) = default;
-  Future& operator=(Future&& other) = default;
-
-  void AddListener(Runnable runnable, api::Executor* executor) override {
-    impl_->AddListener(runnable, executor);
-  }
-  bool Set(const T& value) override { return impl_->Set(absl::any(value)); }
-  bool Set(T&& value) override { return impl_->Set(absl::any(value)); }
-  bool SetException(Exception exception) override {
+  virtual bool Set(T value) { return impl_->Set(std::move(value)); }
+  virtual bool SetException(Exception exception) {
     return impl_->SetException(exception);
   }
-  // throws Exception::kInterrupted, Exception::kExecution
-  ExceptionOr<T> Get() override {
-    auto ret_val = impl_->Get();
-    if (ret_val.ok()) {
-      T result = absl::any_cast<T>(ret_val.result());
-      return ExceptionOr<T>{result};
-    } else {
-      return ExceptionOr<T>{ret_val.exception()};
-    }
+  virtual ExceptionOr<T> Get() { return impl_->Get(); }
+  virtual ExceptionOr<T> Get(absl::Duration timeout) {
+    return impl_->Get(timeout);
   }
-
-  // throws Exception::kInterrupted, Exception::kExecution
-  // throws Exception::kTimeout if timeout is exceeded while waiting for
-  // result.
-  ExceptionOr<T> Get(absl::Duration timeout) override {
-    auto ret_val = impl_->Get(timeout);
-    if (ret_val.ok()) {
-      T result = absl::any_cast<T>(ret_val.result());
-      return ExceptionOr<T>{result};
-    } else {
-      return ExceptionOr<T>{ret_val.exception()};
-    }
+  void AddListener(Runnable runnable, api::Executor* executor) {
+    impl_->AddListener(std::move(runnable), executor);
   }
 
  private:
-  std::unique_ptr<api::SettableFuture<absl::any>> impl_;
+  // Instance of future implementation is wrapped in shared_ptr<> to make
+  // it possible to pass Future by value and share the implementation.
+  // This allows for the following constructions:
+  // 1)
+  // Future<bool> future;
+  // RunOnXyzThread([future]() { future.Set(DoTheJobAndReport()); });
+  // if (future.Get().Ok()) { /*...*/ }
+  // 2)
+  // Future<bool> future = DoSomeAsyncWork(); // Returns future, but keeps copy.
+  // if (future.Get().Ok()) { /*...*/ }
+  std::shared_ptr<SettableFuture<T>> impl_{new SettableFuture<T>()};
 };
 
 }  // namespace nearby

cpp/platform_v2/public/logging_test.cc

@@ -20,7 +20,33 @@
 namespace {
 
 TEST(LoggingTest, CanLog) {
-  NEARBY_LOG(INFO, "message");
+  NEARBY_LOG_SET_SEVERITY(INFO);
+  int num = 42;
+  NEARBY_LOG(INFO, "The answer to everything: %d", num++);
+  EXPECT_EQ(num, 43);
 }
 
+TEST(LoggingTest, CanLog_LoggingDisabled) {
+  NEARBY_LOG_SET_SEVERITY(ERROR);
+  int num = 42;
+  NEARBY_LOG(INFO, "The answer to everything: %d", num++);
+  // num++ should not be evaluated
+  EXPECT_EQ(num, 42);
 }
+
+TEST(LoggingTest, CanStream) {
+  NEARBY_LOG_SET_SEVERITY(INFO);
+  int num = 42;
+  NEARBY_LOGS(INFO) << "The answer to everything: " << num++;
+  EXPECT_EQ(num, 43);
+}
+
+TEST(LoggingTest, CanStream_LoggingDisabled) {
+  NEARBY_LOG_SET_SEVERITY(ERROR);
+  int num = 42;
+  NEARBY_LOGS(INFO) << "The answer to everything: " << num++;
+  // num++ should not be evaluated
+  EXPECT_EQ(num, 42);
+}
+
+}  // namespace

cpp/platform_v2/public/mutex_test.cc

@@ -41,7 +41,7 @@ class MutexTest : public testing::Test {
 
  protected:
   SingleThreadExecutor executor_;
-  const absl::Duration kTimeToWait = absl::Milliseconds(200);
+  const absl::Duration kTimeToWait = absl::Milliseconds(500);
   std::atomic_int step_ = 0;
   absl::Mutex step_mutex_;
   absl::CondVar step_cond_;

cpp/platform_v2/public/scheduled_executor_test.cc

@@ -26,6 +26,14 @@
 namespace location {
 namespace nearby {
 
+// kShortDelay must be significant enough to guarantee that OS under heavy load
+// should be able to execute the non-blocking test paths within this time.
+absl::Duration kShortDelay = absl::Milliseconds(100);
+
+// kLongDelay must be long enough to make sure that under OS under heavy load
+// will let kShortDelay fire and jobs scheduled before the kLongDelay fires.
+absl::Duration kLongDelay = 10 * kShortDelay;
+
 TEST(ScheduledExecutorTest, ConsructorDestructorWorks) {
   ScheduledExecutor executor;
 }
@@ -42,7 +50,7 @@ TEST(ScheduledExecutorTest, CanExecute) {
   {
     absl::MutexLock lock(&mutex);
     if (!done) {
-      cond.WaitWithTimeout(&mutex, absl::Seconds(1));
+      cond.WaitWithTimeout(&mutex, kLongDelay);
     }
   }
   EXPECT_TRUE(done);
@@ -53,25 +61,25 @@ TEST(ScheduledExecutorTest, CanSchedule) {
   std::atomic_int value = 0;
   absl::Mutex mutex;
   absl::CondVar cond;
-  // schedule job due in 100 ms.
+  // schedule job due in kLongDelay.
   executor.Schedule(
       [&value, &cond]() {
         EXPECT_EQ(value, 1);
         value = 5;
         cond.Signal();
       },
-      absl::Milliseconds(100));
-  // schedule job due in 10 ms; must fire before the first one.
+      kLongDelay);
+  // schedule job due in kShortDelay; must fire before the first one.
   executor.Schedule(
       [&value]() {
         EXPECT_EQ(value, 0);
         value = 1;
       },
-      absl::Milliseconds(10));
+      kShortDelay);
   {
-    // wait for the final job to unblock us.
+    // wait for the final job to unblock us; wait longer than kLongDelay.
     absl::MutexLock lock(&mutex);
-    cond.WaitWithTimeout(&mutex, absl::Milliseconds(1000));
+    cond.WaitWithTimeout(&mutex, 2 * kLongDelay);
   }
   EXPECT_EQ(value, 5);
 }
@@ -80,10 +88,10 @@ TEST(ScheduledExecutorTest, CanCancel) {
   ScheduledExecutor executor;
   std::atomic_int value = 0;
   Cancelable cancelable =
-      executor.Schedule([&value]() { value += 1; }, absl::Milliseconds(10));
+      executor.Schedule([&value]() { value += 1; }, kShortDelay);
   EXPECT_EQ(value, 0);
   EXPECT_TRUE(cancelable.Cancel());
-  absl::SleepFor(absl::Milliseconds(500));
+  absl::SleepFor(kLongDelay);
   EXPECT_EQ(value, 0);
 }
 
@@ -92,17 +100,17 @@ TEST(ScheduledExecutorTest, FailToCancel) {
   absl::CondVar cond;
   ScheduledExecutor executor;
   std::atomic_int value = 0;
-  // Schedule job in 10ms, which will we will attempt to cancel later.
+  // Schedule job in kShortDelay, which will we will attempt to cancel later.
   Cancelable cancelable =
-      executor.Schedule([&value]() { value += 1; }, absl::Milliseconds(10));
-  // schedule another job to test results of the first one, in 50ms from now.
+      executor.Schedule([&value]() { value += 1; }, kShortDelay);
+  // schedule another job to test results of the first one, in kLongDelay.
   executor.Schedule(
       [&cancelable, &cond]() {
         EXPECT_FALSE(cancelable.Cancel());
         // Wake up main thread.
         cond.Signal();
       },
-      absl::Milliseconds(50));
+      kLongDelay);
   {
     absl::MutexLock lock(&mutex);
     cond.Wait(&mutex);

cpp/platform_v2/public/settable_future.h

@@ -0,0 +1,122 @@
+// 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_V2_PUBLIC_SETTABLE_FUTURE_H_
+#define PLATFORM_V2_PUBLIC_SETTABLE_FUTURE_H_
+
+#include <utility>
+
+#include "platform_v2/public/condition_variable.h"
+#include "platform_v2/public/mutex.h"
+#include "platform_v2/public/mutex_lock.h"
+#include "platform_v2/public/system_clock.h"
+
+namespace location {
+namespace nearby {
+
+template <typename T>
+class SettableFuture : public api::SettableFuture<T> {
+ public:
+  SettableFuture() = default;
+  ~SettableFuture() override = default;
+
+  bool Set(T value) override {
+    MutexLock lock(&mutex_);
+    if (!done_) {
+      value_ = std::move(value);
+      done_ = true;
+      exception_ = {Exception::kSuccess};
+      completed_.Notify();
+      InvokeAllLocked();
+    }
+    return true;
+  }
+
+  void AddListener(Runnable runnable, api::Executor* executor) override {
+    MutexLock lock(&mutex_);
+    if (done_) {
+      executor->Execute(std::move(runnable));
+    } else {
+      listeners_.emplace_back(std::make_pair(executor, std::move(runnable)));
+    }
+  }
+
+  bool SetException(Exception exception) override {
+    MutexLock lock(&mutex_);
+    return SetExceptionLocked(exception);
+  }
+
+  ExceptionOr<T> Get() override {
+    MutexLock lock(&mutex_);
+    while (!done_) {
+      completed_.Wait();
+    }
+    return exception_.value != Exception::kSuccess
+               ? ExceptionOr<T>{exception_.value}
+               : ExceptionOr<T>{value_};
+  }
+
+  ExceptionOr<T> Get(absl::Duration timeout) override {
+    MutexLock lock(&mutex_);
+    while (!done_) {
+      absl::Time start_time = SystemClock::ElapsedRealtime();
+      if (completed_.Wait(timeout).Raised(Exception::kTimeout)) {
+        SetExceptionLocked({Exception::kTimeout});
+        break;
+      }
+      absl::Duration spent = SystemClock::ElapsedRealtime() - start_time;
+      if (spent < timeout) {
+        timeout -= spent;
+      } else if (!done_) {
+        SetExceptionLocked({Exception::kTimeout});
+        break;
+      }
+    }
+    return exception_.value != Exception::kSuccess
+               ? ExceptionOr<T>{exception_.value}
+               : ExceptionOr<T>{value_};
+  }
+
+ private:
+  bool SetExceptionLocked(Exception exception) {
+    if (!done_) {
+      exception_ = exception.value != Exception::kSuccess
+                       ? exception
+                       : Exception{Exception::kFailed};
+      done_ = true;
+      completed_.Notify();
+      InvokeAllLocked();
+    }
+    return true;
+  }
+
+  void InvokeAllLocked() {
+    for (auto& item : listeners_) {
+      item.first->Execute(std::move(item.second));
+    }
+    listeners_.clear();
+  }
+
+  Mutex mutex_;
+  ConditionVariable completed_{&mutex_};
+  std::vector<std::pair<api::Executor*, std::function<void()>>> listeners_;
+  bool done_{false};
+  T value_;
+  Exception exception_{Exception::kFailed};
+};
+
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_PUBLIC_SETTABLE_FUTURE_H_

cpp/platform_v2/public/webrtc.h

@@ -24,6 +24,34 @@
 namespace location {
 namespace nearby {
 
+class WebRtcSignalingMessenger final {
+ public:
+  using OnSignalingMessageCallback =
+      api::WebRtcSignalingMessenger::OnSignalingMessageCallback;
+
+  explicit WebRtcSignalingMessenger(
+      std::unique_ptr<api::WebRtcSignalingMessenger> messenger)
+      : impl_(std::move(messenger)) {}
+  ~WebRtcSignalingMessenger() = default;
+  WebRtcSignalingMessenger(WebRtcSignalingMessenger&&) = default;
+  WebRtcSignalingMessenger operator=(WebRtcSignalingMessenger&&) = delete;
+
+  bool SendMessage(absl::string_view peer_id, const ByteArray& message) {
+    return impl_->SendMessage(peer_id, message);
+  }
+
+  bool StartReceivingMessages(OnSignalingMessageCallback listener) {
+    return impl_->StartReceivingMessages(listener);
+  }
+
+  void StopReceivingMessages() { impl_->StopReceivingMessages(); }
+
+  bool IsValid() const { return impl_ != nullptr; }
+
+ private:
+  std::unique_ptr<api::WebRtcSignalingMessenger> impl_;
+};
+
 class WebRtcMedium final {
  public:
   using PeerConnectionCallback = api::WebRtcMedium::PeerConnectionCallback;
@@ -41,9 +69,10 @@ class WebRtcMedium final {
   }
 
   // Returns a signaling messenger for sending WebRTC signaling messages.
-  std::unique_ptr<api::WebRtcSignalingMessenger> GetSignalingMessenger(
+  std::unique_ptr<WebRtcSignalingMessenger> GetSignalingMessenger(
       absl::string_view self_id) {
-    return impl_->GetSignalingMessenger(self_id);
+    return std::make_unique<WebRtcSignalingMessenger>(
+        impl_->GetSignalingMessenger(self_id));
   }
 
   bool IsValid() const { return impl_ != nullptr; }

cpp/platform_v2/public/wifi_lan.cc

@@ -0,0 +1,134 @@
+// 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 "platform_v2/public/wifi_lan.h"
+
+#include "platform_v2/public/logging.h"
+#include "platform_v2/public/mutex_lock.h"
+
+namespace location {
+namespace nearby {
+
+bool WifiLanMedium::StartAdvertising(
+    const std::string& service_id,
+    const std::string& wifi_lan_service_info_name) {
+  return impl_->StartAdvertising(service_id, wifi_lan_service_info_name);
+}
+
+bool WifiLanMedium::StopAdvertising(const std::string& service_id) {
+  return impl_->StopAdvertising(service_id);
+}
+
+bool WifiLanMedium::StartDiscovery(const std::string& service_id,
+                                   DiscoveredServiceCallback callback) {
+  {
+    MutexLock lock(&mutex_);
+    discovered_service_callback_ = std::move(callback);
+    services_.clear();
+  }
+  return impl_->StartDiscovery(
+      service_id,
+      {
+          .service_discovered_cb =
+              [this](api::WifiLanService& service,
+                     const std::string& service_id) {
+                MutexLock lock(&mutex_);
+                auto pair = services_.emplace(
+                    &service, absl::make_unique<ServiceDiscoveryInfo>());
+                auto& context = *pair.first->second;
+                if (!pair.second) {
+                  NEARBY_LOG(INFO, "Adding (again) service=%p, impl=%p",
+                             &context.service, &service);
+                  return;
+                }
+                context.service = WifiLanService(&service);
+                NEARBY_LOG(INFO, "Adding service=%p, impl=%p", &context.service,
+                           &service);
+                discovered_service_callback_.service_discovered_cb(
+                    context.service, service_id);
+              },
+          .service_lost_cb =
+              [this](api::WifiLanService& service,
+                     const std::string& service_id) {
+                MutexLock lock(&mutex_);
+                auto item = services_.extract(&service);
+                auto& context = *item.mapped();
+                NEARBY_LOG(INFO, "Removing service=%p, impl=%p",
+                           &context.service, &service);
+                discovered_service_callback_.service_lost_cb(context.service,
+                                                             service_id);
+              },
+      });
+}
+
+bool WifiLanMedium::StopDiscovery(const std::string& service_id) {
+  {
+    MutexLock lock(&mutex_);
+    discovered_service_callback_ = {};
+    services_.clear();
+    NEARBY_LOG(INFO, "WifiLan Discovery disabled: impl=%p", &GetImpl());
+  }
+  return impl_->StopDiscovery(service_id);
+}
+
+bool WifiLanMedium::StartAcceptingConnections(
+    const std::string& service_id, AcceptedConnectionCallback callback) {
+  {
+    MutexLock lock(&mutex_);
+    accepted_connection_callback_ = std::move(callback);
+  }
+  return impl_->StartAcceptingConnections(
+      service_id,
+      {
+          .accepted_cb =
+              [this](api::WifiLanSocket& socket,
+                     const std::string& service_id) {
+                MutexLock lock(&mutex_);
+                auto pair = sockets_.emplace(
+                    &socket, absl::make_unique<AcceptedConnectionInfo>());
+                auto& context = *pair.first->second;
+                if (!pair.second) {
+                  NEARBY_LOG(INFO, "Adding (again) socket=%p, impl=%p",
+                             &context.socket, &socket);
+                  return;
+                }
+                context.socket = WifiLanSocket(&socket);
+                NEARBY_LOG(INFO, "Adding socket=%p, impl=%p", &context.socket,
+                           &socket);
+                accepted_connection_callback_.accepted_cb(context.socket,
+                                                          service_id);
+              },
+      });
+}
+
+bool WifiLanMedium::StopAcceptingConnections(const std::string& service_id) {
+  {
+    MutexLock lock(&mutex_);
+    accepted_connection_callback_ = {};
+    sockets_.clear();
+    NEARBY_LOG(INFO, "WifiLan accepted connection disabled: impl=%p",
+               &GetImpl());
+  }
+  return impl_->StopDiscovery(service_id);
+}
+
+WifiLanSocket WifiLanMedium::Connect(WifiLanService& service,
+                                     const std::string& service_id) {
+  NEARBY_LOG(INFO, "WifiLanMedium::Connect: service=%p [impl=%p]", &service,
+             &service.GetImpl());
+  return WifiLanSocket(impl_->Connect(service.GetImpl(), service_id));
+}
+
+}  // namespace nearby
+}  // namespace location

cpp/platform_v2/public/wifi_lan.h

@@ -0,0 +1,174 @@
+// 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_V2_PUBLIC_WIFI_LAN_H_
+#define PLATFORM_V2_PUBLIC_WIFI_LAN_H_
+
+#include "platform_v2/api/platform.h"
+#include "platform_v2/api/wifi_lan.h"
+#include "platform_v2/base/byte_array.h"
+#include "platform_v2/base/input_stream.h"
+#include "platform_v2/base/output_stream.h"
+#include "platform_v2/public/mutex.h"
+#include "absl/container/flat_hash_map.h"
+
+namespace location {
+namespace nearby {
+
+// Opaque wrapper over a WifiLan service which contains encoded service name.
+class WifiLanService final {
+ public:
+  WifiLanService() = default;
+  WifiLanService(const WifiLanService&) = default;
+  WifiLanService& operator=(const WifiLanService&) = default;
+  explicit WifiLanService(api::WifiLanService* service) : impl_(service) {}
+  ~WifiLanService() = default;
+
+  std::string GetName() const { return impl_->GetName(); }
+
+  api::WifiLanService& GetImpl() { return *impl_; }
+  bool IsValid() const { return impl_ != nullptr; }
+
+ private:
+  api::WifiLanService* impl_;
+};
+
+class WifiLanSocket final {
+ public:
+  WifiLanSocket() = default;
+  WifiLanSocket(const WifiLanSocket&) = default;
+  WifiLanSocket& operator=(const WifiLanSocket&) = default;
+  explicit WifiLanSocket(api::WifiLanSocket* socket) : impl_(socket) {}
+  explicit WifiLanSocket(std::unique_ptr<api::WifiLanSocket> socket)
+      : impl_(socket.release()) {}
+  ~WifiLanSocket() = default;
+
+  // Returns the InputStream of the WifiLanSocket.
+  // On error, returned stream will report Exception::kIo on any operation.
+  //
+  // The returned object is not owned by the caller, and can be invalidated once
+  // the WifiLanSocket object is destroyed.
+  InputStream& GetInputStream() { return impl_->GetInputStream(); }
+
+  // Returns the OutputStream of the WifiLanSocket.
+  // On error, returned stream will report Exception::kIo on any operation.
+  //
+  // The returned object is not owned by the caller, and can be invalidated once
+  // the WifiLanSocket object is destroyed.
+  OutputStream& GetOutputStream() { return impl_->GetOutputStream(); }
+
+  // Returns Exception::kIo on error, Exception::kSuccess otherwise.
+  Exception Close() { return impl_->Close(); }
+
+  WifiLanService GetRemoteWifiLanService() {
+    return WifiLanService(impl_->GetRemoteWifiLanService());
+  }
+
+  // Returns true if a socket is usable. If this method returns false,
+  // it is not safe to call any other method.
+  // NOTE(socket validity):
+  // Socket created by a default public constructor is not valid, because
+  // it is missing platform implementation.
+  // The only way to obtain a valid socket is through connection, such as
+  // an object returned by WifiLanMedium::Connect
+  // These methods may also return an invalid socket if connection failed for
+  // any reason.
+  bool IsValid() const { return impl_ != nullptr; }
+
+  // Returns reference to platform implementation.
+  // This is used to communicate with platform code, and for debugging purposes.
+  // Returned reference will remain valid for while WifiLanSocket object is
+  // itself valid. Typically WifiLanSocket lifetime matches duration of the
+  // connection, and is controlled by end user, since they hold the instance.
+  api::WifiLanSocket& GetImpl() { return *impl_; }
+
+ private:
+  std::shared_ptr<api::WifiLanSocket> impl_;
+};
+
+// Container of operations that can be performed over the WifiLan medium.
+class WifiLanMedium final {
+ public:
+  using Platform = api::ImplementationPlatform;
+  struct DiscoveredServiceCallback {
+    std::function<void(WifiLanService& wifi_lan_service,
+                       const std::string& service_id)>
+        service_discovered_cb =
+            DefaultCallback<WifiLanService&, const std::string&>();
+    std::function<void(WifiLanService& wifi_lan_service,
+                       const std::string& service_id)>
+        service_lost_cb =
+            DefaultCallback<WifiLanService&, const std::string&>();
+  };
+  struct ServiceDiscoveryInfo {
+    WifiLanService service;
+  };
+
+  struct AcceptedConnectionCallback {
+    std::function<void(WifiLanSocket& socket, const std::string& service_id)>
+        accepted_cb = DefaultCallback<WifiLanSocket&, const std::string&>();
+  };
+  struct AcceptedConnectionInfo {
+    WifiLanSocket socket;
+  };
+
+  WifiLanMedium() : impl_(Platform::CreateWifiLanMedium()) {}
+  ~WifiLanMedium() = default;
+
+  bool StartAdvertising(const std::string& service_id,
+                        const std::string& wifi_lan_service_info_name);
+  bool StopAdvertising(const std::string& service_id);
+
+  // Returns true once the WifiLan discovery has been initiated.
+  bool StartDiscovery(const std::string& service_id,
+                      DiscoveredServiceCallback callback);
+
+  // Returns true once WifiLan discovery for service_id is well and truly
+  // stopped; after this returns, there must be no more invocations of the
+  // DiscoveredServiceCallback passed in to StartDiscovery() for service_id.
+  bool StopDiscovery(const std::string& service_id);
+
+  // Returns true once WifiLan socket connection requests to service_id can be
+  // accepted.
+  bool StartAcceptingConnections(const std::string& service_id,
+                                 AcceptedConnectionCallback callback);
+  bool StopAcceptingConnections(const std::string& service_id);
+
+  // Returns a new WifiLanSocket. On Success, WifiLanSocket::IsValid()
+  // returns true.
+  WifiLanSocket Connect(WifiLanService& service, const std::string& service_id);
+
+  bool IsValid() const { return impl_ != nullptr; }
+
+  api::WifiLanMedium& GetImpl() { return *impl_; }
+
+ private:
+  Mutex mutex_;
+  std::unique_ptr<api::WifiLanMedium> impl_;
+  absl::flat_hash_map<api::WifiLanService*,
+                      std::unique_ptr<ServiceDiscoveryInfo>>
+      services_ ABSL_GUARDED_BY(mutex_);
+  absl::flat_hash_map<api::WifiLanSocket*,
+                      std::unique_ptr<AcceptedConnectionInfo>>
+      sockets_ ABSL_GUARDED_BY(mutex_);
+  DiscoveredServiceCallback discovered_service_callback_
+      ABSL_GUARDED_BY(mutex_);
+  AcceptedConnectionCallback accepted_connection_callback_
+      ABSL_GUARDED_BY(mutex_);
+};
+
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_PUBLIC_WIFI_LAN_H_

cpp/platform_v2/public/wifi_lan_test.cc

@@ -0,0 +1,116 @@
+// 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 "platform_v2/public/wifi_lan.h"
+
+#include <memory>
+
+#include "platform_v2/base/medium_environment.h"
+#include "platform_v2/public/count_down_latch.h"
+#include "platform_v2/public/logging.h"
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+namespace location {
+namespace nearby {
+namespace {
+
+constexpr absl::string_view kServiceID{"com.google.location.nearby.apps.test"};
+
+class WifiLanMediumTest : public ::testing::Test {
+ protected:
+  using DiscoveredServiceCallback = WifiLanMedium::DiscoveredServiceCallback;
+
+  WifiLanMediumTest() { env_.Stop(); }
+
+  MediumEnvironment& env_{MediumEnvironment::Instance()};
+};
+
+TEST_F(WifiLanMediumTest, ConstructorDestructorWorks) {
+  env_.Start();
+  WifiLanMedium medium_a;
+  WifiLanMedium medium_b;
+
+  // Make sure we can create functional mediums.
+  ASSERT_TRUE(medium_a.IsValid());
+  ASSERT_TRUE(medium_b.IsValid());
+
+  // Make sure we can create 2 distinct mediums.
+  EXPECT_NE(&medium_a.GetImpl(), &medium_b.GetImpl());
+  env_.Stop();
+}
+
+TEST_F(WifiLanMediumTest, CanStartDiscoveryAndServiceIndeedDiscovered) {
+  env_.Start();
+  WifiLanMedium medium;
+  CountDownLatch found_latch(1);
+  CountDownLatch lost_latch(1);
+
+  medium.StartDiscovery(std::string(kServiceID),
+                        DiscoveredServiceCallback{
+                            .service_discovered_cb =
+                                [&found_latch](WifiLanService& service,
+                                               const std::string& service_id) {
+                                  NEARBY_LOG(INFO, "Service discovered: %s",
+                                             service.GetName().c_str());
+                                  EXPECT_EQ(kServiceID, service_id);
+                                  found_latch.CountDown();
+                                },
+                            .service_lost_cb =
+                                [&lost_latch](WifiLanService& service,
+                                              const std::string& service_id) {
+                                  NEARBY_LOG(INFO, "Service lost: %s",
+                                             service.GetName().c_str());
+                                  EXPECT_EQ(kServiceID, service_id);
+                                  lost_latch.CountDown();
+                                },
+                        });
+  EXPECT_TRUE(found_latch.Await(absl::Milliseconds(1000)).result());
+  env_.Stop();
+}
+
+TEST_F(WifiLanMediumTest, CanStopDiscovery) {
+  env_.Start();
+  WifiLanMedium medium;
+  CountDownLatch found_latch(1);
+  CountDownLatch lost_latch(1);
+
+  medium.StartDiscovery(std::string(kServiceID),
+                        DiscoveredServiceCallback{
+                            .service_discovered_cb =
+                                [&found_latch](WifiLanService& service,
+                                               const std::string& service_id) {
+                                  NEARBY_LOG(INFO, "Service discovered: %s",
+                                             service.GetName().c_str());
+                                  EXPECT_EQ(kServiceID, service_id);
+                                  found_latch.CountDown();
+                                },
+                            .service_lost_cb =
+                                [&lost_latch](WifiLanService& service,
+                                              const std::string& service_id) {
+                                  NEARBY_LOG(INFO, "Service lost: %s",
+                                             service.GetName().c_str());
+                                  EXPECT_EQ(kServiceID, service_id);
+                                  lost_latch.CountDown();
+                                },
+                        });
+  EXPECT_TRUE(found_latch.Await(absl::Milliseconds(1000)).result());
+  bool stop = medium.StopDiscovery(std::string(kServiceID));
+  EXPECT_TRUE(stop);
+  env_.Stop();
+}
+
+}  // namespace
+}  // namespace nearby
+}  // namespace location