Merge branch 'master' into release.

Change-Id: Id627ceca5ef60281e024ff80fc1848b3ccb6c14d

cpp/core_v2/internal/mediums/BUILD

@@ -19,6 +19,7 @@ cc_library(
         "ble_advertisement.cc",
         "ble_advertisement_header.cc",
         "ble_packet.cc",
+        "bloom_filter.cc",
         "bluetooth_radio.cc",
         "uuid.cc",
     ],
@@ -28,6 +29,7 @@ cc_library(
         "ble_advertisement_header.h",
         "ble_packet.h",
         "ble_peripheral.h",
+        "bloom_filter.h",
         "bluetooth_radio.h",
         "lost_entity_tracker.h",
         "uuid.h",
@@ -37,12 +39,15 @@ cc_library(
     ],
     deps = [
         "//platform_v2/base",
-        "//platform_v2/public",
+        "//platform_v2/public:comm",
         "//platform_v2/public:logging",
+        "//platform_v2/public:types",
         "//absl/container:flat_hash_map",
         "//absl/container:flat_hash_set",
+        "//absl/numeric:int128",
         "//absl/strings",
         "//absl/time",
+        "//smhasher:libmurmur3",
     ],
 )
 
@@ -55,7 +60,8 @@ cc_library(
     ],
     deps = [
         "//platform_v2/base",
-        "//platform_v2/public",
+        "//platform_v2/public:comm",
+        "//platform_v2/public:types",
     ],
 )
 
