Roll forward to cl/314747126

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

cpp/platform_v2/public/BUILD

@@ -0,0 +1,128 @@
+# 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.
+
+cc_library(
+    name = "types",
+    srcs = [
+        "pipe.cc",
+    ],
+    hdrs = [
+        "atomic_boolean.h",
+        "atomic_reference.h",
+        "cancelable.h",
+        "cancelable_alarm.h",
+        "condition_variable.h",
+        "count_down_latch.h",
+        "crypto.h",
+        "file.h",
+        "future.h",
+        "multi_thread_executor.h",
+        "mutex.h",
+        "mutex_lock.h",
+        "pipe.h",
+        "scheduled_executor.h",
+        "single_thread_executor.h",
+        "submittable_executor.h",
+        "system_clock.h",
+    ],
+    visibility = [
+        "//core_v2:__subpackages__",
+        "//platform_v2/base:__pkg__",
+        "//platform_v2/public:__pkg__",
+    ],
+    deps = [
+        ":logging",
+        "//platform_v2/api:platform",
+        "//platform_v2/api:types",
+        "//platform_v2/base",
+        "//platform_v2/base:util",
+        "//absl/base:core_headers",
+        "//absl/container:flat_hash_map",
+        "//absl/time",
+        "//absl/types:any",
+    ],
+)
+
+cc_library(
+    name = "comm",
+    srcs = [
+        "bluetooth_classic.cc",
+    ],
+    hdrs = [
+        "bluetooth_adapter.h",
+        "bluetooth_classic.h",
+        "webrtc.h",
+    ],
+    visibility = [
+        "//core_v2:__subpackages__",
+        "//platform_v2/public:__pkg__",
+    ],
+    deps = [
+        ":logging",
+        ":types",
+        "//platform_v2/api:comm",
+        "//platform_v2/api:platform",
+        "//platform_v2/base",
+        "//absl/container:flat_hash_map",
+        "//absl/strings",
+        "//webrtc/api:libjingle_peerconnection_api",
+    ],
+)
+
+cc_library(
+    name = "logging",
+    hdrs = [
+        "logging.h",
+    ],
+    visibility = [
+        "//googlemac/iPhone/Shared/Nearby/Connections:__subpackages__",
+        "//core_v2:__subpackages__",
+        "//platform_v2:__subpackages__",
+    ],
+    deps = [
+        "//platform_v2/base:logging",
+    ],
+)
+
+cc_test(
+    name = "public_test",
+    size = "small",
+    srcs = [
+        "atomic_boolean_test.cc",
+        "atomic_reference_test.cc",
+        "bluetooth_adapter_test.cc",
+        "bluetooth_classic_test.cc",
+        "count_down_latch_test.cc",
+        "crypto_test.cc",
+        "future_test.cc",
+        "logging_test.cc",
+        "multi_thread_executor_test.cc",
+        "mutex_test.cc",
+        "pipe_test.cc",
+        "scheduled_executor_test.cc",
+        "single_thread_executor_test.cc",
+    ],
+    shard_count = 16,
+    deps = [
+        ":comm",
+        ":logging",
+        ":types",
+        "//platform_v2/base",
+        "//platform_v2/base:test_util",
+        "//platform_v2/impl/g3",  # build_cleaner: keep
+        "//testing/base/public:gunit_main",
+        "//absl/synchronization",
+        "//absl/time",
+    ],
+)

cpp/platform_v2/public/atomic_boolean.h

@@ -0,0 +1,48 @@
+// 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_ATOMIC_BOOLEAN_H_
+#define PLATFORM_V2_PUBLIC_ATOMIC_BOOLEAN_H_
+
+#include <memory>
+
+#include "platform_v2/api/atomic_boolean.h"
+#include "platform_v2/api/platform.h"
+
+namespace location {
+namespace nearby {
+
+// A boolean value that may be updated atomically.
+// See documentation in
+// cpp/platform_v2/api/atomic_boolean.h
+class AtomicBoolean final  : public api::AtomicBoolean {
+ public:
+  using Platform = api::ImplementationPlatform;
+  explicit AtomicBoolean(bool value = false)
+      : impl_(Platform::CreateAtomicBoolean(value)) {}
+  ~AtomicBoolean() override = default;
+  AtomicBoolean(AtomicBoolean&&) = default;
+  AtomicBoolean& operator=(AtomicBoolean&&) = default;
+
+  bool Get() const override { return impl_->Get(); }
+  bool Set(bool value) override { return impl_->Set(value); }
+
+ private:
+  std::unique_ptr<api::AtomicBoolean> impl_;
+};
+
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_PUBLIC_ATOMIC_BOOLEAN_H_

cpp/platform_v2/public/atomic_boolean_test.cc

@@ -0,0 +1,38 @@
+// 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/atomic_boolean.h"
+
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+namespace location {
+namespace nearby {
+namespace {
+
+TEST(AtomicBooleanTest, SetReturnsPrevoiusValue) {
+  AtomicBoolean value(false);
+  EXPECT_FALSE(value.Set(true));
+  EXPECT_TRUE(value.Set(true));
+}
+
+TEST(AtomicBooleanTest, GetReturnsWhatWasSet) {
+  AtomicBoolean value(false);
+  EXPECT_FALSE(value.Set(true));
+  EXPECT_TRUE(value.Get());
+}
+
+}  // namespace
+}  // namespace nearby
+}  // namespace location

cpp/platform_v2/public/atomic_reference.h

@@ -0,0 +1,54 @@
+// 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_ATOMIC_REFERENCE_H_
+#define PLATFORM_V2_PUBLIC_ATOMIC_REFERENCE_H_
+
+#include <memory>
+
+#include "platform_v2/api/atomic_reference.h"
+#include "platform_v2/api/platform.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 T>
+class AtomicReference final : public api::AtomicReference<T> {
+ 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;
+  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)); }
+
+ private:
+  std::unique_ptr<api::AtomicReference<absl::any>> impl_;
+};
+
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_PUBLIC_ATOMIC_REFERENCE_H_

cpp/platform_v2/public/atomic_reference_test.cc

@@ -0,0 +1,89 @@
+// 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/atomic_reference.h"
+
+#include "gtest/gtest.h"
+
+namespace location {
+namespace nearby {
+namespace {
+
+struct BigSizedStruct {
+  int data[100]{};
+};
+
+enum TestEnum {
+  kValue1 = 1,
+  kValue2 = 2,
+};
+
+enum class ScopedTestEnum {
+  kValue1 = 1,
+  kValue2 = 2,
+};
+
+bool operator==(const BigSizedStruct& a, const BigSizedStruct& b) {
+  return memcmp(a.data, b.data, sizeof(BigSizedStruct::data)) == 0;
+}
+
+bool operator!=(const BigSizedStruct& a, const BigSizedStruct& b) {
+  return !(a == b);
+}
+
+}  // namespace
+
+TEST(AtomicReferenceTest, SupportIntegralTypes) {
+  AtomicReference<int> atomic_ref({});
+  atomic_ref.Set(5);
+  EXPECT_EQ(atomic_ref.Get(), 5);
+}
+
+TEST(AtomicReferenceTest, SupportEnum) {
+  AtomicReference<TestEnum> atomic_ref({});
+  atomic_ref.Set(TestEnum::kValue1);
+  EXPECT_EQ(atomic_ref.Get(), TestEnum::kValue1);
+}
+
+TEST(AtomicReferenceTest, SupportScopedEnum) {
+  AtomicReference<ScopedTestEnum> atomic_ref({});
+  atomic_ref.Set(ScopedTestEnum::kValue1);
+  EXPECT_EQ(atomic_ref.Get(), ScopedTestEnum::kValue1);
+}
+
+TEST(AtomicReferenceTest, SetTakesCopyOfValue) {
+  // Default constructor is zero-initalizing all data in BigSizedStruct.
+  BigSizedStruct v1;
+  AtomicReference<BigSizedStruct> atomic_ref({});
+  v1.data[0] = 5;  // Changing value before calling set() will affect stored
+  v1.data[7] = 3;  // value.
+  atomic_ref.Set(v1);
+  v1.data[1] = 6;  // Changing value after calling set() will not affect stored
+  v1.data[5] = 4;  // value.
+  BigSizedStruct v2 = atomic_ref.Get();
+  EXPECT_NE(v1, v2);
+  v1.data[1] = 0;
+  v1.data[5] = 0;
+  EXPECT_EQ(v2, v1);
+}
+
+TEST(AtomicReferenceTest, SupportObjects) {
+  std::string s{"test"};
+  AtomicReference<std::string> atomic_ref({});
+  atomic_ref.Set(s);
+  EXPECT_EQ(s, atomic_ref.Get());
+}
+
+}  // namespace nearby
+}  // namespace location

cpp/platform_v2/public/bluetooth_adapter.h

