Migrate DataManager to third_party/nearby/internal

PiperOrigin-RevId: 544395891

Package.swift

@@ -410,6 +410,7 @@ let package = Package(
         "internal/flags/BUILD",
         "internal/network/BUILD",
         "internal/base/BUILD",
+        "internal/data/BUILD",
         "internal/test/BUILD",
         // tests
         "connections/listeners_test.cc",
@@ -476,6 +477,8 @@ let package = Package(
         "internal/crypto/sha2_unittest.cc",
         "internal/crypto/signature_verifier_unittest.cc",
         "internal/crypto/symmetric_key_unittest.cc",
+        "internal/data/leveldb_data_set_test.cc",
+        "internal/data/memory_data_set_test.cc",
         "internal/flags/nearby_flags_test.cc",
         "internal/proto/analytics/connections_log_test.cc",
         "internal/platform/feature_flags_test.cc",
@@ -539,6 +542,7 @@ let package = Package(
         "internal/test/fake_timer_test.cc",
         "internal/test/fake_device_info_test.cc",
         "internal/test/fake_task_runner_test.cc",
+        "internal/test/fake_data_set_test.cc",
         "internal/weave/base_socket_test.cc",
         "internal/weave/control_packet_write_request_test.cc",
         "internal/weave/message_write_request_test.cc",
@@ -552,6 +556,7 @@ let package = Package(
         "connections/implementation/proto",
         "internal/proto",
         "proto",
+        "internal/data/leveldb_data_set_test.proto",
         // webrtc
         "connections/implementation/webrtc_bwu_handler.cc",
         "connections/implementation/webrtc_endpoint_channel.cc",

internal/data/BUILD

@@ -0,0 +1,58 @@
+load("@rules_cc//cc:defs.bzl", "cc_proto_library")
+
+licenses(["notice"])
+
+package(default_visibility = [
+    "//visibility:public",
+])
+
+cc_library(
+    name = "data_manager",
+    hdrs = [
+        "data_manager.h",
+        "data_set.h",
+        "leveldb_data_set.h",
+        "memory_data_set.h",
+    ],
+    deps = [
+        "//internal/platform:logging",
+        "//third_party/leveldb:db",
+        "//third_party/leveldb:table",
+        "//third_party/leveldb:util",
+        "//third_party/protobuf:protobuf_lite",
+        "@com_google_absl//absl/container:flat_hash_map",
+        "@com_google_absl//absl/strings",
+        "@com_google_absl//absl/synchronization",
+    ],
+)
+
+proto_library(
+    name = "leveldb_data_set_test_proto",
+    srcs = ["leveldb_data_set_test.proto"],
+)
+
+cc_proto_library(
+    name = "leveldb_data_set_test_cc_proto",
+    deps = [":leveldb_data_set_test_proto"],
+)
+
+cc_test(
+    name = "data_manager_test",
+    size = "small",
+    timeout = "short",
+    srcs = [
+        "leveldb_data_set_test.cc",
+        "memory_data_set_test.cc",
+    ],
+    shard_count = 8,
+    deps = [
+        ":data_manager",
+        ":leveldb_data_set_test_cc_proto",
+        "//internal/platform/implementation/g3",  # fixdeps: keep
+        "@com_github_protobuf_matchers//protobuf-matchers",
+        "@com_google_absl//absl/container:flat_hash_map",
+        "@com_google_absl//absl/synchronization",
+        "@com_google_absl//absl/time",
+        "@com_google_googletest//:gtest_main",
+    ],
+)

internal/data/data_manager.h

@@ -0,0 +1,53 @@
+// Copyright 2022 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 THIRD_PARTY_NEARBY_INTERNAL_DATA_DATA_MANAGER_H_
+#define THIRD_PARTY_NEARBY_INTERNAL_DATA_DATA_MANAGER_H_
+
+#include <memory>
+
+#include "absl/strings/string_view.h"
+#include "internal/data/data_set.h"
+#include "internal/data/leveldb_data_set.h"
+#include "internal/data/memory_data_set.h"
+
+namespace nearby {
+namespace data {
+
+class DataManager {
+ public:
+  enum class DataStorageType : int { kMemory = 0, kLevelDb = 1 };
+  explicit DataManager(DataStorageType data_storage_type)
+      : data_storage_type_(data_storage_type) {}
+  ~DataManager() = default;
+
+  template <typename T>
+  std::unique_ptr<DataSet<T>> GetDataSet(absl::string_view path) {
+    if (data_storage_type_ == DataStorageType::kMemory) {
+      return std::make_unique<MemoryDataSet<T>>(path);
+    } else if (data_storage_type_ == DataStorageType::kLevelDb) {
+      return std::make_unique<LeveldbDataSet<T>>(path);
+    } else {
+      return nullptr;
+    }
+  }
+
+ private:
+  DataStorageType data_storage_type_;
+};
+
+}  // namespace data
+}  // namespace nearby
+
+#endif  // THIRD_PARTY_NEARBY_INTERNAL_DATA_DATA_MANAGER_H_

internal/data/data_set.h

@@ -0,0 +1,70 @@
+// Copyright 2022 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 THIRD_PARTY_NEARBY_INTERNAL_DATA_DATA_SET_H_
+#define THIRD_PARTY_NEARBY_INTERNAL_DATA_DATA_SET_H_
+
+#include <functional>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+namespace nearby {
+namespace data {
+
+enum class InitStatus {
+  kOK = 0,
+  kNotInitialized = 1,
+  kError = 2,
+  kCorrupt = 3,
+  kInvalidOperation = 4,
+  kMaxValue = kInvalidOperation
+};
+
+template <typename T>
+class DataSet {
+ public:
+  using KeyEntryVector = std::vector<std::pair<std::string, T>>;
+
+  virtual ~DataSet() = default;
+
+  // Asynchronously initializes the object, which must have been created by the
+  // DataManager::GetDataSet<T> function. |callback| will be invoked on the
+  // calling thread when complete.
+  virtual void Initialize(std::function<void(InitStatus)> callback) = 0;
+
+  // Asynchronously loads all entries from the database and invokes |callback|
+  // when complete.
+  virtual void LoadEntries(
+      std::function<void(bool, std::unique_ptr<std::vector<T>>)> callback) = 0;
+
+  // Asynchronously saves |entries_to_save| and deletes entries from
+  // |keys_to_remove| from the database. |callback| will be invoked on the
+  // calling thread when complete. |entries_to_save| and |keys_to_remove| must
+  // be non-null.
+  virtual void UpdateEntries(
+      std::unique_ptr<KeyEntryVector> entries_to_save,
+      std::unique_ptr<std::vector<std::string>> keys_to_remove,
+      std::function<void(bool)> callback) = 0;
+
+  // Asynchronously destroys the database. Use this call only if the database
+  // needs to be destroyed for this particular profile.
+  virtual void Destroy(std::function<void(bool)> callback) = 0;
+};
+
+}  // namespace data
+}  // namespace nearby
+
+#endif  // THIRD_PARTY_NEARBY_INTERNAL_DATA_DATA_SET_H_

internal/data/leveldb_data_set.h

@@ -0,0 +1,226 @@
+// Copyright 2022 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 THIRD_PARTY_NEARBY_INTERNAL_DATA_LEVELDB_DATA_SET_H_
+#define THIRD_PARTY_NEARBY_INTERNAL_DATA_LEVELDB_DATA_SET_H_
+
+#include <functional>
+#include <memory>
+#include <ostream>
+#include <string>
+#include <type_traits>
+#include <utility>
+#include <vector>
+
+#include "absl/strings/string_view.h"
+#include "third_party/leveldb/include/db.h"
+#include "third_party/leveldb/include/iterator.h"
+#include "third_party/leveldb/include/options.h"
+#include "third_party/leveldb/include/slice.h"
+#include "third_party/leveldb/include/status.h"
+#include "internal/data/data_set.h"
+#include "internal/platform/logging.h"
+#include "third_party/protobuf/message_lite.h"
+
+namespace nearby {
+namespace data {
+// DataSet implementation using leveldb as its persistent storage. Values are
+// serialized and stored in leveldb databases.
+template <typename T,
+          std::enable_if_t<std::is_base_of<proto2::MessageLite, T>::value,
+                           bool> = true>
+class LeveldbDataSet : public DataSet<T> {
+ public:
+  using KeyEntryVector = std::vector<std::pair<std::string, T>>;
+
+  explicit LeveldbDataSet(absl::string_view path) : path_(path) {}
+  ~LeveldbDataSet() override = default;
+
+  void Initialize(std::function<void(InitStatus)> callback) override;
+  void LoadEntries(std::function<void(bool, std::unique_ptr<std::vector<T>>)>
+                       callback) override;
+  void LoadEntriesWithKeys(
+      std::function<
+          void(bool, std::unique_ptr<std::vector<std::pair<std::string, T>>>)>
+          callback);
+  void UpdateEntries(std::unique_ptr<KeyEntryVector> entries_to_save,
+                     std::unique_ptr<std::vector<std::string>> keys_to_remove,
+                     std::function<void(bool)> callback) override;
+  void Destroy(std::function<void(bool)> callback) override;
+
+ private:
+  void Serialize(T const& value, std::string& str);
+  void Deserialize(absl::string_view str, T& value);
+
+ private:
+  std::string path_;
+  std::unique_ptr<leveldb::DB> db_ = nullptr;
+  InitStatus status_ = InitStatus::kNotInitialized;
+};
+
+template <typename T,
+          std::enable_if_t<std::is_base_of<proto2::MessageLite, T>::value, bool>
+              isMessageLite>
+void LeveldbDataSet<T, isMessageLite>::Initialize(
+    std::function<void(InitStatus)> callback) {
+  leveldb::Options options;
+  options.create_if_missing = true;
+
+  leveldb::DB* db;
+  leveldb::Status status = leveldb::DB::Open(options, path_, &db);
+  db_ = std::unique_ptr<leveldb::DB>(db);
+
+  if (status.ok()) {
+    status_ = InitStatus::kOK;
+    NEARBY_LOGS(INFO) << "Database is initialized successfully..";
+  } else if (status.IsCorruption() || status.IsIOError()) {
+    status_ = InitStatus::kCorrupt;
+    NEARBY_LOGS(INFO) << "Database is corrupt.";
+
+  } else {
+    status_ = InitStatus::kError;
+    NEARBY_LOGS(INFO) << "Failed to initialize database due to unknown error.";
+  }
+  std::move(callback)(status_);
+}
+
+template <typename T,
+          std::enable_if_t<std::is_base_of<proto2::MessageLite, T>::value, bool>
+              isMessageLite>
+void LeveldbDataSet<T, isMessageLite>::LoadEntries(
+    std::function<void(bool, std::unique_ptr<std::vector<T>>)> callback) {
+  auto result = std::make_unique<std::vector<T>>();
+  if (status_ != InitStatus::kOK) {
+    std::move(callback)(false, std::move(result));
+    return;
+  }
+
+  std::unique_ptr<leveldb::Iterator> it(
+      db_->NewIterator(leveldb::ReadOptions()));
+
+  for (it->SeekToFirst(); it->Valid(); it->Next()) {
+    T value;
+    Deserialize(it->value().ToString(), value);
+    result->push_back(value);
+  }
+
+  if (it->status().ok()) {
+    NEARBY_LOGS(INFO) << "Loaded " << result->size()
+                      << " entries from database.";
+    std::move(callback)(true, std::move(result));
+  } else {
+    NEARBY_LOGS(INFO) << "Failed to load entries from database.";
+    result->clear();
+    std::move(callback)(false, std::move(result));
+  }
+}
+
+template <typename T,
+          std::enable_if_t<std::is_base_of<proto2::MessageLite, T>::value, bool>
+              isMessageLite>
+void LeveldbDataSet<T, isMessageLite>::LoadEntriesWithKeys(
+    std::function<void(bool,
+                       std::unique_ptr<std::vector<std::pair<std::string, T>>>)>
+        callback) {
+  auto result = std::make_unique<std::vector<std::pair<std::string, T>>>();
+  if (status_ != InitStatus::kOK) {
+    std::move(callback)(false, std::move(result));
+    return;
+  }
+
+  std::unique_ptr<leveldb::Iterator> it(
+      db_->NewIterator(leveldb::ReadOptions()));
+
+  for (it->SeekToFirst(); it->Valid(); it->Next()) {
+    T value;
+    Deserialize(it->value().ToString(), value);
+    result->push_back({it->key().ToString(), value});
+  }
+
+  if (it->status().ok()) {
+    NEARBY_LOGS(INFO) << "Loaded " << result->size()
+                      << " entries from database.";
+    std::move(callback)(true, std::move(result));
+  } else {
+    NEARBY_LOGS(INFO) << "Failed to load entries from database.";
+    result->clear();
+    std::move(callback)(false, std::move(result));
+  }
+}
+
+template <typename T,
+          std::enable_if_t<std::is_base_of<proto2::MessageLite, T>::value, bool>
+              isMessageLite>
+void LeveldbDataSet<T, isMessageLite>::UpdateEntries(
+    std::unique_ptr<KeyEntryVector> entries_to_save,
+    std::unique_ptr<std::vector<std::string>> keys_to_remove,
+    std::function<void(bool)> callback) {
+  NEARBY_LOGS(INFO) << "UpdateEntries is called.";
+  if (status_ != InitStatus::kOK) {
+    std::move(callback)(false);
+    return;
+  }
+
+  if (entries_to_save != nullptr) {
+    for (const auto& [key, value] : *entries_to_save) {
+      std::string str;
+      Serialize(value, str);
+      db_->Put(leveldb::WriteOptions(), key, leveldb::Slice(str));
+    }
+  }
+
+  if (keys_to_remove != nullptr) {
+    for (const auto& it : *keys_to_remove) {
+      db_->Delete(leveldb::WriteOptions(), it);
+    }
+  }
+
+  std::move(callback)(true);
+}
+
+template <typename T,
+          std::enable_if_t<std::is_base_of<proto2::MessageLite, T>::value, bool>
+              isMessageLite>
+void LeveldbDataSet<T, isMessageLite>::Destroy(
+    std::function<void(bool)> callback) {
+  NEARBY_LOGS(INFO) << "Destroy is called.";
+  db_.reset();
+  leveldb::DestroyDB(path_, leveldb::Options());
+  std::move(callback)(true);
+}
+
+// Functions for serializing/deserializing data values to/from strings. Strings
+// are used as a convenient container that manages its memory. They don't need
+// to be human-readable.
+
+template <typename T,
+          std::enable_if_t<std::is_base_of<proto2::MessageLite, T>::value, bool>
+              isMessageLite>
+void LeveldbDataSet<T, isMessageLite>::Serialize(T const& value,
+                                                 std::string& str) {
+  value.SerializeToString(&str);
+}
+
+template <typename T,
+          std::enable_if_t<std::is_base_of<proto2::MessageLite, T>::value, bool>
+              isMessageLite>
+void LeveldbDataSet<T, isMessageLite>::Deserialize(absl::string_view str,
+                                                   T& value) {
+  value.ParseFromString(str);
+}
+
+}  // namespace data
+}  // namespace nearby
+
+#endif  // THIRD_PARTY_NEARBY_INTERNAL_DATA_LEVELDB_DATA_SET_H_

internal/data/leveldb_data_set_test.cc

@@ -0,0 +1,259 @@
+// Copyright 2022 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 "internal/data/leveldb_data_set.h"
+
+#include <stdint.h>
+
+#include <filesystem>  // NOLINT(build/c++17)
+#include <ios>
+#include <memory>
+#include <random>
+#include <sstream>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "gmock/gmock.h"
+#include "protobuf-matchers/protocol-buffer-matchers.h"
+#include "gtest/gtest.h"
+#include "absl/container/flat_hash_map.h"
+#include "absl/synchronization/notification.h"
+#include "absl/time/time.h"
+#include "internal/data/data_set.h"
+#include "internal/data/leveldb_data_set_test.proto.h"
+
+namespace nearby {
+namespace data {
+namespace {
+
+using ::testing::SizeIs;
+
+// Generate a unique directory under temp directory for leveldb storage
+std::filesystem::path GenerateLeveldbPath() {
+  auto temp_directory_path = std::filesystem::temp_directory_path();
+  std::random_device dev;
+  std::mt19937 prng(dev());
+  std::uniform_int_distribution<uint64_t> rand(0);
+  std::filesystem::path path;
+  do {
+    std::stringstream leveldb_directory;
+    leveldb_directory << std::hex << "nearby_db_" << rand(prng);
+    path = temp_directory_path / leveldb_directory.str();
+  } while (std::filesystem::exists(path));
+  return path;
+}
+
+// Helper functions to synchronize LeveldbDataSet function calls for testing
+template <typename T>
+std::unique_ptr<LeveldbDataSet<T>> CreateDataSet(
+    const std::filesystem::path& path) {
+  return std::make_unique<LeveldbDataSet<T>>(path.string());
+}
+
+template <typename T>
+InitStatus InitializeAndWait(std::unique_ptr<LeveldbDataSet<T>>& dataset) {
+  InitStatus status;
+  absl::Notification notification;
+  dataset->Initialize([&notification, &status](InitStatus s) {
+    status = s;
+    notification.Notify();
+  });
+  notification.WaitForNotificationWithTimeout(absl::Seconds(5));
+  return status;
+}
+
+template <typename T>
+bool UpdateEntriesAndWait(
+    std::unique_ptr<LeveldbDataSet<T>>& dataset,
+    std::unique_ptr<typename LeveldbDataSet<T>::KeyEntryVector> entries_to_save,
+    std::unique_ptr<std::vector<std::string>> keys_to_remove) {
+  bool result = false;
+  absl::Notification notification;
+  dataset->UpdateEntries(std::move(entries_to_save), std::move(keys_to_remove),
+                         [&result, &notification](bool res) {
+                           result = res;
+                           notification.Notify();
+                         });
+  notification.WaitForNotificationWithTimeout(absl::Seconds(5));
+  return result;
+}
+
+template <typename T>
+std::unique_ptr<std::vector<T>> LoadEntriesAndWait(
+    std::unique_ptr<LeveldbDataSet<T>>& dataset) {
+  auto result = std::make_unique<std::vector<T>>();
+  absl::Notification notification;
+  dataset->LoadEntries(
+      [&result, &notification](bool, std::unique_ptr<std::vector<T>> res) {
+        for (const auto& it : *res) {
+          result->push_back(it);
+        }
+        notification.Notify();
+      });
+  notification.WaitForNotificationWithTimeout(absl::Seconds(5));
+  return result;
+}
+
+template <typename T>
+absl::flat_hash_map<std::string, T> LoadEntriesWithKeysAndWait(
+    std::unique_ptr<LeveldbDataSet<T>>& dataset) {
+  auto result = std::make_unique<std::vector<std::pair<std::string, T>>>();
+  absl::Notification notification;
+  dataset->LoadEntriesWithKeys(
+      [&result, &notification](
+          bool, std::unique_ptr<std::vector<std::pair<std::string, T>>> res) {
+        for (const auto& it : *res) {
+          result->push_back(it);
+        }
+        notification.Notify();
+      });
+  notification.WaitForNotificationWithTimeout(absl::Seconds(5));
+
+  absl::flat_hash_map<std::string, T> entry_map;
+  for (const auto& it : *result) {
+    entry_map[it.first] = it.second;
+  }
+  return entry_map;
+}
+
+template <typename T>
+void WipeCleanAndWait(std::unique_ptr<LeveldbDataSet<T>>& dataset,
+                      std::filesystem::path path) {
+  absl::Notification notification;
+  dataset->Destroy([&notification](bool) { notification.Notify(); });
+  notification.WaitForNotificationWithTimeout(absl::Seconds(5));
+  // Call the destructor before removing leveldb storage directory
+  dataset.reset();
+  std::filesystem::remove_all(path);
+}
+
+DiceRoll GenerateDiceRoll(int value) {
+  DiceRoll result;
+  result.set_value(value);
+  if (value == 12) {
+    result.set_nickname("boxcars");
+  }
+
+  if (value == 2) {
+    result.set_nickname("snake eyes");
+  }
+
+  return result;
+}
+
+TEST(LeveldbDataSet, UpdateEntriesDiceRoll) {
+  std::filesystem::path path = GenerateLeveldbPath();
+  std::unique_ptr<LeveldbDataSet<DiceRoll>> diceroll_set =
+      CreateDataSet<DiceRoll>(path);
+
+  InitStatus status = InitializeAndWait(diceroll_set);
+  ASSERT_EQ(status, InitStatus::kOK);
+
+  DiceRoll diceroll1 = GenerateDiceRoll(2);
+  DiceRoll diceroll2 = GenerateDiceRoll(12);
+
+  auto entries = LeveldbDataSet<DiceRoll>::KeyEntryVector(
+      {{"id1", diceroll1}, {"id2", diceroll2}});
+  auto data =
+      std::make_unique<LeveldbDataSet<DiceRoll>::KeyEntryVector>(entries);
+
+  bool result = UpdateEntriesAndWait(diceroll_set, std::move(data), nullptr);
+  WipeCleanAndWait(diceroll_set, path);
+
+  EXPECT_TRUE(result);
+}
+
+TEST(LeveldbDataSet, LoadEntriesDiceRoll) {
+  std::filesystem::path path = GenerateLeveldbPath();
+  std::unique_ptr<LeveldbDataSet<DiceRoll>> diceroll_set =
+      CreateDataSet<DiceRoll>(path);
+
+  InitializeAndWait(diceroll_set);
+
+  DiceRoll diceroll1 = GenerateDiceRoll(2);
+  DiceRoll diceroll2 = GenerateDiceRoll(12);
+
+  auto entries = LeveldbDataSet<DiceRoll>::KeyEntryVector(
+      {{"id1", diceroll1}, {"id2", diceroll2}});
+  auto data =
+      std::make_unique<LeveldbDataSet<DiceRoll>::KeyEntryVector>(entries);
+  UpdateEntriesAndWait(diceroll_set, std::move(data), nullptr);
+
+  auto result = LoadEntriesAndWait(diceroll_set);
+  WipeCleanAndWait(diceroll_set, path);
+
+  EXPECT_THAT(*result, SizeIs(2));
+
+  // EqualsProto is only available internally
+  // EXPECT_THAT((*result)[0], protobuf_matchers::EqualsProto<DiceRoll>(
+  //                               "value: 2 nickname: 'snake eyes'"));
+  // EXPECT_THAT((*result)[1],
+  //             protobuf_matchers::EqualsProto<DiceRoll>("value: 12 nickname:
+  //             'boxcars'"));
+
+  EXPECT_EQ((*result)[0].value(), 2);
+  EXPECT_EQ((*result)[0].nickname(), "snake eyes");
+
+  EXPECT_EQ((*result)[1].value(), 12);
+  EXPECT_EQ((*result)[1].nickname(), "boxcars");
+}
+
+TEST(LeveldbDataSet, RemoveEntriesDiceRoll) {
+  std::filesystem::path path = GenerateLeveldbPath();
+  std::unique_ptr<LeveldbDataSet<DiceRoll>> diceroll_set =
+      CreateDataSet<DiceRoll>(path);
+
+  InitializeAndWait(diceroll_set);
+
+  DiceRoll diceroll1 = GenerateDiceRoll(2);
+  DiceRoll diceroll2 = GenerateDiceRoll(12);
+  DiceRoll diceroll3 = GenerateDiceRoll(5);
+  DiceRoll diceroll4 = GenerateDiceRoll(7);
+
+  auto entries_to_add1 = LeveldbDataSet<DiceRoll>::KeyEntryVector(
+      {{"id1", diceroll1}, {"id2", diceroll2}});
+  auto data_to_add1 =
+      std::make_unique<LeveldbDataSet<DiceRoll>::KeyEntryVector>(
+          entries_to_add1);
+  UpdateEntriesAndWait(diceroll_set, std::move(data_to_add1), nullptr);
+
+  auto entries_to_add2 = LeveldbDataSet<DiceRoll>::KeyEntryVector(
+      {{"id3", diceroll3}, {"id4", diceroll4}});
+  auto data_to_add2 =
+      std::make_unique<LeveldbDataSet<DiceRoll>::KeyEntryVector>(
+          entries_to_add2);
+
+  auto keys_to_remove = std::make_unique<std::vector<std::string>>(
+      std::vector<std::string>({"id1"}));
+
+  UpdateEntriesAndWait(diceroll_set, std::move(data_to_add2),
+                       std::move(keys_to_remove));
+
+  auto result = LoadEntriesWithKeysAndWait(diceroll_set);
+  WipeCleanAndWait(diceroll_set, path);
+
+  EXPECT_THAT(result, SizeIs(3));
+
+  EXPECT_EQ(result["id2"].value(), diceroll2.value());
+  EXPECT_EQ(result["id2"].nickname(), diceroll2.nickname());
+  EXPECT_EQ(result["id3"].value(), diceroll3.value());
+  EXPECT_EQ(result["id3"].nickname(), diceroll3.nickname());
+  EXPECT_EQ(result["id4"].value(), diceroll4.value());
+  EXPECT_EQ(result["id4"].nickname(), diceroll4.nickname());
+}
+
+}  // namespace
+}  // namespace data
+}  // namespace nearby

internal/data/leveldb_data_set_test.proto

@@ -0,0 +1,22 @@
+// Copyright 2022 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.
+
+syntax = "proto3";
+
+package nearby.data;
+
+message DiceRoll {
+  optional int32 value = 1;      // value of this roll, e.g. 2..12
+  optional string nickname = 2;  // string nickname, e.g. "snake eyes"
+}

internal/data/memory_data_set.h

@@ -0,0 +1,106 @@
+// Copyright 2022 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 THIRD_PARTY_NEARBY_INTERNAL_DATA_MEMORY_DATA_SET_H_
+#define THIRD_PARTY_NEARBY_INTERNAL_DATA_MEMORY_DATA_SET_H_
+
+#include <functional>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "absl/container/flat_hash_map.h"
+#include "absl/strings/string_view.h"
+#include "absl/synchronization/mutex.h"
+#include "internal/data/data_set.h"
+
+namespace nearby {
+namespace data {
+
+template <typename T>
+class MemoryDataSet : public DataSet<T> {
+ public:
+  using KeyEntryVector = std::vector<std::pair<std::string, T>>;
+
+  explicit MemoryDataSet(absl::string_view path) : path_(path) {}
+  ~MemoryDataSet() override = default;
+
+  void Initialize(std::function<void(InitStatus)> callback) override;
+  void LoadEntries(std::function<void(bool, std::unique_ptr<std::vector<T>>)>
+                       callback) override;
+  void UpdateEntries(std::unique_ptr<KeyEntryVector> entries_to_save,
+                     std::unique_ptr<std::vector<std::string>> keys_to_remove,
+                     std::function<void(bool)> callback) override;
+  void Destroy(std::function<void(bool)> callback) override;
+
+ private:
+  std::string path_;
+
+  absl::Mutex mutex_;
+  absl::flat_hash_map<std::string, T> entries_;
+};
+
+template <typename T>
+void MemoryDataSet<T>::Initialize(std::function<void(InitStatus)> callback) {
+  std::move(callback)(InitStatus::kOK);
+}
+
+template <typename T>
+void MemoryDataSet<T>::LoadEntries(
+    std::function<void(bool, std::unique_ptr<std::vector<T>>)> callback) {
+  auto result = std::make_unique<std::vector<T>>();
+  auto it = entries_.begin();
+  while (it != entries_.end()) {
+    result->push_back(it->second);
+    ++it;
+  }
+
+  std::move(callback)(true, std::move(result));
+}
+
+template <typename T>
+void MemoryDataSet<T>::UpdateEntries(
+    std::unique_ptr<KeyEntryVector> entries_to_save,
+    std::unique_ptr<std::vector<std::string>> keys_to_remove,
+    std::function<void(bool)> callback) {
+  if (entries_to_save != nullptr) {
+    auto it = entries_to_save->begin();
+    while (it != entries_to_save->end()) {
+      entries_.emplace(it->first, it->second);
+      ++it;
+    }
+  }
+
+  if (keys_to_remove != nullptr) {
+    auto it = keys_to_remove->begin();
+    while (it != keys_to_remove->end()) {
+      entries_.erase(*it);
+      ++it;
+    }
+  }
+
+  std::move(callback)(true);
+}
+
+template <typename T>
+void MemoryDataSet<T>::Destroy(std::function<void(bool)> callback) {
+  entries_.clear();
+  std::move(callback)(true);
+}
+
+}  // namespace data
+}  // namespace nearby
+
+#endif  // THIRD_PARTY_NEARBY_INTERNAL_DATA_MEMORY_DATA_SET_H_

internal/data/memory_data_set_test.cc

@@ -0,0 +1,71 @@
+// Copyright 2022 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 "internal/data/memory_data_set.h"
+
+#include <algorithm>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "gmock/gmock.h"
+#include "protobuf-matchers/protocol-buffer-matchers.h"
+#include "gtest/gtest.h"
+
+namespace nearby {
+namespace data {
+namespace {
+
+TEST(MemoryDataSet, TestUpdateEntries) {
+  bool result = false;
+  MemoryDataSet<std::string> string_set{""};
+
+  auto temp = MemoryDataSet<std::string>::KeyEntryVector(
+      {{"id1", "string1"}, {"id2", "string2"}});
+
+  auto data =
+      std::make_unique<MemoryDataSet<std::string>::KeyEntryVector>(temp);
+  string_set.UpdateEntries(std::move(data), nullptr,
+                           [&result](bool res) { result = res; });
+  EXPECT_TRUE(result);
+}
+
+TEST(MemoryDataSet, TestLoadEntries) {
+  std::vector<std::string> result = {};
+  MemoryDataSet<std::string> string_set{""};
+
+  auto temp = MemoryDataSet<std::string>::KeyEntryVector(
+      {{"id1", "string1"}, {"id2", "string2"}});
+  auto data =
+      std::make_unique<MemoryDataSet<std::string>::KeyEntryVector>(temp);
+
+  string_set.UpdateEntries(std::move(data), nullptr, [](bool ans) {});
+  string_set.LoadEntries(
+      [&result](bool ans, std::unique_ptr<std::vector<std::string>> res) {
+        auto it = res->begin();
+        while (it != res->end()) {
+          result.push_back(*it);
+          ++it;
+        }
+      });
+
+  EXPECT_THAT(result, testing::SizeIs(2));
+  std::sort(result.begin(), result.end());
+  EXPECT_EQ(result, std::vector<std::string>({"string1", "string2"}));
+}
+
+}  // namespace
+}  // namespace data
+}  // namespace nearby

internal/platform/BUILD

@@ -103,6 +103,7 @@ cc_library(
         "//fastpair:__subpackages__",
         "//internal/auth:__subpackages__",
         "//internal/auth/credential_store:__subpackages__",
+        "//internal/data:__subpackages__",
         "//internal/interop:__pkg__",
         "//internal/network:__subpackages__",
         "//internal/platform:__subpackages__",

internal/platform/implementation/g3/BUILD

@@ -142,6 +142,7 @@ cc_library(
         "//fastpair:__subpackages__",
         "//internal/account:__subpackages__",
         "//internal/auth:__subpackages__",
+        "//internal/data:__subpackages__",
         "//internal/flags:__subpackages__",
         "//internal/network:__subpackages__",
         "//internal/platform:__subpackages__",

internal/test/BUILD

@@ -25,6 +25,7 @@ cc_library(
     ],
     hdrs = [
         "fake_clock.h",
+        "fake_data_set.h",
         "fake_device_info.h",
         "fake_single_thread_executor.h",
         "fake_task_runner.h",
@@ -37,6 +38,7 @@ cc_library(
     visibility = ["//visibility:public"],
     deps = [
         "//internal/base:bluetooth_address",
+        "//internal/data:data_manager",
         "//internal/platform:comm",
         "//internal/platform:types",
         "//internal/platform/implementation:types",
@@ -55,6 +57,7 @@ cc_test(
     timeout = "short",
     srcs = [
         "fake_clock_test.cc",
+        "fake_data_set_test.cc",
         "fake_device_info_test.cc",
         "fake_task_runner_test.cc",
         "fake_timer_test.cc",
@@ -65,9 +68,9 @@ cc_test(
     shard_count = 8,
     deps = [
         ":test",
-        "//internal/platform:types",
+        "//internal/data:data_manager",
         "//internal/platform/implementation:types",
-        "//internal/platform/implementation/g3",
+        "//internal/platform/implementation/g3",  # fixdeps: keep
         "@com_github_protobuf_matchers//protobuf-matchers",
         "@com_google_absl//absl/synchronization",
         "@com_google_absl//absl/time",

internal/test/fake_data_set.h

@@ -0,0 +1,131 @@
+// Copyright 2022 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 THIRD_PARTY_NEARBY_INTERNAL_TEST_FAKE_DATA_SET_H_
+#define THIRD_PARTY_NEARBY_INTERNAL_TEST_FAKE_DATA_SET_H_
+
+#include <functional>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "absl/container/flat_hash_map.h"
+#include "internal/data/data_set.h"
+
+namespace nearby {
+namespace data {
+
+template <typename T>
+class FakeDataSet : public DataSet<T> {
+ public:
+  using KeyEntryVector = std::vector<std::pair<std::string, T>>;
+
+  explicit FakeDataSet(const absl::flat_hash_map<std::string, T>& entries_map)
+      : entries_map_(entries_map) {}
+
+  void Initialize(std::function<void(InitStatus)> callback) override {
+    init_callback_ = std::move(callback);
+  }
+
+  void LoadEntries(std::function<void(bool, std::unique_ptr<std::vector<T>>)>
+                       callback) override {
+    load_callback_ = std::move(callback);
+  }
+
+  void UpdateEntries(std::unique_ptr<KeyEntryVector> entries_to_save,
+                     std::unique_ptr<std::vector<std::string>> keys_to_remove,
+                     std::function<void(bool)> callback) override {
+    entries_to_save_ = std::move(entries_to_save);
+    keys_to_remove_ = std::move(keys_to_remove);
+    update_callback_ = std::move(callback);
+  }
+
+  void Destroy(std::function<void(bool)> callback) override {
+    destroy_callback_ = std::move(callback);
+  }
+
+  // Mocked methods
+  void InitStatusCallback(InitStatus status) {
+    if (init_callback_ != nullptr) {
+      init_callback_(status);
+    }
+  }
+
+  void LoadCallback(bool success) {
+    if (load_callback_ != nullptr) {
+      auto entries = std::make_unique<std::vector<T>>();
+      for (auto it = entries_map_.begin(); it != entries_map_.end(); ++it) {
+        entries->push_back(it->second);
+      }
+      load_callback_(success, std::move(entries));
+    }
+  }
+
+  void UpdateCallback(bool success) {
+    if (success) {
+      if (entries_to_save_ != nullptr) {
+        for (auto it = entries_to_save_->begin(); it != entries_to_save_->end();
+             ++it) {
+          auto entry = entries_map_.find(it->first);
+          if (entry == entries_map_.end()) {
+            entries_map_.emplace(it->first, it->second);
+          } else {
+            entry->second = it->second;
+          }
+        }
+      }
+
+      if (keys_to_remove_ != nullptr) {
+        for (auto it = keys_to_remove_->begin(); it != keys_to_remove_->end();
+             ++it) {
+          entries_map_.erase(*it);
+        }
+      }
+    }
+
+    entries_to_save_ = nullptr;
+    keys_to_remove_ = nullptr;
+    if (update_callback_ != nullptr) {
+      update_callback_(success);
+    }
+  }
+
+  void DestroyCallback(bool success) {
+    if (success) {
+      entries_map_.clear();
+    }
+
+    if (destroy_callback_ != nullptr) {
+      destroy_callback_(success);
+    }
+  }
+
+  absl::flat_hash_map<std::string, T>& entries_map() { return entries_map_; }
+
+ private:
+  absl::flat_hash_map<std::string, T> entries_map_ = nullptr;
+  std::function<void(InitStatus)> init_callback_ = nullptr;
+  std::function<void(bool, std::unique_ptr<std::vector<T>>)> load_callback_ =
+      nullptr;
+  std::unique_ptr<KeyEntryVector> entries_to_save_ = nullptr;
+  std::unique_ptr<std::vector<std::string>> keys_to_remove_ = nullptr;
+  std::function<void(bool)> update_callback_ = nullptr;
+  std::function<void(bool)> destroy_callback_ = nullptr;
+};
+
+}  // namespace data
+}  // namespace nearby
+
+#endif  // THIRD_PARTY_NEARBY_INTERNAL_TEST_FAKE_DATA_SET_H_

internal/test/fake_data_set_test.cc

@@ -0,0 +1,107 @@
+// Copyright 2022 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 "internal/test/fake_data_set.h"
+
+#include <algorithm>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "gmock/gmock.h"
+#include "protobuf-matchers/protocol-buffer-matchers.h"
+#include "gtest/gtest.h"
+#include "internal/data/data_set.h"
+
+namespace nearby {
+namespace data {
+namespace {
+
+TEST(FakeDataSet, TestInitialize) {
+  InitStatus result = InitStatus::kNotInitialized;
+  FakeDataSet<std::string> string_set({});
+
+  string_set.Initialize([&result](InitStatus res) { result = res; });
+  string_set.InitStatusCallback(InitStatus::kOK);
+  EXPECT_EQ(result, InitStatus::kOK);
+}
+
+TEST(FakeDataSet, TestUpdateEntries) {
+  bool result = false;
+  FakeDataSet<std::string> string_set({});
+
+  auto temp = FakeDataSet<std::string>::KeyEntryVector(
+      {{"id1", "string1"}, {"id2", "string2"}});
+
+  auto data = std::make_unique<FakeDataSet<std::string>::KeyEntryVector>(temp);
+
+  string_set.UpdateEntries(std::move(data), nullptr,
+                           [&result](bool res) { result = res; });
+
+  string_set.UpdateCallback(true);
+  EXPECT_TRUE(result);
+  data = std::make_unique<FakeDataSet<std::string>::KeyEntryVector>(temp);
+  string_set.UpdateEntries(std::move(data), nullptr,
+                           [&result](bool res) { result = res; });
+  string_set.UpdateCallback(false);
+  EXPECT_FALSE(result);
+}
+
+TEST(FakeDataSet, TestLoadEntries) {
+  std::vector<std::string> result = {};
+  FakeDataSet<std::string> string_set({});
+
+  auto temp = FakeDataSet<std::string>::KeyEntryVector(
+      {{"id1", "string1"}, {"id2", "string2"}});
+  auto data = std::make_unique<FakeDataSet<std::string>::KeyEntryVector>(temp);
+
+  string_set.UpdateEntries(std::move(data), nullptr, [](bool ans) {});
+  string_set.UpdateCallback(true);
+  string_set.LoadEntries(
+      [&result](bool ans, std::unique_ptr<std::vector<std::string>> res) {
+        auto it = res->begin();
+        while (it != res->end()) {
+          result.push_back(*it);
+          ++it;
+        }
+      });
+  string_set.LoadCallback(true);
+  EXPECT_THAT(result, testing::SizeIs(2));
+  std::sort(result.begin(), result.end());
+  EXPECT_EQ(result, std::vector<std::string>({"string1", "string2"}));
+}
+
+TEST(MockDataSet, TestDestroy) {
+  bool result;
+  std::vector<std::string> data = {};
+  FakeDataSet<std::string> string_set({{"id1", "string1"}, {"id2", "string2"}});
+  string_set.Destroy([&result](bool res) { result = res; });
+  string_set.DestroyCallback(true);
+  EXPECT_TRUE(result);
+  string_set.LoadEntries(
+      [&data](bool ans, std::unique_ptr<std::vector<std::string>> res) {
+        auto it = res->begin();
+        while (it != res->end()) {
+          data.push_back(*it);
+          ++it;
+        }
+      });
+  string_set.LoadCallback(true);
+  EXPECT_THAT(data, ::testing::SizeIs(0));
+}
+
+}  // namespace
+}  // namespace data
+}  // namespace nearby