@@ -67,6 +73,7 @@ cc_test(
         "ble_advertisement_test.cc",
         "ble_packet_test.cc",
         "ble_peripheral_test.cc",
+        "bloom_filter_test.cc",
         "bluetooth_radio_test.cc",
         "lost_entity_tracker_test.cc",
         "uuid_test.cc",
@@ -76,8 +83,9 @@ cc_test(
         ":mediums",
         "//platform_v2/base",
         "//platform_v2/impl/g3",  # build_cleaner: keep
-        "//platform_v2/public",
+        "//platform_v2/public:comm",
         "//platform_v2/public:logging",
+        "//platform_v2/public:types",
         "//testing/base/public:gunit_main",
         "//absl/time",
     ],

cpp/core_v2/internal/mediums/ble_advertisement_header.h

@@ -57,12 +57,11 @@ class BleAdvertisementHeader {
                          const ByteArray &advertisement_hash);
   explicit BleAdvertisementHeader(
       const std::string &ble_advertisement_header_string);
-  ~BleAdvertisementHeader() = default;
-
   BleAdvertisementHeader(const BleAdvertisementHeader &) = default;
   BleAdvertisementHeader &operator=(const BleAdvertisementHeader &) = default;
   BleAdvertisementHeader(BleAdvertisementHeader &&) = default;
   BleAdvertisementHeader &operator=(BleAdvertisementHeader &&) = default;
+  ~BleAdvertisementHeader() = default;
 
   // Produces an encoded binary string which can be decoded by the explicit
   // constructor. The returned string is empty if BleAdvertisementHeader is not

cpp/core_v2/internal/mediums/ble_advertisement_header_test.cc

@@ -30,11 +30,11 @@ constexpr char kServiceIDBloomFilter[] =
 constexpr char kAdvertisementHash[] = "\x0a\x0b\x0c\x0d";
 
 TEST(BleAdvertisementHeaderTest, ConstructionWorks) {
-  ByteArray service_id_bloom_filter(kServiceIDBloomFilter);
-  ByteArray advertisement_hash(kAdvertisementHash);
+  ByteArray service_id_bloom_filter{kServiceIDBloomFilter};
+  ByteArray advertisement_hash{kAdvertisementHash};
 
-  BleAdvertisementHeader ble_advertisement_header(
-      kVersion, kNumSlots, service_id_bloom_filter, advertisement_hash);
+  BleAdvertisementHeader ble_advertisement_header{
+      kVersion, kNumSlots, service_id_bloom_filter, advertisement_hash};
 
   EXPECT_TRUE(ble_advertisement_header.IsValid());
   EXPECT_EQ(kVersion, ble_advertisement_header.GetVersion());
@@ -48,11 +48,11 @@ TEST(BleAdvertisementHeaderTest, ConstructionWorks) {
 TEST(BleAdvertisementHeaderTest, ConstructionFailsWithBadVersion) {
   auto bad_version = static_cast<BleAdvertisementHeader::Version>(666);
 
-  ByteArray service_id_bloom_filter(kServiceIDBloomFilter);
-  ByteArray advertisement_hash(kAdvertisementHash);
+  ByteArray service_id_bloom_filter{kServiceIDBloomFilter};
+  ByteArray advertisement_hash{kAdvertisementHash};
 
-  BleAdvertisementHeader ble_advertisement_header(
-      bad_version, kNumSlots, service_id_bloom_filter, advertisement_hash);
+  BleAdvertisementHeader ble_advertisement_header{
+      bad_version, kNumSlots, service_id_bloom_filter, advertisement_hash};
 
   EXPECT_FALSE(ble_advertisement_header.IsValid());
 }
@@ -61,12 +61,12 @@ TEST(BleAdvertisementHeaderTest,
      ConstructionFailsWithShortServiceIdBloomFilter) {
   char short_service_id_bloom_filter[] = "\x01\x02\x03\x04\x05\x06\x07\x08\x09";
 
-  ByteArray short_service_id_bloom_filter_bytes(short_service_id_bloom_filter);
-  ByteArray advertisement_hash(kAdvertisementHash);
+  ByteArray short_service_id_bloom_filter_bytes{short_service_id_bloom_filter};
+  ByteArray advertisement_hash{kAdvertisementHash};
 
-  BleAdvertisementHeader ble_advertisement_header(
+  BleAdvertisementHeader ble_advertisement_header{
       kVersion, kNumSlots, short_service_id_bloom_filter_bytes,
-      advertisement_hash);
+      advertisement_hash};
 
   EXPECT_FALSE(ble_advertisement_header.IsValid());
 }
@@ -76,11 +76,11 @@ TEST(BleAdvertisementHeaderTest,
   char long_service_id_bloom_filter[] =
       "\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b";
 
-  ByteArray service_id_bloom_filter(long_service_id_bloom_filter);
-  ByteArray advertisement_hash(kAdvertisementHash);
+  ByteArray service_id_bloom_filter{long_service_id_bloom_filter};
+  ByteArray advertisement_hash{kAdvertisementHash};
 
-  BleAdvertisementHeader ble_advertisement_header(
-      kVersion, kNumSlots, service_id_bloom_filter, advertisement_hash);
+  BleAdvertisementHeader ble_advertisement_header{
+      kVersion, kNumSlots, service_id_bloom_filter, advertisement_hash};
 
   EXPECT_FALSE(ble_advertisement_header.IsValid());
 }
@@ -88,38 +88,37 @@ TEST(BleAdvertisementHeaderTest,
 TEST(BleAdvertisementHeaderTest, ConstructionFailsWithShortAdvertisementHash) {
   char short_advertisement_hash[] = "\x0a\x0b\x0c";
 
-  ByteArray service_id_bloom_filter(kServiceIDBloomFilter);
-  ByteArray advertisement_hash(short_advertisement_hash);
+  ByteArray service_id_bloom_filter{kServiceIDBloomFilter};
+  ByteArray advertisement_hash{short_advertisement_hash};
 
-  BleAdvertisementHeader ble_advertisement_header(
-      kVersion, kNumSlots, service_id_bloom_filter, advertisement_hash);
+  BleAdvertisementHeader ble_advertisement_header{
+      kVersion, kNumSlots, service_id_bloom_filter, advertisement_hash};
 
   EXPECT_FALSE(ble_advertisement_header.IsValid());
 }
 
 TEST(BleAdvertisementHeaderTest, ConstructionFailsWithLongAdvertisementHash) {
-  char long_advertisement_hash[] = "\x0a\x0b\x0c\x0d\0x0e";
+  char long_advertisement_hash[] = "\x0a\x0b\x0c\x0d\x0e";
 
-  ByteArray service_id_bloom_filter(kServiceIDBloomFilter);
-  ByteArray advertisement_hash(long_advertisement_hash,
-                               sizeof(long_advertisement_hash) / sizeof(char));
-  BleAdvertisementHeader ble_advertisement_header(
-      kVersion, kNumSlots, service_id_bloom_filter, advertisement_hash);
+  ByteArray service_id_bloom_filter{kServiceIDBloomFilter};
+  ByteArray advertisement_hash{long_advertisement_hash};
+  BleAdvertisementHeader ble_advertisement_header{
+      kVersion, kNumSlots, service_id_bloom_filter, advertisement_hash};
 
   EXPECT_FALSE(ble_advertisement_header.IsValid());
 }
 
 TEST(BleAdvertisementHeaderTest, ConstructionFromSerializedStringWorks) {
-  ByteArray service_id_bloom_filter(kServiceIDBloomFilter);
-  ByteArray advertisement_hash(kAdvertisementHash);
+  ByteArray service_id_bloom_filter{kServiceIDBloomFilter};
+  ByteArray advertisement_hash{kAdvertisementHash};
 
-  BleAdvertisementHeader org_ble_advertisement_header(
-      kVersion, kNumSlots, service_id_bloom_filter, advertisement_hash);
+  BleAdvertisementHeader org_ble_advertisement_header{
+      kVersion, kNumSlots, service_id_bloom_filter, advertisement_hash};
   auto ble_advertisement_header_string =
       std::string(org_ble_advertisement_header);
 
-  auto ble_advertisement_header =
-      BleAdvertisementHeader(ble_advertisement_header_string);
+  BleAdvertisementHeader ble_advertisement_header{
+      ble_advertisement_header_string};
 
   EXPECT_TRUE(ble_advertisement_header.IsValid());
   EXPECT_EQ(kVersion, ble_advertisement_header.GetVersion());
@@ -131,24 +130,24 @@ TEST(BleAdvertisementHeaderTest, ConstructionFromSerializedStringWorks) {
 }
 
 TEST(BleAdvertisementHeaderTest, ConstructionFromExtraBytesWorks) {
-  ByteArray service_id_bloom_filter(kServiceIDBloomFilter);
-  ByteArray advertisement_hash(kAdvertisementHash);
+  ByteArray service_id_bloom_filter{kServiceIDBloomFilter};
+  ByteArray advertisement_hash{kAdvertisementHash};
 
-  BleAdvertisementHeader ble_advertisement_header(
-      kVersion, kNumSlots, service_id_bloom_filter, advertisement_hash);
+  BleAdvertisementHeader ble_advertisement_header{
+      kVersion, kNumSlots, service_id_bloom_filter, advertisement_hash};
   auto ble_advertisement_header_string = std::string(ble_advertisement_header);
 
   // Base64 decode the string, add a character, and then re-encode it.
   ByteArray ble_advertisement_header_bytes =
       Base64Utils::Decode(ble_advertisement_header_string);
-  ByteArray long_ble_advertisement_header_bytes(
-      ble_advertisement_header_bytes.size() + 1);
+  ByteArray long_ble_advertisement_header_bytes{
+      ble_advertisement_header_bytes.size() + 1};
   long_ble_advertisement_header_bytes.CopyAt(0, ble_advertisement_header_bytes);
-  std::string long_ble_advertisement_header_string =
-      Base64Utils::Encode(long_ble_advertisement_header_bytes);
+  std::string long_ble_advertisement_header_string{
+      Base64Utils::Encode(long_ble_advertisement_header_bytes)};
 
-  auto long_ble_advertisement_header =
-      BleAdvertisementHeader(long_ble_advertisement_header_string);
+  BleAdvertisementHeader long_ble_advertisement_header{
+      long_ble_advertisement_header_string};
 
   EXPECT_TRUE(long_ble_advertisement_header.IsValid());
   EXPECT_EQ(kVersion, long_ble_advertisement_header.GetVersion());
@@ -160,25 +159,25 @@ TEST(BleAdvertisementHeaderTest, ConstructionFromExtraBytesWorks) {
 }
 
 TEST(BleAdvertisementHeaderTest, ConstructionFromShortLengthFails) {
-  ByteArray service_id_bloom_filter(kServiceIDBloomFilter);
-  ByteArray advertisement_hash(kAdvertisementHash);
+  ByteArray service_id_bloom_filter{kServiceIDBloomFilter};
+  ByteArray advertisement_hash{kAdvertisementHash};
 
-  BleAdvertisementHeader ble_advertisement_header(
-      kVersion, kNumSlots, service_id_bloom_filter, advertisement_hash);
+  BleAdvertisementHeader ble_advertisement_header{
+      kVersion, kNumSlots, service_id_bloom_filter, advertisement_hash};
   auto ble_advertisement_header_string = std::string(ble_advertisement_header);
 
   // Base64 decode the string, remove a character, and then re-encode it.
   ByteArray ble_advertisement_header_bytes =
       Base64Utils::Decode(ble_advertisement_header_string);
-  ByteArray short_ble_advertisement_header_bytes(
-      ble_advertisement_header_bytes.size() - 1);
+  ByteArray short_ble_advertisement_header_bytes{
+      ble_advertisement_header_bytes.size() - 1};
   short_ble_advertisement_header_bytes.CopyAt(0,
                                               ble_advertisement_header_bytes);
-  std::string short_ble_advertisement_header_string =
-      Base64Utils::Encode(short_ble_advertisement_header_bytes);
+  std::string short_ble_advertisement_header_string{
+      Base64Utils::Encode(short_ble_advertisement_header_bytes)};
 
-  auto short_ble_advertisement_header =
-      BleAdvertisementHeader(short_ble_advertisement_header_string);
+  BleAdvertisementHeader short_ble_advertisement_header{
+      short_ble_advertisement_header_string};
 
   EXPECT_FALSE(short_ble_advertisement_header.IsValid());
 }

cpp/core_v2/internal/mediums/ble_packet.h

@@ -36,12 +36,11 @@ class BlePacket {
   BlePacket() = default;
   BlePacket(const ByteArray& service_id_hash, const ByteArray& data);
   explicit BlePacket(const ByteArray& ble_packet_byte);
-  ~BlePacket() = default;
-
   BlePacket(const BlePacket&) = default;
   BlePacket& operator=(const BlePacket&) = default;
   BlePacket(BlePacket&&) = default;
   BlePacket& operator=(BlePacket&&) = default;
+  ~BlePacket() = default;
 
   explicit operator ByteArray() const;
 

cpp/core_v2/internal/mediums/ble_packet_test.cc

@@ -25,10 +25,10 @@ constexpr char kServiceIDHash[] = "\x0a\x0b\x0c";
 constexpr char kData[] = "\x01\x02\x03\x04\x05";
 
 TEST(BlePacketTest, ConstructionWorks) {
-  ByteArray service_id_hash(kServiceIDHash);
-  ByteArray data(kData);
+  ByteArray service_id_hash{kServiceIDHash};
+  ByteArray data{kData};
 
-  BlePacket ble_packet(service_id_hash, data);
+  BlePacket ble_packet{service_id_hash, data};
 
   EXPECT_TRUE(ble_packet.IsValid());
   EXPECT_EQ(service_id_hash, ble_packet.GetServiceIdHash());
@@ -36,12 +36,12 @@ TEST(BlePacketTest, ConstructionWorks) {
 }
 
 TEST(BlePacketTest, ConstructionWorksWithEmptyData) {
-  char empty_data[] = {};
+  char empty_data[] = "";
 
-  ByteArray service_id_hash(kServiceIDHash);
-  ByteArray data(empty_data);
+  ByteArray service_id_hash{kServiceIDHash};
+  ByteArray data{empty_data};
 
-  BlePacket ble_packet(service_id_hash, data);
+  BlePacket ble_packet{service_id_hash, data};
 
   EXPECT_TRUE(ble_packet.IsValid());
   EXPECT_EQ(service_id_hash, ble_packet.GetServiceIdHash());
@@ -51,8 +51,8 @@ TEST(BlePacketTest, ConstructionWorksWithEmptyData) {
 TEST(BlePacketTest, ConstructionFailsWithShortServiceIdHash) {
   char short_service_id_hash[] = "\x0a\x0b";
 
-  ByteArray service_id_hash(short_service_id_hash);
-  ByteArray data(kData);
+  ByteArray service_id_hash{short_service_id_hash};
+  ByteArray data{kData};
 
   BlePacket ble_packet(service_id_hash, data);
 
@@ -62,22 +62,22 @@ TEST(BlePacketTest, ConstructionFailsWithShortServiceIdHash) {
 TEST(BlePacketTest, ConstructionFailsWithLongServiceIdHash) {
   char long_service_id_hash[] = "\x0a\x0b\x0c\x0d";
 
-  ByteArray service_id_hash(long_service_id_hash);
-  ByteArray data(kData);
+  ByteArray service_id_hash{long_service_id_hash};
+  ByteArray data{kData};
 
-  BlePacket ble_packet(service_id_hash, data);
+  BlePacket ble_packet{service_id_hash, data};
 
   EXPECT_FALSE(ble_packet.IsValid());
 }
 
 TEST(BlePacketTest, ConstructionFromSerializedBytesWorks) {
-  ByteArray service_id_hash(kServiceIDHash);
-  ByteArray data(kData);
+  ByteArray service_id_hash{kServiceIDHash};
+  ByteArray data{kData};
 
-  BlePacket org_ble_packet(service_id_hash, data);
-  ByteArray ble_packet_bytes(org_ble_packet);
+  BlePacket org_ble_packet{service_id_hash, data};
+  ByteArray ble_packet_bytes{org_ble_packet};
 
-  BlePacket ble_packet(ble_packet_bytes);
+  BlePacket ble_packet{ble_packet_bytes};
 
   EXPECT_TRUE(ble_packet.IsValid());
   EXPECT_EQ(service_id_hash, ble_packet.GetServiceIdHash());
@@ -85,22 +85,22 @@ TEST(BlePacketTest, ConstructionFromSerializedBytesWorks) {
 }
 
 TEST(BlePacketTest, ConstructionFromNullBytesFails) {
-  BlePacket ble_packet(ByteArray{});
+  BlePacket ble_packet{ByteArray{}};
 
   EXPECT_FALSE(ble_packet.IsValid());
 }
 
 TEST(BlePacketTest, ConstructionFromShortLengthDataFails) {
-  ByteArray service_id_hash(kServiceIDHash);
-  ByteArray data(kData);
+  ByteArray service_id_hash{kServiceIDHash};
+  ByteArray data{kData};
 
-  BlePacket org_ble_packet(service_id_hash, data);
-  ByteArray org_ble_packet_bytes(org_ble_packet);
+  BlePacket org_ble_packet{service_id_hash, data};
+  ByteArray org_ble_packet_bytes{org_ble_packet};
 
   // Cut off the packet so that it's too short
-  ByteArray short_ble_packet_bytes(ByteArray(org_ble_packet_bytes.data(), 2));
+  ByteArray short_ble_packet_bytes{ByteArray{org_ble_packet_bytes.data(), 2}};
 
-  BlePacket short_ble_packet(short_ble_packet_bytes);
+  BlePacket short_ble_packet{short_ble_packet_bytes};
 
   EXPECT_FALSE(short_ble_packet.IsValid());
 }

cpp/core_v2/internal/mediums/ble_peripheral.h

@@ -26,12 +26,11 @@ class BlePeripheral {
  public:
   BlePeripheral() = default;
   explicit BlePeripheral(const ByteArray& id) : id_(id) {}
-  ~BlePeripheral() = default;
-
   BlePeripheral(const BlePeripheral&) = default;
   BlePeripheral& operator=(const BlePeripheral&) = default;
   BlePeripheral(BlePeripheral&&) = default;
   BlePeripheral& operator=(BlePeripheral&&) = default;
+  ~BlePeripheral() = default;
 
   bool IsValid() const { return !id_.Empty(); }
   ByteArray GetId() const { return id_; }

cpp/core_v2/internal/mediums/ble_peripheral_test.cc

@@ -25,9 +25,9 @@ namespace {
 const char kId[] = "AB12";
 
 TEST(BlePeripheralTest, ConstructionWorks) {
-  ByteArray id(kId);
+  ByteArray id{kId};
 
-  BlePeripheral ble_peripheral(id);
+  BlePeripheral ble_peripheral{id};
 
   EXPECT_TRUE(ble_peripheral.IsValid());
   EXPECT_EQ(id, ble_peripheral.GetId());

cpp/core_v2/internal/mediums/bloom_filter.cc

@@ -0,0 +1,105 @@
+// 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 "core_v2/internal/mediums/bloom_filter.h"
+
+#include "absl/numeric/int128.h"
+#include "absl/strings/numbers.h"
+#include "smhasher/MurmurHash3.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+namespace mediums {
+
+BloomFilterBase::BloomFilterBase(const ByteArray& bytes, BitSet* bit_set)
+    : bits_(bit_set) {
+  const char* bytes_read_ptr = bytes.data();
+  for (size_t byte_index = 0; byte_index < bytes.size(); byte_index++) {
+    for (size_t bit_index = 0; bit_index < 8; bit_index++) {
+      bits_->Set((byte_index * 8) + bit_index,
+                 (*bytes_read_ptr >> bit_index) & 0x01);
+    }
+    bytes_read_ptr++;
+  }
+}
+
+BloomFilterBase::operator ByteArray() const {
+  // Gets a binary string representation of the bitset where the leftmost
+  // character corresponds to bitset position (total size) - 1.
+  //
+  // If the bitset's internal representation is:
+  //     [position 0]     0 0 1 1 0 0 0 1 0 1 0 1     [position 11]
+  // The string representation will be outputted like this:
+  //                     "1 0 1 0 1 0 0 0 1 1 0 0"
+  std::string bitset_binary_string = bits_->ToString();
+
+  ByteArray result_bytes(GetMinBytesForBits());
+  char* result_bytes_write_ptr = result_bytes.data();
+  // We go through the string backwards because the rightmost character
+  // corresponds to position 0 in the bitset.
+  for (size_t i = bits_->Size(); i > 0; i -= 8) {
+    std::string byte_binary_string = bitset_binary_string.substr(i - 8, 8);
+    std::uint32_t byte_value;
+    absl::numbers_internal::safe_strtou32_base(byte_binary_string, &byte_value,
+                                               /* base= */ 2);
+    *result_bytes_write_ptr = static_cast<char>(byte_value & 0x000000FF);
+    result_bytes_write_ptr++;
+  }
+  return result_bytes;
+}
+
+void BloomFilterBase::Add(const std::string& s) {
+  std::vector<std::int32_t> hashes = GetHashes(s);
+  for (int32_t hash : hashes) {
+    size_t position = static_cast<size_t>(hash) % bits_->Size();
+    bits_->Set(position, true);
+  }
+}
+
+bool BloomFilterBase::PossiblyContains(const std::string& s) {
+  std::vector<std::int32_t> hashes = GetHashes(s);
+  for (int32_t hash : hashes) {
+    size_t position = static_cast<size_t>(hash) % bits_->Size();
+    if (!bits_->Test(position)) {
+      return false;
+    }
+  }
+  return true;
+}
+
+std::vector<std::int32_t> BloomFilterBase::GetHashes(const std::string& s) {
+  std::vector<std::int32_t> hashes(kHasherNumberOfRepetitions, 0);
+
+  absl::uint128 hash128;
+  MurmurHash3_x64_128(s.data(), s.size(), 0, &hash128);
+  std::uint64_t hash64 =
+      absl::Uint128Low64(hash128);  // the lower 64 bits of the 128-bit hash
+  std::int32_t hash1 = static_cast<std::int32_t>(
+      hash64 & 0x00000000FFFFFFFF);  // the lower 32 bits of the 64-bit hash
+  std::int32_t hash2 = static_cast<std::int32_t>(
+      (hash64 >> 32) & 0x0FFFFFFFF);  // the upper 32 bits of the 64-bit hash
+  for (size_t i = 1; i <= kHasherNumberOfRepetitions; i++) {
+    std::int32_t combinedHash = static_cast<std::int32_t>(hash1 + (i * hash2));
+    // Flip all the bits if it's negative (guaranteed positive number)
+    if (combinedHash < 0) combinedHash = ~combinedHash;
+    hashes[i - 1] = combinedHash;
+  }
+  return hashes;
+}
+
+}  // namespace mediums
+}  // namespace connections
+}  // namespace nearby
+}  // namespace location

cpp/core_v2/internal/mediums/bloom_filter.h

@@ -0,0 +1,101 @@
+// 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 CORE_V2_INTERNAL_MEDIUMS_BLOOM_FILTER_H_
+#define CORE_V2_INTERNAL_MEDIUMS_BLOOM_FILTER_H_
+
+#include <bitset>
+#include <vector>
+
+#include "platform_v2/base/byte_array.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+namespace mediums {
+
+/**
+ * A bloom filter that gives access to the underlying BitSet. The implementation
+ * is copied from our Java version of Bloom filter, which in turn copies from
+ * Guava's BloomFilter.
+ *
+ * BloomFilter is templatized on the size of the byte array and not the size of
+ * the bit set to ensure the bit set's length is a multiple of 8 (and can
+ * neatly be returned as a ByteArray).
+ */
+class BloomFilterBase {
+ public:
+  explicit operator ByteArray() const;
+
+  void Add(const std::string& s);
+  bool PossiblyContains(const std::string& s);
+
+ protected:
+  class BitSet {
+   public:
+    virtual ~BitSet() = default;
+    virtual std::string ToString() const = 0;
+    virtual void Set(size_t pos, bool value) = 0;
+    virtual bool Test(size_t pos) const = 0;
+    virtual size_t Size() const = 0;
+  };
+
+  BloomFilterBase(const ByteArray& bytes, BitSet* bit_set);
+  virtual ~BloomFilterBase() = default;
+
+  constexpr static int kHasherNumberOfRepetitions = 5;
+  std::vector<std::int32_t> GetHashes(const std::string& s);
+
+ private:
+  int GetMinBytesForBits() const { return (bits_->Size() + 7) >> 3; }
+
+  BitSet* bits_;
+};
+
+template <size_t CapacityInBytes>
+class BloomFilter final : public BloomFilterBase {
+ public:
+  BloomFilter() : BloomFilterBase(ByteArray{}, &bits_) {}
+  explicit BloomFilter(const ByteArray& bytes)
+      : BloomFilterBase(bytes, &bits_) {}
+  BloomFilter(const BloomFilter&) = default;
+  BloomFilter& operator=(const BloomFilter&) = default;
+  BloomFilter(BloomFilter&& other) : BloomFilterBase(ByteArray{}, &bits_) {
+    *this = std::move(other);
+  }
+  BloomFilter& operator=(BloomFilter&& other) {
+    std::swap((*this).bits_, other.bits_);
+    return *this;
+  }
+  ~BloomFilter() override = default;
+
+ private:
+  class BitSetImpl final : public BitSet {
+   public:
+    std::string ToString() const override { return bits_.to_string(); }
+    void Set(size_t pos, bool value) override { bits_.set(pos, value); }
+    bool Test(size_t pos) const override { return bits_.test(pos); }
+    size_t Size() const override { return bits_.size(); }
+
+   private:
+    std::bitset<CapacityInBytes * 8> bits_;
+  } bits_;
+};
+
+}  // namespace mediums
+}  // namespace connections
+}  // namespace nearby
+}  // namespace location
+
+#endif  // CORE_V2_INTERNAL_MEDIUMS_BLOOM_FILTER_H_

cpp/core_v2/internal/mediums/bloom_filter_test.cc

@@ -0,0 +1,207 @@
+// 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 "core_v2/internal/mediums/bloom_filter.h"
+
+#include <algorithm>
+
+#include "gtest/gtest.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+namespace mediums {
+namespace {
+
+const size_t kByteArrayLength = 100;
+
+TEST(BloomFilterTest, EmptyFilterReturnsEmptyArray) {
+  BloomFilter<kByteArrayLength> bloom_filter;
+
+  ByteArray bloom_filter_bytes(bloom_filter);
+  std::string empty_string(kByteArrayLength, '\0');
+
+  EXPECT_EQ(empty_string, std::string(bloom_filter_bytes));
+}
+
+TEST(BloomFilterTest, EmptyFilterNeverContains) {
+  BloomFilter<kByteArrayLength> bloom_filter;
+
+  EXPECT_FALSE(bloom_filter.PossiblyContains("ELEMENT_1"));
+  EXPECT_FALSE(bloom_filter.PossiblyContains("ELEMENT_2"));
+  EXPECT_FALSE(bloom_filter.PossiblyContains("ELEMENT_3"));
+}
+
+TEST(BloomFilterTest, AddSuccess) {
+  BloomFilter<kByteArrayLength> bloom_filter;
+
+  EXPECT_FALSE(bloom_filter.PossiblyContains("ELEMENT_1"));
+
+  bloom_filter.Add("ELEMENT_1");
+
+  EXPECT_TRUE(bloom_filter.PossiblyContains("ELEMENT_1"));
+}
+
+TEST(BloomFilterTest, AddOnlyGivenArg) {
+  BloomFilter<kByteArrayLength> bloom_filter;
+
+  bloom_filter.Add("ELEMENT_1");
+
+  EXPECT_TRUE(bloom_filter.PossiblyContains("ELEMENT_1"));
+  EXPECT_FALSE(bloom_filter.PossiblyContains("ELEMENT_2"));
+  EXPECT_FALSE(bloom_filter.PossiblyContains("ELEMENT_3"));
+}
+
+TEST(BloomFilterTest, AddMultipleArgs) {
+  BloomFilter<kByteArrayLength> bloom_filter;
+
+  bloom_filter.Add("ELEMENT_1");
+  bloom_filter.Add("ELEMENT_2");
+
+  EXPECT_TRUE(bloom_filter.PossiblyContains("ELEMENT_1"));
+  EXPECT_TRUE(bloom_filter.PossiblyContains("ELEMENT_2"));
+  EXPECT_FALSE(bloom_filter.PossiblyContains("ELEMENT_3"));
+}
+
+TEST(BloomFilterTest, AddMultipleArgsReturnsNonemptyArray) {
+  BloomFilter<10> bloom_filter;
+
+  bloom_filter.Add("ELEMENT_1");
+  bloom_filter.Add("ELEMENT_2");
+  bloom_filter.Add("ELEMENT_3");
+
+  ByteArray bloom_filter_bytes(bloom_filter);
+  std::string empty_string(kByteArrayLength, '\0');
+
+  EXPECT_NE(std::string(bloom_filter_bytes), empty_string);
+}
+
+TEST(BloomFilterTest, CopyConstructorAndAssignmentSuccess) {
+  BloomFilter<kByteArrayLength> bloom_filter;
+
+  EXPECT_FALSE(bloom_filter.PossiblyContains("ELEMENT_1"));
+
+  bloom_filter.Add("ELEMENT_1");
+
+  BloomFilter<kByteArrayLength> bloom_filter_copy_1{bloom_filter};
+  BloomFilter<kByteArrayLength> bloom_filter_copy_2 = bloom_filter;
+
+  EXPECT_TRUE(bloom_filter.PossiblyContains("ELEMENT_1"));
+  EXPECT_TRUE(bloom_filter_copy_1.PossiblyContains("ELEMENT_1"));
+  EXPECT_TRUE(bloom_filter_copy_2.PossiblyContains("ELEMENT_1"));
+}
+
+TEST(BloomFilterTest, MoveConstructorSuccess) {
+  BloomFilter<kByteArrayLength> bloom_filter;
+
+  bloom_filter.Add("ELEMENT_1");
+
+  BloomFilter<kByteArrayLength> bloom_filter_move{std::move(bloom_filter)};
+
+  EXPECT_TRUE(bloom_filter_move.PossiblyContains("ELEMENT_1"));
+}
+
+TEST(BloomFilterTest, MoveAssignmentSuccess) {
+  BloomFilter<kByteArrayLength> bloom_filter;
+
+  bloom_filter.Add("ELEMENT_1");
+
+  BloomFilter<kByteArrayLength> bloom_filter_move = std::move(bloom_filter);
+
+  EXPECT_TRUE(bloom_filter_move.PossiblyContains("ELEMENT_1"));
+}
+
+/**
+ * This test was added because of a bug where the BloomFilter doesn't utilize
+ * all bits given. Functionally, the filter still works, but we just have a much
+ * higher false positive rate. The bug was caused by confusing bit length and
+ * byte length, which made our BloomFilter only set bits on the first byteLength
+ * (bitLength / 8) bits rather than the whole bitLength bits.
+ *
+ * <p>Here, we're verifying that the bits set are somewhat scattered. So instead
+ * of something like [ 0, 1, 1, 0, 0, 0, 0, ..., 0 ], we should be getting
+ * something like [ 0, 1, 0, 0, 1, 1, 0, 0, 0, 1, ..., 1, 0].
+ */
+TEST(BloomFilterTest, RandomnessNoEndBias) {
+  BloomFilter<kByteArrayLength> bloom_filter;
+
+  // Add one element to our BloomFilter.
+  bloom_filter.Add("ELEMENT_1");
+
+  std::int32_t non_zero_count = 0;
+  std::int32_t longest_zero_streak = 0;
+  std::int32_t current_zero_streak = 0;
+
+  // Record the amount of non-zero bytes and the longest streak of zero bytes in
+  // the resulting BloomFilter. This is an approximation of reasonable
+  // distribution since we're recording by bytes instead of bits.
+  ByteArray bloom_filter_bytes(bloom_filter);
+  const char* bloom_filter_bytes_read_ptr = bloom_filter_bytes.data();
+  for (int i = 0; i < bloom_filter_bytes.size(); i++) {
+    if (*bloom_filter_bytes_read_ptr == '\0') {
+      current_zero_streak++;
+    } else {
+      // Increment the number of non-zero bytes we've seen, update the longest
+      // zero streak, and then reset the current zero streak.
+      non_zero_count++;
+      longest_zero_streak = std::max(longest_zero_streak, current_zero_streak);
+      current_zero_streak = 0;
+    }
+    bloom_filter_bytes_read_ptr++;
+  }
+  // Update the longest zero streak again for the tail case.
+  longest_zero_streak = std::min(longest_zero_streak, current_zero_streak);
+
+  // Since randomness is hard to measure within one unit test, we instead do a
+  // sanity check. All non-zero bytes should not be packed into one end of the
+  // array.
+  //
+  // In this case, the size of one end is approximated to be:
+  //     kByteArrayLength / nonZeroCount.
+  // Therefore, the longest zero streak should be less than:
+  //     kByteArrayLength - one end of the array.
+  std::int32_t longest_acceptable_zero_streak =
+      kByteArrayLength - (kByteArrayLength / non_zero_count);
+
+  EXPECT_TRUE(longest_zero_streak <= longest_acceptable_zero_streak);
+}
+
+TEST(BloomFilterTest, RandomnessFalsePositiveRate) {
+  BloomFilter<10> bloom_filter;
+
+  // Add 5 distinct elements to the BloomFilter.
+  bloom_filter.Add("ELEMENT_1");
+  bloom_filter.Add("ELEMENT_2");
+  bloom_filter.Add("ELEMENT_3");
+  bloom_filter.Add("ELEMENT_4");
+  bloom_filter.Add("ELEMENT_5");
+
+  std::int32_t false_positives = 0;
+  // Now test 100 other elements and record the number of false positives.
+  for (int i = 5; i < 105; i++) {
+    false_positives +=
+        bloom_filter.PossiblyContains("ELEMENT_" + std::to_string(i)) ? 1 : 0;
+  }
+
+  // We expect the false positive rate to be 3% with 5 elements in a 10 byte
+  // filter. Thus, we give a little leeway and verify that the false positive
+  // rate is no more than 5%.
+  EXPECT_LE(false_positives, 5);
+}
+
+}  // namespace
+}  // namespace mediums
+}  // namespace connections
+}  // namespace nearby
+}  // namespace location

cpp/core_v2/internal/mediums/webrtc/BUILD

@@ -15,27 +15,39 @@
 cc_library(
     name = "webrtc",
     srcs = [
+        "connection_flow.cc",
+        "peer_connection_observer_impl.cc",
         "webrtc_socket.cc",
     ],
     hdrs = [
+        "connection_flow.h",
+        "data_channel_listener.h",
+        "local_ice_candidate_listener.h",
+        "peer_connection_observer_impl.h",
         "webrtc_socket.h",
     ],
     deps = [
         "//core_v2:core_types",
         "//platform_v2/base",
-        "//platform_v2/public",
+        "//platform_v2/public:comm",
         "//platform_v2/public:logging",
+        "//platform_v2/public:types",
+        "//absl/memory",
         "//webrtc/api:libjingle_peerconnection_api",
     ],
 )
 
 cc_test(
     name = "webrtc_test",
-    srcs = ["webrtc_socket_test.cc"],
+    srcs = [
+        "connection_flow_test.cc",
+        "webrtc_socket_test.cc",
+    ],
     deps = [
         ":webrtc",
         "//platform_v2/base",
         "//platform_v2/impl/g3",  # buildcleaner: keep
+        "//platform_v2/public:comm",
         "//testing/base/public:gunit_main",
         "//webrtc/api:libjingle_peerconnection_api",
     ],
@@ -48,7 +60,8 @@ cc_test(
         ":peer_id",
         "//platform_v2/base",
         "//platform_v2/impl/g3",  #buildcleaner: keep
-        "//platform_v2/public",
+        "//platform_v2/public:comm",
+        "//platform_v2/public:types",
         "//testing/base/public:gunit_main",
     ],
 )

cpp/core_v2/internal/mediums/webrtc/connection_flow.cc

@@ -0,0 +1,148 @@
+// 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 "core_v2/internal/mediums/webrtc/connection_flow.h"
+
+#include <memory>
+
+#include "platform_v2/public/mutex_lock.h"
+#include "platform_v2/public/webrtc.h"
+#include "absl/memory/memory.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+namespace mediums {
+
+std::unique_ptr<ConnectionFlow> ConnectionFlow::Create(
+    LocalIceCandidateListener local_ice_candidate_listener,
+    DataChannelListener data_channel_listener,
+    SingleThreadExecutor* single_threaded_executor,
+    WebRtcMedium& webrtc_medium) {
+  auto connection_flow = absl::WrapUnique(new ConnectionFlow(
+      std::move(local_ice_candidate_listener), std::move(data_channel_listener),
+      single_threaded_executor));
+  if (connection_flow->InitPeerConnection(webrtc_medium)) {
+    return connection_flow;
+  }
+
+  return nullptr;
+}
+
+ConnectionFlow::ConnectionFlow(
+    LocalIceCandidateListener local_ice_candidate_listener,
+    DataChannelListener data_channel_listener,
+    SingleThreadExecutor* single_threaded_executor)
+    : data_channel_listener_(std::move(data_channel_listener)),
+      peer_connection_observer_(this, std::move(local_ice_candidate_listener),
+                                single_threaded_executor) {}
+
+std::unique_ptr<webrtc::SessionDescriptionInterface>
+ConnectionFlow::CreateOffer() {
+  MutexLock lock(&mutex_);
+
+  // TODO(bfranz): Implement
+
+  return std::unique_ptr<webrtc::SessionDescriptionInterface>();
+}
+
+std::unique_ptr<webrtc::SessionDescriptionInterface>
+ConnectionFlow::CreateAnswer() {
+  MutexLock lock(&mutex_);
+
+  // TODO(bfranz): Implement
+
+  return std::unique_ptr<webrtc::SessionDescriptionInterface>();
+}
+
+bool ConnectionFlow::SetLocalSessionDescription(
+    std::unique_ptr<webrtc::SessionDescriptionInterface> sdp) {
+  MutexLock lock(&mutex_);
+
+  // TODO(bfranz): Implement
+
+  return false;
+}
+
+void ConnectionFlow::OnOfferReceived(
+    std::unique_ptr<webrtc::SessionDescriptionInterface> offer) {
+  MutexLock lock(&mutex_);
+
+  // TODO(bfranz): Implement
+}
+
+void ConnectionFlow::OnAnswerReceived(
+    std::unique_ptr<webrtc::SessionDescriptionInterface> answer) {
+  MutexLock lock(&mutex_);
+
+  // TODO(bfranz): Implement
+}
+
+bool ConnectionFlow::OnRemoteIceCandidatesReceived(
+    std::vector<webrtc::IceCandidateInterface*> ice_candidates) {
+  MutexLock lock(&mutex_);
+
+  // TODO(bfranz): Implement
+
+  return false;
+}
+
+api::ListenableFuture<rtc::scoped_refptr<webrtc::DataChannelInterface>>*
+ConnectionFlow::GetDataChannel() {
+  return static_cast<
+      api::ListenableFuture<rtc::scoped_refptr<webrtc::DataChannelInterface>>*>(
+      &data_channel_future_);
+}
+
+bool ConnectionFlow::Close() {
+  MutexLock lock(&mutex_);
+
+  // TODO(bfranz): Implement
+
+  return false;
+}
+
+bool ConnectionFlow::InitPeerConnection(WebRtcMedium& webrtc_medium) {
+  Future<bool> success_future;
+  webrtc_medium.CreatePeerConnection(
+      &peer_connection_observer_,
+      [this, &success_future](
+          rtc::scoped_refptr<webrtc::PeerConnectionInterface> peer_connection) {
+        peer_connection_ = peer_connection;
+        success_future.Set(true);
+      });
+
+  ExceptionOr<bool> result = success_future.Get(kTimeout);
+  return result.ok() && result.result();
+}
+
+void ConnectionFlow::OnSignalingStable() {
+  // TODO(bfranz): Implement
+}
+
+void ConnectionFlow::ProcessOnPeerConnectionChange(
+    webrtc::PeerConnectionInterface::PeerConnectionState new_state) {
+  // TODO(bfranz): Implement
+}
+
+webrtc::DataChannelObserver* ConnectionFlow::CreateDataChannelObserver(
+    rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel) {
+  // TODO(bfranz): Implement
+
+  return nullptr;
+}
+}  // namespace mediums
+}  // namespace connections
+}  // namespace nearby
+}  // namespace location

