Roll forward to cl/338482889

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

cpp/platform/impl/g3/BUILD

@@ -13,28 +13,112 @@
 # limitations under the License.
 
 cc_library(
+    name = "types",
+    testonly = True,
+    srcs = [
+        "log_message.cc",
+        "scheduled_executor.cc",
+        "system_clock.cc",
+    ],
+    hdrs = [
+        "atomic_boolean.h",
+        "atomic_reference.h",
+        "condition_variable.h",
+        "count_down_latch.h",
+        "log_message.h",
+        "multi_thread_executor.h",
+        "mutex.h",
+        "pipe.h",
+        "scheduled_executor.h",
+        "single_thread_executor.h",
+    ],
+    visibility = ["//visibility:private"],
+    deps = [
+        "//base",
+        "//platform/api:platform",
+        "//platform/api:types",
+        "//platform/base",
+        "//platform/base:util",
+        "//platform/impl/shared:posix_mutex",
+        "//absl/base:core_headers",
+        "//absl/synchronization",
+        "//absl/time",
+        "//thread",
+    ],
+)
+
+cc_library(
+    name = "comm",
+    testonly = True,
+    srcs = [
+        "ble.cc",
+        "bluetooth_adapter.cc",
+        "bluetooth_classic.cc",
+        "webrtc.cc",
+        "wifi_lan.cc",
+    ],
+    hdrs = [
+        "ble.h",
+        "bluetooth_adapter.h",
+        "bluetooth_classic.h",
+        "webrtc.h",
+        "wifi_lan.h",
+    ],
+    visibility = ["//visibility:private"],
+    deps = [
+        ":types",
+        "//platform/api:comm",
+        "//platform/base",
+        "//platform/base:logging",
+        "//platform/base:test_util",
+        "//absl/base:core_headers",
+        "//absl/container:flat_hash_map",
+        "//absl/container:flat_hash_set",
+        "//absl/strings",
+        "//absl/synchronization",
+        "//webrtc/api:create_peerconnection_factory",  #buildcleaner: keep
+        "//webrtc/api:libjingle_peerconnection_api",
+        "//webrtc/api/task_queue:default_task_queue_factory",
+    ],
+)
+
+cc_library(
+    name = "crypto",
+    testonly = True,
+    srcs = [
+        "crypto.cc",
+    ],
+    visibility = ["//visibility:private"],
+    deps = [
+        "//platform/api:types",
+        "//platform/base",
+        "//absl/strings",
+        "//openssl:crypto",
+    ],
+)
+
+cc_library(
     name = "g3",
+    testonly = True,
     srcs = [
-        "atomic_reference_impl.h",
         "platform.cc",
-        "settable_future_impl.h",
-        "system_clock_impl.h",
     ],
     visibility = [
         "//core:__subpackages__",
         "//platform:__subpackages__",
     ],
     deps = [
-        "//platform:types",
-        "//platform/api",
-        "//platform/impl/shared:atomic_boolean",
+        ":comm",
+        ":crypto",  # build_cleaner: keep
+        ":types",
+        "//platform/api:comm",
+        "//platform/api:platform",
+        "//platform/api:types",
+        "//platform/base:test_util",
         "//platform/impl/shared:file",
-        "//platform/impl/shared:posix_condition_variable",
-        "//platform/impl/shared:posix_lock",
-        "//platform/port:string",
         "//absl/base:core_headers",
-        "//absl/synchronization",
+        "//absl/memory",
+        "//absl/strings",
         "//absl/time",
-        "//absl/types:any",
     ],
 )

cpp/platform/impl/g3/CMakeLists.txt

@@ -1,42 +0,0 @@
-# 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.
-
-add_library(platform_impl_g3 STATIC)
-
-target_sources(platform_impl_g3
-  PRIVATE
-    "atomic_reference_impl.h"
-    "platform.cc"
-    "settable_future_impl.h"
-    "system_clock_impl.h"
-)
-
-target_include_directories(platform_impl_g3
-  PUBLIC
-    ${CMAKE_CURRENT_SOURCE_DIR}
-)
-
-target_link_libraries(platform_impl_g3
-  PUBLIC
-    "platform_types"
-    "platform_api"
-    "platform_impl_shared_atomic_boolean"
-    "platform_impl_shared_file"
-    "platform_impl_shared_posix_condition_variable"
-    "platform_impl_shared_posix_lock"
-    "platform_port_string"
-    "absl::base"
-    "absl::synchronization"
-    "absl::time"
-)

cpp/platform/impl/g3/atomic_boolean.h