@@ -0,0 +1,103 @@
+// 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_BLUETOOTH_ADAPTER_H_
+#define PLATFORM_V2_PUBLIC_BLUETOOTH_ADAPTER_H_
+
+#include <string>
+
+#include "platform_v2/api/bluetooth_adapter.h"
+#include "platform_v2/api/bluetooth_classic.h"
+#include "platform_v2/api/platform.h"
+#include "absl/strings/string_view.h"
+
+namespace location {
+namespace nearby {
+
+// https://developer.android.com/reference/android/bluetooth/BluetoothDevice.html.
+class BluetoothDevice final {
+ public:
+  BluetoothDevice() = default;
+  BluetoothDevice(const BluetoothDevice&) = default;
+  BluetoothDevice& operator=(const BluetoothDevice&) = default;
+  explicit BluetoothDevice(api::BluetoothDevice* device) : impl_(device) {}
+  ~BluetoothDevice() = default;
+
+  // https://developer.android.com/reference/android/bluetooth/BluetoothDevice.html#getName()
+  std::string GetName() const { return impl_->GetName(); }
+
+  api::BluetoothDevice& GetImpl() { return *impl_; }
+  bool IsValid() const { return impl_ != nullptr; }
+
+ private:
+  api::BluetoothDevice* impl_;
+};
+
+// https://developer.android.com/reference/android/bluetooth/BluetoothAdapter.html
+class BluetoothAdapter final {
+ public:
+  using Status = api::BluetoothAdapter::Status;
+  using ScanMode = api::BluetoothAdapter::ScanMode;
+
+  BluetoothAdapter()
+      : impl_(api::ImplementationPlatform::CreateBluetoothAdapter()) {}
+  ~BluetoothAdapter() = default;
+  BluetoothAdapter(BluetoothAdapter&&) = default;
+  BluetoothAdapter& operator=(BluetoothAdapter&&) = default;
+
+  // Synchronously sets the status of the BluetoothAdapter to 'status', and
+  // returns true if the operation was a success.
+  bool SetStatus(Status status) { return impl_->SetStatus(status); }
+  Status GetStatus() const {
+    return IsEnabled() ? Status::kEnabled : Status::kDisabled;
+  }
+
+  // Returns true if the BluetoothAdapter's current status is
+  // Status::Value::kEnabled.
+  bool IsEnabled() const { return impl_->IsEnabled(); }
+
+  // https://developer.android.com/reference/android/bluetooth/BluetoothAdapter.html#getScanMode()
+  //
+  // Returns ScanMode::kUnknown on error.
+  ScanMode GetScanMode() const { return impl_->GetScanMode(); }
+
+  // Synchronously sets the scan mode of the adapter, and returns true if the
+  // operation was a success.
+  bool SetScanMode(ScanMode scan_mode) {
+    return impl_->SetScanMode(scan_mode);
+  }
+
+  // https://developer.android.com/reference/android/bluetooth/BluetoothAdapter.html#getName()
+  // Returns an empty string on error
+  std::string GetName() const { return impl_->GetName(); }
+
+  // https://developer.android.com/reference/android/bluetooth/BluetoothAdapter.html#setName(java.lang.String)
+  bool SetName(absl::string_view name) { return impl_->SetName(name); }
+
+  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 BluetoothAdapter object is
+  // itself valid. It matches Core() object lifetime.
+  api::BluetoothAdapter& GetImpl() { return *impl_; }
+
+ private:
+  std::unique_ptr<api::BluetoothAdapter> impl_;
+};
+
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_PUBLIC_BLUETOOTH_ADAPTER_H_

cpp/platform_v2/public/bluetooth_adapter_test.cc

@@ -0,0 +1,58 @@
+// 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/bluetooth_adapter.h"
+
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+namespace location {
+namespace nearby {
+namespace {
+
+TEST(BluetoothAdapterTest, ConstructorDestructorWorks) {
+  BluetoothAdapter adapter;
+  EXPECT_TRUE(adapter.IsValid());
+}
+
+TEST(BluetoothAdapterTest, CanSetName) {
+  constexpr char kAdapterName[] = "MyBtAdapter";
+  BluetoothAdapter adapter;
+  EXPECT_EQ(adapter.GetStatus(), BluetoothAdapter::Status::kDisabled);
+  EXPECT_TRUE(adapter.SetName(kAdapterName));
+  EXPECT_EQ(adapter.GetName(), std::string(kAdapterName));
+}
+
+TEST(BluetoothAdapterTest, CanSetStatus) {
+  BluetoothAdapter adapter;
+  EXPECT_EQ(adapter.GetStatus(), BluetoothAdapter::Status::kDisabled);
+  EXPECT_TRUE(adapter.SetStatus(BluetoothAdapter::Status::kEnabled));
+  EXPECT_EQ(adapter.GetStatus(), BluetoothAdapter::Status::kEnabled);
+}
+
+TEST(BluetoothAdapterTest, CanSetMode) {
+  BluetoothAdapter adapter;
+  EXPECT_TRUE(adapter.SetScanMode(BluetoothAdapter::ScanMode::kConnectable));
+  EXPECT_EQ(adapter.GetScanMode(), BluetoothAdapter::ScanMode::kConnectable);
+  EXPECT_TRUE(adapter.SetScanMode(
+      BluetoothAdapter::ScanMode::kConnectableDiscoverable));
+  EXPECT_EQ(adapter.GetScanMode(),
+            BluetoothAdapter::ScanMode::kConnectableDiscoverable);
+  EXPECT_TRUE(adapter.SetScanMode(BluetoothAdapter::ScanMode::kNone));
+  EXPECT_EQ(adapter.GetScanMode(), BluetoothAdapter::ScanMode::kNone);
+}
+
+}  // namespace
+}  // namespace nearby
+}  // namespace location

cpp/platform_v2/public/bluetooth_classic.cc

@@ -0,0 +1,99 @@
+// 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/bluetooth_classic.h"
+
+#include "platform_v2/public/logging.h"
+#include "platform_v2/public/mutex_lock.h"
+
+namespace location {
+namespace nearby {
+
+BluetoothClassicMedium::~BluetoothClassicMedium() { StopDiscovery(); }
+
+BluetoothSocket BluetoothClassicMedium::ConnectToService(
+    BluetoothDevice& remote_device, const std::string& service_uuid) {
+  NEARBY_LOG(INFO,
+             "BluetoothClassicMedium::ConnectToService: device=%p [impl=%p]",
+             &remote_device, &remote_device.GetImpl());
+  return BluetoothSocket(
+      impl_->ConnectToService(remote_device.GetImpl(), service_uuid));
+}
+
+bool BluetoothClassicMedium::StartDiscovery(DiscoveryCallback callback) {
+  {
+    MutexLock lock(&mutex_);
+    if (discovery_enabled_) {
+      NEARBY_LOG(INFO, "BT Discovery already enabled; impl=%p", &GetImpl());
+      return false;
+    }
+    discovery_callback_ = std::move(callback);
+    devices_.clear();
+    discovery_enabled_ = true;
+    NEARBY_LOG(INFO, "BT Discovery enabled; impl=%p", &GetImpl());
+  }
+  return impl_->StartDiscovery({
+      .device_discovered_cb =
+          [this](api::BluetoothDevice& device) {
+            MutexLock lock(&mutex_);
+            auto pair = devices_.emplace(
+                &device, absl::make_unique<DeviceDiscoveryInfo>());
+            auto& context = *pair.first->second;
+            if (!pair.second) {
+              NEARBY_LOG(INFO, "Adding (again) device=%p, impl=%p",
+                         &context.device, &device);
+              return;
+            }
+            context.device = BluetoothDevice(&device);
+            NEARBY_LOG(INFO, "Adding device=%p, impl=%p", &context.device,
+                       &device);
+            if (!discovery_enabled_) return;
+            discovery_callback_.device_discovered_cb(context.device);
+          },
+      .device_name_changed_cb =
+          [this](api::BluetoothDevice& device) {
+            MutexLock lock(&mutex_);
+            auto& context = *devices_[&device];
+            NEARBY_LOG(INFO, "Renaming device=%p, impl=%p", &context.device,
+                       &device);
+            if (!discovery_enabled_) return;
+            discovery_callback_.device_name_changed_cb(context.device);
+          },
+      .device_lost_cb =
+          [this](api::BluetoothDevice& device) {
+            MutexLock lock(&mutex_);
+            auto item = devices_.extract(&device);
+            auto& context = *item.mapped();
+            NEARBY_LOG(INFO, "Removing device=%p, impl=%p", &context.device,
+                       &device);
+            if (!discovery_enabled_) return;
+            discovery_callback_.device_lost_cb(context.device);
+          },
+  });
+}
+
+bool BluetoothClassicMedium::StopDiscovery() {
+  {
+    MutexLock lock(&mutex_);
+    if (!discovery_enabled_) return true;
+    discovery_enabled_ = false;
+    discovery_callback_ = {};
+    devices_.clear();
+    NEARBY_LOG(INFO, "BT Discovery disabled: impl=%p", &GetImpl());
+  }
+  return impl_->StopDiscovery();
+}
+
+}  // namespace nearby
+}  // namespace location

cpp/platform_v2/public/bluetooth_classic.h

@@ -0,0 +1,219 @@
+// 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_BLUETOOTH_CLASSIC_H_
+#define PLATFORM_V2_PUBLIC_BLUETOOTH_CLASSIC_H_
+
+#include <memory>
+#include <string>
+
+#include "platform_v2/api/bluetooth_classic.h"
+#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/listeners.h"
+#include "platform_v2/base/output_stream.h"
+#include "platform_v2/public/bluetooth_adapter.h"
+#include "platform_v2/public/mutex.h"
+#include "absl/container/flat_hash_map.h"
+
+namespace location {
+namespace nearby {
+
+// https://developer.android.com/reference/android/bluetooth/BluetoothSocket.html.
+class BluetoothSocket final {
+ public:
+  BluetoothSocket() = default;
+  BluetoothSocket(const BluetoothSocket&) = default;
+  BluetoothSocket& operator=(const BluetoothSocket&) = default;
+  explicit BluetoothSocket(std::unique_ptr<api::BluetoothSocket> socket)
+      : impl_(socket.release()) {}
+  ~BluetoothSocket() = default;
+
+  // Returns the InputStream of this connected BluetoothSocket.
+  InputStream& GetInputStream() { return impl_->GetInputStream(); }
+
+  // Returns the OutputStream of this connected BluetoothSocket.
+  OutputStream& GetOutputStream() { return impl_->GetOutputStream(); }
+
+  // Closes both input and output streams, marks Socket as closed.
+  // After this call object should be treated as not connected.
+  // Returns Exception::kIo on error, Exception::kSuccess otherwise.
+  Exception Close() { return impl_->Close(); }
+
+  // https://developer.android.com/reference/android/bluetooth/BluetoothSocket.html#getRemoteDevice()
+  BluetoothDevice GetRemoteDevice() {
+    return BluetoothDevice(impl_->GetRemoteDevice());
+  }
+
+  // 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 either BluetoothClassicMedium::ConnectTotService or
+  // BluetoothServerSocket::Accept().
+  // 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 BluetoothSocket object is
+  // itself valid. Typically BluetoothSocket lifetime matches duration of the
+  // connection, and is controlled by end user, since they hold the instance.
+  api::BluetoothSocket& GetImpl() { return *impl_; }
+
+ private:
+  std::shared_ptr<api::BluetoothSocket> impl_;
+};
+
+// https://developer.android.com/reference/android/bluetooth/BluetoothServerSocket.html.
+class BluetoothServerSocket final {
+ public:
+  BluetoothServerSocket() = default;
+  BluetoothServerSocket(const BluetoothServerSocket&) = default;
+  BluetoothServerSocket& operator=(const BluetoothServerSocket&) = default;
+  ~BluetoothServerSocket() = default;
+  explicit BluetoothServerSocket(
+      std::unique_ptr<api::BluetoothServerSocket> socket)
+      : impl_(std::move(socket)) {}
+
+  // https://developer.android.com/reference/android/bluetooth/BluetoothServerSocket.html#accept()
+  //
+  // 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.
+  BluetoothSocket Accept() { return BluetoothSocket(impl_->Accept()); }
+
+  // https://developer.android.com/reference/android/bluetooth/BluetoothServerSocket.html#close()
+  //
+  // Returns Exception::kIo on error, Exception::kSuccess otherwise.
+  Exception Close() { return impl_->Close(); }
+
+  bool IsValid() const { return impl_ != nullptr; }
+  api::BluetoothServerSocket& GetImpl() { return *impl_; }
+
+ private:
+  std::shared_ptr<api::BluetoothServerSocket> impl_;
+};
+
+// Container of operations that can be performed over the Bluetooth Classic
+// medium.
+class BluetoothClassicMedium final {
+ public:
+  using Platform = api::ImplementationPlatform;
+  struct DiscoveryCallback {
+    // BluetoothDevice is a proxy object created as a result of BT discovery.
+    // Its lifetime spans between calls to device_discovered_cb and
+    // device_lost_cb.
+    // It is safe to use BluetoothDevice in device_discovered_cb() callback
+    // and at any time afterwards, until device_lost_cb() is called.
+    // It is not safe to use BluetoothDevice after returning from
+    // device_lost_cb() callback.
+    std::function<void(BluetoothDevice& device)> device_discovered_cb =
+        DefaultCallback<BluetoothDevice&>();
+    std::function<void(BluetoothDevice& device)> device_name_changed_cb =
+        DefaultCallback<BluetoothDevice&>();
+    std::function<void(BluetoothDevice& device)> device_lost_cb =
+        DefaultCallback<BluetoothDevice&>();
+  };
+  struct DeviceDiscoveryInfo {
+    BluetoothDevice device;
+  };
+
+  explicit BluetoothClassicMedium(BluetoothAdapter& adapter)
+      : impl_(Platform::CreateBluetoothClassicMedium(adapter.GetImpl())),
+        adapter_(adapter) {}
+
+  ~BluetoothClassicMedium();
+
+  // NOTE(DiscoveryCallback):
+  // BluetoothDevice is a proxy object created as a result of BT discovery.
+  // Its lifetime spans between calls to device_discovered_cb and
+  // device_lost_cb.
+  // It is safe to use BluetoothDevice in device_discovered_cb() callback
+  // and at any time afterwards, until device_lost_cb() is called.
+  // It is not safe to use BluetoothDevice after returning from
+  // device_lost_cb() callback.
+
+  // https://developer.android.com/reference/android/bluetooth/BluetoothAdapter.html#startDiscovery()
+  //
+  // Returns true once the process of discovery has been initiated.
+  bool StartDiscovery(DiscoveryCallback callback);
+
+  // https://developer.android.com/reference/android/bluetooth/BluetoothAdapter.html#cancelDiscovery()
+  //
+  // Returns true once discovery is well and truly stopped; after this returns,
+  // there must be no more invocations of the DiscoveryCallback passed in to
+  // StartDiscovery().
+  bool StopDiscovery();
+
+  // A combination of
+  // https://developer.android.com/reference/android/bluetooth/BluetoothDevice.html#createInsecureRfcommSocketToServiceRecord
+  // followed by
+  // https://developer.android.com/reference/android/bluetooth/BluetoothSocket.html#connect().
+  //
+  // service_uuid is the canonical textual representation
+  // (https://en.wikipedia.org/wiki/Universally_unique_identifier#Format) of a
+  // type 3 name-based
+  // (https://en.wikipedia.org/wiki/Universally_unique_identifier#Versions_3_and_5_(namespace_name-based))
+  // UUID.
+  //
+  // Returns a new BluetoothSocket. On Success, BluetoothSocket::IsValid()
+  // returns true.
+  BluetoothSocket ConnectToService(BluetoothDevice& remote_device,
+                                   const std::string& service_uuid);
+
+  // https://developer.android.com/reference/android/bluetooth/BluetoothAdapter.html#listenUsingInsecureRfcommWithServiceRecord
+  //
+  // service_uuid is the canonical textual representation
+  // (https://en.wikipedia.org/wiki/Universally_unique_identifier#Format) of a
+  // type 3 name-based
+  // (https://en.wikipedia.org/wiki/Universally_unique_identifier#Versions_3_and_5_(namespace_name-based))
+  // UUID.
+  //
+  // Returns a new BluetoothServerSocket.
+  // On Success, BluetoothServerSocket::IsValid() returns true.
+  BluetoothServerSocket ListenForService(const std::string& service_name,
+                                         const std::string& service_uuid) {
+    return BluetoothServerSocket(
+        impl_->ListenForService(service_name, service_uuid));
+  }
+
+  bool IsValid() const { return impl_ != nullptr; }
+
+  api::BluetoothClassicMedium& GetImpl() { return *impl_; }
+  BluetoothAdapter& GetAdapter() { return adapter_; }
+
+ private:
+  Mutex mutex_;
+  std::unique_ptr<api::BluetoothClassicMedium> impl_;
+  BluetoothAdapter& adapter_;
+  absl::flat_hash_map<api::BluetoothDevice*,
+                      std::unique_ptr<DeviceDiscoveryInfo>>
+      devices_ ABSL_GUARDED_BY(mutex_);
+  DiscoveryCallback discovery_callback_ ABSL_GUARDED_BY(mutex_);
+  bool discovery_enabled_ ABSL_GUARDED_BY(mutex_) = false;
+};
+
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_PUBLIC_BLUETOOTH_CLASSIC_H_

cpp/platform_v2/public/bluetooth_classic_test.cc

@@ -0,0 +1,211 @@
+// 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/bluetooth_classic.h"
+
+#include <memory>
+
+#include "platform_v2/base/medium_environment.h"
+#include "platform_v2/public/bluetooth_adapter.h"
+#include "platform_v2/public/count_down_latch.h"
+#include "platform_v2/public/logging.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 {
+
+class BluetoothClassicMediumTest : public ::testing::Test {
+ protected:
+  using DiscoveryCallback = BluetoothClassicMedium::DiscoveryCallback;
+  BluetoothClassicMediumTest() {
+    env_.Reset();
+    adapter_a_ = std::make_unique<BluetoothAdapter>();
+    adapter_b_ = std::make_unique<BluetoothAdapter>();
+    bt_a_ = std::make_unique<BluetoothClassicMedium>(*adapter_a_);
+    bt_b_ = std::make_unique<BluetoothClassicMedium>(*adapter_b_);
+    adapter_a_->SetName("Device-A");
+    adapter_b_->SetName("Device-B");
+    adapter_a_->SetStatus(BluetoothAdapter::Status::kEnabled);
+    adapter_b_->SetStatus(BluetoothAdapter::Status::kEnabled);
+    env_.Sync();
+  }
+  ~BluetoothClassicMediumTest() override {
+    env_.Sync(false);
+    adapter_a_->SetStatus(BluetoothAdapter::Status::kDisabled);
+    adapter_b_->SetStatus(BluetoothAdapter::Status::kDisabled);
+    bt_a_.reset();
+    bt_b_.reset();
+    env_.Sync(false);
+    adapter_a_.reset();
+    adapter_b_.reset();
+    env_.Reset();
+  }
+
+  MediumEnvironment& env_{MediumEnvironment::Instance()};
+
+  std::unique_ptr<BluetoothAdapter> adapter_a_;
+  std::unique_ptr<BluetoothAdapter> adapter_b_;
+  std::unique_ptr<BluetoothClassicMedium> bt_a_;
+  std::unique_ptr<BluetoothClassicMedium> bt_b_;
+};
+
+TEST_F(BluetoothClassicMediumTest, ConstructorDestructorWorks) {
+  // Make sure we can create functional adapters.
+  ASSERT_TRUE(adapter_a_->IsValid());
+  ASSERT_TRUE(adapter_b_->IsValid());
+
+  // Make sure we can create 2 distinct adapters.
+  // NOTE: multiple adapters are supported on a test platform, but not
+  // necessarily on every available HW platform.
+  // Often, HW platform supports only one BT adapter.
+  EXPECT_NE(&adapter_a_->GetImpl(), &adapter_b_->GetImpl());
+
+  // Make sure we can create functional mediums.
+  ASSERT_TRUE(bt_a_->IsValid());
+  ASSERT_TRUE(bt_b_->IsValid());
+
+  // Make sure we can create 2 distinct mediums.
+  EXPECT_NE(&bt_a_->GetImpl(), &bt_b_->GetImpl());
+}
+
+TEST_F(BluetoothClassicMediumTest, CanStartDiscovery) {
+  adapter_a_->SetScanMode(BluetoothAdapter::ScanMode::kConnectable);
+  CountDownLatch found_latch(1);
+  CountDownLatch lost_latch(1);
+  bt_a_->StartDiscovery(DiscoveryCallback{
+      .device_discovered_cb =
+          [this, &found_latch](BluetoothDevice& device) {
+            NEARBY_LOG(INFO, "Device discovered: %s", device.GetName().c_str());
+            EXPECT_EQ(device.GetName(), adapter_b_->GetName());
+            found_latch.CountDown();
+          },
+      .device_lost_cb =
+          [this, &lost_latch](BluetoothDevice& device) {
+            NEARBY_LOG(INFO, "Device lost: %s", device.GetName().c_str());
+            EXPECT_EQ(device.GetName(), adapter_b_->GetName());
+            lost_latch.CountDown();
+          },
+  });
+  adapter_b_->SetScanMode(BluetoothAdapter::ScanMode::kConnectableDiscoverable);
+  EXPECT_EQ(adapter_b_->GetScanMode(),
+            BluetoothAdapter::ScanMode::kConnectableDiscoverable);
+  EXPECT_TRUE(found_latch.Await(absl::Milliseconds(1000)).result());
+  adapter_b_->SetStatus(BluetoothAdapter::Status::kDisabled);
+  EXPECT_FALSE(adapter_b_->IsEnabled());
+  EXPECT_TRUE(lost_latch.Await(absl::Milliseconds(1000)).result());
+}
+
+TEST_F(BluetoothClassicMediumTest, CanStopDiscovery) {
+  adapter_a_->SetScanMode(BluetoothAdapter::ScanMode::kConnectable);
+  CountDownLatch found_latch(1);
+  CountDownLatch lost_latch(1);
+  bt_a_->StartDiscovery(DiscoveryCallback{
+      .device_discovered_cb =
+          [this, &found_latch](BluetoothDevice& device) {
+            NEARBY_LOG(INFO, "Device discovered: %s", device.GetName().c_str());
+            EXPECT_EQ(device.GetName(), adapter_b_->GetName());
+            found_latch.CountDown();
+          },
+      .device_lost_cb =
+          [this, &lost_latch](BluetoothDevice& device) {
+            NEARBY_LOG(INFO, "Device lost: %s", device.GetName().c_str());
+            EXPECT_EQ(device.GetName(), adapter_b_->GetName());
+            lost_latch.CountDown();
+          },
+  });
+  adapter_b_->SetScanMode(BluetoothAdapter::ScanMode::kConnectableDiscoverable);
+  EXPECT_EQ(adapter_b_->GetScanMode(),
+            BluetoothAdapter::ScanMode::kConnectableDiscoverable);
+  EXPECT_TRUE(found_latch.Await(absl::Milliseconds(1000)).result());
+  bt_a_->StopDiscovery();
+  adapter_b_->SetStatus(BluetoothAdapter::Status::kDisabled);
+  EXPECT_FALSE(adapter_b_->IsEnabled());
+  EXPECT_FALSE(lost_latch.Await(absl::Milliseconds(1000)).result());
+}
+
+TEST_F(BluetoothClassicMediumTest, CanListenForService) {
+  adapter_a_->SetScanMode(BluetoothAdapter::ScanMode::kConnectable);
+  CountDownLatch found_latch(1);
+  bt_a_->StartDiscovery(DiscoveryCallback{
+      .device_discovered_cb =
+          [this, &found_latch](BluetoothDevice& device) {
+            NEARBY_LOG(INFO, "Device discovered: %s", device.GetName().c_str());
+            EXPECT_EQ(device.GetName(), adapter_b_->GetName());
+            found_latch.CountDown();
+          },
+  });
+  adapter_b_->SetScanMode(BluetoothAdapter::ScanMode::kConnectableDiscoverable);
+  EXPECT_EQ(adapter_b_->GetScanMode(),
+            BluetoothAdapter::ScanMode::kConnectableDiscoverable);
+  EXPECT_TRUE(found_latch.Await(absl::Milliseconds(1000)).result());
+  std::string service_name{"service"};
+  std::string service_uuid("service-uuid");
+  BluetoothServerSocket server_socket =
+      bt_b_->ListenForService(service_name, service_uuid);
+  EXPECT_TRUE(server_socket.IsValid());
+  server_socket.Close();
+}
+
+TEST_F(BluetoothClassicMediumTest, CanConnectToService) {
+  adapter_a_->SetScanMode(BluetoothAdapter::ScanMode::kConnectable);
+  CountDownLatch found_latch(1);
+  BluetoothDevice* discovered_device = nullptr;
+  bt_a_->StartDiscovery(DiscoveryCallback{
+      .device_discovered_cb =
+          [this, &found_latch, &discovered_device](BluetoothDevice& device) {
+            NEARBY_LOG(INFO, "Device discovered: %s", device.GetName().c_str());
+            EXPECT_EQ(device.GetName(), adapter_b_->GetName());
+            discovered_device = &device;
+            found_latch.CountDown();
+          },
+  });
+  adapter_b_->SetScanMode(BluetoothAdapter::ScanMode::kConnectableDiscoverable);
+  EXPECT_EQ(adapter_b_->GetScanMode(),
+            BluetoothAdapter::ScanMode::kConnectableDiscoverable);
+  EXPECT_TRUE(found_latch.Await(absl::Milliseconds(1000)).result());
+  std::string service_name{"service"};
+  std::string service_uuid("service-uuid");
+  BluetoothServerSocket server_socket =
+      bt_b_->ListenForService(service_name, service_uuid);
+  EXPECT_TRUE(server_socket.IsValid());
+  BluetoothSocket socket_a;
+  BluetoothSocket socket_b;
+  EXPECT_FALSE(socket_a.IsValid());
+  EXPECT_FALSE(socket_b.IsValid());
+  {
+    SingleThreadExecutor server_executor;
+    SingleThreadExecutor client_executor;
+    client_executor.Execute(
+        [this, &socket_a, discovered_device, &service_uuid, &server_socket]() {
+          socket_a = bt_a_->ConnectToService(*discovered_device, service_uuid);
+          if (!socket_a.IsValid()) server_socket.Close();
+        });
+    server_executor.Execute(
+        [&socket_b, &server_socket]() {
+          socket_b = server_socket.Accept();
+          if (!socket_b.IsValid()) server_socket.Close();
+        });
+  }
+  EXPECT_TRUE(socket_a.IsValid());
+  EXPECT_TRUE(socket_b.IsValid());
+  server_socket.Close();
+}
+
+}  // namespace
+}  // namespace nearby
+}  // namespace location

cpp/platform_v2/public/cancelable.h

@@ -0,0 +1,50 @@
+// 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_CANCELABLE_H_
+#define PLATFORM_V2_PUBLIC_CANCELABLE_H_
+
+#include <memory>
+#include <utility>
+
+#include "platform_v2/api/cancelable.h"
+
+namespace location {
+namespace nearby {
+
+// An interface to provide a cancellation mechanism for objects that represent
+// long-running operations.
+class Cancelable final {
+ public:
+  Cancelable() = default;
+  Cancelable(const Cancelable&) = default;
+  Cancelable& operator=(const Cancelable& other) = default;
+
+  ~Cancelable() = default;
+
+  // This constructor is used internally only,
+  // by other classes in "//platform_v2/public/".
+  explicit Cancelable(std::shared_ptr<api::Cancelable> impl)
+      : impl_(std::move(impl)) {}
+
+  bool Cancel() { return impl_->Cancel(); }
+
+ private:
+  std::shared_ptr<api::Cancelable> impl_;
+};
+
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_PUBLIC_CANCELABLE_H_

cpp/platform_v2/public/cancelable_alarm.h

@@ -0,0 +1,70 @@
+// 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_CANCELABLE_ALARM_H_
+#define PLATFORM_V2_PUBLIC_CANCELABLE_ALARM_H_
+
+#include <cstdint>
+#include <functional>
+#include <memory>
+#include <string>
+
+#include "platform_v2/public/cancelable.h"
+#include "platform_v2/public/mutex.h"
+#include "platform_v2/public/mutex_lock.h"
+#include "platform_v2/public/scheduled_executor.h"
+
+namespace location {
+namespace nearby {
+
+/**
+ * A cancelable alarm with a name. This is a simple wrapper around the logic
+ * for posting a Runnable on a ScheduledExecutor and (possibly) later
+ * canceling it.
+ */
+class CancelableAlarm {
+ public:
+  CancelableAlarm(absl::string_view name, std::function<void()>&& runnable,
+                  absl::Duration delay, ScheduledExecutor* scheduled_executor)
+      : name_(name),
+        cancelable_(scheduled_executor->Schedule(std::move(runnable), delay)) {}
+  ~CancelableAlarm() = default;
+  CancelableAlarm(CancelableAlarm&& other) {
+    *this = std::move(other);
+  }
+  CancelableAlarm& operator=(CancelableAlarm&& other) {
+    MutexLock lock(&mutex_);
+    {
+      MutexLock other_lock(&other.mutex_);
+      name_ = std::move(other.name_);
+      cancelable_ = std::move(other.cancelable_);
+    }
+    return *this;
+  }
+
+  bool Cancel() {
+    MutexLock lock(&mutex_);
+    return cancelable_.Cancel();
+  }
+
+ private:
+  Mutex mutex_;
+  std::string name_;
+  Cancelable cancelable_;
+};
+
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_PUBLIC_CANCELABLE_ALARM_H_

cpp/platform_v2/public/condition_variable.h

@@ -0,0 +1,50 @@
+// 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_CONDITION_VARIABLE_H_
+#define PLATFORM_V2_PUBLIC_CONDITION_VARIABLE_H_
+
+#include "platform_v2/api/condition_variable.h"
+#include "platform_v2/api/platform.h"
+#include "platform_v2/base/exception.h"
+#include "platform_v2/public/mutex.h"
+
+namespace location {
+namespace nearby {
+
+// The ConditionVariable class is a synchronization primitive that can be used
+// to block a thread, or multiple threads at the same time, until another thread
+// both modifies a shared variable (the condition), and notifies the
+// ConditionVariable.
+class ConditionVariable final {
+ public:
+  using Platform = api::ImplementationPlatform;
+  explicit ConditionVariable(Mutex* mutex)
+      : impl_(Platform::CreateConditionVariable(mutex->impl_.get())) {}
+  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(); }
+
+ private:
+  std::unique_ptr<api::ConditionVariable> impl_;
+};
+
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_PUBLIC_CONDITION_VARIABLE_H_

cpp/platform_v2/public/count_down_latch.h

@@ -0,0 +1,54 @@
+// 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_COUNT_DOWN_LATCH_H_
+#define PLATFORM_V2_PUBLIC_COUNT_DOWN_LATCH_H_
+
+#include <cstdint>
+
+#include "platform_v2/api/count_down_latch.h"
+#include "platform_v2/api/platform.h"
+#include "platform_v2/base/exception.h"
+#include "absl/time/time.h"
+
+namespace location {
+namespace nearby {
+
+// A synchronization aid that allows one or more threads to wait until a set of
+// operations being performed in other threads completes.
+//
+// https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CountDownLatch.html
+class CountDownLatch final {
+ public:
+  using Platform = api::ImplementationPlatform;
+  explicit CountDownLatch(int count)
+      : impl_(Platform::CreateCountDownLatch(count)) {}
+  CountDownLatch(CountDownLatch&&) = default;
+  CountDownLatch& operator=(CountDownLatch&&) = default;
+  ~CountDownLatch() = default;
+
+  Exception Await() { return impl_->Await(); }
+  ExceptionOr<bool> Await(absl::Duration timeout) {
+    return impl_->Await(timeout);
+  }
+  void CountDown() { impl_->CountDown(); }
+
+ private:
+  std::unique_ptr<api::CountDownLatch> impl_;
+};
+
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_PUBLIC_COUNT_DOWN_LATCH_H_

cpp/platform_v2/public/count_down_latch_test.cc

@@ -0,0 +1,62 @@
+// 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/count_down_latch.h"
+
+#include "platform_v2/public/single_thread_executor.h"
+#include "gtest/gtest.h"
+
+namespace location {
+namespace nearby {
+namespace {
+
+TEST(CountDownLatch, ConstructorDestructorWorks) { CountDownLatch latch(1); }
+
+TEST(CountDownLatch, LatchAwaitCanWait) {
+  CountDownLatch latch(1);
+  SingleThreadExecutor executor;
+  std::atomic_bool done = false;
+  executor.Execute([&done, &latch]() {
+    done = true;
+    latch.CountDown();
+  });
+  latch.Await();
+  EXPECT_TRUE(done);
+}
+
+TEST(CountDownLatch, LatchExtraCountDownIgnored) {
+  CountDownLatch latch(1);
+  SingleThreadExecutor executor;
+  std::atomic_bool done = false;
+  executor.Execute([&done, &latch]() {
+    done = true;
+    latch.CountDown();
+    latch.CountDown();
+    latch.CountDown();
+  });
+  latch.Await();
+  EXPECT_TRUE(done);
+}
+
+TEST(CountDownLatch, LatchAwaitWithTimeoutCanExpire) {
+  CountDownLatch latch(1);
+  SingleThreadExecutor executor;
+  auto response = latch.Await(absl::Milliseconds(100));
+  EXPECT_TRUE(response.ok());
+  EXPECT_FALSE(response.result());
+}
+
+}  // namespace
+}  // namespace nearby
+}  // namespace location

cpp/platform_v2/public/crypto.h

@@ -0,0 +1,20 @@
+// 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_CRYPTO_H_
+#define PLATFORM_V2_PUBLIC_CRYPTO_H_
+
+#include "platform_v2/api/crypto.h"
+
+#endif  // PLATFORM_V2_PUBLIC_CRYPTO_H_

cpp/platform_v2/public/crypto_test.cc

@@ -0,0 +1,48 @@
+// 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/crypto.h"
+
+#include "platform_v2/base/byte_array.h"
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+namespace location {
+namespace nearby {
+
+TEST(CryptoTest, Md5GeneratesHash) {
+  const ByteArray expected_md5(
+      "\xb4\x5c\xff\xe0\x84\xdd\x3d\x20\xd9\x28\xbe\xe8\x5e\x7b\x0f\x21");
+  ByteArray md5_hash = Crypto::Md5("string");
+  EXPECT_EQ(md5_hash, expected_md5);
+}
+
+TEST(CryptoTest, Md5ReturnsEmptyOnError) {
+  EXPECT_EQ(Crypto::Md5(""), ByteArray{});
+}
+
+TEST(CryptoTest, Sha256GeneratesHash) {
+  const ByteArray expected_sha256(
+      "\x47\x32\x87\xf8\x29\x8d\xba\x71\x63\xa8\x97\x90\x89\x58\xf7\xc0"
+      "\xea\xe7\x33\xe2\x5d\x2e\x02\x79\x92\xea\x2e\xdc\x9b\xed\x2f\xa8");
+  ByteArray sha256_hash = Crypto::Sha256("string");
+  EXPECT_EQ(sha256_hash, expected_sha256);
+}
+
+TEST(CryptoTest, Sha256ReturnsEmptyOnError) {
+  EXPECT_EQ(Crypto::Sha256(""), ByteArray{});
+}
+
+}  // namespace nearby
+}  // namespace location

cpp/platform_v2/public/file.h

@@ -0,0 +1,71 @@
+// 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_FILE_H_
+#define PLATFORM_V2_PUBLIC_FILE_H_
+
+#include <cstdint>
+#include <memory>
+#include <string>
+
+#include "platform_v2/api/input_file.h"
+#include "platform_v2/api/output_file.h"
+#include "platform_v2/api/platform.h"
+#include "platform_v2/base/byte_array.h"
+#include "platform_v2/base/exception.h"
+
+namespace location {
+namespace nearby {
+
+class InputFile final : public api::InputFile {
+ 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(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(); }
+
+ private:
+  std::unique_ptr<api::InputFile> impl_;
+};
+
+class OutputFile final : public api::OutputFile {
+ public:
+  using Platform = api::ImplementationPlatform;
+  explicit OutputFile(std::int64_t payload_id)
+      : impl_(Platform::CreateOutputFile(payload_id)) {}
+  ~OutputFile() override = 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(); }
+
+ private:
+  std::unique_ptr<api::OutputFile> impl_;
+};
+
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_PUBLIC_FILE_H_

cpp/platform_v2/public/future.h

@@ -0,0 +1,77 @@
+// 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_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"
+
+namespace location {
+namespace nearby {
+
+template <typename T>
+class Future final : public api::SettableFuture<T> {
+ 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 {
+    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()};
+    }
+  }
+
+  // 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()};
+    }
+  }
+
+ private:
+  std::unique_ptr<api::SettableFuture<absl::any>> impl_;
+};
+
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_PUBLIC_FUTURE_H_

cpp/platform_v2/public/future_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/future.h"
+
+#include "platform_v2/public/single_thread_executor.h"
+#include "gtest/gtest.h"
+#include "absl/time/clock.h"
+#include "absl/time/time.h"
+
+namespace location {
+namespace nearby {
+
+namespace {
+
+enum TestEnum {
+  kValue1 = 1,
+  kValue2 = 2,
+};
+
+enum class ScopedTestEnum {
+  kValue1 = 1,
+  kValue2 = 2,
+};
+
+struct BigSizedStruct {
+  int data[100]{};
+};
+
+bool operator==(const BigSizedStruct& a, const BigSizedStruct& b) {
+  return memcmp(a.data, b.data, sizeof(BigSizedStruct::data)) == 0;
+}
+
+bool operator!=(const BigSizedStruct& a, const BigSizedStruct& b) {
+  return !(a == b);
+}
+
+}  // namespace
+
+TEST(FutureTest, SupportIntegralTypes) {
+  Future<int> future;
+  future.Set(5);
+  EXPECT_EQ(future.Get().exception(), Exception::kSuccess);
+  EXPECT_EQ(future.Get().result(), 5);
+}
+
+TEST(FutureTest, SetExceptionIsPropagated) {
+  Future<int> future;
+  future.SetException({Exception::kIo});
+  EXPECT_EQ(future.Get().exception(), Exception::kIo);
+}
+
+TEST(FutureTest, SupportEnum) {
+  Future<TestEnum> future;
+  future.Set(TestEnum::kValue1);
+  EXPECT_EQ(future.Get().exception(), Exception::kSuccess);
+  EXPECT_EQ(future.Get().result(), TestEnum::kValue1);
+}
+
+TEST(FutureTest, SupportScopedEnum) {
+  Future<ScopedTestEnum> future;
+  future.Set(ScopedTestEnum::kValue1);
+  EXPECT_EQ(future.Get().exception(), Exception::kSuccess);
+  EXPECT_EQ(future.Get().result(), ScopedTestEnum::kValue1);
+}
+
+TEST(FutureTest, SetTakesCopyOfValue) {
+  // Default constructor is zero-initalizing all data in BigSizedStruct.
+  BigSizedStruct v1;
+  Future<BigSizedStruct> future;
+  v1.data[0] = 5;  // Changing value before calling Set() will affect stored
+  v1.data[7] = 3;  // value.
+  future.Set(v1);
+  v1.data[1] = 6;  // Changing value after calling Set() will not affect stored
+  v1.data[5] = 4;  // value.
+  EXPECT_EQ(future.Get().exception(), Exception::kSuccess);
+  BigSizedStruct v2 = future.Get().result();
+  EXPECT_NE(v1, v2);
+  v1.data[1] = 0;
+  v1.data[5] = 0;
+  EXPECT_EQ(v2, v1);
+}
+
+TEST(FutureTest, SetsExceptionOnTimeout) {
+  Future<int> future;
+  EXPECT_EQ(future.Get(absl::Milliseconds(100)).exception(),
+            Exception::kTimeout);
+}
+
+TEST(FutureTest, GetBlocksWhenNotReady) {
+  Future<int> future;
+  SingleThreadExecutor executor;
+  absl::Time start = absl::Now();
+  executor.Execute([&future](){
+    absl::SleepFor(absl::Milliseconds(500));
+    future.Set(10);
+  });
+  auto response = future.Get();
+  absl::Duration blocked_duration = absl::Now() - start;
+  EXPECT_EQ(response.result(), 10);
+  EXPECT_GE(blocked_duration, absl::Milliseconds(500));
+}
+
+}  // namespace nearby
+}  // namespace location

cpp/platform_v2/public/logging.h

@@ -0,0 +1,20 @@
+// 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_LOGGING_H_
+#define PLATFORM_V2_PUBLIC_LOGGING_H_
+
+#include "platform_v2/base/logging.h"
+
+#endif  // PLATFORM_V2_PUBLIC_LOGGING_H_

cpp/platform_v2/public/logging_test.cc

@@ -0,0 +1,26 @@
+// 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/logging.h"
+
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+namespace {
+
+TEST(LoggingTest, CanLog) {
+  NEARBY_LOG(INFO, "message");
+}
+
+}

cpp/platform_v2/public/multi_thread_executor.h

@@ -0,0 +1,42 @@
+// 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_MULTI_THREAD_EXECUTOR_H_
+#define PLATFORM_V2_PUBLIC_MULTI_THREAD_EXECUTOR_H_
+
+#include "platform_v2/api/platform.h"
+#include "platform_v2/public/submittable_executor.h"
+
+namespace location {
+namespace nearby {
+
+// An Executor that reuses a fixed number of threads operating off a shared
+// unbounded queue.
+//
+// https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executors.html#newFixedThreadPool-int-
+class MultiThreadExecutor final : public SubmittableExecutor {
+ public:
+  using Platform = api::ImplementationPlatform;
+  explicit MultiThreadExecutor(int max_parallelism)
+      : SubmittableExecutor(
+            Platform::CreateMultiThreadExecutor(max_parallelism)) {}
+  MultiThreadExecutor(MultiThreadExecutor&&) = default;
+  MultiThreadExecutor& operator=(MultiThreadExecutor&&) = default;
+  ~MultiThreadExecutor() override = default;
+};
+
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_PUBLIC_MULTI_THREAD_EXECUTOR_H_

cpp/platform_v2/public/multi_thread_executor_test.cc

@@ -0,0 +1,108 @@
+// 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/multi_thread_executor.h"
+
+#include <atomic>
+#include <functional>
+
+#include "platform_v2/base/exception.h"
+#include "gtest/gtest.h"
+#include "absl/synchronization/mutex.h"
+#include "absl/time/clock.h"
+#include "absl/time/time.h"
+
+namespace location {
+namespace nearby {
+
+namespace {
+const int kMaxThreads = 5;
+}
+
+TEST(MultiThreadExecutorTest, ConsructorDestructorWorks) {
+  MultiThreadExecutor executor(kMaxThreads);
+}
+
+TEST(MultiThreadExecutorTest, CanExecute) {
+  absl::CondVar cond;
+  std::atomic_bool done = false;
+  MultiThreadExecutor executor(kMaxThreads);
+  executor.Execute([&done, &cond]() {
+    done = true;
+    cond.SignalAll();
+  });
+  absl::Mutex mutex;
+  {
+    absl::MutexLock lock(&mutex);
+    if (!done) {
+      cond.WaitWithTimeout(&mutex, absl::Seconds(1));
+    }
+  }
+  EXPECT_TRUE(done);
+}
+
+TEST(MultiThreadExecutorTest, JobsExecuteInParallel) {
+  absl::Mutex mutex;
+  absl::CondVar thread_cond;
+  absl::CondVar test_cond;
+  MultiThreadExecutor executor(kMaxThreads);
+  int count = 0;
+
+  for (int i = 0; i < kMaxThreads; ++i) {
+    executor.Execute([&count, &mutex, &test_cond, &thread_cond]() {
+      absl::MutexLock lock(&mutex);
+      count++;
+      test_cond.Signal();
+      thread_cond.Wait(&mutex);
+      count--;
+      test_cond.Signal();
+    });
+  }
+
+  {
+    absl::Duration duration = absl::Milliseconds(kMaxThreads * 100);
+    absl::MutexLock lock(&mutex);
+    while (count < kMaxThreads) {
+      absl::Time start = absl::Now();
+      if (test_cond.WaitWithTimeout(&mutex, duration)) break;
+      duration -= absl::Now() - start;
+    }
+  }
+
+  EXPECT_EQ(count, kMaxThreads);
+  thread_cond.SignalAll();
+
+  {
+    absl::Duration duration = absl::Milliseconds(kMaxThreads * 100);
+    absl::MutexLock lock(&mutex);
+    while (count > 0) {
+      absl::Time start = absl::Now();
+      if (test_cond.WaitWithTimeout(&mutex, duration)) break;
+      duration -= absl::Now() - start;
+    }
+  }
+  EXPECT_EQ(count, 0);
+}
+
+TEST(MultiThreadExecutorTest, CanSubmit) {
+  MultiThreadExecutor executor(kMaxThreads);
+  Future<bool> future;
+  bool submitted =
+      executor.Submit<bool>([]() { return ExceptionOr<bool>{true}; }, &future);
+  EXPECT_TRUE(submitted);
+  EXPECT_TRUE(future.Get().result());
+}
+
+}  // namespace nearby
+}  // namespace location

cpp/platform_v2/public/mutex.h

@@ -0,0 +1,78 @@
+// 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_MUTEX_H_
+#define PLATFORM_V2_PUBLIC_MUTEX_H_
+
+#include <memory>
+
+#include "platform_v2/api/mutex.h"
+#include "platform_v2/api/platform.h"
+#include "absl/base/thread_annotations.h"
+
+namespace location {
+namespace nearby {
+
+// This is a classic mutex can be acquired at most once.
+// Atttempt to acuire mutex from the same thread that is holding it will likely
+// cause a deadlock.
+class ABSL_LOCKABLE Mutex final {
+ public:
+  using Platform = api::ImplementationPlatform;
+  using Mode = api::Mutex::Mode;
+
+  explicit Mutex(bool check = true)
+      : impl_(Platform::CreateMutex(check ? Mode::kRegular
+                                          : Mode::kRegularNoCheck)) {}
+  Mutex(Mutex&&) = default;
+  Mutex& operator=(Mutex&&) = default;
+  ~Mutex() = default;
+
+  void Lock() ABSL_EXCLUSIVE_LOCK_FUNCTION() { impl_->Lock(); }
+  void Unlock() ABSL_UNLOCK_FUNCTION() { impl_->Unlock(); }
+
+ private:
+  friend class ConditionVariable;
+  friend class MutexLock;
+  std::unique_ptr<api::Mutex> impl_;
+};
+
+// This mutex is compatible with Java definition:
+// https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/locks/Lock.html
+// This mutex may be acuired multiple times by a thread that is already holding
+// it without blocking.
+// It needs to be released equal number of times before any other thread could
+// successfully acquire it.
+class ABSL_LOCKABLE RecursiveMutex final {
+ public:
+  using Platform = api::ImplementationPlatform;
+  using Mode = api::Mutex::Mode;
+
+  RecursiveMutex() : impl_(Platform::CreateMutex(Mode::kRecursive)) {}
+  RecursiveMutex(RecursiveMutex&&) = default;
+  RecursiveMutex& operator=(RecursiveMutex&&) = default;
+  ~RecursiveMutex() = default;
+
+  void Lock() ABSL_EXCLUSIVE_LOCK_FUNCTION() { impl_->Lock(); }
+  void Unlock() ABSL_UNLOCK_FUNCTION() { impl_->Unlock(); }
+
+ private:
+  friend class MutexLock;
+  std::unique_ptr<api::Mutex> impl_;
+};
+
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_PUBLIC_MUTEX_H_

cpp/platform_v2/public/mutex_lock.h

@@ -0,0 +1,45 @@
+// 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_MUTEX_LOCK_H_
+#define PLATFORM_V2_PUBLIC_MUTEX_LOCK_H_
+
+#include "platform_v2/api/mutex.h"
+#include "platform_v2/public/mutex.h"
+#include "absl/base/thread_annotations.h"
+
+namespace location {
+namespace nearby {
+
+// An RAII mechanism to acquire a Lock over a block of code.
+class ABSL_SCOPED_LOCKABLE MutexLock final {
+ public:
+  explicit MutexLock(Mutex* mutex) ABSL_EXCLUSIVE_LOCK_FUNCTION(mutex)
+      : mutex_(mutex->impl_.get()) {
+    mutex_->Lock();
+  }
+  explicit MutexLock(RecursiveMutex* mutex) ABSL_EXCLUSIVE_LOCK_FUNCTION(mutex)
+      : mutex_(mutex->impl_.get()) {
+    mutex_->Lock();
+  }
+  ~MutexLock() ABSL_UNLOCK_FUNCTION() { mutex_->Unlock(); }
+
+ private:
+  api::Mutex* mutex_;
+};
+
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_PUBLIC_MUTEX_LOCK_H_

cpp/platform_v2/public/mutex_test.cc

@@ -0,0 +1,117 @@
+// 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/mutex.h"
+
+#include "platform_v2/public/condition_variable.h"
+#include "platform_v2/public/single_thread_executor.h"
+#include "gtest/gtest.h"
+#include "absl/synchronization/mutex.h"
+#include "absl/time/clock.h"
+#include "absl/time/time.h"
+
+namespace location {
+namespace nearby {
+namespace {
+
+class MutexTest : public testing::Test {
+ public:
+  void VerifyStepReached(int expected) {
+    absl::MutexLock lock(&step_mutex_);
+    absl::Time deadline = absl::Now() + kTimeToWait;
+    while (step_ != expected) {
+      if (step_cond_.WaitWithDeadline(&step_mutex_, deadline)) break;
+    }
+    EXPECT_EQ(step_, expected);
+    // Make sure we are not progressing further.
+    absl::SleepFor(kTimeToWait);
+    EXPECT_EQ(step_, expected);
+  }
+
+ protected:
+  SingleThreadExecutor executor_;
+  const absl::Duration kTimeToWait = absl::Milliseconds(200);
+  std::atomic_int step_ = 0;
+  absl::Mutex step_mutex_;
+  absl::CondVar step_cond_;
+};
+
+TEST_F(MutexTest, ConstructorDestructorWorks) {
+  Mutex test_mutex;
+  SUCCEED();
+}
+
+TEST_F(MutexTest, BasicLockingWorks) {
+  Mutex test_mutex;
+  test_mutex.Lock();
+  executor_.Execute([this, &test_mutex]() {
+    step_ = 1;
+    step_cond_.Signal();
+    test_mutex.Lock();
+    test_mutex.Unlock();
+    step_ = 2;
+    step_cond_.Signal();
+  });
+  VerifyStepReached(1);
+  test_mutex.Unlock();
+  VerifyStepReached(2);
+}
+
+#ifdef THREAD_SANITIZER
+TEST_F(MutexTest, DISABLED_DoubleLockIsDeadlock)
+ABSL_NO_THREAD_SAFETY_ANALYSIS {
+#else
+TEST_F(MutexTest, DoubleLockIsDeadlock) ABSL_NO_THREAD_SAFETY_ANALYSIS {
+#endif
+  Mutex test_mutex{/*check=*/false};  // Disable run-time deadlock detection.
+  test_mutex.Lock();
+  executor_.Execute([this, &test_mutex]() ABSL_NO_THREAD_SAFETY_ANALYSIS {
+    step_ = 1;
+    step_cond_.Signal();  // We entered executor.
+    test_mutex.Lock();
+    step_ = 2;
+    step_cond_.Signal();  // We acquired the test lock.
+    test_mutex.Lock();    // Deadlock. (Main thread should save us).
+    step_ = 3;
+    step_cond_.Signal();  // We are done.
+  });
+  VerifyStepReached(1);
+  test_mutex.Unlock();  // Let executor proceed to step 2.
+  VerifyStepReached(2);
+  test_mutex.Unlock();  // Bring executor out of deadlock.
+  VerifyStepReached(3);
+  test_mutex.Unlock();  // Unlock before shutdown.
+}
+
+TEST_F(MutexTest, DoubleLockIsNotDeadlock) {
+  RecursiveMutex test_mutex;
+  test_mutex.Lock();
+  executor_.Execute([this, &test_mutex]() ABSL_NO_THREAD_SAFETY_ANALYSIS {
+    step_ = 1;
+    step_cond_.Signal();  // We entered executor.
+    test_mutex.Lock();
+    test_mutex.Lock();
+    test_mutex.Unlock();
+    test_mutex.Unlock();
+    step_ = 2;
+    step_cond_.Signal();  // We are done.
+  });
+  VerifyStepReached(1);
+  test_mutex.Unlock();  // Let executor continue.
+  VerifyStepReached(2);
+}
+
+}  // namespace
+}  // namespace nearby
+}  // namespace location

cpp/platform_v2/public/pipe.cc

@@ -0,0 +1,35 @@
+// 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/pipe.h"
+
+#include "platform_v2/api/condition_variable.h"
+#include "platform_v2/api/mutex.h"
+#include "platform_v2/api/platform.h"
+
+namespace location {
+namespace nearby {
+
+namespace {
+using Platform = api::ImplementationPlatform;
+}
+
+Pipe::Pipe() {
+  auto mutex = Platform::CreateMutex(api::Mutex::Mode::kRegular);
+  auto cond = Platform::CreateConditionVariable(mutex.get());
+  Setup(std::move(mutex), std::move(cond));
+}
+
+}  // namespace nearby
+}  // namespace location

cpp/platform_v2/public/pipe.h

@@ -0,0 +1,36 @@
+// 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_PIPE_H_
+#define PLATFORM_V2_PUBLIC_PIPE_H_
+
+#include "platform_v2/base/base_pipe.h"
+
+namespace location {
+namespace nearby {
+
+// See for details:
+// cpp/platform_v2/base/base_pipe.h
+class Pipe final : public BasePipe {
+ public:
+  Pipe();
+  ~Pipe() override = default;
+  Pipe(Pipe&&) = delete;
+  Pipe& operator=(Pipe&&) = delete;
+};
+
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_PUBLIC_PIPE_H_

cpp/platform_v2/public/pipe_test.cc

@@ -0,0 +1,346 @@
+// 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/pipe.h"
+
+#include <pthread.h>
+
+#include <atomic>
+#include <cstring>
+#include <string>
+
+#include "platform_v2/base/prng.h"
+#include "platform_v2/base/runnable.h"
+#include "gtest/gtest.h"
+
+namespace location {
+namespace nearby {
+
+TEST(PipeTest, ConstructorDestructorWorks) {
+  Pipe pipe;
+  SUCCEED();
+}
+
+TEST(PipeTest, SimpleWriteRead) {
+  Pipe pipe;
+  InputStream& input_stream{pipe.GetInputStream()};
+  OutputStream& output_stream{pipe.GetOutputStream()};
+
+  std::string data("ABCD");
+  EXPECT_TRUE(output_stream.Write(ByteArray(data)).Ok());
+
+  ExceptionOr<ByteArray> read_data = input_stream.Read(Pipe::kChunkSize);
+  EXPECT_TRUE(read_data.ok());
+  EXPECT_EQ(data, std::string(read_data.result()));
+}
+
+TEST(PipeTest, WriteEndClosedBeforeRead) {
+  Pipe pipe;
+  InputStream& input_stream{pipe.GetInputStream()};
+  OutputStream& output_stream{pipe.GetOutputStream()};
+
+  std::string data("ABCD");
+  EXPECT_TRUE(output_stream.Write(ByteArray(data)).Ok());
+
+  // Close the write end before the read end has even begun reading.
+  EXPECT_TRUE(output_stream.Close().Ok());
+
+  // We should still be able to read what was written.
+  ExceptionOr<ByteArray> read_data = input_stream.Read(Pipe::kChunkSize);
+  EXPECT_TRUE(read_data.ok());
+  EXPECT_EQ(data, std::string(read_data.result()));
+
+  // And after that, we should get our indication that all the data that could
+  // ever be read, has already been read.
+  read_data = input_stream.Read(Pipe::kChunkSize);
+  EXPECT_TRUE(read_data.ok());
+  EXPECT_TRUE(read_data.result().Empty());
+}
+
+TEST(PipeTest, ReadEndClosedBeforeWrite) {
+  Pipe pipe;
+  InputStream& input_stream{pipe.GetInputStream()};
+  OutputStream& output_stream{pipe.GetOutputStream()};
+
+  // Close the read end before the write end has even begun writing.
+  EXPECT_TRUE(input_stream.Close().Ok());
+
+  std::string data("ABCD");
+  EXPECT_TRUE(output_stream.Write(ByteArray(data)).Raised(Exception::kIo));
+}
+
+TEST(PipeTest, SizedReadMoreThanFirstChunkSize) {
+  Pipe pipe;
+  InputStream& input_stream{pipe.GetInputStream()};
+  OutputStream& output_stream{pipe.GetOutputStream()};
+
+  std::string data("ABCD");
+  EXPECT_TRUE(output_stream.Write(ByteArray(data)).Ok());
+
+  // Even though we ask for double of what's there in the first chunk, we should
+  // get back only what's there in that first chunk, and that's alright.
+  ExceptionOr<ByteArray> read_data = input_stream.Read(data.size() * 2);
+  EXPECT_TRUE(read_data.ok());
+  EXPECT_EQ(data, std::string(read_data.result()));
+}
+
+TEST(PipeTest, SizedReadLessThanFirstChunkSize) {
+  Pipe pipe;
+  InputStream& input_stream{pipe.GetInputStream()};
+  OutputStream& output_stream{pipe.GetOutputStream()};
+
+  std::string data_first_part("ABCD");
+  std::string data_second_part("EFGHIJ");
+  std::string data = data_first_part + data_second_part;
+  EXPECT_TRUE(output_stream.Write(ByteArray(data)).Ok());
+
+  // When we ask for less than what's there in the first chunk, we should get
+  // back exactly what we asked for, with the remainder still being available
+  // for the next read.
+  std::int64_t desired_size = data_first_part.size();
+  ExceptionOr<ByteArray> first_read_data = input_stream.Read(desired_size);
+  EXPECT_TRUE(first_read_data.ok());
+  EXPECT_EQ(data_first_part, std::string(first_read_data.result()));
+
+  // Now read the remainder, and get everything that ought to have been left.
+  ExceptionOr<ByteArray> second_read_data = input_stream.Read(Pipe::kChunkSize);
+  EXPECT_TRUE(second_read_data.ok());
+  EXPECT_EQ(data_second_part, std::string(second_read_data.result()));
+}
+
+TEST(PipeTest, ReadAfterInputStreamClosed) {
+  Pipe pipe;
+  InputStream& input_stream{pipe.GetInputStream()};
+
+  input_stream.Close();
+
+  ExceptionOr<ByteArray> read_data = input_stream.Read(Pipe::kChunkSize);
+  EXPECT_TRUE(!read_data.ok());
+  EXPECT_TRUE(read_data.GetException().Raised(Exception::kIo));
+}
+
+TEST(PipeTest, WriteAfterOutputStreamClosed) {
+  Pipe pipe;
+  OutputStream& output_stream{pipe.GetOutputStream()};
+
+  output_stream.Close();
+
+  std::string data("ABCD");
+  EXPECT_TRUE(output_stream.Write(ByteArray(data)).Raised(Exception::kIo));
+}
+
+TEST(PipeTest, RepeatedClose) {
+  Pipe pipe;
+  InputStream& input_stream{pipe.GetInputStream()};
+  OutputStream& output_stream{pipe.GetOutputStream()};
+
+  EXPECT_TRUE(output_stream.Close().Ok());
+  EXPECT_TRUE(output_stream.Close().Ok());
+  EXPECT_TRUE(output_stream.Close().Ok());
+
+  EXPECT_TRUE(input_stream.Close().Ok());
+  EXPECT_TRUE(input_stream.Close().Ok());
+  EXPECT_TRUE(input_stream.Close().Ok());
+}
+
+class Thread {
+ public:
+  Thread() : thread_(), attr_(), runnable_() {
+    pthread_attr_init(&attr_);
+    pthread_attr_setdetachstate(&attr_, PTHREAD_CREATE_JOINABLE);
+  }
+  ~Thread() { pthread_attr_destroy(&attr_); }
+
+  void Start(Runnable runnable) {
+    runnable_ = runnable;
+
+    pthread_create(&thread_, &attr_, Thread::Body, this);
+  }
+
+  void Join() { pthread_join(thread_, nullptr); }
+
+ private:
+  static void* Body(void* args) {
+    reinterpret_cast<Thread*>(args)->runnable_();
+    return nullptr;
+  }
+
+  pthread_t thread_;
+  pthread_attr_t attr_;
+  Runnable runnable_;
+};
+
+TEST(PipeTest, ReadBlockedUntilWrite) {
+  using CrossThreadBool = std::atomic_bool;
+
+  class ReaderRunnable {
+   public:
+    ReaderRunnable(InputStream* input_stream,
+                   absl::string_view expected_read_data,
+                   CrossThreadBool* ok_for_read_to_unblock)
+        : input_stream_(input_stream),
+          expected_read_data_(expected_read_data),
+          ok_for_read_to_unblock_(ok_for_read_to_unblock) {}
+    ~ReaderRunnable() = default;
+
+    // Signature "void()" satisfies Runnable.
+    void operator()() {
+      ExceptionOr<ByteArray> read_data = input_stream_->Read(Pipe::kChunkSize);
+
+      // Make sure read() doesn't return before it's appropriate.
+      if (!*ok_for_read_to_unblock_) {
+        FAIL() << "read() unblocked before it was supposed to.";
+      }
+
+      // And then run our normal set of checks to make sure the read() was
+      // successful.
+      EXPECT_TRUE(read_data.ok());
+      EXPECT_EQ(expected_read_data_, std::string(read_data.result()));
+    }
+
+   private:
+    InputStream* input_stream_;
+    const std::string expected_read_data_;
+    CrossThreadBool* ok_for_read_to_unblock_;
+  };
+
+  Pipe pipe;
+  OutputStream& output_stream{pipe.GetOutputStream()};
+
+  // State shared between this thread (the writer) and reader_thread.
+  CrossThreadBool ok_for_read_to_unblock = false;
+  std::string data("ABCD");
+
+  // Kick off reader_thread.
+  Thread reader_thread;
+  reader_thread.Start(
+      ReaderRunnable(&pipe.GetInputStream(), data, &ok_for_read_to_unblock));
+
+  // Introduce a delay before we actually write anything.
+  absl::SleepFor(absl::Seconds(5));
+  // Mark that we're done with the delay, and that the write is about to occur
+  // (this is slightly earlier than it ought to be, but there's no way to
+  // atomically set this from within the implementation of write(), and doing it
+  // after is too late for the purposes of this test).
+  ok_for_read_to_unblock = true;
+
+  // Perform the actual write.
+  EXPECT_TRUE(output_stream.Write(ByteArray(data)).Ok());
+
+  // And wait for reader_thread to finish.
+  reader_thread.Join();
+}
+
+TEST(PipeTest, ConcurrentWriteAndRead) {
+  class BaseRunnable {
+   protected:
+    explicit BaseRunnable(const std::vector<std::string>& chunks)
+        : chunks_(chunks), prng_() {}
+    virtual ~BaseRunnable() = default;
+
+    void RandomSleep() {
+      // Generate a random sleep between 100 and 1000 milliseconds.
+      absl::SleepFor(absl::Milliseconds(BoundedUint32(100, 1000)));
+    }
+
+    const std::vector<std::string>& chunks_;
+
+   private:
+    // Both ends of the bounds are inclusive.
+    std::uint32_t BoundedUint32(std::uint32_t lower_bound,
+                                std::uint32_t upper_bound) {
+      return (prng_.NextUint32() % (upper_bound - lower_bound + 1)) +
+             lower_bound;
+    }
+
+    Prng prng_;
+  };
+
+  class WriterRunnable : public BaseRunnable {
+   public:
+    WriterRunnable(OutputStream* output_stream,
+                   const std::vector<std::string>& chunks)
+        : BaseRunnable(chunks), output_stream_(output_stream) {}
+    ~WriterRunnable() override = default;
+
+    void operator()() {
+      for (auto& chunk : chunks_) {
+        RandomSleep();  // Random pauses before each write.
+        EXPECT_TRUE(output_stream_->Write(ByteArray(chunk)).Ok());
+      }
+
+      RandomSleep();  // A random pause before closing the writer end.
+      EXPECT_TRUE(output_stream_->Close().Ok());
+    }
+
+   private:
+    OutputStream* output_stream_;
+  };
+
+  class ReaderRunnable : public BaseRunnable {
+   public:
+    ReaderRunnable(InputStream* input_stream,
+                   const std::vector<std::string>& chunks)
+        : BaseRunnable(chunks), input_stream_(input_stream) {}
+    ~ReaderRunnable() override = default;
+
+    void operator()() {
+      // First, calculate what we expect to receive, in total.
+      std::string expected_data;
+      for (auto& chunk : chunks_) {
+        expected_data += chunk;
+      }
+
+      // Then, start actually receiving.
+      std::string actual_data;
+      while (true) {
+        RandomSleep();  // Random pauses before each read.
+        ExceptionOr<ByteArray> read_data =
+            input_stream_->Read(Pipe::kChunkSize);
+        if (read_data.ok()) {
+          ByteArray result = read_data.result();
+          if (result.Empty()) {
+            break;  // Normal exit from the read loop.
+          }
+          actual_data += std::string(result);
+        } else {
+          break;  // Erroneous exit from the read loop.
+        }
+      }
+
+      // And once we're done, check that we got everything we expected.
+      EXPECT_EQ(expected_data, actual_data);
+    }
+
+   private:
+    InputStream* input_stream_;
+  };
+
+  Pipe pipe;
+
+  std::vector<std::string> chunks;
+  chunks.push_back("ABCD");
+  chunks.push_back("EFGH");
+  chunks.push_back("IJKL");
+
+  Thread writer_thread;
+  Thread reader_thread;
+  writer_thread.Start(WriterRunnable(&pipe.GetOutputStream(), chunks));
+  reader_thread.Start(ReaderRunnable(&pipe.GetInputStream(), chunks));
+  writer_thread.Join();
+  reader_thread.Join();
+}
+
+}  // namespace nearby
+}  // namespace location

cpp/platform_v2/public/scheduled_executor.h

@@ -0,0 +1,89 @@
+// 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_SCHEDULED_EXECUTOR_H_
+#define PLATFORM_V2_PUBLIC_SCHEDULED_EXECUTOR_H_
+
+#include <cstdint>
+#include <functional>
+#include <memory>
+
+#include "platform_v2/api/platform.h"
+#include "platform_v2/api/scheduled_executor.h"
+#include "platform_v2/base/runnable.h"
+#include "platform_v2/public/cancelable.h"
+#include "platform_v2/public/mutex.h"
+#include "platform_v2/public/mutex_lock.h"
+#include "absl/time/time.h"
+
+namespace location {
+namespace nearby {
+
+// An Executor that can schedule commands to run after a given delay, or to
+// execute periodically.
+//
+// https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ScheduledExecutorService.html
+class ScheduledExecutor final {
+ public:
+  using Platform = api::ImplementationPlatform;
+
+  ScheduledExecutor() : impl_(Platform::CreateScheduledExecutor()) {}
+  ScheduledExecutor(ScheduledExecutor&& other) { *this = std::move(other); }
+  ~ScheduledExecutor() {
+    MutexLock lock(&mutex_);
+    DoShutdown();
+  }
+
+  ScheduledExecutor& operator=(ScheduledExecutor&& other)
+      ABSL_LOCKS_EXCLUDED(mutex_) {
+    MutexLock lock(&mutex_);
+    {
+      MutexLock other_lock(&other.mutex_);
+      impl_ = std::move(other.impl_);
+    }
+    return *this;
+  }
+  void Execute(Runnable&& runnable) ABSL_LOCKS_EXCLUDED(mutex_) {
+    MutexLock lock(&mutex_);
+    if (impl_) impl_->Execute(std::move(runnable));
+  }
+
+  void Shutdown() ABSL_LOCKS_EXCLUDED(mutex_) {
+    MutexLock lock(&mutex_);
+    DoShutdown();
+  }
+
+  Cancelable Schedule(Runnable&& runnable, absl::Duration duration)
+      ABSL_LOCKS_EXCLUDED(mutex_) {
+    MutexLock lock(&mutex_);
+    return impl_ ? Cancelable(impl_->Schedule(std::move(runnable), duration))
+                 : Cancelable();
+  }
+
+ private:
+  void DoShutdown() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_) {
+    if (impl_) {
+      impl_->Shutdown();
+      impl_.reset();
+    }
+  }
+
+  Mutex mutex_;
+  std::unique_ptr<api::ScheduledExecutor> ABSL_GUARDED_BY(mutex_) impl_;
+};
+
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_PUBLIC_SCHEDULED_EXECUTOR_H_

cpp/platform_v2/public/scheduled_executor_test.cc

@@ -0,0 +1,114 @@
+// 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/scheduled_executor.h"
+
+#include <atomic>
+#include <functional>
+
+#include "platform_v2/base/exception.h"
+#include "gtest/gtest.h"
+#include "absl/synchronization/mutex.h"
+#include "absl/time/clock.h"
+#include "absl/time/time.h"
+
+namespace location {
+namespace nearby {
+
+TEST(ScheduledExecutorTest, ConsructorDestructorWorks) {
+  ScheduledExecutor executor;
+}
+
+TEST(ScheduledExecutorTest, CanExecute) {
+  absl::Mutex mutex;
+  absl::CondVar cond;
+  std::atomic_bool done = false;
+  ScheduledExecutor executor;
+  executor.Execute([&done, &cond]() {
+    done = true;
+    cond.SignalAll();
+  });
+  {
+    absl::MutexLock lock(&mutex);
+    if (!done) {
+      cond.WaitWithTimeout(&mutex, absl::Seconds(1));
+    }
+  }
+  EXPECT_TRUE(done);
+}
+
+TEST(ScheduledExecutorTest, CanSchedule) {
+  ScheduledExecutor executor;
+  std::atomic_int value = 0;
+  absl::Mutex mutex;
+  absl::CondVar cond;
+  // schedule job due in 100 ms.
+  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.
+  executor.Schedule(
+      [&value]() {
+        EXPECT_EQ(value, 0);
+        value = 1;
+      },
+      absl::Milliseconds(10));
+  {
+    // wait for the final job to unblock us.
+    absl::MutexLock lock(&mutex);
+    cond.WaitWithTimeout(&mutex, absl::Milliseconds(1000));
+  }
+  EXPECT_EQ(value, 5);
+}
+
+TEST(ScheduledExecutorTest, CanCancel) {
+  ScheduledExecutor executor;
+  std::atomic_int value = 0;
+  Cancelable cancelable =
+      executor.Schedule([&value]() { value += 1; }, absl::Milliseconds(10));
+  EXPECT_EQ(value, 0);
+  EXPECT_TRUE(cancelable.Cancel());
+  absl::SleepFor(absl::Milliseconds(500));
+  EXPECT_EQ(value, 0);
+}
+
+TEST(ScheduledExecutorTest, FailToCancel) {
+  absl::Mutex mutex;
+  absl::CondVar cond;
+  ScheduledExecutor executor;
+  std::atomic_int value = 0;
+  // Schedule job in 10ms, 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(
+      [&cancelable, &cond]() {
+        EXPECT_FALSE(cancelable.Cancel());
+        // Wake up main thread.
+        cond.Signal();
+      },
+      absl::Milliseconds(50));
+  {
+    absl::MutexLock lock(&mutex);
+    cond.Wait(&mutex);
+  }
+  EXPECT_EQ(value, 1);
+}
+
+}  // namespace nearby
+}  // namespace location

cpp/platform_v2/public/single_thread_executor.h

@@ -0,0 +1,40 @@
+// 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_SINGLE_THREAD_EXECUTOR_H_
+#define PLATFORM_V2_PUBLIC_SINGLE_THREAD_EXECUTOR_H_
+
+#include "platform_v2/public/submittable_executor.h"
+
+namespace location {
+namespace nearby {
+
+// An Executor that uses a single worker thread operating off an unbounded
+// queue.
+//
+// https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executors.html#newSingleThreadExecutor--
+class SingleThreadExecutor final : public SubmittableExecutor {
+ public:
+  using Platform = api::ImplementationPlatform;
+  SingleThreadExecutor()
+      : SubmittableExecutor(Platform::CreateSingleThreadExecutor()) {}
+  ~SingleThreadExecutor() override = default;
+  SingleThreadExecutor(SingleThreadExecutor&&) = default;
+  SingleThreadExecutor& operator=(SingleThreadExecutor&&) = default;
+};
+
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_PUBLIC_SINGLE_THREAD_EXECUTOR_H_

cpp/platform_v2/public/single_thread_executor_test.cc

@@ -0,0 +1,85 @@
+// 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/single_thread_executor.h"
+
+#include <atomic>
+#include <functional>
+
+#include "platform_v2/base/exception.h"
+#include "gtest/gtest.h"
+#include "absl/synchronization/mutex.h"
+#include "absl/time/clock.h"
+
+namespace location {
+namespace nearby {
+
+TEST(SingleThreadExecutorTest, ConsructorDestructorWorks) {
+  SingleThreadExecutor executor;
+}
+
+TEST(SingleThreadExecutorTest, CanExecute) {
+  absl::CondVar cond;
+  std::atomic_bool done = false;
+  SingleThreadExecutor executor;
+  executor.Execute([&done, &cond]() {
+    done = true;
+    cond.SignalAll();
+  });
+  absl::Mutex mutex;
+  {
+    absl::MutexLock lock(&mutex);
+    if (!done) {
+      cond.WaitWithTimeout(&mutex, absl::Seconds(1));
+    }
+  }
+  EXPECT_TRUE(done);
+}
+
+TEST(SingleThreadExecutorTest, JobsExecuteInOrder) {
+  std::vector<int> results;
+  SingleThreadExecutor executor;
+
+  for (int i = 0; i < 10; ++i) {
+    executor.Execute([i, &results]() { results.push_back(i); });
+  }
+
+  absl::CondVar cond;
+  std::atomic_bool done = false;
+  executor.Execute([&done, &cond]() {
+    done = true;
+    cond.SignalAll();
+  });
+  absl::Mutex mutex;
+  {
+    absl::MutexLock lock(&mutex);
+    if (!done) {
+      cond.WaitWithTimeout(&mutex, absl::Seconds(1));
+    }
+  }
+  EXPECT_TRUE(done);
+  EXPECT_EQ(results, (std::vector<int>{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}));
+}
+
+TEST(SingleThreadExecutorTest, CanSubmit) {
+  SingleThreadExecutor executor;
+  Future<bool> future;
+  bool submitted =
+      executor.Submit<bool>([]() { return ExceptionOr<bool>{true}; }, &future);
+  EXPECT_TRUE(submitted);
+  EXPECT_TRUE(future.Get().result());
+}
+
+}  // namespace nearby
+}  // namespace location

cpp/platform_v2/public/submittable_executor.h

@@ -0,0 +1,110 @@
+// 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_SUBMITTABLE_EXECUTOR_H_
+#define PLATFORM_V2_PUBLIC_SUBMITTABLE_EXECUTOR_H_
+
+#include <cstddef>
+#include <functional>
+#include <memory>
+#include <utility>
+
+#include "platform_v2/api/executor.h"
+#include "platform_v2/api/submittable_executor.h"
+#include "platform_v2/base/callable.h"
+#include "platform_v2/base/runnable.h"
+#include "platform_v2/public/future.h"
+#include "platform_v2/public/mutex.h"
+#include "platform_v2/public/mutex_lock.h"
+
+namespace location {
+namespace nearby {
+
+// Main interface to be used by platform as a base class for
+// - MultiThreadExecutor
+// - SingleThreadExecutor
+class SubmittableExecutor : public api::SubmittableExecutor {
+ public:
+  ~SubmittableExecutor() override {
+    MutexLock lock(&mutex_);
+    DoShutdown();
+  }
+  SubmittableExecutor(SubmittableExecutor&& other) { *this = std::move(other); }
+  SubmittableExecutor& operator=(SubmittableExecutor&& other)
+      ABSL_LOCKS_EXCLUDED(mutex_) {
+    MutexLock lock(&mutex_);
+    {
+      MutexLock other_lock(&other.mutex_);
+      impl_ = std::move(other.impl_);
+    }
+    return *this;
+  }
+  void Execute(Runnable&& runnable) ABSL_LOCKS_EXCLUDED(mutex_) override {
+    MutexLock lock(&mutex_);
+    if (impl_) impl_->Execute(std::move(runnable));
+  }
+
+  void Shutdown() ABSL_LOCKS_EXCLUDED(mutex_) override {
+    MutexLock lock(&mutex_);
+    DoShutdown();
+  }
+
+  // Submits a callable for execution.
+  // When execution completes, return value is assigned to the passed future.
+  // Future must outlive the whole execution chain.
+  template <typename T>
+  bool Submit(Callable<T>&& callable, Future<T>* future)
+      ABSL_LOCKS_EXCLUDED(mutex_) {
+    MutexLock lock(&mutex_);
+    bool submitted = DoSubmit([callable{std::move(callable)}, future]() {
+      ExceptionOr<T> result = callable();
+      if (result.ok()) {
+        future->Set(result.result());
+      } else {
+        future->SetException({result.exception()});
+      }
+    });
+    if (!submitted) {
+      // complete immediately with kExecution exception value.
+      future->SetException({Exception::kExecution});
+    }
+    return submitted;
+  }
+
+ protected:
+  explicit SubmittableExecutor(std::unique_ptr<api::SubmittableExecutor> impl)
+      : impl_(std::move(impl)) {}
+
+ private:
+  void DoShutdown() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_) {
+    if (impl_) {
+      impl_->Shutdown();
+      impl_.reset();
+    }
+  }
+  // Submit a callable (with no delay).
+  // Returns true, if callable was submitted, false otherwise.
+  // Callable is not submitted if shutdown is in progress.
+  bool DoSubmit(Runnable&& wrapped_callable)
+      ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_) override {
+    return impl_ ? impl_->DoSubmit(std::move(wrapped_callable)) : false;
+  }
+  Mutex mutex_;
+  std::unique_ptr<api::SubmittableExecutor> ABSL_GUARDED_BY(mutex_) impl_;
+};
+
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_PUBLIC_SUBMITTABLE_EXECUTOR_H_

cpp/platform_v2/public/system_clock.h

@@ -0,0 +1,20 @@
+// 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_SYSTEM_CLOCK_H_
+#define PLATFORM_V2_PUBLIC_SYSTEM_CLOCK_H_
+
+#include "platform_v2/api/system_clock.h"
+
+#endif  // PLATFORM_V2_PUBLIC_SYSTEM_CLOCK_H_

cpp/platform_v2/public/webrtc.h

@@ -0,0 +1,58 @@
+// 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_WEBRTC_H_
+#define PLATFORM_V2_PUBLIC_WEBRTC_H_
+
+#include <memory>
+
+#include "platform_v2/api/platform.h"
+#include "platform_v2/api/webrtc.h"
+#include "webrtc/api/peer_connection_interface.h"
+
+namespace location {
+namespace nearby {
+
+class WebRtcMedium final {
+ public:
+  using PeerConnectionCallback = api::WebRtcMedium::PeerConnectionCallback;
+
+  WebRtcMedium() : impl_(api::ImplementationPlatform::CreateWebRtcMedium()) {}
+  ~WebRtcMedium() = default;
+  WebRtcMedium(WebRtcMedium&&) = delete;
+  WebRtcMedium& operator=(WebRtcMedium&&) = delete;
+
+  // Creates and returns a new webrtc::PeerConnectionInterface object via
+  // |callback|.
+  void CreatePeerConnection(webrtc::PeerConnectionObserver* observer,
+                            PeerConnectionCallback callback) {
+    impl_->CreatePeerConnection(observer, std::move(callback));
+  }
+
+  // Returns a signaling messenger for sending WebRTC signaling messages.
+  std::unique_ptr<api::WebRtcSignalingMessenger> GetSignalingMessenger(
+      absl::string_view self_id) {
+    return impl_->GetSignalingMessenger(self_id);
+  }
+
+  bool IsValid() const { return impl_ != nullptr; }
+
+ private:
+  std::unique_ptr<api::WebRtcMedium> impl_;
+};
+
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_V2_PUBLIC_WEBRTC_H_