cpp/core_v2/internal/mediums/webrtc/connection_flow.h

@@ -0,0 +1,147 @@
+// 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 CORE_V2_INTERNAL_MEDIUMS_WEBRTC_CONNECTION_FLOW_H_
+#define CORE_V2_INTERNAL_MEDIUMS_WEBRTC_CONNECTION_FLOW_H_
+
+#include <memory>
+
+#include "core_v2/internal/mediums/webrtc/data_channel_listener.h"
+#include "core_v2/internal/mediums/webrtc/local_ice_candidate_listener.h"
+#include "core_v2/internal/mediums/webrtc/peer_connection_observer_impl.h"
+#include "platform_v2/base/runnable.h"
+#include "platform_v2/public/future.h"
+#include "platform_v2/public/single_thread_executor.h"
+#include "platform_v2/public/webrtc.h"
+#include "webrtc/api/data_channel_interface.h"
+#include "webrtc/api/peer_connection_interface.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+namespace mediums {
+
+/**
+ * Flow for an offerer:
+ *
+ * <ul>
+ *   <li>INITIALIZED: After construction.
+ *   <li>CREATING_OFFER: After CreateOffer(). Local ice candidate collection
+ * begins.
+ *   <li>WAITING_FOR_ANSWER: Until the remote peer sends their answer.
+ *   <li>WAITING_TO_CONNECT: Until the data channel actually connects. Remote
+ * ice candidates should be added with OnRemoteIceCandidatesReceived as they are
+ * gathered.
+ *   <li>CONNECTED: We successfully connected to the remote data
+ * channel.
+ *   <li>ENDED: The final state that can occur from any of the previous
+ * states if we disconnect at any point in the flow.
+ * </ul>
+ *
+ * <p>Flow for an answerer:
+ *
+ * <ul>
+ *   <li>INITIALIZED: After construction.
+ *   <li>RECEIVED_OFFER: After onOfferReceived().
+ *   <li>CREATING_ANSWER: After CreateAnswer(). Local ice candidate collection
+ * begins.
+ *   <li>WAITING_TO_CONNECT: Until the data channel actually connects.
+ * Remote ice candidates should be added with OnRemoteIceCandidatesReceived as
+ * they are gathered.
+ *   <li>CONNECTED: We successfully connected to the remote
+ * data channel.
+ *   <li>ENDED: The final state that can occur from any of the
+ * previous states if we disconnect at any point in the flow.
+ * </ul>
+ */
+class ConnectionFlow {
+ public:
+  // This method blocks on the creation of the peer connection object.
+  static std::unique_ptr<ConnectionFlow> Create(
+      LocalIceCandidateListener local_ice_candidate_listener,
+      DataChannelListener data_channel_listener,
+      SingleThreadExecutor* single_threaded_executor,
+      WebRtcMedium& webrtc_medium);
+  ~ConnectionFlow() = default;
+
+  // Create the offer that will be sent to the remote. Mirrors the behaviour of
+  // PeerConnectionInterface::CreateOffer.
+  std::unique_ptr<webrtc::SessionDescriptionInterface> CreateOffer()
+      ABSL_LOCKS_EXCLUDED(mutex_);
+  // Create the answer that will be sent to the remote. Mirrors the behaviour of
+  // PeerConnectionInterface::CreateAnswer.
+  std::unique_ptr<webrtc::SessionDescriptionInterface> CreateAnswer()
+      ABSL_LOCKS_EXCLUDED(mutex_);
+  // Set the local session description. |sdp| was created via CreateOffer()
+  // or CreateAnswer().
+  bool SetLocalSessionDescription(
+      std::unique_ptr<webrtc::SessionDescriptionInterface> sdp)
+      ABSL_LOCKS_EXCLUDED(mutex_);
+  // Invoked when an offer was received from a remote; this will set the remote
+  // session description on the peer connection.
+  void OnOfferReceived(
+      std::unique_ptr<webrtc::SessionDescriptionInterface> offer)
+      ABSL_LOCKS_EXCLUDED(mutex_);
+  // Invoked when an answer was received from a remote; this will set the remote
+  // session description on the peer connection.
+  void OnAnswerReceived(
+      std::unique_ptr<webrtc::SessionDescriptionInterface> answer)
+      ABSL_LOCKS_EXCLUDED(mutex_);
+  // Invoked when an ice candidate was received from a remote; this will add the
+  // ice candidate to the peer connection if ready or cache it otherwise.
+  bool OnRemoteIceCandidatesReceived(
+      std::vector<webrtc::IceCandidateInterface*> ice_candidates)
+      ABSL_LOCKS_EXCLUDED(mutex_);
+  // Get a future for the data channel.
+  api::ListenableFuture<rtc::scoped_refptr<webrtc::DataChannelInterface>>*
+  GetDataChannel();
+  // Close the peer connection and data channel.
+  bool Close() ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // Invoked when the peer connection indicates that signaling is stable.
+  void OnSignalingStable();
+  webrtc::DataChannelObserver* CreateDataChannelObserver(
+      rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel);
+
+  // Invoked upon changes in the state of peer connection, e.g. react to
+  // disconnect.
+  void ProcessOnPeerConnectionChange(
+      webrtc::PeerConnectionInterface::PeerConnectionState new_state);
+
+ private:
+  ConnectionFlow(LocalIceCandidateListener local_ice_candidate_listener,
+                 DataChannelListener data_channel_listener,
+                 SingleThreadExecutor* single_threaded_executor);
+
+  // TODO(bfranz): Consider whether this needs to be configurable per platform
+  static constexpr absl::Duration kTimeout = absl::Milliseconds(250);
+
+  bool InitPeerConnection(WebRtcMedium& webrtc_medium);
+
+  DataChannelListener data_channel_listener_;
+
+  Future<rtc::scoped_refptr<webrtc::DataChannelInterface>> data_channel_future_;
+
+  PeerConnectionObserverImpl peer_connection_observer_;
+  rtc::scoped_refptr<webrtc::PeerConnectionInterface> peer_connection_;
+
+  Mutex mutex_;
+};
+
+}  // namespace mediums
+}  // namespace connections
+}  // namespace nearby
+}  // namespace location
+
+#endif  // CORE_V2_INTERNAL_MEDIUMS_WEBRTC_CONNECTION_FLOW_H_