@@ -0,0 +1,44 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef PLATFORM_IMPL_G3_ATOMIC_BOOLEAN_H_
+#define PLATFORM_IMPL_G3_ATOMIC_BOOLEAN_H_
+
+#include <atomic>
+
+#include "platform/api/atomic_boolean.h"
+
+namespace location {
+namespace nearby {
+namespace g3 {
+
+// See documentation in
+// cpp/platform/api/atomic_boolean.h
+class AtomicBoolean : public api::AtomicBoolean {
+ public:
+  explicit AtomicBoolean(bool initial_value) : value_(initial_value) {}
+  ~AtomicBoolean() override = default;
+
+  bool Get() const override { return value_.load(); }
+  bool Set(bool value) override { return value_.exchange(value); }
+
+ private:
+  std::atomic_bool value_;
+};
+
+}  // namespace g3
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_IMPL_G3_ATOMIC_BOOLEAN_H_

cpp/platform/impl/g3/atomic_reference.h

@@ -12,41 +12,36 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#ifndef PLATFORM_IMPL_G3_ATOMIC_REFERENCE_IMPL_H_
-#define PLATFORM_IMPL_G3_ATOMIC_REFERENCE_IMPL_H_
+#ifndef PLATFORM_IMPL_G3_ATOMIC_REFERENCE_H_
+#define PLATFORM_IMPL_G3_ATOMIC_REFERENCE_H_
+
+#include <atomic>
+#include <cstdint>
 
 #include "platform/api/atomic_reference.h"
-#include "platform/ptr.h"
-#include "absl/synchronization/mutex.h"
-#include "absl/types/any.h"
 
 namespace location {
 namespace nearby {
-namespace platform {
+namespace g3 {
 
-// Provide implementation for absl::any.
-class AtomicReferenceImpl : public AtomicReference<absl::any> {
+class AtomicUint32 : public api::AtomicUint32 {
  public:
-  explicit AtomicReferenceImpl(absl::any initial_value)
-      : value_(std::move(initial_value)) {}
-  ~AtomicReferenceImpl() override = default;
+  explicit AtomicUint32(std::int32_t value) : value_(value) {}
+  ~AtomicUint32() override = default;
 
-  absl::any get() override {
-    absl::MutexLock lock(&mutex_);
+  std::uint32_t Get() const override {
     return value_;
   }
-  void set(absl::any value) override {
-    absl::MutexLock lock(&mutex_);
-    value_ = std::move(value);
+  void Set(std::uint32_t value) override {
+    value_ = value;
   }
 
  private:
-  absl::Mutex mutex_;
-  absl::any value_;
+  std::atomic<std::uint32_t> value_;
 };
 
-}  // namespace platform
+}  // namespace g3
 }  // namespace nearby
 }  // namespace location
 
-#endif  // PLATFORM_IMPL_G3_ATOMIC_REFERENCE_IMPL_H_
+#endif  // PLATFORM_IMPL_G3_ATOMIC_REFERENCE_H_

cpp/platform/impl/g3/ble.cc

@@ -0,0 +1,365 @@
+// 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/impl/g3/ble.h"
+
+#include <iostream>
+#include <memory>
+#include <string>
+
+#include "platform/api/ble.h"
+#include "platform/base/logging.h"
+#include "platform/base/medium_environment.h"
+#include "absl/synchronization/mutex.h"
+
+namespace location {
+namespace nearby {
+namespace g3 {
+
+BleSocket::~BleSocket() {
+  absl::MutexLock lock(&mutex_);
+  DoClose();
+}
+
+void BleSocket::Connect(BleSocket& other) {
+  absl::MutexLock lock(&mutex_);
+  remote_socket_ = &other;
+  input_ = other.output_;
+}
+
+InputStream& BleSocket::GetInputStream() {
+  auto* remote_socket = GetRemoteSocket();
+  CHECK(remote_socket != nullptr);
+  return remote_socket->GetLocalInputStream();
+}
+
+OutputStream& BleSocket::GetOutputStream() {
+  return GetLocalOutputStream();
+}
+
+BleSocket* BleSocket::GetRemoteSocket() {
+  absl::MutexLock lock(&mutex_);
+  return remote_socket_;
+}
+
+bool BleSocket::IsConnected() const {
+  absl::MutexLock lock(&mutex_);
+  return IsConnectedLocked();
+}
+
+bool BleSocket::IsClosed() const {
+  absl::MutexLock lock(&mutex_);
+  return closed_;
+}
+
+Exception BleSocket::Close() {
+  absl::MutexLock lock(&mutex_);
+  DoClose();
+  return {Exception::kSuccess};
+}
+
+BlePeripheral* BleSocket::GetRemotePeripheral() {
+  absl::MutexLock lock(&mutex_);
+  return peripheral_;
+}
+
+void BleSocket::DoClose() {
+  if (!closed_) {
+    remote_socket_ = nullptr;
+    output_->GetOutputStream().Close();
+    output_->GetInputStream().Close();
+    if (IsConnectedLocked()) {
+      input_->GetOutputStream().Close();
+      input_->GetInputStream().Close();
+    }
+    closed_ = true;
+  }
+}
+
+bool BleSocket::IsConnectedLocked() const { return input_ != nullptr; }
+
+InputStream& BleSocket::GetLocalInputStream() {
+  absl::MutexLock lock(&mutex_);
+  return output_->GetInputStream();
+}
+
+OutputStream& BleSocket::GetLocalOutputStream() {
+  absl::MutexLock lock(&mutex_);
+  return output_->GetOutputStream();
+}
+
+std::unique_ptr<api::BleSocket> BleServerSocket::Accept(
+    BlePeripheral* peripheral) {
+  absl::MutexLock lock(&mutex_);
+  if (closed_) return {};
+  while (pending_sockets_.empty()) {
+    cond_.Wait(&mutex_);
+    if (closed_) break;
+  }
+  if (closed_) return {};
+  auto* remote_socket =
+      pending_sockets_.extract(pending_sockets_.begin()).value();
+  CHECK(remote_socket);
+  auto local_socket = std::make_unique<BleSocket>(peripheral);
+  local_socket->Connect(*remote_socket);
+  remote_socket->Connect(*local_socket);
+  cond_.SignalAll();
+  return local_socket;
+}
+
+bool BleServerSocket::Connect(BleSocket& socket) {
+  absl::MutexLock lock(&mutex_);
+  if (closed_) return false;
+  if (socket.IsConnected()) {
+    NEARBY_LOG(ERROR,
+               "Failed to connect to Ble server socket: already connected");
+    return true;  // already connected.
+  }
+  // add client socket to the pending list
+  pending_sockets_.emplace(&socket);
+  cond_.SignalAll();
+  while (!socket.IsConnected()) {
+    cond_.Wait(&mutex_);
+    if (closed_) return false;
+  }
+  return true;
+}
+
+void BleServerSocket::SetCloseNotifier(std::function<void()> notifier) {
+  absl::MutexLock lock(&mutex_);
+  close_notifier_ = std::move(notifier);
+}
+
+BleServerSocket::~BleServerSocket() {
+  absl::MutexLock lock(&mutex_);
+  DoClose();
+}
+
+Exception BleServerSocket::Close() {
+  absl::MutexLock lock(&mutex_);
+  return DoClose();
+}
+
+Exception BleServerSocket::DoClose() {
+  bool should_notify = !closed_;
+  closed_ = true;
+  if (should_notify) {
+    cond_.SignalAll();
+    if (close_notifier_) {
+      auto notifier = std::move(close_notifier_);
+      mutex_.Unlock();
+      // Notifier may contain calls to public API, and may cause deadlock, if
+      // mutex_ is held during the call.
+      notifier();
+      mutex_.Lock();
+    }
+  }
+  return {Exception::kSuccess};
+}
+
+BleMedium::BleMedium(api::BluetoothAdapter& adapter)
+    : adapter_(static_cast<BluetoothAdapter*>(&adapter)) {
+  adapter_->SetBleMedium(this);
+  auto& env = MediumEnvironment::Instance();
+  env.RegisterBleMedium(*this);
+}
+
+BleMedium::~BleMedium() {
+  adapter_->SetBleMedium(nullptr);
+  auto& env = MediumEnvironment::Instance();
+  env.UnregisterBleMedium(*this);
+
+  StopAdvertising(advertising_info_.service_id);
+  StopScanning(scanning_info_.service_id);
+
+  accept_loops_runner_.Shutdown();
+  NEARBY_LOG(INFO, "BleMedium dtor advertising_accept_thread_running_ = %d",
+             acceptance_thread_running_.load());
+  // If acceptance thread is still running, wait to finish.
+  if (acceptance_thread_running_) {
+    while (acceptance_thread_running_) {
+      CountDownLatch latch(1);
+      close_accept_loops_runner_.Execute([&latch]() { latch.CountDown(); });
+      latch.Await();
+    }
+  }
+}
+
+bool BleMedium::StartAdvertising(
+    const std::string& service_id, const ByteArray& advertisement_bytes,
+    const std::string& fast_advertisement_service_uuid) {
+  NEARBY_LOGS(INFO) << "G3 Ble StartAdvertising: service_id=" << service_id
+                    << ", advertisement bytes=" << advertisement_bytes.data()
+                    << "(" << advertisement_bytes.size() << "),"
+                    << " fast advertisement service uuid="
+                    << fast_advertisement_service_uuid;
+  auto& env = MediumEnvironment::Instance();
+  auto& peripheral = adapter_->GetPeripheral();
+  peripheral.SetAdvertisementBytes(service_id, advertisement_bytes);
+  bool fast_advertisement = !fast_advertisement_service_uuid.empty();
+  env.UpdateBleMediumForAdvertising(*this, peripheral, service_id,
+                                    fast_advertisement, true);
+
+  absl::MutexLock lock(&mutex_);
+  if (server_socket_ != nullptr) server_socket_.release();
+  server_socket_ = std::make_unique<BleServerSocket>();
+
+  acceptance_thread_running_.exchange(true);
+  accept_loops_runner_.Execute([&env, this, service_id]() mutable {
+    if (!accept_loops_runner_.InShutdown()) {
+      while (true) {
+        auto client_socket =
+            server_socket_->Accept(&(this->adapter_->GetPeripheral()));
+        if (client_socket == nullptr) break;
+        env.CallBleAcceptedConnectionCallback(*this, *(client_socket.release()),
+                                              service_id);
+      }
+    }
+    acceptance_thread_running_.exchange(false);
+  });
+  advertising_info_.service_id = service_id;
+  return true;
+}
+
+bool BleMedium::StopAdvertising(const std::string& service_id) {
+  NEARBY_LOGS(INFO) << "G3 Ble StopAdvertising: service_id=" << service_id;
+  {
+    absl::MutexLock lock(&mutex_);
+    if (advertising_info_.Empty()) {
+      NEARBY_LOGS(INFO) << "G3 Ble StopAdvertising: Can't stop advertising "
+                           "because we never started advertising.";
+      return false;
+    }
+    advertising_info_.Clear();
+  }
+
+  auto& env = MediumEnvironment::Instance();
+  env.UpdateBleMediumForAdvertising(*this, adapter_->GetPeripheral(),
+                                    service_id, /*fast_advertisement=*/false,
+                                    /*enabled=*/false);
+  accept_loops_runner_.Shutdown();
+  if (server_socket_ == nullptr) {
+    NEARBY_LOGS(ERROR) << "G3 Ble StopAdvertising: Failed to find Ble Server "
+                          "socket: service_id="
+                       << service_id;
+    // Fall through for server socket not found.
+    return true;
+  }
+
+  if (!server_socket_->Close().Ok()) {
+    NEARBY_LOGS(INFO)
+        << "G3 Ble StopAdvertising: Failed to close Ble server socket for "
+        << service_id;
+    return false;
+  }
+  return true;
+}
+
+bool BleMedium::StartScanning(
+    const std::string& service_id,
+    const std::string& fast_advertisement_service_uuid,
+    DiscoveredPeripheralCallback callback) {
+  NEARBY_LOGS(INFO) << "G3 Ble StartScanning: service_id=" << service_id;
+  auto& env = MediumEnvironment::Instance();
+  env.UpdateBleMediumForScanning(*this, service_id,
+                                 fast_advertisement_service_uuid,
+                                 std::move(callback), true);
+  {
+    absl::MutexLock lock(&mutex_);
+    scanning_info_.service_id = service_id;
+  }
+  return true;
+}
+
+bool BleMedium::StopScanning(const std::string& service_id) {
+  NEARBY_LOGS(INFO) << "G3 Ble StopScanning: service_id=" << service_id;
+  {
+    absl::MutexLock lock(&mutex_);
+    if (scanning_info_.Empty()) {
+      NEARBY_LOGS(INFO) << "G3 Ble StopDiscovery: Can't stop scanning because "
+                           "we never started scanning.";
+      return false;
+    }
+    scanning_info_.Clear();
+  }
+
+  auto& env = MediumEnvironment::Instance();
+  env.UpdateBleMediumForScanning(*this, service_id, {}, {}, false);
+  return true;
+}
+
+bool BleMedium::StartAcceptingConnections(const std::string& service_id,
+                                          AcceptedConnectionCallback callback) {
+  NEARBY_LOGS(INFO) << "G3 Ble StartAcceptingConnections: service_id="
+                    << service_id;
+  auto& env = MediumEnvironment::Instance();
+  env.UpdateBleMediumForAcceptedConnection(*this, service_id, callback);
+  return true;
+}
+
+bool BleMedium::StopAcceptingConnections(const std::string& service_id) {
+  NEARBY_LOGS(INFO) << "G3 Ble StopAcceptingConnections: service_id="
+                    << service_id;
+  auto& env = MediumEnvironment::Instance();
+  env.UpdateBleMediumForAcceptedConnection(*this, service_id, {});
+  return true;
+}
+
+std::unique_ptr<api::BleSocket> BleMedium::Connect(
+    api::BlePeripheral& remote_peripheral, const std::string& service_id) {
+  NEARBY_LOG(INFO,
+             "G3 Ble Connect [self]: medium=%p, adapter=%p, peripheral=%p, "
+             "service_id=%s",
+             this, &GetAdapter(), &GetAdapter().GetPeripheral(),
+             service_id.c_str());
+  // First, find an instance of remote medium, that exposed this peripheral.
+  auto& adapter = static_cast<BlePeripheral&>(remote_peripheral).GetAdapter();
+  auto* medium = static_cast<BleMedium*>(adapter.GetBleMedium());
+
+  if (!medium) return {};  // Can't find medium. Bail out.
+
+  BleServerSocket* remote_server_socket = nullptr;
+  NEARBY_LOG(INFO,
+             "G3 Ble Connect [peer]: medium=%p, adapter=%p, peripheral=%p, "
+             "service_id=%s",
+             medium, &adapter, &remote_peripheral, service_id.c_str());
+  // Then, find our server socket context in this medium.
+  {
+    absl::MutexLock medium_lock(&medium->mutex_);
+    remote_server_socket = medium->server_socket_.get();
+    if (remote_server_socket == nullptr) {
+      NEARBY_LOGS(ERROR)
+          << "G3 Ble Connect: Failed to find Ble Server socket: service_id="
+          << service_id;
+      return {};
+    }
+  }
+
+  BlePeripheral peripheral = static_cast<BlePeripheral&>(remote_peripheral);
+  auto socket = std::make_unique<BleSocket>(&peripheral);
+  // Finally, Request to connect to this socket.
+  if (!remote_server_socket->Connect(*socket)) {
+    NEARBY_LOGS(ERROR) << "G3 Ble Connect: Failed to connect to existing Ble "
+                          "Server socket: service_id="
+                       << service_id;
+    return {};
+  }
+
+  NEARBY_LOG(INFO, "G3 Ble Connect: connected: socket=%p", socket.get());
+  return socket;
+}
+
+}  // namespace g3
+}  // namespace nearby
+}  // namespace location

cpp/platform/impl/g3/ble.h

@@ -0,0 +1,229 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef PLATFORM_IMPL_G3_BLE_H_
+#define PLATFORM_IMPL_G3_BLE_H_
+
+#include <memory>
+#include <string>
+
+#include "platform/api/ble.h"
+#include "platform/base/byte_array.h"
+#include "platform/base/input_stream.h"
+#include "platform/base/output_stream.h"
+#include "platform/impl/g3/bluetooth_adapter.h"
+#include "platform/impl/g3/bluetooth_classic.h"
+#include "platform/impl/g3/multi_thread_executor.h"
+#include "platform/impl/g3/pipe.h"
+#include "absl/container/flat_hash_map.h"
+#include "absl/container/flat_hash_set.h"
+#include "absl/strings/escaping.h"
+#include "absl/synchronization/mutex.h"
+
+namespace location {
+namespace nearby {
+namespace g3 {
+
+class BleMedium;
+
+class BleSocket : public api::BleSocket {
+ public:
+  BleSocket() = default;
+  explicit BleSocket(BlePeripheral* peripheral) : peripheral_(peripheral) {}
+  ~BleSocket() override;
+
+  // Connect to another BleSocket, to form a functional low-level channel.
+  // from this point on, and until Close is called, connection exists.
+  void Connect(BleSocket& other) ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns the InputStream of this connected BleSocket.
+  InputStream& GetInputStream() override ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns the OutputStream of this connected BleSocket.
+  // This stream is for local side to write.
+  OutputStream& GetOutputStream() override ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns address of a remote BleSocket or nullptr.
+  BleSocket* GetRemoteSocket() ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns true if connection exists to the (possibly closed) remote socket.
+  bool IsConnected() const ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns true if socket is closed.
+  bool IsClosed() const ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns Exception::kIo on error, Exception::kSuccess otherwise.
+  Exception Close() override ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns valid BlePeripheral pointer if there is a connection, and
+  // nullptr otherwise.
+  BlePeripheral* GetRemotePeripheral() override
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+ private:
+  void DoClose() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
+  // Returns true if connection exists to the (possibly closed) remote socket.
+  bool IsConnectedLocked() const ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
+  // Returns InputStream of our side of a connection.
+  // This is what the remote side is supposed to read from.
+  // This is a helper for GetInputStream() method.
+  InputStream& GetLocalInputStream() ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns OutputStream of our side of a connection.
+  // This is what the local size is supposed to write to.
+  // This is a helper for GetOutputStream() method.
+  OutputStream& GetLocalOutputStream() ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Output pipe is initialized by constructor, it remains always valid, until
+  // it is closed. it represents output part of a local socket. Input part of a
+  // local socket comes from the peer socket, after connection.
+  std::shared_ptr<Pipe> output_ {new Pipe};
+  std::shared_ptr<Pipe> input_;
+  mutable absl::Mutex mutex_;
+  BlePeripheral* peripheral_;
+  BleSocket* remote_socket_ ABSL_GUARDED_BY(mutex_) = nullptr;
+  bool closed_ ABSL_GUARDED_BY(mutex_) = false;
+};
+
+class BleServerSocket {
+ public:
+  ~BleServerSocket();
+
+  // Blocks until either:
+  // - at least one incoming connection request is available, or
+  // - ServerSocket is closed.
+  // On success, returns connected socket, ready to exchange data.
+  // Returns nullptr on error.
+  // Once error is reported, it is permanent, and ServerSocket has to be closed.
+  //
+  // Called by the server side of a connection.
+  // Returns BleSocket to the server side.
+  // If not null, returned socket is connected to its remote (client-side) peer.
+  std::unique_ptr<api::BleSocket> Accept(BlePeripheral* peripheral)
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Blocks until either:
+  // - connection is available, or
+  // - server socket is closed, or
+  // - error happens.
+  //
+  // Called by the client side of a connection.
+  // Returns true, if socket is successfully connected.
+  bool Connect(BleSocket& socket) ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Called by the server side of a connection before passing ownership of
+  // BleServerSocker to user, to track validity of a pointer to this
+  // server socket,
+  void SetCloseNotifier(std::function<void()> notifier)
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns Exception::kIo on error, Exception::kSuccess otherwise.
+  // Calls close_notifier if it was previously set, and marks socket as closed.
+  Exception Close() ABSL_LOCKS_EXCLUDED(mutex_);
+
+ private:
+  Exception DoClose() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
+  absl::Mutex mutex_;
+  absl::CondVar cond_;
+  absl::flat_hash_set<BleSocket*> pending_sockets_ ABSL_GUARDED_BY(mutex_);
+  std::function<void()> close_notifier_ ABSL_GUARDED_BY(mutex_);
+  bool closed_ ABSL_GUARDED_BY(mutex_) = false;
+};
+
+// Container of operations that can be performed over the BLE medium.
+class BleMedium : public api::BleMedium {
+ public:
+  explicit BleMedium(api::BluetoothAdapter& adapter);
+  ~BleMedium() override;
+
+  // Returns true once the Ble advertising has been initiated.
+  bool StartAdvertising(
+      const std::string& service_id, const ByteArray& advertisement_bytes,
+      const std::string& fast_advertisement_service_uuid) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
+  bool StopAdvertising(const std::string& service_id) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns true once the Ble scanning has been initiated.
+  bool StartScanning(const std::string& service_id,
+                     const std::string& fast_advertisement_service_uuid,
+                     DiscoveredPeripheralCallback callback) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns true once Ble scanning for service_id is well and truly
+  // stopped; after this returns, there must be no more invocations of the
+  // DiscoveredPeripheralCallback passed in to StartScanning() for service_id.
+  bool StopScanning(const std::string& service_id) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns true once Ble socket connection requests to service_id can be
+  // accepted.
+  bool StartAcceptingConnections(const std::string& service_id,
+                                 AcceptedConnectionCallback callback)
+      override ABSL_LOCKS_EXCLUDED(mutex_);
+  bool StopAcceptingConnections(const std::string& service_id) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Connects to existing remote Ble peripheral.
+  //
+  // On success, returns a new BleSocket.
+  // On error, returns nullptr.
+  std::unique_ptr<api::BleSocket> Connect(
+      api::BlePeripheral& remote_peripheral,
+      const std::string& service_id) override ABSL_LOCKS_EXCLUDED(mutex_);
+
+  BluetoothAdapter& GetAdapter() { return *adapter_; }
+
+ private:
+  static constexpr int kMaxConcurrentAcceptLoops = 5;
+
+  struct AdvertisingInfo {
+    bool Empty() const { return service_id.empty(); }
+    void Clear() { service_id.clear(); }
+
+    std::string service_id;
+  };
+
+  struct ScanningInfo {
+    bool Empty() const { return service_id.empty(); }
+    void Clear() { service_id.clear(); }
+
+    std::string service_id;
+  };
+
+  absl::Mutex mutex_;
+  BluetoothAdapter* adapter_;  // Our device adapter; read-only.
+
+  // A thread pool dedicated to running all the accept loops from
+  // StartAdvertising().
+  MultiThreadExecutor accept_loops_runner_{kMaxConcurrentAcceptLoops};
+  std::atomic_bool acceptance_thread_running_ = false;
+
+  // A thread pool dedicated to wait to complete the accept_loops_runner_.
+  MultiThreadExecutor close_accept_loops_runner_{kMaxConcurrentAcceptLoops};
+
+  // A server socket is established when start advertising.
+  std::unique_ptr<BleServerSocket> server_socket_;
+  AdvertisingInfo advertising_info_ ABSL_GUARDED_BY(mutex_);
+  ScanningInfo scanning_info_ ABSL_GUARDED_BY(mutex_);
+};
+
+}  // namespace g3
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_IMPL_G3_BLE_H_

cpp/platform/impl/g3/bluetooth_adapter.cc

@@ -0,0 +1,135 @@
+// 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/impl/g3/bluetooth_adapter.h"
+
+#include <string>
+
+#include "platform/base/medium_environment.h"
+#include "platform/base/prng.h"
+#include "platform/impl/g3/bluetooth_classic.h"
+
+namespace location {
+namespace nearby {
+namespace g3 {
+
+BlePeripheral::BlePeripheral(BluetoothAdapter* adapter) : adapter_(*adapter) {}
+
+std::string BlePeripheral::GetName() const { return adapter_.GetName(); }
+
+ByteArray BlePeripheral::GetAdvertisementBytes(
+    const std::string& service_id) const {
+  return advertisement_bytes_;
+}
+
+void BlePeripheral::SetAdvertisementBytes(
+    const std::string& service_id, const ByteArray& advertisement_bytes) {
+  advertisement_bytes_ = advertisement_bytes;
+}
+
+BluetoothDevice::BluetoothDevice(BluetoothAdapter* adapter)
+    : adapter_(*adapter) {}
+
+std::string BluetoothDevice::GetName() const { return adapter_.GetName(); }
+std::string BluetoothDevice::GetMacAddress() const {
+  return adapter_.GetMacAddress();
+}
+
+BluetoothAdapter::BluetoothAdapter() {
+  std::string mac_address;
+  mac_address.resize(6);
+  int64_t raw_mac_addr = Prng().NextInt64();
+  mac_address[0] = static_cast<char>(raw_mac_addr >> 40);
+  mac_address[1] = static_cast<char>(raw_mac_addr >> 32);
+  mac_address[2] = static_cast<char>(raw_mac_addr >> 24);
+  mac_address[3] = static_cast<char>(raw_mac_addr >> 16);
+  mac_address[4] = static_cast<char>(raw_mac_addr >> 8);
+  mac_address[5] = static_cast<char>(raw_mac_addr >> 0);
+  SetMacAddress(mac_address);
+}
+
+BluetoothAdapter::~BluetoothAdapter() { SetStatus(Status::kDisabled); }
+
+void BluetoothAdapter::SetBluetoothClassicMedium(
+    api::BluetoothClassicMedium* medium) {
+  bluetooth_classic_medium_ = medium;
+}
+
+void BluetoothAdapter::SetBleMedium(api::BleMedium* medium) {
+  ble_medium_ = medium;
+}
+
+bool BluetoothAdapter::SetStatus(Status status) {
+  BluetoothAdapter::ScanMode mode;
+  bool enabled = status == Status::kEnabled;
+  std::string name;
+  {
+    absl::MutexLock lock(&mutex_);
+    enabled_ = enabled;
+    name = name_;
+    mode = mode_;
+  }
+  auto& env = MediumEnvironment::Instance();
+  env.OnBluetoothAdapterChangedState(*this, device_, name, enabled, mode);
+  return true;
+}
+
+bool BluetoothAdapter::IsEnabled() const {
+  absl::MutexLock lock(&mutex_);
+  return enabled_;
+}
+
+BluetoothAdapter::ScanMode BluetoothAdapter::GetScanMode() const {
+  absl::MutexLock lock(&mutex_);
+  return mode_;
+}
+
+bool BluetoothAdapter::SetScanMode(BluetoothAdapter::ScanMode mode) {
+  bool enabled;
+  std::string name;
+  {
+    absl::MutexLock lock(&mutex_);
+    mode_ = mode;
+    name = name_;
+    enabled = enabled_;
+  }
+  auto& env = MediumEnvironment::Instance();
+  env.OnBluetoothAdapterChangedState(*this, device_, std::move(name), enabled,
+                                     mode);
+  return true;
+}
+
+std::string BluetoothAdapter::GetName() const {
+  absl::MutexLock lock(&mutex_);
+  return name_;
+}
+
+bool BluetoothAdapter::SetName(absl::string_view name) {
+  BluetoothAdapter::ScanMode mode;
+  bool enabled;
+  {
+    absl::MutexLock lock(&mutex_);
+    name_ = name;
+    enabled = enabled_;
+    mode = mode_;
+  }
+  auto& env = MediumEnvironment::Instance();
+  env.OnBluetoothAdapterChangedState(*this, device_, std::string(name), enabled,
+                                     mode);
+  return true;
+}
+
+}  // namespace g3
+}  // namespace nearby
+}  // namespace location

cpp/platform/impl/g3/bluetooth_adapter.h

@@ -0,0 +1,142 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef PLATFORM_IMPL_G3_BLUETOOTH_ADAPTER_H_
+#define PLATFORM_IMPL_G3_BLUETOOTH_ADAPTER_H_
+
+#include <string>
+
+#include "platform/api/ble.h"
+#include "platform/api/bluetooth_adapter.h"
+#include "platform/api/bluetooth_classic.h"
+#include "platform/impl/g3/single_thread_executor.h"
+#include "absl/base/thread_annotations.h"
+#include "absl/strings/string_view.h"
+#include "absl/synchronization/mutex.h"
+
+namespace location {
+namespace nearby {
+namespace g3 {
+
+// BluetoothDevice and BluetoothAdapter have a mutual dependency.
+class BluetoothAdapter;
+
+// Opaque wrapper over a Ble peripheral. Must contain enough data about a
+// particular Ble device to connect to its GATT server.
+class BlePeripheral : public api::BlePeripheral {
+ public:
+  ~BlePeripheral() override = default;
+
+  std::string GetName() const override;
+  ByteArray GetAdvertisementBytes(const std::string& service_id) const override;
+  void SetAdvertisementBytes(const std::string& service_id,
+                             const ByteArray& advertisement_bytes);
+  BluetoothAdapter& GetAdapter() { return adapter_; }
+
+ private:
+  // Only BluetoothAdapter may instantiate BlePeripheral.
+  friend class BluetoothAdapter;
+
+  explicit BlePeripheral(BluetoothAdapter* adapter);
+
+  BluetoothAdapter& adapter_;
+  ByteArray advertisement_bytes_;
+};
+
+// https://developer.android.com/reference/android/bluetooth/BluetoothDevice.html.
+class BluetoothDevice : public api::BluetoothDevice {
+ public:
+  ~BluetoothDevice() override = default;
+
+  // https://developer.android.com/reference/android/bluetooth/BluetoothDevice.html#getName()
+  std::string GetName() const override;
+  std::string GetMacAddress() const override;
+  BluetoothAdapter& GetAdapter() { return adapter_; }
+
+ private:
+  // Only BluetoothAdapter may instantiate BluetoothDevice.
+  friend class BluetoothAdapter;
+
+  explicit BluetoothDevice(BluetoothAdapter* adapter);
+
+  BluetoothAdapter& adapter_;
+};
+
+// https://developer.android.com/reference/android/bluetooth/BluetoothAdapter.html
+class BluetoothAdapter : public api::BluetoothAdapter {
+ public:
+  using Status = api::BluetoothAdapter::Status;
+  using ScanMode = api::BluetoothAdapter::ScanMode;
+
+  BluetoothAdapter();
+  ~BluetoothAdapter() override;
+
+  // Synchronously sets the status of the BluetoothAdapter to 'status', and
+  // returns true if the operation was a success.
+  bool SetStatus(Status status) override ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns true if the BluetoothAdapter's current status is
+  // Status::Value::kEnabled.
+  bool IsEnabled() const override ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // https://developer.android.com/reference/android/bluetooth/BluetoothAdapter.html#getScanMode()
+  //
+  // Returns ScanMode::kUnknown on error.
+  ScanMode GetScanMode() const override ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Synchronously sets the scan mode of the adapter, and returns true if the
+  // operation was a success.
+  bool SetScanMode(ScanMode mode) override ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // https://developer.android.com/reference/android/bluetooth/BluetoothAdapter.html#getName()
+  // Returns an empty string on error
+  std::string GetName() const override ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // https://developer.android.com/reference/android/bluetooth/BluetoothAdapter.html#setName(java.lang.String)
+  bool SetName(absl::string_view name) override ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns BT MAC address assigned to this adapter.
+  std::string GetMacAddress() const override { return mac_address_; }
+
+  BluetoothDevice& GetDevice() { return device_; }
+
+  void SetBluetoothClassicMedium(api::BluetoothClassicMedium* medium);
+  api::BluetoothClassicMedium* GetBluetoothClassicMedium() {
+    return bluetooth_classic_medium_;
+  }
+
+  BlePeripheral& GetPeripheral() { return peripheral_; }
+
+  void SetBleMedium(api::BleMedium* medium);
+  api::BleMedium* GetBleMedium() { return ble_medium_; }
+
+  void SetMacAddress(std::string& mac_address) { mac_address_ = mac_address; }
+
+ private:
+  mutable absl::Mutex mutex_;
+  BluetoothDevice device_{this};
+  BlePeripheral peripheral_{this};
+  api::BluetoothClassicMedium* bluetooth_classic_medium_ = nullptr;
+  api::BleMedium* ble_medium_ = nullptr;
+  std::string mac_address_;
+  ScanMode mode_ ABSL_GUARDED_BY(mutex_) = ScanMode::kNone;
+  std::string name_ ABSL_GUARDED_BY(mutex_) = "unknown G3 BT device";
+  bool enabled_ ABSL_GUARDED_BY(mutex_) = false;
+};
+
+}  // namespace g3
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_IMPL_G3_BLUETOOTH_ADAPTER_H_

cpp/platform/impl/g3/bluetooth_classic.cc

@@ -0,0 +1,265 @@
+// 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/impl/g3/bluetooth_classic.h"
+
+#include <memory>
+#include <string>
+
+#include "platform/api/bluetooth_classic.h"
+#include "platform/base/logging.h"
+#include "platform/base/medium_environment.h"
+#include "platform/impl/g3/bluetooth_adapter.h"
+#include "absl/synchronization/mutex.h"
+
+namespace location {
+namespace nearby {
+namespace g3 {
+
+BluetoothSocket::~BluetoothSocket() {
+  absl::MutexLock lock(&mutex_);
+  DoClose();
+}
+
+void BluetoothSocket::Connect(BluetoothSocket& other) {
+  absl::MutexLock lock(&mutex_);
+  remote_socket_ = &other;
+  input_ = other.output_;
+}
+
+bool BluetoothSocket::IsConnected() const {
+  absl::MutexLock lock(&mutex_);
+  return IsConnectedLocked();
+}
+
+bool BluetoothSocket::IsClosed() const {
+  absl::MutexLock lock(&mutex_);
+  return closed_;
+}
+
+bool BluetoothSocket::IsConnectedLocked() const { return input_ != nullptr; }
+
+InputStream& BluetoothSocket::GetInputStream() {
+  auto* remote_socket = GetRemoteSocket();
+  CHECK(remote_socket != nullptr);
+  return remote_socket->GetLocalInputStream();
+}
+
+OutputStream& BluetoothSocket::GetOutputStream() {
+  return GetLocalOutputStream();
+}
+
+InputStream& BluetoothSocket::GetLocalInputStream() {
+  absl::MutexLock lock(&mutex_);
+  return output_->GetInputStream();
+}
+
+OutputStream& BluetoothSocket::GetLocalOutputStream() {
+  absl::MutexLock lock(&mutex_);
+  return output_->GetOutputStream();
+}
+
+Exception BluetoothSocket::Close() {
+  absl::MutexLock lock(&mutex_);
+  DoClose();
+  return {Exception::kSuccess};
+}
+
+void BluetoothSocket::DoClose() {
+  if (!closed_) {
+    remote_socket_ = nullptr;
+    output_->GetOutputStream().Close();
+    output_->GetInputStream().Close();
+    input_->GetOutputStream().Close();
+    input_->GetInputStream().Close();
+    closed_ = true;
+  }
+}
+
+BluetoothSocket* BluetoothSocket::GetRemoteSocket() {
+  absl::MutexLock lock(&mutex_);
+  return remote_socket_;
+}
+
+BluetoothDevice* BluetoothSocket::GetRemoteDevice() {
+  BluetoothAdapter* remote_adapter = nullptr;
+  {
+    absl::MutexLock lock(&mutex_);
+    if (remote_socket_ == nullptr || remote_socket_->adapter_ == nullptr) {
+      return nullptr;
+    }
+    remote_adapter = remote_socket_->adapter_;
+  }
+  return remote_adapter ? &remote_adapter->GetDevice() : nullptr;
+}
+
+std::unique_ptr<api::BluetoothSocket> BluetoothServerSocket::Accept() {
+  absl::MutexLock lock(&mutex_);
+  while (pending_sockets_.empty()) {
+    cond_.Wait(&mutex_);
+    if (closed_) break;
+  }
+  // whether or not we were running in the wait loop, return early if closed.
+  if (closed_) return {};
+  auto* remote_socket =
+      pending_sockets_.extract(pending_sockets_.begin()).value();
+  CHECK(remote_socket);
+  auto local_socket = std::make_unique<BluetoothSocket>(adapter_);
+  local_socket->Connect(*remote_socket);
+  remote_socket->Connect(*local_socket);
+  cond_.SignalAll();
+  return local_socket;
+}
+
+bool BluetoothServerSocket::Connect(BluetoothSocket& socket) {
+  absl::MutexLock lock(&mutex_);
+  if (closed_) return false;
+  if (socket.IsConnected()) {
+    NEARBY_LOG(ERROR,
+               "Failed to connect to BT server socket: already connected");
+    return true;  // already connected.
+  }
+  // add client socket to the pending list
+  pending_sockets_.emplace(&socket);
+  cond_.SignalAll();
+  while (!socket.IsConnected()) {
+    cond_.Wait(&mutex_);
+    if (closed_) return false;
+  }
+  return true;
+}
+
+void BluetoothServerSocket::SetCloseNotifier(std::function<void()> notifier) {
+  absl::MutexLock lock(&mutex_);
+  close_notifier_ = std::move(notifier);
+}
+
+BluetoothServerSocket::~BluetoothServerSocket() {
+  absl::MutexLock lock(&mutex_);
+  DoClose();
+}
+
+Exception BluetoothServerSocket::Close() {
+  absl::MutexLock lock(&mutex_);
+  return DoClose();
+}
+
+Exception BluetoothServerSocket::DoClose() {
+  bool should_notify = !closed_;
+  closed_ = true;
+  if (should_notify) {
+    cond_.SignalAll();
+    if (close_notifier_) {
+      auto notifier = std::move(close_notifier_);
+      mutex_.Unlock();
+      // Notifier may contain calls to public API, and may cause deadlock, if
+      // mutex_ is held during the call.
+      notifier();
+      mutex_.Lock();
+    }
+  }
+  return {Exception::kSuccess};
+}
+
+BluetoothClassicMedium::BluetoothClassicMedium(api::BluetoothAdapter& adapter)
+    // TODO(apolyudov): implement and use downcast<> with static assertions.
+    : adapter_(static_cast<BluetoothAdapter*>(&adapter)) {
+  adapter_->SetBluetoothClassicMedium(this);
+  auto& env = MediumEnvironment::Instance();
+  env.RegisterBluetoothMedium(*this, GetAdapter());
+}
+
+BluetoothClassicMedium::~BluetoothClassicMedium() {
+  adapter_->SetBluetoothClassicMedium(nullptr);
+  auto& env = MediumEnvironment::Instance();
+  env.UnregisterBluetoothMedium(*this);
+}
+
+bool BluetoothClassicMedium::StartDiscovery(DiscoveryCallback callback) {
+  auto& env = MediumEnvironment::Instance();
+  env.UpdateBluetoothMedium(*this, std::move(callback));
+  return true;
+}
+
+bool BluetoothClassicMedium::StopDiscovery() {
+  auto& env = MediumEnvironment::Instance();
+  env.UpdateBluetoothMedium(*this, {});
+  return true;
+}
+
+std::unique_ptr<api::BluetoothSocket> BluetoothClassicMedium::ConnectToService(
+    api::BluetoothDevice& remote_device, const std::string& service_uuid) {
+  NEARBY_LOG(INFO,
+             "G3 ConnectToService [self]: medium=%p, adapter=%p, device=%p",
+             this, &GetAdapter(), &GetAdapter().GetDevice());
+  // First, find an instance of remote medium, that exposed this device.
+  auto& adapter = static_cast<BluetoothDevice&>(remote_device).GetAdapter();
+  auto* medium =
+      static_cast<BluetoothClassicMedium*>(adapter.GetBluetoothClassicMedium());
+
+  if (!medium) return {};  // Adapter is not bound to medium. Bail out.
+
+  BluetoothServerSocket* server_socket = nullptr;
+  NEARBY_LOG(
+      INFO,
+      "G3 ConnectToService [peer]: medium=%p, adapter=%p, device=%p, uuid=%s",
+      medium, &adapter, &remote_device, service_uuid.c_str());
+  // Then, find our server socket context in this medium.
+  {
+    absl::MutexLock medium_lock(&medium->mutex_);
+    auto item = medium->sockets_.find(service_uuid);
+    server_socket = item != sockets_.end() ? item->second : nullptr;
+    if (server_socket == nullptr) {
+      NEARBY_LOG(ERROR, "Failed to find BT Server socket: uuid=%s",
+                 service_uuid.c_str());
+      return {};
+    }
+  }
+
+  auto socket = std::make_unique<BluetoothSocket>(&GetAdapter());
+  // Finally, Request to connect to this socket.
+  if (!server_socket->Connect(*socket)) {
+    NEARBY_LOG(ERROR, "Failed to connect to existing BT Server socket: uuid=%s",
+               service_uuid.c_str());
+    return {};
+  }
+
+  NEARBY_LOG(INFO, "G3 ConnectToService: connected: socket=%p", socket.get());
+  return socket;
+}
+
+std::unique_ptr<api::BluetoothServerSocket>
+BluetoothClassicMedium::ListenForService(const std::string& service_name,
+                                         const std::string& service_uuid) {
+  auto socket = std::make_unique<BluetoothServerSocket>(GetAdapter());
+  socket->SetCloseNotifier([this, uuid = service_uuid]() {
+    absl::MutexLock lock(&mutex_);
+    sockets_.erase(uuid);
+  });
+  NEARBY_LOG(INFO, "Adding service: medium=%p, uuid=%s", this,
+             service_uuid.c_str());
+  absl::MutexLock lock(&mutex_);
+  sockets_.emplace(service_uuid, socket.get());
+  return socket;
+}
+
+api::BluetoothDevice* BluetoothClassicMedium::GetRemoteDevice(
+    const std::string& mac_address) {
+  auto& env = MediumEnvironment::Instance();
+  return env.FindBluetoothDevice(mac_address);
+}
+
+}  // namespace g3
+}  // namespace nearby
+}  // namespace location

cpp/platform/impl/g3/bluetooth_classic.h

@@ -0,0 +1,238 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef PLATFORM_IMPL_G3_BLUETOOTH_CLASSIC_H_
+#define PLATFORM_IMPL_G3_BLUETOOTH_CLASSIC_H_
+
+#include <memory>
+#include <string>
+
+#include "platform/api/bluetooth_classic.h"
+#include "platform/base/byte_array.h"
+#include "platform/base/exception.h"
+#include "platform/base/input_stream.h"
+#include "platform/base/listeners.h"
+#include "platform/base/output_stream.h"
+#include "platform/impl/g3/bluetooth_adapter.h"
+#include "platform/impl/g3/pipe.h"
+#include "absl/container/flat_hash_map.h"
+#include "absl/container/flat_hash_set.h"
+#include "absl/synchronization/mutex.h"
+
+namespace location {
+namespace nearby {
+namespace g3 {
+
+// https://developer.android.com/reference/android/bluetooth/BluetoothSocket.html.
+class BluetoothSocket : public api::BluetoothSocket {
+ public:
+  BluetoothSocket() = default;
+  explicit BluetoothSocket(BluetoothAdapter* adapter) : adapter_(adapter) {}
+  ~BluetoothSocket() override;
+
+  // Connects to another BluetoothSocket, to form a functional low-level
+  // channel. From this point on, and until Close is called, connection exists.
+  void Connect(BluetoothSocket& other);
+
+  // NOTE:
+  // It is an undefined behavior if GetInputStream() or GetOutputStream() is
+  // called for a not-connected BluetoothSocket, i.e. any object that is not
+  // returned by BluetoothClassicMedium::ConnectToService() for client side or
+  // BluetoothServerSocket::Accept() for server side of connection.
+
+  // Returns the InputStream of this connected BluetoothSocket.
+  InputStream& GetInputStream() override;
+
+  // Returns the OutputStream of this connected BluetoothSocket.
+  // This stream is for local side to write.
+  OutputStream& GetOutputStream() override;
+
+  // Returns address of a remote BluetoothSocket or nullptr.
+  BluetoothSocket* GetRemoteSocket() ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns true if connection exists to the (possibly closed) remote socket.
+  bool IsConnected() const ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns true if socket is closed.
+  bool IsClosed() const ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // 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() override ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // https://developer.android.com/reference/android/bluetooth/BluetoothSocket.html#getRemoteDevice()
+  // Returns valid BluetoothDevice pointer if there is a connection, and
+  // nullptr otherwise.
+  BluetoothDevice* GetRemoteDevice() override ABSL_LOCKS_EXCLUDED(mutex_);
+
+ private:
+  void DoClose() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
+  // Returns true if connection exists to the (possibly closed) remote socket.
+  bool IsConnectedLocked() const ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
+  // Returns InputStream of our side of a connection.
+  // This is what the remote side is supposed to read from.
+  // This is a helper for GetInputStream() method.
+  InputStream& GetLocalInputStream() ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns OutputStream of our side of a connection.
+  // This is what the local size is supposed to write to.
+  // This is a helper for GetOutputStream() method.
+  OutputStream& GetLocalOutputStream() ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Output pipe is initialized by constructor, it remains always valid, until
+  // it is closed. it represents output part of a local socket. Input part of a
+  // local socket comes from the peer socket, after connection.
+  std::shared_ptr<Pipe> output_{new Pipe};
+  std::shared_ptr<Pipe> input_;
+  mutable absl::Mutex mutex_;
+  BluetoothAdapter* adapter_ = nullptr;  // Our Adapter. Read only.
+  BluetoothSocket* remote_socket_ ABSL_GUARDED_BY(mutex_) = nullptr;
+  bool closed_ ABSL_GUARDED_BY(mutex_) = false;
+};
+
+// https://developer.android.com/reference/android/bluetooth/BluetoothServerSocket.html.
+class BluetoothServerSocket : public api::BluetoothServerSocket {
+ public:
+  explicit BluetoothServerSocket(BluetoothAdapter& adapter)
+      : adapter_(&adapter) {}
+  ~BluetoothServerSocket() override;
+
+  // Blocks until either:
+  // - at least one incoming connection request is available, or
+  // - ServerSocket is closed.
+  // On success, returns connected socket, ready to exchange data.
+  // Returns nullptr on error.
+  // Once error is reported, it is permanent, and ServerSocket has to be closed.
+  //
+  // Called by the server side of a connection.
+  // Returns BluetoothSocket to the server side.
+  // If not null, returned socket is connected to its remote (client-side) peer.
+  std::unique_ptr<api::BluetoothSocket> Accept() override
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Blocks until either:
+  // - connection is available, or
+  // - server socket is closed, or
+  // - error happens.
+  //
+  // Called by the client side of a connection.
+  // socket is an initialized BluetoothSocket, associated with a client
+  // BluetoothAdapter.
+  // Returns true, if socket is successfully connected.
+  bool Connect(BluetoothSocket& socket) ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Called by the server side of a connection before passing ownership of
+  // BluetoothServerSocker to user, to track validity of a pointer to this
+  // server socket,
+  void SetCloseNotifier(std::function<void()> notifier)
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns Exception::kIo on error, Exception::kSuccess otherwise.
+  // Calls close_notifier if it was previously set, and marks socket as closed.
+  Exception Close() override ABSL_LOCKS_EXCLUDED(mutex_);
+
+ private:
+  Exception DoClose() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
+  absl::Mutex mutex_;
+  absl::CondVar cond_;
+  BluetoothAdapter* adapter_ = nullptr;  // Our Adapter. Read only.
+  absl::flat_hash_set<BluetoothSocket*> pending_sockets_
+      ABSL_GUARDED_BY(mutex_);
+  std::function<void()> close_notifier_ ABSL_GUARDED_BY(mutex_);
+  bool closed_ ABSL_GUARDED_BY(mutex_) = false;
+};
+
+// Container of operations that can be performed over the Bluetooth Classic
+// medium.
+class BluetoothClassicMedium : public api::BluetoothClassicMedium {
+ public:
+  explicit BluetoothClassicMedium(api::BluetoothAdapter& adapter);
+  ~BluetoothClassicMedium() override;
+
+  // 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) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // 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() override ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Connects to existing remote BT service.
+  //
+  // 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.
+  //
+  // On success, returns a new BluetoothSocket.
+  // On error, returns nullptr.
+  std::unique_ptr<api::BluetoothSocket> ConnectToService(
+      api::BluetoothDevice& remote_device,
+      const std::string& service_uuid) override ABSL_LOCKS_EXCLUDED(mutex_);
+
+  BluetoothAdapter& GetAdapter() { return *adapter_; }
+
+  // Creates BT service, and begins listening for remote attempts to connect.
+  //
+  // 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 nullptr on error.
+  std::unique_ptr<api::BluetoothServerSocket> ListenForService(
+      const std::string& service_name, const std::string& service_uuid) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+  api::BluetoothDevice* GetRemoteDevice(
+      const std::string& mac_address) override;
+
+ private:
+  absl::Mutex mutex_;
+  BluetoothAdapter* adapter_;  // Our device adapter; read-only.
+  absl::flat_hash_map<std::string, BluetoothServerSocket*> sockets_
+      ABSL_GUARDED_BY(mutex_);
+};
+
+}  // namespace g3
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_IMPL_G3_BLUETOOTH_CLASSIC_H_

cpp/platform/impl/g3/condition_variable.h

@@ -0,0 +1,51 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef PLATFORM_IMPL_G3_CONDITION_VARIABLE_H_
+#define PLATFORM_IMPL_G3_CONDITION_VARIABLE_H_
+
+#include "platform/api/condition_variable.h"
+#include "platform/base/exception.h"
+#include "platform/impl/g3/mutex.h"
+#include "absl/synchronization/mutex.h"
+
+namespace location {
+namespace nearby {
+namespace g3 {
+
+class ConditionVariable : public api::ConditionVariable {
+ public:
+  explicit ConditionVariable(g3::Mutex* mutex) : mutex_(&mutex->mutex_) {}
+  ~ConditionVariable() override = default;
+
+  Exception Wait() override {
+    cond_var_.Wait(mutex_);
+    return {Exception::kSuccess};
+  }
+  Exception Wait(absl::Duration timeout) override {
+    cond_var_.WaitWithTimeout(mutex_, timeout);
+    return {Exception::kSuccess};
+  }
+  void Notify() override { cond_var_.SignalAll(); }
+
+ private:
+  absl::Mutex* mutex_;
+  absl::CondVar cond_var_;
+};
+
+}  // namespace g3
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_IMPL_G3_CONDITION_VARIABLE_H_

cpp/platform/impl/g3/count_down_latch.h

@@ -0,0 +1,73 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef PLATFORM_IMPL_G3_COUNT_DOWN_LATCH_H_
+#define PLATFORM_IMPL_G3_COUNT_DOWN_LATCH_H_
+
+#include "platform/api/count_down_latch.h"
+#include "absl/base/thread_annotations.h"
+#include "absl/synchronization/mutex.h"
+#include "absl/time/clock.h"
+
+namespace location {
+namespace nearby {
+namespace g3 {
+
+// 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 api::CountDownLatch {
+ public:
+  explicit CountDownLatch(int count) : count_(count) {}
+  CountDownLatch(const CountDownLatch&) = delete;
+  CountDownLatch& operator=(const CountDownLatch&) = delete;
+  CountDownLatch(CountDownLatch&&) = delete;
+  CountDownLatch& operator=(CountDownLatch&&) = delete;
+  ExceptionOr<bool> Await(absl::Duration timeout) override {
+    absl::MutexLock lock(&mutex_);
+    absl::Time deadline = absl::Now() + timeout;
+    while (count_ > 0) {
+      if (cond_.WaitWithDeadline(&mutex_, deadline)) {
+        return ExceptionOr<bool>(false);
+      }
+    }
+    return ExceptionOr<bool>(true);
+  }
+  Exception Await() override {
+    absl::MutexLock lock(&mutex_);
+    while (count_ > 0) {
+      cond_.Wait(&mutex_);
+    }
+    return {Exception::kSuccess};
+  }
+  void CountDown() override {
+    absl::MutexLock lock(&mutex_);
+    if (count_ > 0 && --count_ == 0) {
+      cond_.SignalAll();
+    }
+  }
+
+ private:
+  absl::Mutex mutex_;   // Mutex to be used with cond_.Wait...() method family.
+  absl::CondVar cond_;  // Condition to synchronize up to N waiting threads.
+  int count_
+      ABSL_GUARDED_BY(mutex_);  // When zero, latch should release all waiters.
+};
+
+}  // namespace g3
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_IMPL_G3_COUNT_DOWN_LATCH_H_

cpp/platform/impl/g3/crypto.cc

@@ -0,0 +1,53 @@
+// 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/api/crypto.h"
+
+#include <cstdint>
+#include <string>
+
+#include "platform/base/byte_array.h"
+#include "absl/strings/string_view.h"
+#include "openssl/digest.h"
+
+namespace location {
+namespace nearby {
+
+// Initialize global crypto state.
+void Crypto::Init() {}
+
+static ByteArray Hash(absl::string_view input, const EVP_MD* algo) {
+  unsigned int md_out_size = EVP_MAX_MD_SIZE;
+  uint8_t digest_buffer[EVP_MAX_MD_SIZE];
+  if (input.empty()) return {};
+
+  if (!EVP_Digest(input.data(), input.size(), digest_buffer, &md_out_size, algo,
+                  nullptr))
+    return {};
+
+  return ByteArray{reinterpret_cast<char*>(digest_buffer), md_out_size};
+}
+
+// Return MD5 hash of input.
+ByteArray Crypto::Md5(absl::string_view input) {
+  return Hash(input, EVP_md5());
+}
+
+// Return SHA256 hash of input.
+ByteArray Crypto::Sha256(absl::string_view input) {
+  return Hash(input, EVP_sha256());
+}
+
+}  // namespace nearby
+}  // namespace location

cpp/platform/impl/g3/log_message.cc

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

cpp/platform/impl/g3/log_message.h

@@ -0,0 +1,44 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef PLATFORM_IMPL_G3_LOG_MESSAGE_H_
+#define PLATFORM_IMPL_G3_LOG_MESSAGE_H_
+
+#include "base/logging.h"
+#include "platform/api/log_message.h"
+
+namespace location {
+namespace nearby {
+namespace g3 {
+
+// See documentation in
+// cpp/platform/api/log_message.h
+class LogMessage : public api::LogMessage {
+ public:
+  LogMessage(const char* file, int line, Severity severity);
+  ~LogMessage() override;
+
+  void Print(const char* format, ...) override;
+
+  std::ostream& Stream() override;
+
+ private:
+  absl::LogStreamer log_streamer_;
+};
+
+}  // namespace g3
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_IMPL_G3_LOG_MESSAGE_H_

cpp/platform/impl/g3/multi_thread_executor.h

@@ -0,0 +1,73 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef PLATFORM_IMPL_G3_MULTI_THREAD_EXECUTOR_H_
+#define PLATFORM_IMPL_G3_MULTI_THREAD_EXECUTOR_H_
+
+#include <atomic>
+
+#include "platform/api/submittable_executor.h"
+#include "platform/impl/g3/count_down_latch.h"
+#include "absl/time/clock.h"
+#include "thread/threadpool.h"
+
+namespace location {
+namespace nearby {
+namespace g3 {
+
+// An Executor that reuses a fixed number of threads operating off a shared
+// unbounded queue.
+class MultiThreadExecutor : public api::SubmittableExecutor {
+ public:
+  explicit MultiThreadExecutor(int max_parallelism)
+      : thread_pool_(max_parallelism) {
+    thread_pool_.StartWorkers();
+  }
+  void Execute(Runnable&& runnable) override {
+    if (!shutdown_) {
+      thread_pool_.Schedule(std::move(runnable));
+    }
+  }
+  bool DoSubmit(Runnable&& runnable) override {
+    if (shutdown_) return false;
+    thread_pool_.Schedule(std::move(runnable));
+    return true;
+  }
+  void Shutdown() override { DoShutdown(); }
+  ~MultiThreadExecutor() override { DoShutdown(); }
+
+  int GetTid(int index) const override {
+    const auto* thread = thread_pool_.thread(index);
+    return thread ? thread->tid() : 0;
+  }
+
+  void ScheduleAfter(absl::Duration delay, Runnable&& runnable) {
+    if (shutdown_) return;
+    thread_pool_.ScheduleAt(absl::Now() + delay, std::move(runnable));
+  }
+  bool InShutdown() const { return shutdown_; }
+
+ private:
+  void DoShutdown() {
+    shutdown_ = true;
+  }
+  std::atomic_bool shutdown_ = false;
+  ThreadPool thread_pool_;
+};
+
+}  // namespace g3
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_IMPL_G3_MULTI_THREAD_EXECUTOR_H_

cpp/platform/impl/g3/mutex.h

@@ -0,0 +1,61 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef PLATFORM_IMPL_G3_MUTEX_H_
+#define PLATFORM_IMPL_G3_MUTEX_H_
+
+#include "platform/api/mutex.h"
+#include "platform/impl/shared/posix_mutex.h"
+#include "absl/synchronization/mutex.h"
+
+namespace location {
+namespace nearby {
+namespace g3 {
+
+class ABSL_LOCKABLE Mutex : public api::Mutex {
+ public:
+  explicit Mutex(bool check) : check_(check) {}
+  ~Mutex() override = default;
+  Mutex(Mutex&&) = delete;
+  Mutex& operator=(Mutex&&) = delete;
+  Mutex(const Mutex&) = delete;
+  Mutex& operator=(const Mutex&) = delete;
+
+  void Lock() ABSL_EXCLUSIVE_LOCK_FUNCTION() override {
+    mutex_.Lock();
+    if (!check_) mutex_.ForgetDeadlockInfo();
+  }
+  void Unlock() ABSL_UNLOCK_FUNCTION() override { mutex_.Unlock(); }
+
+ private:
+  friend class ConditionVariable;
+  absl::Mutex mutex_;
+  bool check_;
+};
+
+class ABSL_LOCKABLE RecursiveMutex : public posix::Mutex {
+ public:
+  ~RecursiveMutex() override = default;
+  RecursiveMutex() = default;
+  RecursiveMutex(RecursiveMutex&&) = delete;
+  RecursiveMutex& operator=(RecursiveMutex&&) = delete;
+  RecursiveMutex(const RecursiveMutex&) = delete;
+  RecursiveMutex& operator=(const RecursiveMutex&) = delete;
+};
+
+}  // namespace g3
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_IMPL_G3_MUTEX_H_

cpp/platform/impl/g3/pipe.h

@@ -0,0 +1,44 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef PLATFORM_IMPL_G3_PIPE_H_
+#define PLATFORM_IMPL_G3_PIPE_H_
+
+#include <memory>
+
+#include "platform/base/base_pipe.h"
+#include "platform/impl/g3/condition_variable.h"
+#include "platform/impl/g3/mutex.h"
+
+namespace location {
+namespace nearby {
+namespace g3 {
+
+class Pipe : public BasePipe {
+ public:
+  Pipe() {
+    auto mutex = std::make_unique<g3::Mutex>(/*check=*/true);
+    auto cond = std::make_unique<g3::ConditionVariable>(mutex.get());
+    Setup(std::move(mutex), std::move(cond));
+  }
+  ~Pipe() override = default;
+  Pipe(Pipe &&) = delete;
+  Pipe& operator=(Pipe&&) = delete;
+};
+
+}  // namespace g3
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_IMPL_G3_PIPE_H_

cpp/platform/impl/g3/platform.cc

@@ -19,149 +19,155 @@
 
 #include "platform/api/atomic_boolean.h"
 #include "platform/api/atomic_reference.h"
-#include "platform/api/ble.h"
 #include "platform/api/ble_v2.h"
 #include "platform/api/bluetooth_adapter.h"
 #include "platform/api/bluetooth_classic.h"
 #include "platform/api/condition_variable.h"
 #include "platform/api/count_down_latch.h"
-#include "platform/api/hash_utils.h"
-#include "platform/api/lock.h"
+#include "platform/api/log_message.h"
+#include "platform/api/mutex.h"
 #include "platform/api/scheduled_executor.h"
 #include "platform/api/server_sync.h"
-#include "platform/api/settable_future.h"
 #include "platform/api/submittable_executor.h"
-#include "platform/api/system_clock.h"
-#include "platform/api/thread_utils.h"
 #include "platform/api/webrtc.h"
 #include "platform/api/wifi.h"
-#include "platform/impl/g3/atomic_reference_impl.h"
-#include "platform/impl/g3/settable_future_impl.h"
-#include "platform/impl/g3/system_clock_impl.h"
-#include "platform/impl/shared/atomic_boolean_impl.h"
-#include "platform/impl/shared/file_impl.h"
-#include "platform/impl/shared/posix_condition_variable.h"
-#include "platform/impl/shared/posix_lock.h"
-#include "platform/port/string.h"
-#include "platform/ptr.h"
-#include "absl/synchronization/mutex.h"
+#include "platform/base/medium_environment.h"
+#include "platform/impl/g3/atomic_boolean.h"
+#include "platform/impl/g3/atomic_reference.h"
+#include "platform/impl/g3/ble.h"
+#include "platform/impl/g3/bluetooth_adapter.h"
+#include "platform/impl/g3/bluetooth_classic.h"
+#include "platform/impl/g3/condition_variable.h"
+#include "platform/impl/g3/count_down_latch.h"
+#include "platform/impl/g3/log_message.h"
+#include "platform/impl/g3/multi_thread_executor.h"
+#include "platform/impl/g3/mutex.h"
+#include "platform/impl/g3/scheduled_executor.h"
+#include "platform/impl/g3/single_thread_executor.h"
+#include "platform/impl/g3/webrtc.h"
+#include "platform/impl/g3/wifi_lan.h"
+#include "platform/impl/shared/file.h"
+#include "absl/base/integral_types.h"
+#include "absl/memory/memory.h"
+#include "absl/strings/str_cat.h"
 #include "absl/time/time.h"
 
 namespace location {
 namespace nearby {
-namespace platform {
+namespace api {
 
 namespace {
-std::string getPayloadPath(std::int64_t payload_id) {
-  return "/tmp/" + std::to_string(payload_id);
+std::string GetPayloadPath(PayloadId payload_id) {
+  return absl::StrCat("/tmp/", payload_id);
 }
 }  // namespace
 
-Ptr<SubmittableExecutor> ImplementationPlatform::createSingleThreadExecutor() {
-  return Ptr<SubmittableExecutor>(/*new SingleThreadExecutorImpl()*/);
+int GetCurrentTid() {
+  const LiveThread* my = Thread_GetMyLiveThread();
+  return LiveThread_Pthread_TID(my);
 }
 
-Ptr<SubmittableExecutor> ImplementationPlatform::createMultiThreadExecutor(
-    int max_concurrency) {
-  return Ptr<SubmittableExecutor>(/*new MultiThreadExecutorImpl()*/);
+std::unique_ptr<SubmittableExecutor>
+ImplementationPlatform::CreateSingleThreadExecutor() {
+  return absl::make_unique<g3::SingleThreadExecutor>();
 }
 
-Ptr<ScheduledExecutor> ImplementationPlatform::createScheduledExecutor() {
-  return Ptr<ScheduledExecutor>(/*new ScheduledExecutorImpl()*/);
+std::unique_ptr<SubmittableExecutor>
+ImplementationPlatform::CreateMultiThreadExecutor(int max_concurrency) {
+  return absl::make_unique<g3::MultiThreadExecutor>(max_concurrency);
 }
 
-Ptr<AtomicReference<absl::any>>
-ImplementationPlatform::createAtomicReferenceAny(absl::any initial_value) {
-  return Ptr<AtomicReference<absl::any>>(
-      new AtomicReferenceImpl(initial_value));
+std::unique_ptr<ScheduledExecutor>
+ImplementationPlatform::CreateScheduledExecutor() {
+  return absl::make_unique<g3::ScheduledExecutor>();
 }
 
-Ptr<SettableFuture<absl::any>>
-ImplementationPlatform::createSettableFutureAny() {
-  return Ptr<SettableFuture<std::any>>(new SettableFutureImpl{});
+std::unique_ptr<AtomicUint32>
+ImplementationPlatform::CreateAtomicUint32(std::uint32_t value) {
+  return absl::make_unique<g3::AtomicUint32>(value);
 }
 
-Ptr<BluetoothAdapter> ImplementationPlatform::createBluetoothAdapter() {
-  return Ptr<BluetoothAdapter>{};
+std::unique_ptr<BluetoothAdapter>
+ImplementationPlatform::CreateBluetoothAdapter() {
+  return absl::make_unique<g3::BluetoothAdapter>();
 }
 
-Ptr<WifiMedium> ImplementationPlatform::createWifiMedium() {
-  return Ptr<WifiMedium>();
-}
-
-Ptr<CountDownLatch> ImplementationPlatform::createCountDownLatch(
+std::unique_ptr<CountDownLatch> ImplementationPlatform::CreateCountDownLatch(
     std::int32_t count) {
-  return Ptr<CountDownLatch>(/*new CountDownLatchImpl(count)*/);
+  return absl::make_unique<g3::CountDownLatch>(count);
 }
 
-Ptr<ThreadUtils> ImplementationPlatform::createThreadUtils() {
-  return Ptr<ThreadUtils>(/*new ThreadUtilsImpl()*/);
-}
-
-Ptr<SystemClock> ImplementationPlatform::createSystemClock() {
-  return Ptr<SystemClock>(new SystemClockImpl());
-}
-
-Ptr<AtomicBoolean> ImplementationPlatform::createAtomicBoolean(
+std::unique_ptr<AtomicBoolean> ImplementationPlatform::CreateAtomicBoolean(
     bool initial_value) {
-  return Ptr<AtomicBoolean>(new AtomicBooleanImpl(initial_value));
+  return absl::make_unique<g3::AtomicBoolean>(initial_value);
 }
 
-Ptr<InputFile> ImplementationPlatform::createInputFile(
-    std::int64_t payload_id, std::int64_t total_size) {
-  return MakePtr(new InputFileImpl(getPayloadPath(payload_id), total_size));
+std::unique_ptr<InputFile> ImplementationPlatform::CreateInputFile(
+    PayloadId payload_id, std::int64_t total_size) {
+  return absl::make_unique<shared::InputFile>(GetPayloadPath(payload_id),
+                                              total_size);
 }
 
-Ptr<OutputFile> ImplementationPlatform::createOutputFile(
-    std::int64_t payload_id) {
-  return MakePtr(new OutputFileImpl(getPayloadPath(payload_id)));
+std::unique_ptr<OutputFile> ImplementationPlatform::CreateOutputFile(
+    PayloadId payload_id) {
+  return absl::make_unique<shared::OutputFile>(GetPayloadPath(payload_id));
 }
 
-Ptr<BluetoothClassicMedium>
-ImplementationPlatform::createBluetoothClassicMedium() {
-  return Ptr<BluetoothClassicMedium>();
+std::unique_ptr<LogMessage> ImplementationPlatform::CreateLogMessage(
+    const char* file, int line, LogMessage::Severity severity) {
+  return absl::make_unique<g3::LogMessage>(file, line, severity);
 }
 
-Ptr<BLEMedium> ImplementationPlatform::createBLEMedium() {
-  return Ptr<BLEMedium>();
+std::unique_ptr<BluetoothClassicMedium>
+ImplementationPlatform::CreateBluetoothClassicMedium(
+    api::BluetoothAdapter& adapter) {
+  return absl::make_unique<g3::BluetoothClassicMedium>(adapter);
 }
 
-Ptr<BLEMediumV2> ImplementationPlatform::createBLEMediumV2() {
-  return Ptr<BLEMediumV2>();
+std::unique_ptr<BleMedium> ImplementationPlatform::CreateBleMedium(
+    api::BluetoothAdapter& adapter) {
+  return absl::make_unique<g3::BleMedium>(adapter);
 }
 
-Ptr<ServerSyncMedium> ImplementationPlatform::createServerSyncMedium() {
-  return Ptr<ServerSyncMedium>(/*new ServerSyncMediumImpl()*/);
+std::unique_ptr<ble_v2::BleMedium> ImplementationPlatform::CreateBleV2Medium(
+    api::BluetoothAdapter& adapter) {
+  return std::unique_ptr<ble_v2::BleMedium>();
 }
 
-Ptr<WifiLanMedium> ImplementationPlatform::createWifiLanMedium() {
-  return Ptr<WifiLanMedium>();
+std::unique_ptr<ServerSyncMedium>
+ImplementationPlatform::CreateServerSyncMedium() {
+  return std::unique_ptr<ServerSyncMedium>(/*new ServerSyncMediumImpl()*/);
 }
 
-//Ptr<WebRtcSignalingMessenger>
-//ImplementationPlatform::createWebRtcSignalingMessenger(
-//    const std::string& self_id) {
-//  return Ptr<WebRtcSignalingMessenger>(/*new FCMSignalingMessenger()*/);
-//}
-
-Ptr<Lock> ImplementationPlatform::createLock() {
-  return Ptr<Lock>(new PosixLock());
+std::unique_ptr<WifiMedium> ImplementationPlatform::CreateWifiMedium() {
+  return std::unique_ptr<WifiMedium>();
 }
 
-Ptr<ConditionVariable> ImplementationPlatform::createConditionVariable(
-    Ptr<Lock> lock) {
-  return Ptr<ConditionVariable>(new PosixConditionVariable(lock));
+std::unique_ptr<WifiLanMedium> ImplementationPlatform::CreateWifiLanMedium() {
+  return absl::make_unique<g3::WifiLanMedium>();
 }
 
-Ptr<HashUtils> ImplementationPlatform::createHashUtils() {
-  return Ptr<HashUtils>(/*new HashUtilsImpl()*/);
+std::unique_ptr<WebRtcMedium> ImplementationPlatform::CreateWebRtcMedium() {
+  if (MediumEnvironment::Instance().GetEnvironmentConfig().webrtc_enabled) {
+    return absl::make_unique<g3::WebRtcMedium>();
+  } else {
+    return nullptr;
+  }
 }
 
-std::string ImplementationPlatform::getDeviceId() {
-  // TODO(alexchau): Get deviceId from base
-  return "google3";
+std::unique_ptr<Mutex> ImplementationPlatform::CreateMutex(Mutex::Mode mode) {
+  if (mode == Mutex::Mode::kRecursive)
+    return absl::make_unique<g3::RecursiveMutex>();
+  else
+    return absl::make_unique<g3::Mutex>(mode == Mutex::Mode::kRegular);
 }
 
-}  // namespace platform
+std::unique_ptr<ConditionVariable>
+ImplementationPlatform::CreateConditionVariable(Mutex* mutex) {
+  return std::unique_ptr<ConditionVariable>(
+      new g3::ConditionVariable(static_cast<g3::Mutex*>(mutex)));
+}
+
+}  // namespace api
 }  // namespace nearby
 }  // namespace location

cpp/platform/impl/g3/scheduled_executor.cc

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

cpp/platform/impl/g3/scheduled_executor.h

@@ -0,0 +1,59 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef PLATFORM_IMPL_G3_SCHEDULED_EXECUTOR_H_
+#define PLATFORM_IMPL_G3_SCHEDULED_EXECUTOR_H_
+
+#include <atomic>
+#include <memory>
+
+#include "platform/api/cancelable.h"
+#include "platform/api/scheduled_executor.h"
+#include "platform/base/runnable.h"
+#include "platform/impl/g3/single_thread_executor.h"
+#include "absl/time/clock.h"
+#include "thread/threadpool.h"
+
+namespace location {
+namespace nearby {
+namespace g3 {
+
+// An Executor that reuses a fixed number of threads operating off a shared
+// unbounded queue.
+class ScheduledExecutor final : public api::ScheduledExecutor {
+ public:
+  ScheduledExecutor() = default;
+  ~ScheduledExecutor() override {
+    executor_.Shutdown();
+  }
+
+  void Execute(Runnable&& runnable) override {
+    executor_.Execute(std::move(runnable));
+  }
+  std::shared_ptr<api::Cancelable> Schedule(Runnable&& runnable,
+                                            absl::Duration delay) override;
+  void Shutdown() override { executor_.Shutdown(); }
+
+  int GetTid(int index) const override {
+    return executor_.GetTid(index);
+  }
+ private:
+  SingleThreadExecutor executor_;
+};
+
+}  // namespace g3
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_IMPL_G3_SCHEDULED_EXECUTOR_H_

cpp/platform/impl/g3/settable_future_impl.h

@@ -1,108 +0,0 @@
-// Copyright 2020 Google LLC
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     https://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-#ifndef PLATFORM_IMPL_G3_SETTABLE_FUTURE_IMPL_H_
-#define PLATFORM_IMPL_G3_SETTABLE_FUTURE_IMPL_H_
-
-#include <utility>
-
-#include "platform/api/platform.h"
-#include "platform/api/settable_future.h"
-#include "absl/synchronization/mutex.h"
-#include "absl/time/clock.h"
-#include "absl/types/any.h"
-
-namespace location {
-namespace nearby {
-namespace platform {
-
-class SettableFutureImpl : public SettableFuture<absl::any> {
- public:
-  explicit SettableFutureImpl() = default;
-  ~SettableFutureImpl() override = default;
-
-  bool set(absl::any value) override {
-    absl::MutexLock lock(&mutex_);
-    if (!done_) {
-      value_ = std::move(value);
-      done_ = true;
-      exception_ = {Exception::kSuccess};
-      completed_.SignalAll();
-    }
-    return true;
-  }
-
-  bool setException(Exception exception) override {
-    absl::MutexLock lock(&mutex_);
-    return SetExceptionLocked(exception);
-  }
-
-  void addListener(Ptr<Runnable> runnable, Executor* executor) override {}
-
-  ExceptionOr<std::any> get() override {
-    absl::MutexLock lock(&mutex_);
-    while (!done_) {
-      completed_.Wait(&mutex_);
-    }
-    return exception_.value != Exception::kSuccess
-               ? ExceptionOr<std::any>{exception_.value}
-               : ExceptionOr<std::any>{value_};
-  }
-
-  ExceptionOr<std::any> get(std::int64_t timeout_ms) override {
-    absl::MutexLock lock(&mutex_);
-    absl::Duration timeout = absl::Milliseconds(timeout_ms);
-    while (!done_) {
-      absl::Time start_time = absl::Now();
-      if (completed_.WaitWithTimeout(&mutex_, timeout)) {
-        SetExceptionLocked({Exception::kTimeout});
-        break;
-      }
-      absl::Duration spent = absl::Now() - start_time;
-      if (spent < timeout) {
-        timeout -= spent;
-      } else if (!done_) {
-        SetExceptionLocked({Exception::kTimeout});
-        break;
-      }
-    }
-    return exception_.value != Exception::kSuccess
-               ? ExceptionOr<std::any>{exception_.value}
-               : ExceptionOr<std::any>{value_};
-  }
-
- private:
-  bool SetExceptionLocked(Exception exception) {
-    if (!done_) {
-      exception_ = exception.value != Exception::kSuccess
-                       ? exception
-                       : Exception{Exception::kFailed};
-      done_ = true;
-      completed_.SignalAll();
-    }
-    return true;
-  }
-
-  absl::Mutex mutex_;
-  absl::CondVar completed_;
-  bool done_{false};
-  absl::any value_;
-  Exception exception_{Exception::kFailed};
-};
-
-}  // namespace platform
-}  // namespace nearby
-}  // namespace location
-
-#endif  // PLATFORM_IMPL_G3_SETTABLE_FUTURE_IMPL_H_

cpp/platform/impl/g3/single_thread_executor.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_IMPL_G3_SINGLE_THREAD_EXECUTOR_H_
+#define PLATFORM_IMPL_G3_SINGLE_THREAD_EXECUTOR_H_
+
+#include "platform/impl/g3/multi_thread_executor.h"
+
+namespace location {
+namespace nearby {
+namespace g3 {
+
+// An Executor that uses a single worker thread operating off an unbounded
+// queue.
+class SingleThreadExecutor final : public MultiThreadExecutor {
+ public:
+  SingleThreadExecutor() : MultiThreadExecutor(1) {}
+  ~SingleThreadExecutor() override = default;
+};
+
+}  // namespace g3
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_IMPL_G3_SINGLE_THREAD_EXECUTOR_H_

cpp/platform/impl/g3/system_clock.cc

@@ -12,26 +12,19 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#ifndef PLATFORM_IMPL_G3_SYSTEM_CLOCK_IMPL_H_
-#define PLATFORM_IMPL_G3_SYSTEM_CLOCK_IMPL_H_
-
-#include <cstdint>
-
 #include "platform/api/system_clock.h"
+
+#include "platform/base/exception.h"
 #include "absl/time/clock.h"
-#include "absl/time/time.h"
 
 namespace location {
 namespace nearby {
 
-class SystemClockImpl : public SystemClock {
- public:
-  std::int64_t elapsedRealtime() override {
-    return absl::ToUnixMillis(absl::Now());
-  }
-};
+absl::Time SystemClock::ElapsedRealtime() { return absl::Now(); }
+Exception SystemClock::Sleep(absl::Duration duration) {
+  absl::SleepFor(duration);
+  return {Exception::kSuccess};
+}
 
 }  // namespace nearby
 }  // namespace location
-
-#endif  // PLATFORM_IMPL_G3_SYSTEM_CLOCK_IMPL_H_

cpp/platform/impl/g3/webrtc.cc

@@ -0,0 +1,82 @@
+// 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/impl/g3/webrtc.h"
+
+#include <memory>
+
+#include "platform/base/medium_environment.h"
+#include "webrtc/api/task_queue/default_task_queue_factory.h"
+
+namespace location {
+namespace nearby {
+namespace g3 {
+
+WebRtcSignalingMessenger::WebRtcSignalingMessenger(
+    absl::string_view self_id, const connections::LocationHint& location_hint)
+    : self_id_(self_id), location_hint_(location_hint) {}
+
+bool WebRtcSignalingMessenger::SendMessage(absl::string_view peer_id,
+                                           const ByteArray& message) {
+  auto& env = MediumEnvironment::Instance();
+  env.SendWebRtcSignalingMessage(peer_id, message);
+  return true;
+}
+
+bool WebRtcSignalingMessenger::StartReceivingMessages(
+    OnSignalingMessageCallback listener) {
+  auto& env = MediumEnvironment::Instance();
+  env.RegisterWebRtcSignalingMessenger(self_id_, listener);
+  return true;
+}
+
+void WebRtcSignalingMessenger::StopReceivingMessages() {
+  auto& env = MediumEnvironment::Instance();
+  env.UnregisterWebRtcSignalingMessenger(self_id_);
+}
+
+void WebRtcMedium::CreatePeerConnection(
+    webrtc::PeerConnectionObserver* observer, PeerConnectionCallback callback) {
+  auto& env = MediumEnvironment::Instance();
+  if (!env.GetUseValidPeerConnection()) {
+    callback(nullptr);
+    return;
+  }
+
+  webrtc::PeerConnectionInterface::RTCConfiguration rtc_config;
+  webrtc::PeerConnectionDependencies dependencies(observer);
+
+  signaling_thread_ = rtc::Thread::Create();
+  signaling_thread_->SetName("signaling_thread", nullptr);
+  RTC_CHECK(signaling_thread_->Start()) << "Failed to start thread";
+
+  webrtc::PeerConnectionFactoryDependencies factory_dependencies;
+  factory_dependencies.task_queue_factory =
+      webrtc::CreateDefaultTaskQueueFactory();
+  factory_dependencies.signaling_thread = signaling_thread_.get();
+
+  callback(webrtc::CreateModularPeerConnectionFactory(
+               std::move(factory_dependencies))
+               ->CreatePeerConnection(rtc_config, std::move(dependencies)));
+}
+
+std::unique_ptr<api::WebRtcSignalingMessenger>
+WebRtcMedium::GetSignalingMessenger(
+    absl::string_view self_id, const connections::LocationHint& location_hint) {
+  return std::make_unique<WebRtcSignalingMessenger>(self_id, location_hint);
+}
+
+}  // namespace g3
+}  // namespace nearby
+}  // namespace location

cpp/platform/impl/g3/webrtc.h

@@ -0,0 +1,73 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef PLATFORM_IMPL_G3_WEBRTC_H_
+#define PLATFORM_IMPL_G3_WEBRTC_H_
+
+#include <memory>
+
+#include "platform/api/webrtc.h"
+#include "absl/strings/string_view.h"
+#include "webrtc/api/peer_connection_interface.h"
+
+namespace location {
+namespace nearby {
+namespace g3 {
+
+class WebRtcSignalingMessenger : public api::WebRtcSignalingMessenger {
+ public:
+  using OnSignalingMessageCallback =
+      api::WebRtcSignalingMessenger::OnSignalingMessageCallback;
+
+  explicit WebRtcSignalingMessenger(
+      absl::string_view self_id,
+      const connections::LocationHint& location_hint);
+  ~WebRtcSignalingMessenger() override = default;
+
+  bool SendMessage(absl::string_view peer_id,
+                   const ByteArray& message) override;
+  bool StartReceivingMessages(OnSignalingMessageCallback listener) override;
+  void StopReceivingMessages() override;
+
+ private:
+  absl::string_view self_id_;
+  connections::LocationHint location_hint_;
+};
+
+class WebRtcMedium : public api::WebRtcMedium {
+ public:
+  using PeerConnectionCallback = api::WebRtcMedium::PeerConnectionCallback;
+
+  WebRtcMedium() = default;
+  ~WebRtcMedium() override = default;
+
+  // Creates and returns a new webrtc::PeerConnectionInterface object via
+  // |callback|.
+  void CreatePeerConnection(webrtc::PeerConnectionObserver* observer,
+                            PeerConnectionCallback callback) override;
+
+  // Returns a signaling messenger for sending WebRTC signaling messages.
+  std::unique_ptr<api::WebRtcSignalingMessenger> GetSignalingMessenger(
+      absl::string_view self_id,
+      const connections::LocationHint& location_hint) override;
+
+ private:
+  std::unique_ptr<rtc::Thread> signaling_thread_;
+};
+
+}  // namespace g3
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_IMPL_G3_WEBRTC_H_

cpp/platform/impl/g3/wifi_lan.cc

@@ -0,0 +1,379 @@
+// 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/impl/g3/wifi_lan.h"
+
+#include <iostream>
+#include <memory>
+#include <string>
+
+#include "platform/api/wifi_lan.h"
+#include "platform/base/logging.h"
+#include "platform/base/medium_environment.h"
+#include "platform/base/prng.h"
+#include "absl/synchronization/mutex.h"
+
+namespace location {
+namespace nearby {
+namespace g3 {
+
+WifiLanSocket::~WifiLanSocket() {
+  absl::MutexLock lock(&mutex_);
+  DoClose();
+}
+
+void WifiLanSocket::Connect(WifiLanSocket& other) {
+  absl::MutexLock lock(&mutex_);
+  remote_socket_ = &other;
+  input_ = other.output_;
+}
+
+InputStream& WifiLanSocket::GetInputStream() {
+  auto* remote_socket = GetRemoteSocket();
+  CHECK(remote_socket != nullptr);
+  return remote_socket->GetLocalInputStream();
+}
+
+OutputStream& WifiLanSocket::GetOutputStream() {
+  return GetLocalOutputStream();
+}
+
+WifiLanSocket* WifiLanSocket::GetRemoteSocket() {
+  absl::MutexLock lock(&mutex_);
+  return remote_socket_;
+}
+
+bool WifiLanSocket::IsConnected() const {
+  absl::MutexLock lock(&mutex_);
+  return IsConnectedLocked();
+}
+
+bool WifiLanSocket::IsClosed() const {
+  absl::MutexLock lock(&mutex_);
+  return closed_;
+}
+
+Exception WifiLanSocket::Close() {
+  absl::MutexLock lock(&mutex_);
+  DoClose();
+  return {Exception::kSuccess};
+}
+
+WifiLanService* WifiLanSocket::GetRemoteWifiLanService() {
+  absl::MutexLock lock(&mutex_);
+  return service_;
+}
+
+void WifiLanSocket::DoClose() {
+  if (!closed_) {
+    remote_socket_ = nullptr;
+    output_->GetOutputStream().Close();
+    output_->GetInputStream().Close();
+    if (IsConnectedLocked()) {
+      input_->GetOutputStream().Close();
+      input_->GetInputStream().Close();
+    }
+    closed_ = true;
+  }
+}
+
+bool WifiLanSocket::IsConnectedLocked() const { return input_ != nullptr; }
+
+InputStream& WifiLanSocket::GetLocalInputStream() {
+  absl::MutexLock lock(&mutex_);
+  return output_->GetInputStream();
+}
+
+OutputStream& WifiLanSocket::GetLocalOutputStream() {
+  absl::MutexLock lock(&mutex_);
+  return output_->GetOutputStream();
+}
+
+std::unique_ptr<api::WifiLanSocket> WifiLanServerSocket::Accept(
+    WifiLanService* service) {
+  absl::MutexLock lock(&mutex_);
+  if (closed_) return {};
+  while (pending_sockets_.empty()) {
+    cond_.Wait(&mutex_);
+    if (closed_) break;
+  }
+  if (closed_) return {};
+  auto* remote_socket =
+      pending_sockets_.extract(pending_sockets_.begin()).value();
+  CHECK(remote_socket);
+  auto local_socket = std::make_unique<WifiLanSocket>(service);
+  local_socket->Connect(*remote_socket);
+  remote_socket->Connect(*local_socket);
+  cond_.SignalAll();
+  return local_socket;
+}
+
+bool WifiLanServerSocket::Connect(WifiLanSocket& socket) {
+  absl::MutexLock lock(&mutex_);
+  if (closed_) return false;
+  if (socket.IsConnected()) {
+    NEARBY_LOG(ERROR,
+               "Failed to connect to WifiLan server socket: already connected");
+    return true;  // already connected.
+  }
+  // add client socket to the pending list
+  pending_sockets_.emplace(&socket);
+  cond_.SignalAll();
+  while (!socket.IsConnected()) {
+    cond_.Wait(&mutex_);
+    if (closed_) return false;
+  }
+  return true;
+}
+
+void WifiLanServerSocket::SetCloseNotifier(std::function<void()> notifier) {
+  absl::MutexLock lock(&mutex_);
+  close_notifier_ = std::move(notifier);
+}
+
+WifiLanServerSocket::~WifiLanServerSocket() {
+  absl::MutexLock lock(&mutex_);
+  DoClose();
+}
+
+Exception WifiLanServerSocket::Close() {
+  absl::MutexLock lock(&mutex_);
+  return DoClose();
+}
+
+Exception WifiLanServerSocket::DoClose() {
+  bool should_notify = !closed_;
+  closed_ = true;
+  if (should_notify) {
+    cond_.SignalAll();
+    if (close_notifier_) {
+      auto notifier = std::move(close_notifier_);
+      mutex_.Unlock();
+      // Notifier may contain calls to public API, and may cause deadlock, if
+      // mutex_ is held during the call.
+      notifier();
+      mutex_.Lock();
+    }
+  }
+  return {Exception::kSuccess};
+}
+
+WifiLanMedium::WifiLanMedium() {
+  service_.SetMedium(this);
+  std::string ip_address;
+  ip_address.resize(4);
+  uint32_t raw_ip_addr = Prng().NextUint32();
+  uint16_t port = Prng().NextUint32();
+  ip_address[0] = static_cast<char>(raw_ip_addr >> 24);
+  ip_address[1] = static_cast<char>(raw_ip_addr >> 16);
+  ip_address[2] = static_cast<char>(raw_ip_addr >> 8);
+  ip_address[3] = static_cast<char>(raw_ip_addr >> 0);
+  service_.SetServiceAddress(ip_address, port);
+  auto& env = MediumEnvironment::Instance();
+  env.RegisterWifiLanMedium(*this);
+}
+
+WifiLanMedium::~WifiLanMedium() {
+  service_.SetMedium(nullptr);
+  auto& env = MediumEnvironment::Instance();
+  env.UnregisterWifiLanMedium(*this);
+
+  StopAdvertising(advertising_info_.service_id);
+  StopDiscovery(discovering_info_.service_id);
+
+  NEARBY_LOG(INFO, "WifiLanMedium dtor advertising_accept_thread_running_ = %d",
+             acceptance_thread_running_.load());
+  // If acceptance thread is still running, wait to finish.
+  if (acceptance_thread_running_) {
+    while (acceptance_thread_running_) {
+      CountDownLatch latch(1);
+      close_accept_loops_runner_.Execute([&latch]() { latch.CountDown(); });
+      latch.Await();
+    }
+  }
+}
+
+bool WifiLanMedium::StartAdvertising(const std::string& service_id,
+                                     const std::string& service_info_name,
+                                     const std::string& endpoint_info_name) {
+  NEARBY_LOG(INFO,
+             "G3 WifiLan StartAdvertising: service_id=%s, service_info_name=%s",
+             service_id.c_str(), service_info_name.c_str());
+  auto& env = MediumEnvironment::Instance();
+  service_.SetServiceName(service_info_name);
+  service_.SetTxtRecord("n", endpoint_info_name);
+  env.UpdateWifiLanMediumForAdvertising(*this, service_, service_id, true);
+
+  absl::MutexLock lock(&mutex_);
+  if (server_socket_ != nullptr) server_socket_.release();
+  server_socket_ = std::make_unique<WifiLanServerSocket>();
+
+  acceptance_thread_running_.exchange(true);
+  accept_loops_runner_.Execute([&env, this, service_id]() mutable {
+    if (!accept_loops_runner_.InShutdown()) {
+      while (true) {
+        auto client_socket = server_socket_->Accept(&service_);
+        if (client_socket == nullptr) break;
+        env.CallWifiLanAcceptedConnectionCallback(
+            *this, *(client_socket.release()), service_id);
+      }
+    }
+    acceptance_thread_running_.exchange(false);
+  });
+  advertising_info_.service_id = service_id;
+  return true;
+}
+
+bool WifiLanMedium::StopAdvertising(const std::string& service_id) {
+  NEARBY_LOG(INFO, "G3 WifiLan StopAdvertising: service_id=%s",
+             service_id.c_str());
+  {
+    absl::MutexLock lock(&mutex_);
+    if (advertising_info_.Empty()) {
+      NEARBY_LOG(INFO,
+                 "G3 WifiLan StopAdvertising: Can't stop advertising because "
+                 "we never started advertising.");
+      return false;
+    }
+    advertising_info_.Clear();
+  }
+
+  auto& env = MediumEnvironment::Instance();
+  env.UpdateWifiLanMediumForAdvertising(*this, service_, service_id, false);
+  accept_loops_runner_.Shutdown();
+  if (server_socket_ == nullptr) {
+    NEARBY_LOGS(ERROR) << "G3 WifiLan StopAdvertising: failed to find WifiLan "
+                          "Server socket: service_id="
+                       << service_id;
+    // Fall through for server socket not found.
+    return true;
+  }
+
+  if (!server_socket_->Close().Ok()) {
+    NEARBY_LOG(INFO,
+               "G3 WifiLan StopAdvertising: Failed to close WifiLan server "
+               "socket for %s.",
+               service_id.c_str());
+    return false;
+  }
+
+  return true;
+}
+
+bool WifiLanMedium::StartDiscovery(const std::string& service_id,
+                                   DiscoveredServiceCallback callback) {
+  NEARBY_LOG(INFO, "G3 WifiLan StartDiscovery: service_id=%s",
+             service_id.c_str());
+  auto& env = MediumEnvironment::Instance();
+  env.UpdateWifiLanMediumForDiscovery(*this, service_id, std::move(callback),
+                                      true);
+  {
+    absl::MutexLock lock(&mutex_);
+    discovering_info_.service_id = service_id;
+  }
+  return true;
+}
+
+bool WifiLanMedium::StopDiscovery(const std::string& service_id) {
+  NEARBY_LOG(INFO, "G3 WifiLan StopDiscovery: service_id=%s",
+             service_id.c_str());
+  {
+    absl::MutexLock lock(&mutex_);
+    if (discovering_info_.Empty()) {
+      NEARBY_LOG(INFO,
+                 "G3 WifiLan StopDiscovery: Can't stop discovering because we "
+                 "never started discovering.");
+      return false;
+    }
+    discovering_info_.Clear();
+  }
+
+  auto& env = MediumEnvironment::Instance();
+  env.UpdateWifiLanMediumForDiscovery(*this, service_id, {}, false);
+  return true;
+}
+
+bool WifiLanMedium::StartAcceptingConnections(
+    const std::string& service_id, AcceptedConnectionCallback callback) {
+  NEARBY_LOG(INFO, "G3 WifiLan StartAcceptingConnections: service_id=%s",
+             service_id.c_str());
+  auto& env = MediumEnvironment::Instance();
+  env.UpdateWifiLanMediumForAcceptedConnection(*this, service_id, callback);
+  return true;
+}
+
+bool WifiLanMedium::StopAcceptingConnections(const std::string& service_id) {
+  NEARBY_LOG(INFO, "G3 WifiLan StopAcceptingConnections: service_id=%s",
+             service_id.c_str());
+  auto& env = MediumEnvironment::Instance();
+  env.UpdateWifiLanMediumForAcceptedConnection(*this, service_id, {});
+  return true;
+}
+
+std::unique_ptr<api::WifiLanSocket> WifiLanMedium::Connect(
+    api::WifiLanService& remote_service, const std::string& service_id) {
+  NEARBY_LOG(INFO,
+             "G3 WifiLan Connect: medium=%p, service=%p, service_info_name=%s, "
+             "service_id=%s",
+             this, &service_, remote_service.GetServiceName().c_str(),
+             service_id.c_str());
+  // First, find an instance of remote medium, that exposed this service.
+  auto* medium = static_cast<WifiLanService&>(remote_service).GetMedium();
+
+  if (!medium) return {};  // Can't find medium. Bail out.
+
+  WifiLanServerSocket* remote_server_socket = nullptr;
+  NEARBY_LOG(INFO,
+             "G3 WifiLan Connect [peer]: medium=%p, service=%p, "
+             "service_info_name=%s, service_id=%s",
+             medium, &remote_service, remote_service.GetServiceName().c_str(),
+             service_id.c_str());
+  // Then, find our server socket context in this medium.
+  {
+    absl::MutexLock medium_lock(&medium->mutex_);
+    remote_server_socket = medium->server_socket_.get();
+    if (remote_server_socket == nullptr) {
+      NEARBY_LOG(ERROR,
+                 "G3 WifiLan Connect: Failed to find WifiLan Server socket: "
+                 "service_id=%s",
+                 service_id.c_str());
+      // Fall through for server socket not found.
+      return {};
+    }
+  }
+
+  WifiLanService service = static_cast<WifiLanService&>(remote_service);
+  auto socket = std::make_unique<WifiLanSocket>(&service);
+  // Finally, Request to connect to this socket.
+  if (!remote_server_socket->Connect(*socket)) {
+    NEARBY_LOG(ERROR,
+               "G3 WifiLan Connect: Failed to connect to existing WifiLan "
+               "Server socket: service_id=%s",
+               service_id.c_str());
+    return {};
+  }
+
+  NEARBY_LOG(INFO, "G3 WifiLan Connect: connected: socket=%p", socket.get());
+  return socket;
+}
+
+api::WifiLanService* WifiLanMedium::FindRemoteService(
+    const std::string& ip_address, int port) {
+  auto& env = MediumEnvironment::Instance();
+  return env.FindWifiLanService(ip_address, port);
+}
+
+}  // namespace g3
+}  // namespace nearby
+}  // namespace location

cpp/platform/impl/g3/wifi_lan.h

@@ -0,0 +1,273 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef PLATFORM_IMPL_G3_WIFI_LAN_H_
+#define PLATFORM_IMPL_G3_WIFI_LAN_H_
+
+#include <memory>
+#include <string>
+#include <utility>
+
+#include "platform/api/wifi_lan.h"
+#include "platform/base/byte_array.h"
+#include "platform/base/input_stream.h"
+#include "platform/base/output_stream.h"
+#include "platform/impl/g3/multi_thread_executor.h"
+#include "platform/impl/g3/pipe.h"
+#include "absl/container/flat_hash_map.h"
+#include "absl/container/flat_hash_set.h"
+#include "absl/synchronization/mutex.h"
+
+namespace location {
+namespace nearby {
+namespace g3 {
+
+class WifiLanMedium;
+
+// Opaque wrapper over a WifiLan service which contains packed
+// |WifiLanServiceInfo| string name.
+class WifiLanService : public api::WifiLanService {
+ public:
+  explicit WifiLanService(std::string service_info_name)
+      : service_info_name_(std::move(service_info_name)) {}
+  ~WifiLanService() override = default;
+
+  std::string GetServiceName() const override { return service_info_name_; }
+
+  void SetServiceName(std::string service_info_name) {
+    service_info_name_ = std::move(service_info_name);
+  }
+
+  std::string GetTxtRecord(const std::string& txt_record_key) const override {
+    if (txt_records_.empty()) return {};
+    auto record = txt_records_.find(txt_record_key);
+    if (record == txt_records_.end()) return {};
+    return record->second;
+  }
+
+  void SetTxtRecord(const std::string& txt_record_key,
+                    const std::string& txt_record_value) {
+    txt_records_.emplace(txt_record_key, txt_record_value);
+  }
+
+  std::pair<std::string, int> GetServiceAddress() const override {
+    return std::make_pair(ip_address_, port_);
+  }
+
+  void SetServiceAddress(const std::string& ip_address, int port) {
+    ip_address_ = ip_address;
+    port_ = port;
+  }
+
+  WifiLanMedium* GetMedium() { return medium_; }
+
+  void SetMedium(WifiLanMedium* medium) { medium_ = medium; }
+
+ private:
+  std::string service_info_name_;
+  absl::flat_hash_map<std::string, std::string> txt_records_;
+  WifiLanMedium* medium_ = nullptr;
+  std::string ip_address_;
+  int port_;
+};
+
+class WifiLanSocket : public api::WifiLanSocket {
+ public:
+  WifiLanSocket() = default;
+  explicit WifiLanSocket(WifiLanService* service) : service_(service) {}
+  ~WifiLanSocket() override;
+
+  // Connect to another WifiLanSocket, to form a functional low-level channel.
+  // from this point on, and until Close is called, connection exists.
+  void Connect(WifiLanSocket& other) ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns the InputStream of this connected WifiLanSocket.
+  InputStream& GetInputStream() override ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns the OutputStream of this connected WifiLanSocket.
+  // This stream is for local side to write.
+  OutputStream& GetOutputStream() override ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns address of a remote WifiLanSocket or nullptr.
+  WifiLanSocket* GetRemoteSocket() ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns true if connection exists to the (possibly closed) remote socket.
+  bool IsConnected() const ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns true if socket is closed.
+  bool IsClosed() const ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns Exception::kIo on error, Exception::kSuccess otherwise.
+  Exception Close() override ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns valid WifiLanService pointer if there is a connection, and
+  // nullptr otherwise.
+  WifiLanService* GetRemoteWifiLanService() override
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+ private:
+  void DoClose() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
+  // Returns true if connection exists to the (possibly closed) remote socket.
+  bool IsConnectedLocked() const ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
+  // Returns InputStream of our side of a connection.
+  // This is what the remote side is supposed to read from.
+  // This is a helper for GetInputStream() method.
+  InputStream& GetLocalInputStream() ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns OutputStream of our side of a connection.
+  // This is what the local size is supposed to write to.
+  // This is a helper for GetOutputStream() method.
+  OutputStream& GetLocalOutputStream() ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Output pipe is initialized by constructor, it remains always valid, until
+  // it is closed. it represents output part of a local socket. Input part of a
+  // local socket comes from the peer socket, after connection.
+  std::shared_ptr<Pipe> output_{new Pipe};
+  std::shared_ptr<Pipe> input_;
+  mutable absl::Mutex mutex_;
+  WifiLanService* service_;
+  WifiLanSocket* remote_socket_ ABSL_GUARDED_BY(mutex_) = nullptr;
+  bool closed_ ABSL_GUARDED_BY(mutex_) = false;
+};
+
+class WifiLanServerSocket {
+ public:
+  ~WifiLanServerSocket();
+
+  // Blocks until either:
+  // - at least one incoming connection request is available, or
+  // - ServerSocket is closed.
+  // On success, returns connected socket, ready to exchange data.
+  // Returns nullptr on error.
+  // Once error is reported, it is permanent, and ServerSocket has to be closed.
+  //
+  // Called by the server side of a connection.
+  // Returns WifiLanSocket to the server side.
+  // If not null, returned socket is connected to its remote (client-side) peer.
+  std::unique_ptr<api::WifiLanSocket> Accept(WifiLanService* service)
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Blocks until either:
+  // - connection is available, or
+  // - server socket is closed, or
+  // - error happens.
+  //
+  // Called by the client side of a connection.
+  // Returns true, if socket is successfully connected.
+  bool Connect(WifiLanSocket& socket) ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Called by the server side of a connection before passing ownership of
+  // WifiLanServerSocker to user, to track validity of a pointer to this
+  // server socket,
+  void SetCloseNotifier(std::function<void()> notifier)
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns Exception::kIo on error, Exception::kSuccess otherwise.
+  // Calls close_notifier if it was previously set, and marks socket as closed.
+  Exception Close() ABSL_LOCKS_EXCLUDED(mutex_);
+
+ private:
+  Exception DoClose() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
+  absl::Mutex mutex_;
+  absl::CondVar cond_;
+  absl::flat_hash_set<WifiLanSocket*> pending_sockets_ ABSL_GUARDED_BY(mutex_);
+  std::function<void()> close_notifier_ ABSL_GUARDED_BY(mutex_);
+  bool closed_ ABSL_GUARDED_BY(mutex_) = false;
+};
+
+// Container of operations that can be performed over the WifiLan medium.
+class WifiLanMedium : public api::WifiLanMedium {
+ public:
+  WifiLanMedium();
+  ~WifiLanMedium() override;
+
+  bool StartAdvertising(const std::string& service_id,
+                        const std::string& service_info_name,
+                        const std::string& endpoint_info_name) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
+  bool StopAdvertising(const std::string& service_id) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns true once the WifiLan discovery has been initiated.
+  bool StartDiscovery(const std::string& service_id,
+                      DiscoveredServiceCallback callback) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns true once WifiLan discovery for service_id is well and truly
+  // stopped; after this returns, there must be no more invocations of the
+  // DiscoveredServiceCallback passed in to StartDiscovery() for service_id.
+  bool StopDiscovery(const std::string& service_id) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Returns true once WifiLan socket connection requests to service_id can be
+  // accepted.
+  bool StartAcceptingConnections(const std::string& service_id,
+                                 AcceptedConnectionCallback callback) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
+  bool StopAcceptingConnections(const std::string& service_id) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Connects to existing remote WifiLan service.
+  //
+  // On success, returns a new WifiLanSocket.
+  // On error, returns nullptr.
+  std::unique_ptr<api::WifiLanSocket> Connect(
+      api::WifiLanService& remote_service,
+      const std::string& service_id) override ABSL_LOCKS_EXCLUDED(mutex_);
+
+  api::WifiLanService* FindRemoteService(const std::string& ip_address,
+                                         int port) override;
+
+ private:
+  static constexpr int kMaxConcurrentAcceptLoops = 5;
+
+  struct AdvertisingInfo {
+    bool Empty() const { return service_id.empty(); }
+    void Clear() { service_id.clear(); }
+
+    std::string service_id;
+  };
+
+  struct DiscoveringInfo {
+    bool Empty() const { return service_id.empty(); }
+    void Clear() { service_id.clear(); }
+
+    std::string service_id;
+  };
+
+  absl::Mutex mutex_;
+  WifiLanService service_{"unknown G3 WifiLan service"};
+
+  // A thread pool dedicated to running all the accept loops from
+  // StartAdvertising().
+  MultiThreadExecutor accept_loops_runner_{kMaxConcurrentAcceptLoops};
+  std::atomic_bool acceptance_thread_running_ = false;
+
+  // A thread pool dedicated to wait to complete the accept_loops_runner_.
+  MultiThreadExecutor close_accept_loops_runner_{kMaxConcurrentAcceptLoops};
+
+  // A server socket is established when start advertising.
+  std::unique_ptr<WifiLanServerSocket> server_socket_;
+  AdvertisingInfo advertising_info_ ABSL_GUARDED_BY(mutex_);
+  DiscoveringInfo discovering_info_ ABSL_GUARDED_BY(mutex_);
+};
+
+}  // namespace g3
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_IMPL_G3_WIFI_LAN_H_