cpp/core_v2/internal/mediums/webrtc/connection_flow_test.cc

@@ -0,0 +1,46 @@
+// 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 "core_v2/internal/mediums/webrtc/connection_flow.h"
+
+#include <memory>
+
+#include "platform_v2/public/webrtc.h"
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+namespace mediums {
+namespace {
+
+TEST(ConnectionFlowTest, Create) {
+  LocalIceCandidateListener local_ice_candidate_listener;
+  DataChannelListener data_channel_listener;
+  SingleThreadExecutor executor;
+  WebRtcMedium webrtc_medium;
+
+  std::unique_ptr<ConnectionFlow> connection_flow = ConnectionFlow::Create(
+      std::move(local_ice_candidate_listener), std::move(data_channel_listener),
+      &executor, webrtc_medium);
+
+  EXPECT_NE(connection_flow, nullptr);
+}
+
+}  // namespace
+}  // namespace mediums
+}  // namespace connections
+}  // namespace nearby
+}  // namespace location

cpp/core_v2/internal/mediums/webrtc/data_channel_listener.h

@@ -0,0 +1,45 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef CORE_V2_INTERNAL_MEDIUMS_WEBRTC_DATA_CHANNEL_LISTENER_H_
+#define CORE_V2_INTERNAL_MEDIUMS_WEBRTC_DATA_CHANNEL_LISTENER_H_
+
+#include "core_v2/listeners.h"
+#include "platform_v2/base/byte_array.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+namespace mediums {
+
+// Callbacks from the data channel.
+struct DataChannelListener {
+  std::function<void()> data_channel_closed_cb = DefaultCallback<>();
+
+  // Called when a new message was received on the data channel.
+  std::function<void(ByteArray)> data_channel_message_received_cb =
+      DefaultCallback<ByteArray>();
+
+  // Called when the data channel indicates that the buffered amount has
+  // changed.
+  std::function<void()> data_channel_buffered_amount_changed_cb =
+      DefaultCallback<>();
+};
+
+}  // namespace mediums
+}  // namespace connections
+}  // namespace nearby
+}  // namespace location
+
+#endif  // CORE_V2_INTERNAL_MEDIUMS_WEBRTC_DATA_CHANNEL_LISTENER_H_

cpp/core_v2/internal/mediums/webrtc/local_ice_candidate_listener.h

@@ -0,0 +1,39 @@
+// 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 CORE_V2_INTERNAL_MEDIUMS_WEBRTC_LOCAL_ICE_CANDIDATE_LISTENER_H_
+#define CORE_V2_INTERNAL_MEDIUMS_WEBRTC_LOCAL_ICE_CANDIDATE_LISTENER_H_
+
+#include "core_v2/listeners.h"
+#include "webrtc/api/peer_connection_interface.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+namespace mediums {
+
+// Callbacks from local ice candidate collection.
+struct LocalIceCandidateListener {
+  // Called when a new local ice candidate has been found.
+  std::function<void(const webrtc::IceCandidateInterface*)>
+      local_ice_candidate_found_cb = location::nearby::DefaultCallback<
+          const webrtc::IceCandidateInterface*>();
+};
+
+}  // namespace mediums
+}  // namespace connections
+}  // namespace nearby
+}  // namespace location
+
+#endif  // CORE_V2_INTERNAL_MEDIUMS_WEBRTC_LOCAL_ICE_CANDIDATE_LISTENER_H_

cpp/core_v2/internal/mediums/webrtc/peer_connection_observer_impl.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 "core_v2/internal/mediums/webrtc/peer_connection_observer_impl.h"
+
+#include "core_v2/internal/mediums/webrtc/connection_flow.h"
+#include "platform_v2/public/logging.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+namespace mediums {
+
+PeerConnectionObserverImpl::PeerConnectionObserverImpl(
+    ConnectionFlow* connection_flow,
+    LocalIceCandidateListener local_ice_candidate_listener,
+    SingleThreadExecutor* executor)
+    : connection_flow_(connection_flow),
+      local_ice_candidate_listener_(std::move(local_ice_candidate_listener)),
+      single_threaded_signaling_offloader_(executor) {}
+
+void PeerConnectionObserverImpl::OnIceCandidate(
+    const webrtc::IceCandidateInterface* candidate) {
+  NEARBY_LOG(INFO, "OnIceCandidate");
+  local_ice_candidate_listener_.local_ice_candidate_found_cb(candidate);
+}
+
+void PeerConnectionObserverImpl::OnSignalingChange(
+    webrtc::PeerConnectionInterface::SignalingState new_state) {
+  NEARBY_LOG(INFO, "OnSignalingChange: %d", new_state);
+
+  OffloadFromSignalingThread([this, new_state]() {
+    if (new_state == webrtc::PeerConnectionInterface::SignalingState::kStable)
+      connection_flow_->OnSignalingStable();
+  });
+}
+
+void PeerConnectionObserverImpl::OnDataChannel(
+    rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel) {
+  NEARBY_LOG(INFO, "OnDataChannel");
+
+  data_channel->RegisterObserver(
+      connection_flow_->CreateDataChannelObserver(data_channel));
+}
+
+void PeerConnectionObserverImpl::OnIceGatheringChange(
+    webrtc::PeerConnectionInterface::IceGatheringState new_state) {
+  NEARBY_LOG(INFO, "OnIceGatheringChange: %d", new_state);
+}
+
+void PeerConnectionObserverImpl::OnConnectionChange(
+    webrtc::PeerConnectionInterface::PeerConnectionState new_state) {
+  NEARBY_LOG(INFO, "OnConnectionChange: %d", new_state);
+
+  OffloadFromSignalingThread([this, new_state]() {
+    connection_flow_->ProcessOnPeerConnectionChange(new_state);
+  });
+}
+
+void PeerConnectionObserverImpl ::OnRenegotiationNeeded() {
+  NEARBY_LOG(INFO, "OnRenegotiationNeeded");
+}
+
+void PeerConnectionObserverImpl::OffloadFromSignalingThread(Runnable runnable) {
+  single_threaded_signaling_offloader_->Execute(std::move(runnable));
+}
+
+}  // namespace mediums
+}  // namespace connections
+}  // namespace nearby
+}  // namespace location

cpp/core_v2/internal/mediums/webrtc/peer_connection_observer_impl.h

@@ -0,0 +1,62 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef CORE_V2_INTERNAL_MEDIUMS_WEBRTC_PEER_CONNECTION_OBSERVER_IMPL_H_
+#define CORE_V2_INTERNAL_MEDIUMS_WEBRTC_PEER_CONNECTION_OBSERVER_IMPL_H_
+
+#include "core_v2/internal/mediums/webrtc/local_ice_candidate_listener.h"
+#include "platform_v2/public/single_thread_executor.h"
+#include "webrtc/api/peer_connection_interface.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+namespace mediums {
+
+class ConnectionFlow;
+
+class PeerConnectionObserverImpl : public webrtc::PeerConnectionObserver {
+ public:
+  ~PeerConnectionObserverImpl() override = default;
+  PeerConnectionObserverImpl(
+      ConnectionFlow* connection_flow,
+      LocalIceCandidateListener local_ice_candidate_listener,
+      SingleThreadExecutor* executor);
+
+  // webrtc::PeerConnectionObserver:
+  void OnIceCandidate(const webrtc::IceCandidateInterface* candidate) override;
+  void OnSignalingChange(
+      webrtc::PeerConnectionInterface::SignalingState new_state) override;
+  void OnDataChannel(
+      rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel) override;
+  void OnIceGatheringChange(
+      webrtc::PeerConnectionInterface::IceGatheringState new_state) override;
+  void OnConnectionChange(
+      webrtc::PeerConnectionInterface::PeerConnectionState new_state) override;
+  void OnRenegotiationNeeded() override;
+
+ private:
+  void OffloadFromSignalingThread(Runnable runnable);
+
+  ConnectionFlow* connection_flow_;
+  LocalIceCandidateListener local_ice_candidate_listener_;
+  SingleThreadExecutor* single_threaded_signaling_offloader_;
+};
+
+}  // namespace mediums
+}  // namespace connections
+}  // namespace nearby
+}  // namespace location
+
+#endif  // CORE_V2_INTERNAL_MEDIUMS_WEBRTC_PEER_CONNECTION_OBSERVER_IMPL_H_