Merge branch 'google:main' into linux-platform

connections/BUILD

@@ -78,7 +78,7 @@ cc_library(
         "//internal/platform:types",
         "//internal/platform:util",
         "//proto:connections_enums_cc_proto",
-        "@com_google_absl//absl/strings",
+        "@com_google_absl//absl/functional:any_invocable",
         "@com_google_absl//absl/types:variant",
     ],
 )

connections/c/core_adapter.cc

@@ -213,7 +213,8 @@ void AcceptConnection(connections::Core *pCore, const char *endpoint_id,
   }
   connections::PayloadListener payload_listener =
       std::move(*listener.GetImpl());
-  pCore->AcceptConnection(endpoint_id, payload_listener, *callback.GetImpl());
+  pCore->AcceptConnection(endpoint_id, std::move(payload_listener),
+                          *callback.GetImpl());
 }
 
 void RejectConnection(connections::Core *pCore, const char *endpoint_id,

connections/c/listeners_w.cc

@@ -193,7 +193,7 @@ PayloadListenerW::PayloadListenerW(PayloadCB payloadCB,
           new connections::PayloadListener())) {
   CHECK(payload_cb != nullptr);
   auto pcb = payload_cb;
-  impl_->payload_cb = [pcb](const std::string &endpoint_id,
+  impl_->payload_cb = [pcb](absl::string_view endpoint_id,
                             connections::Payload payload) {
     PayloadW payloadW;
 
@@ -221,13 +221,13 @@ PayloadListenerW::PayloadListenerW(PayloadCB payloadCB,
         break;
       }
     }
-    pcb(endpoint_id.c_str(), payloadW);
+    pcb(std::string(endpoint_id).c_str(), payloadW);
   };
 
   CHECK(payload_progress_cb != nullptr);
   auto ppcb = payload_progress_cb;
   impl_->payload_progress_cb =
-      [ppcb](const std::string &endpoint_id,
+      [ppcb](absl::string_view endpoint_id,
              connections::PayloadProgressInfo payload_progress_info) {
         PayloadProgressInfoW payload_progress_info_w;
         payload_progress_info_w.payload_id = payload_progress_info.payload_id;
@@ -254,7 +254,7 @@ PayloadListenerW::PayloadListenerW(PayloadCB payloadCB,
             break;
         }
 
-        ppcb(endpoint_id.c_str(), payload_progress_info_w);
+        ppcb(std::string(endpoint_id).c_str(), payload_progress_info_w);
       };
 }
 

connections/clients/ios/Internal/GNCPayloadListener.h

@@ -41,8 +41,8 @@ class GNCPayloadListener : public PayloadListener {
                      GNCPayloadsProvider payloadsProvider)
       : core_(core), handlers_provider_(handlersProvider), payloads_provider_(payloadsProvider) {}
 
-  void OnPayload(const std::string &endpoint_id, Payload payload);
-  void OnPayloadProgress(const std::string &endpoint_id, const PayloadProgressInfo &info);
+  void OnPayload(absl::string_view endpoint_id, Payload payload);
+  void OnPayloadProgress(absl::string_view endpoint_id, const PayloadProgressInfo &info);
 
  private:
   GNCCore *core_;

connections/clients/ios/Internal/GNCPayloadListener.mm

@@ -35,7 +35,7 @@ NS_ASSUME_NONNULL_BEGIN
 namespace nearby {
 namespace connections {
 
-void GNCPayloadListener::OnPayload(const std::string &endpoint_id, Payload payload) {
+void GNCPayloadListener::OnPayload(absl::string_view endpoint_id, Payload payload) {
   GNCConnectionHandlers *handlers = handlers_provider_();
   int64_t payloadId = payload.GetId();
 
@@ -180,7 +180,7 @@ void GNCPayloadListener::OnPayload(const std::string &endpoint_id, Payload paylo
   }
 }
 
-void GNCPayloadListener::OnPayloadProgress(const std::string &endpoint_id,
+void GNCPayloadListener::OnPayloadProgress(absl::string_view endpoint_id,
                                            const PayloadProgressInfo &info) {
   // Note: The logic in this callback for handling progress updates and payload completion is
   // identical for Bytes, Stream and File payloads.

connections/core.cc

@@ -128,7 +128,8 @@ void Core::AcceptConnection(absl::string_view endpoint_id,
                             PayloadListener listener, ResultCallback callback) {
   assert(!endpoint_id.empty());
 
-  router_->AcceptConnection(&client_, endpoint_id, listener, callback);
+  router_->AcceptConnection(&client_, endpoint_id, std::move(listener),
+                            callback);
 }
 
 void Core::RejectConnection(absl::string_view endpoint_id,

connections/core.h

@@ -17,6 +17,7 @@
 
 #include <memory>
 #include <string>
+#include <utility>
 
 #include "absl/strings/string_view.h"
 #include "absl/types/span.h"
@@ -507,7 +508,9 @@ class Core {
 
   // Registers a DeviceProvider to provide functionality for Nearby Connections
   // to interact with the DeviceProvider for retrieving the local device.
-  void RegisterDeviceProvider(std::unique_ptr<NearbyDeviceProvider> provider);
+  void RegisterDeviceProvider(std::unique_ptr<NearbyDeviceProvider> provider) {
+    client_.RegisterDeviceProvider(std::move(provider));
+  }
 
  private:
   ClientProxy client_;

connections/implementation/base_pcp_handler.cc

@@ -870,13 +870,14 @@ bool BasePcpHandler::AutoUpgradeBandwidth(
   return local_advertising_options.auto_upgrade_bandwidth;
 }
 
-Status BasePcpHandler::AcceptConnection(
-    ClientProxy* client, const std::string& endpoint_id,
-    const PayloadListener& payload_listener) {
+Status BasePcpHandler::AcceptConnection(ClientProxy* client,
+                                        const std::string& endpoint_id,
+                                        PayloadListener payload_listener) {
   Future<Status> response;
   RunOnPcpHandlerThread(
-      "accept-connection", [this, client, endpoint_id, payload_listener,
-                            &response]() RUN_ON_PCP_HANDLER_THREAD() {
+      "accept-connection", [this, client, endpoint_id,
+                            payload_listener = std::move(payload_listener),
+                            &response]() RUN_ON_PCP_HANDLER_THREAD() mutable {
         NEARBY_LOGS(INFO) << "AcceptConnection: endpoint_id=" << endpoint_id;
         if (!pending_connections_.count(endpoint_id)) {
           NEARBY_LOGS(INFO)
@@ -918,8 +919,8 @@ Status BasePcpHandler::AcceptConnection(
 
         NEARBY_LOGS(INFO) << "AcceptConnection: accepting locally: endpoint_id="
                           << endpoint_id;
-        connection_info.LocalEndpointAcceptedConnection(endpoint_id,
-                                                        payload_listener);
+        connection_info.LocalEndpointAcceptedConnection(
+            endpoint_id, std::move(payload_listener));
         EvaluateConnectionResult(client, endpoint_id,
                                  false /* can_close_immediately */);
         response.Set({Status::kSuccess});
@@ -1118,29 +1119,41 @@ void BasePcpHandler::OnEndpointFound(
 void BasePcpHandler::OnEndpointLost(
     ClientProxy* client, const BasePcpHandler::DiscoveredEndpoint& endpoint) {
   // Look up the DiscoveredEndpoint we have in our cache.
-  const auto* discovered_endpoint = GetDiscoveredEndpoint(endpoint.endpoint_id);
-  if (discovered_endpoint == nullptr) {
+  NEARBY_LOGS(INFO) << "OnEndpointLost: id=" << endpoint.endpoint_id;
+
+  auto range = discovered_endpoints_.equal_range(endpoint.endpoint_id);
+  bool is_range_empty = range.first == range.second;
+  if (is_range_empty) {
     NEARBY_LOGS(INFO) << "No previous endpoint (nothing to lose): endpoint_id="
                       << endpoint.endpoint_id;
     return;
   }
+  int count = discovered_endpoints_.count(endpoint.endpoint_id);
+  absl::btree_multimap<std::string,
+                       std::shared_ptr<DiscoveredEndpoint>>::iterator item;
+  for (item = range.first; item != range.second; ++item) {
+    auto& discovered_endpoint = item->second;
+    if (discovered_endpoint->medium != endpoint.medium) continue;
 
-  // Validate that the cached endpoint has the same info as the one reported as
-  // onLost. If the info differs, then no-op. This likely means that the remote
-  // device changed their info. We reported onFound for the new info and are
-  // just now figuring out that we lost the old info.
-  if (discovered_endpoint->endpoint_info != endpoint.endpoint_info) {
-    NEARBY_LOGS(INFO) << "Previous endpoint name mismatch; passed="
-                      << absl::BytesToHexString(endpoint.endpoint_info.data())
-                      << "; expected="
-                      << absl::BytesToHexString(
-                             discovered_endpoint->endpoint_info.data());
-    return;
-  }
-
-  auto item = discovered_endpoints_.extract(endpoint.endpoint_id);
-  if (!discovered_endpoints_.count(endpoint.endpoint_id)) {
-    client->OnEndpointLost(endpoint.service_id, endpoint.endpoint_id);
+    // Validate that the cached endpoint has the same info as the one reported
+    // as onLost. If the info differs, we still remove it. This likely means
+    // that the remote device changed their info. We reported onFound for the
+    // new info and are just now figuring out that we lost the old info.
+    if (discovered_endpoint->endpoint_info != endpoint.endpoint_info) {
+      NEARBY_LOGS(INFO) << "Previous endpoint name mismatch; passed="
+                        << absl::BytesToHexString(endpoint.endpoint_info.data())
+                        << "; expected="
+                        << absl::BytesToHexString(
+                               discovered_endpoint->endpoint_info.data());
+    }
+    NEARBY_LOGS(INFO) << "Erase Endpoint with Meduim: "
+                      << location::nearby::proto::connections::Medium_Name(
+                             discovered_endpoint->medium);
+    if (--count == 0) {
+      client->OnEndpointLost(endpoint.service_id, endpoint.endpoint_id);
+    }
+    discovered_endpoints_.erase(item);
+    break;
   }
 }
 
@@ -1722,8 +1735,9 @@ BasePcpHandler::PendingConnectionInfo::~PendingConnectionInfo() {
 }
 
 void BasePcpHandler::PendingConnectionInfo::LocalEndpointAcceptedConnection(
-    const std::string& endpoint_id, const PayloadListener& payload_listener) {
-  client->LocalEndpointAcceptedConnection(endpoint_id, payload_listener);
+    const std::string& endpoint_id, PayloadListener payload_listener) {
+  client->LocalEndpointAcceptedConnection(endpoint_id,
+                                          std::move(payload_listener));
 }
 
 void BasePcpHandler::PendingConnectionInfo::LocalEndpointRejectedConnection(

connections/implementation/base_pcp_handler.h

@@ -69,7 +69,6 @@ class BasePcpHandler : public PcpHandler,
  public:
   using FrameProcessor = EndpointManager::FrameProcessor;
 
-  // TODO(apolyudov): Add SecureRandom.
   BasePcpHandler(Mediums* mediums, EndpointManager* endpoint_manager,
                  EndpointChannelManager* channel_manager,
                  BwuManager* bwu_manager, Pcp pcp);
@@ -118,7 +117,7 @@ class BasePcpHandler : public PcpHandler,
   // Until both parties call it, connection will not reach a data phase.
   // Updates state in ClientProxy.
   Status AcceptConnection(ClientProxy* client, const std::string& endpoint_id,
-                          const PayloadListener& payload_listener) override;
+                          PayloadListener payload_listener) override;
 
   // Called by either party to reject connection on their part.
   // If either party does call it, connection will terminate.
@@ -326,9 +325,8 @@ class BasePcpHandler : public PcpHandler,
     void SetCryptoContext(std::unique_ptr<securegcm::UKey2Handshake> ukey2);
 
     // Pass Accept notification to client.
-    void LocalEndpointAcceptedConnection(
-        const std::string& endpoint_id,
-        const PayloadListener& payload_listener);
+    void LocalEndpointAcceptedConnection(const std::string& endpoint_id,
+                                         PayloadListener payload_listener);
 
     // Pass Reject notification to client.
     void LocalEndpointRejectedConnection(const std::string& endpoint_id);

connections/implementation/base_pcp_handler_test.cc

@@ -17,6 +17,7 @@
 #include <array>
 #include <atomic>
 #include <memory>
+#include <string>
 
 #include "gmock/gmock.h"
 #include "protobuf-matchers/protocol-buffer-matchers.h"
@@ -179,6 +180,10 @@ class MockPcpHandler : public BasePcpHandler {
       ABSL_NO_THREAD_SAFETY_ANALYSIS {
     BasePcpHandler::OnEndpointLost(client, endpoint);
   }
+  BasePcpHandler::DiscoveredEndpoint* GetDiscoveredEndpoint(
+      const std::string& endpoint_id) {
+    return BasePcpHandler::GetDiscoveredEndpoint(endpoint_id);
+  }
   std::vector<BasePcpHandler::DiscoveredEndpoint*> GetDiscoveredEndpoints(
       const std::string& endpoint_id) {
     return BasePcpHandler::GetDiscoveredEndpoints(endpoint_id);
@@ -733,10 +738,15 @@ TEST_P(BasePcpHandlerTest, MultipleMediumsProduceSingleEndpointLostEvent) {
     EXPECT_EQ(pcp_handler.AcceptConnection(&client, endpoint_id, {}),
               Status{Status::kSuccess});
     EXPECT_CALL(mock_connection_listener_.rejected_cb, Call).Times(AtLeast(0));
+    auto endpoint_disc = pcp_handler.GetDiscoveredEndpoint(endpoint_id);
+    pcp_handler.OnEndpointLost(&client, *endpoint_disc);
+    EXPECT_NE(pcp_handler.GetDiscoveredEndpoint(endpoint_id), nullptr);
     for (const auto* endpoint :
          pcp_handler.GetDiscoveredEndpoints(endpoint_id)) {
       pcp_handler.OnEndpointLost(&client, *endpoint);
     }
+    EXPECT_EQ(pcp_handler.GetDiscoveredEndpoint(endpoint_id), nullptr);
+    EXPECT_FALSE(client.IsConnectedToEndpoint(endpoint_id));
     NEARBY_LOG(INFO, "Closing connection: id=%s", endpoint_id.c_str());
     channel_b->Close();
     bwu.Shutdown();

connections/implementation/client_proxy.cc

@@ -83,9 +83,9 @@ std::string ClientProxy::GetLocalEndpointId() {
 }
 
 std::string ClientProxy::GetConnectionToken(const std::string& endpoint_id) {
-  Connection* item = LookupConnection(endpoint_id);
+  ConnectionPair* item = LookupConnection(endpoint_id);
   if (item != nullptr) {
-    return item->connection_token;
+    return item->first.connection_token;
   }
   return {};
 }
@@ -278,12 +278,18 @@ void ClientProxy::OnConnectionInitiated(
   // still need to accept this connection, so set its establishment status to
   // PENDING.
   auto result = connections_.emplace(
-      endpoint_id, Connection{
-                       .is_incoming = info.is_incoming_connection,
-                       .connection_listener = listener,
-                       .connection_options = connection_options,
-                       .connection_token = connection_token,
-                   });
+      endpoint_id, std::make_pair(
+                       Connection{
+                           .is_incoming = info.is_incoming_connection,
+                           .connection_listener = listener,
+                           .connection_options = connection_options,
+                           .connection_token = connection_token,
+                       },
+                       PayloadListener{
+                           .payload_cb = [](absl::string_view, Payload) {},
+                           .payload_progress_cb = [](absl::string_view,
+                                                     PayloadProgressInfo) {},
+                       }));
   // Instead of using structured binding which is nice, but banned
   // (can not use c++17 features, until chromium does) we unpack manually.
   auto& pair_iter = result.first;
@@ -293,12 +299,12 @@ void ClientProxy::OnConnectionInitiated(
       << GetClientId() << "; endpoint_id=" << endpoint_id
       << "; inserted=" << inserted;
   DCHECK(inserted);
-  const Connection& item = pair_iter->second;
+  const ConnectionPair& item = pair_iter->second;
   // Notify the client.
   //
   // Note: we allow devices to connect to an advertiser even after it stops
   // advertising, so no need to check IsAdvertising() here.
-  item.connection_listener.initiated_cb(endpoint_id, info);
+  item.first.connection_listener.initiated_cb(endpoint_id, info);
 
   if (info.is_incoming_connection) {
     // Add CancellationFlag for advertisers once encryption succeeds.
@@ -320,10 +326,10 @@ void ClientProxy::OnConnectionAccepted(const std::string& endpoint_id) {
   }
 
   // Notify the client.
-  Connection* item = LookupConnection(endpoint_id);
+  ConnectionPair* item = LookupConnection(endpoint_id);
   if (item != nullptr) {
-    item->connection_listener.accepted_cb(endpoint_id);
-    item->status = Connection::kConnected;
+    item->first.connection_listener.accepted_cb(endpoint_id);
+    item->first.status = Connection::kConnected;
   }
 }
 
@@ -339,9 +345,9 @@ void ClientProxy::OnConnectionRejected(const std::string& endpoint_id,
   }
 
   // Notify the client.
-  const Connection* item = LookupConnection(endpoint_id);
+  const ConnectionPair* item = LookupConnection(endpoint_id);
   if (item != nullptr) {
-    item->connection_listener.rejected_cb(endpoint_id, status);
+    item->first.connection_listener.rejected_cb(endpoint_id, status);
     OnDisconnected(endpoint_id, false /* notify */);
   }
 }
@@ -350,9 +356,10 @@ void ClientProxy::OnBandwidthChanged(const std::string& endpoint_id,
                                      Medium new_medium) {
   MutexLock lock(&mutex_);
 
-  const Connection* item = LookupConnection(endpoint_id);
+  const ConnectionPair* item = LookupConnection(endpoint_id);
   if (item != nullptr) {
-    item->connection_listener.bandwidth_changed_cb(endpoint_id, new_medium);
+    item->first.connection_listener.bandwidth_changed_cb(endpoint_id,
+                                                         new_medium);
     NEARBY_LOGS(INFO) << "ClientProxy [reporting onBandwidthChanged]: client="
                       << GetClientId() << "; endpoint_id=" << endpoint_id;
   }
@@ -361,10 +368,10 @@ void ClientProxy::OnBandwidthChanged(const std::string& endpoint_id,
 void ClientProxy::OnDisconnected(const std::string& endpoint_id, bool notify) {
   MutexLock lock(&mutex_);
 
-  const Connection* item = LookupConnection(endpoint_id);
+  const ConnectionPair* item = LookupConnection(endpoint_id);
   if (item != nullptr) {
     if (notify) {
-      item->connection_listener.disconnected_cb({endpoint_id});
+      item->first.connection_listener.disconnected_cb({endpoint_id});
     }
     connections_.erase(endpoint_id);
     OnSessionComplete();
@@ -377,9 +384,9 @@ bool ClientProxy::ConnectionStatusMatches(const std::string& endpoint_id,
                                           Connection::Status status) const {
   MutexLock lock(&mutex_);
 
-  const Connection* item = LookupConnection(endpoint_id);
+  const ConnectionPair* item = LookupConnection(endpoint_id);
   if (item != nullptr) {
-    return item->status == status;
+    return item->first.status == status;
   }
   return false;
 }
@@ -388,9 +395,9 @@ BooleanMediumSelector ClientProxy::GetUpgradeMediums(
     const std::string& endpoint_id) const {
   MutexLock lock(&mutex_);
 
-  const Connection* item = LookupConnection(endpoint_id);
+  const ConnectionPair* item = LookupConnection(endpoint_id);
   if (item != nullptr) {
-    return item->connection_options.allowed;
+    return item->first.connection_options.allowed;
   }
   return {};
 }
@@ -398,9 +405,9 @@ BooleanMediumSelector ClientProxy::GetUpgradeMediums(
 bool ClientProxy::Is5GHzSupported(const std::string& endpoint_id) const {
   MutexLock lock(&mutex_);
 
-  const Connection* item = LookupConnection(endpoint_id);
+  const ConnectionPair* item = LookupConnection(endpoint_id);
   if (item != nullptr) {
-    return item->connection_options.connection_info.supports_5_ghz;
+    return item->first.connection_options.connection_info.supports_5_ghz;
   }
   return false;
 }
@@ -408,9 +415,9 @@ bool ClientProxy::Is5GHzSupported(const std::string& endpoint_id) const {
 std::string ClientProxy::GetBssid(const std::string& endpoint_id) const {
   MutexLock lock(&mutex_);
 
-  const Connection* item = LookupConnection(endpoint_id);
+  const ConnectionPair* item = LookupConnection(endpoint_id);
   if (item != nullptr) {
-    return item->connection_options.connection_info.bssid;
+    return item->first.connection_options.connection_info.bssid;
   }
   return {};
 }
@@ -418,9 +425,9 @@ std::string ClientProxy::GetBssid(const std::string& endpoint_id) const {
 std::int32_t ClientProxy::GetApFrequency(const std::string& endpoint_id) const {
   MutexLock lock(&mutex_);
 
-  const Connection* item = LookupConnection(endpoint_id);
+  const ConnectionPair* item = LookupConnection(endpoint_id);
   if (item != nullptr) {
-    return item->connection_options.connection_info.ap_frequency;
+    return item->first.connection_options.connection_info.ap_frequency;
   }
   return -1;
 }
@@ -428,9 +435,9 @@ std::int32_t ClientProxy::GetApFrequency(const std::string& endpoint_id) const {
 std::string ClientProxy::GetIPAddress(const std::string& endpoint_id) const {
   MutexLock lock(&mutex_);
 
-  const Connection* item = LookupConnection(endpoint_id);
+  const ConnectionPair* item = LookupConnection(endpoint_id);
   if (item != nullptr) {
-    return item->connection_options.connection_info.ip_address;
+    return item->first.connection_options.connection_info.ip_address;
   }
   return {};
 }
@@ -447,8 +454,8 @@ std::vector<std::string> ClientProxy::GetMatchingEndpoints(
 
   for (const auto& pair : connections_) {
     const auto& endpoint_id = pair.first;
-    const auto& connection = pair.second;
-    if (pred(connection)) {
+    const auto& connection_pair = pair.second;
+    if (pred(connection_pair.first)) {
       connected_endpoints.push_back(endpoint_id);
     }
   }
@@ -487,9 +494,9 @@ bool ClientProxy::HasPendingConnectionToEndpoint(
     const std::string& endpoint_id) const {
   MutexLock lock(&mutex_);
 
-  const Connection* item = LookupConnection(endpoint_id);
+  const ConnectionPair* item = LookupConnection(endpoint_id);
   if (item != nullptr) {
-    return item->status != Connection::kConnected;
+    return item->first.status != Connection::kConnected;
   }
   return false;
 }
@@ -515,7 +522,7 @@ bool ClientProxy::HasRemoteEndpointResponded(
 }
 
 void ClientProxy::LocalEndpointAcceptedConnection(
-    const std::string& endpoint_id, const PayloadListener& listener) {
+    const std::string& endpoint_id, PayloadListener listener) {
   MutexLock lock(&mutex_);
 
   if (HasLocalEndpointResponded(endpoint_id)) {
@@ -526,9 +533,9 @@ void ClientProxy::LocalEndpointAcceptedConnection(
   }
 
   AppendConnectionStatus(endpoint_id, Connection::kLocalEndpointAccepted);
-  Connection* item = LookupConnection(endpoint_id);
+  ConnectionPair* item = LookupConnection(endpoint_id);
   if (item != nullptr) {
-    item->payload_listener = listener;
+    item->second = std::move(listener);
   }
   analytics_recorder_->OnLocalEndpointAccepted(endpoint_id);
 }
@@ -642,18 +649,18 @@ const OsInfo& ClientProxy::GetLocalOsInfo() const {
 
 std::optional<OsInfo> ClientProxy::GetRemoteOsInfo(
     absl::string_view endpoint_id) const {
-  const Connection* item = LookupConnection(endpoint_id);
+  const ConnectionPair* item = LookupConnection(endpoint_id);
   if (item != nullptr) {
-    return item->os_info;
+    return item->first.os_info;
   }
   return std::nullopt;
 }
 
 void ClientProxy::SetRemoteOsInfo(absl::string_view endpoint_id,
                                   const OsInfo& remote_os_info) {
-  Connection* item = LookupConnection(endpoint_id);
+  ConnectionPair* item = LookupConnection(endpoint_id);
   if (item != nullptr) {
-    item->os_info.emplace(remote_os_info);
+    item->first.os_info.emplace(remote_os_info);
   }
 }
 void ClientProxy::CancelAllEndpoints() {
@@ -671,23 +678,24 @@ void ClientProxy::OnPayload(const std::string& endpoint_id, Payload payload) {
   MutexLock lock(&mutex_);
 
   if (IsConnectedToEndpoint(endpoint_id)) {
-    const Connection* item = LookupConnection(endpoint_id);
+    const std::pair<ClientProxy::Connection, PayloadListener>* item =
+        LookupConnection(endpoint_id);
     if (item != nullptr) {
       NEARBY_LOGS(INFO) << "ClientProxy [reporting onPayloadReceived]: client="
                         << GetClientId() << "; endpoint_id=" << endpoint_id
                         << " ; payload_id=" << payload.GetId();
-      item->payload_listener.payload_cb(endpoint_id, std::move(payload));
+      item->second.payload_cb(endpoint_id, std::move(payload));
     }
   }
 }
 
-const ClientProxy::Connection* ClientProxy::LookupConnection(
+const ClientProxy::ConnectionPair* ClientProxy::LookupConnection(
     absl::string_view endpoint_id) const {
   auto item = connections_.find(endpoint_id);
   return item != connections_.end() ? &item->second : nullptr;
 }
 
-ClientProxy::Connection* ClientProxy::LookupConnection(
+ClientProxy::ConnectionPair* ClientProxy::LookupConnection(
     absl::string_view endpoint_id) {
   auto item = connections_.find(endpoint_id);
   return item != connections_.end() ? &item->second : nullptr;
@@ -698,9 +706,10 @@ void ClientProxy::OnPayloadProgress(const std::string& endpoint_id,
   MutexLock lock(&mutex_);
 
   if (IsConnectedToEndpoint(endpoint_id)) {
-    Connection* item = LookupConnection(endpoint_id);
+    std::pair<ClientProxy::Connection, PayloadListener>* item =
+        LookupConnection(endpoint_id);
     if (item != nullptr) {
-      item->payload_listener.payload_progress_cb(endpoint_id, info);
+      item->second.payload_progress_cb(endpoint_id, info);
 
       if (info.status == PayloadProgressInfo::Status::kInProgress) {
         NEARBY_LOGS(VERBOSE)
@@ -742,19 +751,19 @@ void ClientProxy::OnSessionComplete() {
 
 bool ClientProxy::ConnectionStatusesContains(
     const std::string& endpoint_id, Connection::Status status_to_match) const {
-  const Connection* item = LookupConnection(endpoint_id);
+  const ConnectionPair* item = LookupConnection(endpoint_id);
   if (item != nullptr) {
-    return (item->status & status_to_match) != 0;
+    return (item->first.status & status_to_match) != 0;
   }
   return false;
 }
 
 void ClientProxy::AppendConnectionStatus(const std::string& endpoint_id,
                                          Connection::Status status_to_append) {
-  Connection* item = LookupConnection(endpoint_id);
+  ConnectionPair* item = LookupConnection(endpoint_id);
   if (item != nullptr) {
-    item->status =
-        static_cast<Connection::Status>(item->status | status_to_append);
+    item->first.status =
+        static_cast<Connection::Status>(item->first.status | status_to_append);
   }
 }
 
@@ -870,10 +879,10 @@ std::string ClientProxy::Dump() {
   for (auto it = connections_.begin(); it != connections_.end(); ++it) {
     // TODO(deling): write Connection.ToString()
     sstream << "    " << it->first << " :(connection token) "
-            << it->second.connection_token << ", (remote os type) "
-            << (it->second.os_info.has_value()
+            << it->second.first.connection_token << ", (remote os type) "
+            << (it->second.first.os_info.has_value()
                     ? location::nearby::connections::OsInfo::OsType_Name(
-                          it->second.os_info->type())
+                          it->second.first.os_info->type())
                     : "unknown")
             << std::endl;
   }

connections/implementation/client_proxy.h

@@ -157,7 +157,7 @@ class ClientProxy final {
   bool HasRemoteEndpointResponded(const std::string& endpoint_id) const;
   // Marks the local endpoint as having accepted the connection.
   void LocalEndpointAcceptedConnection(const std::string& endpoint_id,
-                                       const PayloadListener& listener);
+                                       PayloadListener listener);
   // Marks the local endpoint as having rejected the connection.
   void LocalEndpointRejectedConnection(const std::string& endpoint_id);
   // Marks the remote endpoint as having accepted the connection.
@@ -236,13 +236,13 @@ class ClientProxy final {
     bool is_incoming{false};
     Status status{kPending};
     ConnectionListener connection_listener;
-    PayloadListener payload_listener;
     ConnectionOptions connection_options;
     DiscoveryOptions discovery_options;
     AdvertisingOptions advertising_options;
     std::string connection_token;
     std::optional<location::nearby::connections::OsInfo> os_info;
   };
+  using ConnectionPair = std::pair<Connection, PayloadListener>;
 
   struct AdvertisingInfo {
     std::string service_id;
@@ -265,8 +265,8 @@ class ClientProxy final {
   void AppendConnectionStatus(const std::string& endpoint_id,
                               Connection::Status status_to_append);
 
-  const Connection* LookupConnection(absl::string_view endpoint_id) const;
-  Connection* LookupConnection(absl::string_view endpoint_id);
+  const ConnectionPair* LookupConnection(absl::string_view endpoint_id) const;
+  ConnectionPair* LookupConnection(absl::string_view endpoint_id);
   bool ConnectionStatusMatches(const std::string& endpoint_id,
                                Connection::Status status) const;
   std::vector<std::string> GetMatchingEndpoints(
@@ -323,7 +323,7 @@ class ClientProxy final {
   DiscoveryOptions discovery_options_;
 
   // Maps endpoint_id to endpoint connection state.
-  absl::flat_hash_map<std::string, Connection> connections_;
+  absl::flat_hash_map<std::string, ConnectionPair> connections_;
 
   // A cache of endpoint ids that we've already notified the discoverer of. We
   // check this cache before calling onEndpointFound() so that we don't notify

connections/implementation/client_proxy_test.cc

@@ -85,9 +85,9 @@ class ClientProxyTest : public ::testing::TestWithParam<FeatureFlags::Flags> {
   };
   struct MockPayloadListener {
     StrictMock<
-        MockFunction<void(const std::string& endpoint_id, Payload payload)>>
+        MockFunction<void(absl::string_view endpoint_id, Payload payload)>>
         payload_cb;
-    StrictMock<MockFunction<void(const std::string& endpoint_id,
+    StrictMock<MockFunction<void(absl::string_view endpoint_id,
                                  const PayloadProgressInfo& info)>>
         payload_progress_cb;
   };
@@ -178,7 +178,13 @@ class ClientProxyTest : public ::testing::TestWithParam<FeatureFlags::Flags> {
                                           const Endpoint& endpoint) {
     EXPECT_TRUE(client->HasPendingConnectionToEndpoint(endpoint.id));
     EXPECT_FALSE(client->HasLocalEndpointResponded(endpoint.id));
-    client->LocalEndpointAcceptedConnection(endpoint.id, payload_listener_);
+    client->LocalEndpointAcceptedConnection(
+        endpoint.id,
+        {
+            .payload_cb = mock_discovery_payload_.payload_cb.AsStdFunction(),
+            .payload_progress_cb =
+                mock_discovery_payload_.payload_progress_cb.AsStdFunction(),
+        });
     EXPECT_TRUE(client->HasLocalEndpointResponded(endpoint.id));
     EXPECT_TRUE(client->LocalConnectionIsAccepted(endpoint.id));
   }
@@ -292,11 +298,6 @@ class ClientProxyTest : public ::testing::TestWithParam<FeatureFlags::Flags> {
       .endpoint_found_cb = mock_discovery_.endpoint_found_cb.AsStdFunction(),
       .endpoint_lost_cb = mock_discovery_.endpoint_lost_cb.AsStdFunction(),
   };
-  PayloadListener payload_listener_{
-      .payload_cb = mock_discovery_payload_.payload_cb.AsStdFunction(),
-      .payload_progress_cb =
-          mock_discovery_payload_.payload_progress_cb.AsStdFunction(),
-  };
   ConnectionOptions connection_options_;
   AdvertisingOptions advertising_options_;
   DiscoveryOptions discovery_options_;

connections/implementation/endpoint_manager.cc

@@ -296,7 +296,10 @@ EndpointManager::EndpointManager(
 
 EndpointManager::~EndpointManager() {
   NEARBY_LOG(INFO, "Initiating shutdown of EndpointManager.");
-  is_shutdown_ = true;
+  {
+    MutexLock lock(&mutex_);
+    is_shutdown_ = true;
+  }
   analytics::ThroughputRecorderContainer::GetInstance().Shutdown();
   CountDownLatch latch(1);
   RunOnEndpointManagerThread("bring-down-endpoints", [this, &latch]() {
@@ -413,11 +416,12 @@ void EndpointManager::RegisterEndpoint(
         absl::Milliseconds(connection_options.keep_alive_interval_millis);
     absl::Duration keep_alive_timeout =
         absl::Milliseconds(connection_options.keep_alive_timeout_millis);
-    NEARBY_LOGS(INFO)
-        << "Registering endpoint " << endpoint_id << " for client "
-        << client->GetClientId() << " with keep-alive frame as interval="
-        << absl::FormatDuration(keep_alive_interval)
-        << ", timeout=" << absl::FormatDuration(keep_alive_timeout);
+    NEARBY_LOGS(INFO) << "Registering endpoint " << endpoint_id
+                      << " for client " << client->GetClientId()
+                      << " with keep-alive frame as interval="
+                      << absl::FormatDuration(keep_alive_interval)
+                      << ", timeout="
+                      << absl::FormatDuration(keep_alive_timeout);
 
     // Pass ownership of channel to EndpointChannelManager
     NEARBY_LOGS(INFO) << "Registering endpoint with channel manager: endpoint "
@@ -530,48 +534,50 @@ std::vector<std::string> EndpointManager::SendPayloadChunk(
 void EndpointManager::DiscardEndpoint(ClientProxy* client,
                                       const std::string& endpoint_id) {
   NEARBY_LOGS(VERBOSE) << "DiscardEndpoint for endpoint " << endpoint_id;
-  RunOnEndpointManagerThread(
-      "discard-endpoint", [this, client, endpoint_id]() {
-        // `ClientProxy` is destroyed before `EndpointManager` in
-        // `~NearbyConnections`, which means "discard-endpoint" needs to check
-        // if this task is being executing during `~EndpointManager` to
-        // prevent accessing an invalid `ClientProxy` pointer. There are two
-        // cases where "discard-endpoint" can be executed during destruction,
-        // both of which can safely use `is_shutdown_` to check if this is being
-        // executed during the destruction of the object:
-        //
-        // Case 1: "discard-endpoints" is posted to the thread before
-        // destruction, but not executed yet: `~EndpointManager` blocks on
-        // "bring-down-endpoints" and because the executor is a single thread
-        // executor, tasks are guaranteed to execute sequentially
-        // (see
-        // https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executors.html#newSingleThreadExecutor--)
-        // and this means that the "discard-endpoints" will be executed before
-        // "bring-down-endpoints", blocking the destruction of `is_shutdown_`
-        // and therefore `is_shutdown_` is not garbage memory.
-        //
-        // Case 2: "discard-endpoints" is posted to the thread during
-        // destruction, after "bring-down-endpoints" is called: the executor
-        // will be destructed before `is_shutdown_` because of the ordering of
-        // `EndpointManager`'s member variables, and the executor's destructor
-        // blocks on running all pending tasks
-        // (see
-        // https://source.chromium.org/chromium/chromium/src/+/refs/heads/main:chrome/services/sharing/nearby/platform/scheduled_executor.cc;l=67;drc=e0e0d24aaa54727dc0a8bc4b159ccdf80d3f5d8d),
-        // which means that "discard-endpoints" will run during the destruction
-        // of `serial_executor_` and will still have access to a valid
-        // `is_shutdown_`.
-        //
-        // TODO(b/280653613): Develop a more robost solution to prevent
-        // accessing an already destroyed `ClientProxy` during destruction.
-        if (is_shutdown_) {
-          NEARBY_LOGS(VERBOSE)
-              << "DiscardEndpoint called during destruction, returning early.";
-          return;
-        }
+  RunOnEndpointManagerThread("discard-endpoint", [this, client, endpoint_id]() {
+    // `ClientProxy` is destroyed before `EndpointManager` in
+    // `~NearbyConnections`, which means "discard-endpoint" needs to check
+    // if this task is being executing during `~EndpointManager` to
+    // prevent accessing an invalid `ClientProxy` pointer. There are two
+    // cases where "discard-endpoint" can be executed during destruction,
+    // both of which can safely use `is_shutdown_` to check if this is being
+    // executed during the destruction of the object:
+    //
+    // Case 1: "discard-endpoints" is posted to the thread before
+    // destruction, but not executed yet: `~EndpointManager` blocks on
+    // "bring-down-endpoints" and because the executor is a single thread
+    // executor, tasks are guaranteed to execute sequentially
+    // (see
+    // https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executors.html#newSingleThreadExecutor--)
+    // and this means that the "discard-endpoints" will be executed before
+    // "bring-down-endpoints", blocking the destruction of `is_shutdown_`
+    // and therefore `is_shutdown_` is not garbage memory.
+    //
+    // Case 2: "discard-endpoints" is posted to the thread during
+    // destruction, after "bring-down-endpoints" is called: the executor
+    // will be destructed before `is_shutdown_` because of the ordering of
+    // `EndpointManager`'s member variables, and the executor's destructor
+    // blocks on running all pending tasks
+    // (see
+    // https://source.chromium.org/chromium/chromium/src/+/refs/heads/main:chrome/services/sharing/nearby/platform/scheduled_executor.cc;l=67;drc=e0e0d24aaa54727dc0a8bc4b159ccdf80d3f5d8d),
+    // which means that "discard-endpoints" will run during the destruction
+    // of `serial_executor_` and will still have access to a valid
+    // `is_shutdown_`.
+    //
+    // TODO(b/280653613): Develop a more robost solution to prevent
+    // accessing an already destroyed `ClientProxy` during destruction.
+    {
+      MutexLock lock(&mutex_);
+      if (is_shutdown_) {
+        NEARBY_LOGS(VERBOSE)
+            << "DiscardEndpoint called during destruction, returning early.";
+        return;
+      }
+    }
 
-        RemoveEndpoint(client, endpoint_id,
-                       /*notify=*/client->IsConnectedToEndpoint(endpoint_id));
-      });
+    RemoveEndpoint(client, endpoint_id,
+                   /*notify=*/client->IsConnectedToEndpoint(endpoint_id));
+  });
 }
 
 std::vector<std::string> EndpointManager::SendControlMessage(

connections/implementation/endpoint_manager.h

@@ -302,7 +302,8 @@ class EndpointManager {
   // pending tasks during it's destruction, and the "discard-endpoints"
   // task checks `is_shutdown_` to prevent accessing an invalid `ClientProxy`
   // pointer.
-  bool is_shutdown_ = false;
+  mutable RecursiveMutex mutex_;
+  bool is_shutdown_ ABSL_GUARDED_BY(mutex_) = false;
 
   std::unique_ptr<SingleThreadExecutor> serial_executor_;
 };

connections/implementation/mock_service_controller.h

@@ -60,7 +60,7 @@ class MockServiceController : public ServiceController {
 
   MOCK_METHOD(Status, AcceptConnection,
               (ClientProxy * client, const std::string& endpoint_id,
-               const PayloadListener& listener),
+               PayloadListener listener),
               (override));
 
   MOCK_METHOD(Status, RejectConnection,

connections/implementation/mock_service_controller_router.h

@@ -60,7 +60,7 @@ class MockServiceControllerRouter : public ServiceControllerRouter {
 
   MOCK_METHOD(void, AcceptConnection,
               (ClientProxy * client, absl::string_view endpoint_id,
-               const PayloadListener& listener, const ResultCallback& callback),
+               PayloadListener listener, const ResultCallback& callback),
               (override));
 
   MOCK_METHOD(void, RejectConnection,

connections/implementation/offline_service_controller.cc

@@ -89,12 +89,13 @@ Status OfflineServiceController::RequestConnection(
 
 Status OfflineServiceController::AcceptConnection(
     ClientProxy* client, const std::string& endpoint_id,
-    const PayloadListener& listener) {
+    PayloadListener listener) {
   if (stop_) return {Status::kOutOfOrderApiCall};
   NEARBY_LOGS(INFO) << "Client " << client->GetClientId()
                     << " accepted the connection with endpoint_id="
                     << endpoint_id;
-  return pcp_manager_.AcceptConnection(client, endpoint_id, listener);
+  return pcp_manager_.AcceptConnection(client, endpoint_id,
+                                       std::move(listener));
 }
 
 Status OfflineServiceController::RejectConnection(

connections/implementation/offline_service_controller.h

@@ -59,7 +59,7 @@ class OfflineServiceController : public ServiceController {
       const ConnectionRequestInfo& info,
       const ConnectionOptions& connection_options) override;
   Status AcceptConnection(ClientProxy* client, const std::string& endpoint_id,
-                          const PayloadListener& listener) override;
+                          PayloadListener listener) override;
   Status RejectConnection(ClientProxy* client,
                           const std::string& endpoint_id) override;
 

connections/implementation/offline_simulation_user.cc

@@ -72,13 +72,13 @@ void OfflineSimulationUser::OnEndpointLost(const std::string& endpoint_id) {
   if (lost_latch_) lost_latch_->CountDown();
 }
 
-void OfflineSimulationUser::OnPayload(const std::string& endpoint_id,
+void OfflineSimulationUser::OnPayload(absl::string_view endpoint_id,
                                       Payload payload) {
   payload_ = std::move(payload);
   if (payload_latch_) payload_latch_->CountDown();
 }
 
-void OfflineSimulationUser::OnPayloadProgress(const std::string& endpoint_id,
+void OfflineSimulationUser::OnPayloadProgress(absl::string_view endpoint_id,
                                               const PayloadProgressInfo& info) {
   MutexLock lock(&progress_mutex_);
   progress_info_ = info;

connections/implementation/offline_simulation_user.h

@@ -172,8 +172,8 @@ class OfflineSimulationUser {
   void OnEndpointLost(const std::string& endpoint_id);
 
   // PayloadListener callbacks
-  void OnPayload(const std::string& endpoint_id, Payload payload);
-  void OnPayloadProgress(const std::string& endpoint_id,
+  void OnPayload(absl::string_view endpoint_id, Payload payload);
+  void OnPayloadProgress(absl::string_view endpoint_id,
                          const PayloadProgressInfo& info);
 
   std::string service_id_;

connections/implementation/pcp_handler.h

@@ -105,7 +105,7 @@ class PcpHandler {
   // Update state in ClientProxy.
   virtual Status AcceptConnection(ClientProxy* client,
                                   const std::string& endpoint_id,
-                                  const PayloadListener& payload_listener) = 0;
+                                  PayloadListener payload_listener) = 0;
 
   // Either party may call this to reject connection on their part before
   // connection reaches data phase. If either party does call it, connection

connections/implementation/pcp_manager.cc

@@ -109,12 +109,13 @@ Status PcpManager::RequestConnection(
 
 Status PcpManager::AcceptConnection(ClientProxy* client,
                                     const string& endpoint_id,
-                                    const PayloadListener& payload_listener) {
+                                    PayloadListener payload_listener) {
   if (!current_) {
     return {Status::kOutOfOrderApiCall};
   }
 
-  return current_->AcceptConnection(client, endpoint_id, payload_listener);
+  return current_->AcceptConnection(client, endpoint_id,
+                                    std::move(payload_listener));
 }
 
 Status PcpManager::RejectConnection(ClientProxy* client,

connections/implementation/pcp_manager.h

@@ -64,7 +64,7 @@ class PcpManager {
                            const ConnectionRequestInfo& info,
                            const ConnectionOptions& connection_options);
   Status AcceptConnection(ClientProxy* client, const string& endpoint_id,
-                          const PayloadListener& payload_listener);
+                          PayloadListener payload_listener);
   Status RejectConnection(ClientProxy* client, const string& endpoint_id);
 
   location::nearby::proto::connections::Medium GetBandwidthUpgradeMedium();

connections/implementation/service_controller.h

@@ -83,7 +83,7 @@ class ServiceController {
       const ConnectionOptions& connection_options) = 0;
   virtual Status AcceptConnection(ClientProxy* client,
                                   const std::string& endpoint_id,
-                                  const PayloadListener& listener) = 0;
+                                  PayloadListener listener) = 0;
   virtual Status RejectConnection(ClientProxy* client,
                                   const std::string& endpoint_id) = 0;
 

connections/implementation/service_controller_router.cc

@@ -191,12 +191,12 @@ void ServiceControllerRouter::RequestConnection(
 
 void ServiceControllerRouter::AcceptConnection(ClientProxy* client,
                                                absl::string_view endpoint_id,
-                                               const PayloadListener& listener,
+                                               PayloadListener listener,
                                                const ResultCallback& callback) {
   RouteToServiceController(
       "scr-accept-connection",
-      [this, client, endpoint_id = std::string(endpoint_id), listener,
-       callback]() {
+      [this, client, endpoint_id = std::string(endpoint_id),
+       listener = std::move(listener), callback]() mutable {
         if (client->IsConnectedToEndpoint(endpoint_id)) {
           callback.result_cb({Status::kAlreadyConnectedToEndpoint});
           return;
@@ -213,7 +213,7 @@ void ServiceControllerRouter::AcceptConnection(ClientProxy* client,
         }
 
         callback.result_cb(GetServiceController()->AcceptConnection(
-            client, endpoint_id, listener));
+            client, endpoint_id, std::move(listener)));
       });
 }
 

connections/implementation/service_controller_router.h

@@ -85,7 +85,7 @@ class ServiceControllerRouter {
                                  const ResultCallback& callback);
   virtual void AcceptConnection(ClientProxy* client,
                                 absl::string_view endpoint_id,
-                                const PayloadListener& listener,
+                                PayloadListener listener,
                                 const ResultCallback& callback);
   virtual void RejectConnection(ClientProxy* client,
                                 absl::string_view endpoint_id,

connections/implementation/service_controller_router_test.cc

@@ -158,7 +158,6 @@ class ServiceControllerRouterTest : public testing::Test {
   }
 
   void AcceptConnection(ClientProxy* client, const std::string endpoint_id,
-                        const PayloadListener& listener,
                         const ResultCallback& callback) {
     EXPECT_CALL(*mock_, AcceptConnection)
         .WillOnce(Return(Status{Status::kSuccess}));
@@ -167,11 +166,13 @@ class ServiceControllerRouterTest : public testing::Test {
     {
       MutexLock lock(&mutex_);
       complete_ = false;
-      router_.AcceptConnection(client, endpoint_id, listener, callback);
+      router_.AcceptConnection(client, endpoint_id, {},
+                               callback);
       while (!complete_) cond_.Wait();
       EXPECT_EQ(result_, Status{Status::kSuccess});
     }
-    client->LocalEndpointAcceptedConnection(endpoint_id, listener);
+    client->LocalEndpointAcceptedConnection(endpoint_id,
+                                            {});
     client->RemoteEndpointAcceptedConnection(endpoint_id);
     EXPECT_TRUE(client->IsConnectionAccepted(endpoint_id));
     client->OnConnectionAccepted(endpoint_id);
@@ -314,7 +315,6 @@ class ServiceControllerRouterTest : public testing::Test {
   };
 
   DiscoveryListener discovery_listener_;
-  PayloadListener payload_listener_;
 
   Mutex mutex_;
   ConditionVariable cond_{&mutex_};
@@ -372,7 +372,7 @@ TEST_F(ServiceControllerRouterTest, AcceptConnectionCalled) {
   RequestConnection(&client_, kRemoteEndpointId, kConnectionRequestInfo,
                     kCallback);
   // Now, we can accept connection.
-  AcceptConnection(&client_, kRemoteEndpointId, payload_listener_, kCallback);
+  AcceptConnection(&client_, kRemoteEndpointId, kCallback);
 }
 
 TEST_F(ServiceControllerRouterTest, RejectConnectionCalled) {
@@ -394,7 +394,7 @@ TEST_F(ServiceControllerRouterTest, InitiateBandwidthUpgradeCalled) {
   RequestConnection(&client_, kRemoteEndpointId, kConnectionRequestInfo,
                     kCallback);
   // Now, we can accept connection.
-  AcceptConnection(&client_, kRemoteEndpointId, payload_listener_, kCallback);
+  AcceptConnection(&client_, kRemoteEndpointId, kCallback);
   // Now we can change connection bandwidth.
   InitiateBandwidthUpgrade(&client_, kRemoteEndpointId, kCallback);
 }
@@ -407,7 +407,7 @@ TEST_F(ServiceControllerRouterTest, SendPayloadCalled) {
   RequestConnection(&client_, kRemoteEndpointId, kConnectionRequestInfo,
                     kCallback);
   // Now, we can accept connection.
-  AcceptConnection(&client_, kRemoteEndpointId, payload_listener_, kCallback);
+  AcceptConnection(&client_, kRemoteEndpointId, kCallback);
   // Now we can send payload.
   SendPayload(&client_, std::vector<std::string>{kRemoteEndpointId},
               Payload{ByteArray("data")}, kCallback);
@@ -421,7 +421,7 @@ TEST_F(ServiceControllerRouterTest, CancelPayloadCalled) {
   RequestConnection(&client_, kRemoteEndpointId, kConnectionRequestInfo,
                     kCallback);
   // Now, we can accept connection.
-  AcceptConnection(&client_, kRemoteEndpointId, payload_listener_, kCallback);
+  AcceptConnection(&client_, kRemoteEndpointId, kCallback);
   // We have to know payload id, before we can cancel payload transfer.
   // It is either after a call to SendPayload, or after receiving
   // PayloadProgress callback. Let's assume we have it, and proceed.
@@ -436,7 +436,7 @@ TEST_F(ServiceControllerRouterTest, DisconnectFromEndpointCalled) {
   RequestConnection(&client_, kRemoteEndpointId, kConnectionRequestInfo,
                     kCallback);
   // Now, we can accept connection.
-  AcceptConnection(&client_, kRemoteEndpointId, payload_listener_, kCallback);
+  AcceptConnection(&client_, kRemoteEndpointId, kCallback);
   // We can disconnect at any time after RequestConnection.
   DisconnectFromEndpoint(&client_, kRemoteEndpointId, kCallback);
 }

connections/implementation/simulation_user.cc

@@ -63,13 +63,12 @@ void SimulationUser::OnEndpointLost(const std::string& endpoint_id) {
   if (lost_latch_) lost_latch_->CountDown();
 }
 
-void SimulationUser::OnPayload(const std::string& endpoint_id,
-                               Payload payload) {
+void SimulationUser::OnPayload(absl::string_view endpoint_id, Payload payload) {
   payload_ = std::move(payload);
   if (payload_latch_) payload_latch_->CountDown();
 }
 
-void SimulationUser::OnPayloadProgress(const std::string& endpoint_id,
+void SimulationUser::OnPayloadProgress(absl::string_view endpoint_id,
                                        const PayloadProgressInfo& info) {
   MutexLock lock(&progress_mutex_);
   progress_info_ = info;

connections/implementation/simulation_user.h

@@ -141,8 +141,8 @@ class SimulationUser {
   void OnEndpointLost(const std::string& endpoint_id);
 
   // PayloadListener callbacks
-  void OnPayload(const std::string& endpoint_id, Payload payload);
-  void OnPayloadProgress(const std::string& endpoint_id,
+  void OnPayload(absl::string_view, Payload payload);
+  void OnPayloadProgress(absl::string_view endpoint_id,
                          const PayloadProgressInfo& info);
 
   std::string service_id_;

connections/listeners.h

@@ -27,6 +27,7 @@
 //   default-initialized.
 // - callbacks may be initialized with lambdas; lambda definitions are concize.
 
+#include "absl/functional/any_invocable.h"
 #include "connections/connection_options.h"
 #include "connections/payload.h"
 #include "connections/status.h"
@@ -170,8 +171,8 @@ struct PayloadListener {
   // endpoint_id - The identifier for the remote endpoint that sent the
   //               payload.
   // payload     - The Payload object received.
-  std::function<void(const std::string& endpoint_id, Payload payload)>
-      payload_cb = [](const std::string&, Payload) {};
+  absl::AnyInvocable<void(absl::string_view endpoint_id, Payload payload) const>
+      payload_cb = [](absl::string_view, Payload) {};
 
   // Called with progress information about an active Payload transfer, either
   // incoming or outgoing.
@@ -180,10 +181,10 @@ struct PayloadListener {
   //               receiving this payload.
   // info -  The PayloadProgressInfo structure describing the status of
   //         the transfer.
-  std::function<void(const std::string& endpoint_id,
-                     const PayloadProgressInfo& info)>
+  absl::AnyInvocable<void(absl::string_view endpoint_id,
+                          const PayloadProgressInfo& info)>
       payload_progress_cb =
-          [](const std::string&, const PayloadProgressInfo&) {};
+          [](absl::string_view, const PayloadProgressInfo&) {};
 };
 
 }  // namespace connections

connections/listeners_test.cc

@@ -62,7 +62,7 @@ TEST(ListenersTest, PayloadListener_PayloadCb_Works) {
 
   PayloadListener listener{
       .payload_cb =
-          [&](const std::string& endpoint_id, Payload payload) {
+          [&](absl::string_view endpoint_id, Payload payload) {
             if (payload.AsBytes().data() == input_bytes) {
               payload_content_match = true;
             }

connections/swift/NearbyCoreAdapter/Sources/GNCCoreAdapter.mm

@@ -278,14 +278,14 @@ GNCStatus GNCStatusFromCppStatus(Status status) {
   std::string endpoint_id = [endpointID cStringUsingEncoding:[NSString defaultCStringEncoding]];
 
   PayloadListener listener;
-  listener.payload_cb = ^(const std::string &endpoint_id, Payload payload) {
-    NSString *endpointID = @(endpoint_id.c_str());
+  listener.payload_cb = [&delegate](absl::string_view endpoint_id, Payload payload) {
+    NSString *endpointID = @(std::string(endpoint_id).c_str());
     GNCPayload *gncPayload = [GNCPayload fromCpp:std::move(payload)];
     [delegate receivedPayload:gncPayload fromEndpoint:endpointID];
   };
   listener.payload_progress_cb =
-      ^(const std::string &endpoint_id, const PayloadProgressInfo &info) {
-        NSString *endpointID = @(endpoint_id.c_str());
+      [&delegate](absl::string_view endpoint_id, const PayloadProgressInfo &info) {
+        NSString *endpointID = @(std::string(endpoint_id).c_str());
         GNCPayloadStatus status;
         switch (info.status) {
           case PayloadProgressInfo::Status::kSuccess:

connections/v3/BUILD

@@ -3,7 +3,7 @@ cc_library(
     hdrs = [
         "bandwidth_info.h",
         "connection_listening_options.h",
-        "connection_resolution.h",
+        "connection_result.h",
         "connections_device.h",
         "connections_device_provider.h",
         "listeners.h",

connections/v3/connection_result.h

@@ -12,8 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#ifndef THIRD_PARTY_NEARBY_CONNECTIONS_V3_CONNECTION_RESOLUTION_H_
-#define THIRD_PARTY_NEARBY_CONNECTIONS_V3_CONNECTION_RESOLUTION_H_
+#ifndef THIRD_PARTY_NEARBY_CONNECTIONS_V3_CONNECTION_RESULT_H_
+#define THIRD_PARTY_NEARBY_CONNECTIONS_V3_CONNECTION_RESULT_H_
+
+#include <string>
 
 #include "connections/status.h"
 
@@ -25,8 +27,18 @@ struct ConnectionResult {
   nearby::connections::Status status;
 };
 
+// These fields should never be empty.
+struct InitialConnectionInfo {
+  // 4-digit authentication code shown to user derived by UKEY2.
+  std::string authentication_digits;
+  // Raw 32-byte authentication token derived by UKEY2.
+  std::string raw_authentication_token;
+  // Specifies if the connection is incoming or outgoing.
+  bool is_incoming_connection = false;
+};
+
 }  // namespace v3
 }  // namespace connections
 }  // namespace nearby
 
-#endif  // THIRD_PARTY_NEARBY_CONNECTIONS_V3_CONNECTION_RESOLUTION_H_
+#endif  // THIRD_PARTY_NEARBY_CONNECTIONS_V3_CONNECTION_RESULT_H_

connections/v3/listeners.h

@@ -18,7 +18,7 @@
 #include "absl/functional/any_invocable.h"
 #include "connections/listeners.h"
 #include "connections/v3/bandwidth_info.h"
-#include "connections/v3/connection_resolution.h"
+#include "connections/v3/connection_result.h"
 #include "internal/interop/device.h"
 
 namespace nearby {
@@ -44,8 +44,9 @@ struct ConnectionListener {
   // remote_device - The identifier for the remote endpoint.
   // info -  Other relevant information about the connection.
   absl::AnyInvocable<void(const NearbyDevice& remote_device,
-                          const ConnectionResponseInfo& info)>
-      initiated_cb = [](const NearbyDevice&, const ConnectionResponseInfo&) {};
+                          const v3::InitialConnectionInfo& info)>
+      initiated_cb =
+          [](const NearbyDevice&, const v3::InitialConnectionInfo&) {};
 
   // Called when both sides have accepted or either side has rejected the
   // connection. If the {@link ConnectionResolution}'s status is {@link

fastpair/common/BUILD

@@ -3,6 +3,7 @@ licenses(["notice"])
 cc_library(
     name = "common",
     srcs = [
+        "account_key_filter.cc",
         "battery_notification.cc",
         "fast_pair_device.cc",
         "fast_pair_http_result.cc",
@@ -11,6 +12,7 @@ cc_library(
     ],
     hdrs = [
         "account_key.h",
+        "account_key_filter.h",
         "battery_notification.h",
         "constant.h",
         "fast_pair_device.h",
@@ -31,6 +33,22 @@ cc_library(
     ],
 )
 
+cc_test(
+    name = "account_key_filter_test",
+    size = "small",
+    srcs = [
+        "account_key_filter_test.cc",
+    ],
+    shard_count = 16,
+    deps = [
+        ":common",
+        "//internal/platform/implementation/g3",  # build_cleaner: keep
+        "@com_github_protobuf_matchers//protobuf-matchers",
+        "@com_google_absl//absl/strings",
+        "@com_google_googletest//:gtest_main",
+    ],
+)
+
 cc_test(
     name = "fast_pair_device_test",
     size = "small",

fastpair/common/account_key.h

@@ -17,6 +17,7 @@
 
 #include <ostream>
 #include <string>
+#include <vector>
 
 #include "absl/strings/escaping.h"
 #include "absl/strings/string_view.h"
@@ -30,6 +31,9 @@ class AccountKey {
  public:
   AccountKey() = default;
   explicit AccountKey(absl::string_view bytes) : bytes_(bytes) {}
+  explicit AccountKey(const std::vector<uint8_t> bytes) {
+    bytes_ = std::string(bytes.begin(), bytes.end());
+  }
 
   static AccountKey CreateRandomKey() {
     std::string key(kAccountKeySize, 0);

fastpair/common/account_key_filter.cc

@@ -0,0 +1,125 @@
+// Copyright 2023 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 "fastpair/common/account_key_filter.h"
+
+#include <algorithm>
+#include <array>
+#include <cstdint>
+#include <vector>
+
+#include "absl/strings/string_view.h"
+#include "fastpair/common/battery_notification.h"
+#include "fastpair/common/non_discoverable_advertisement.h"
+#include "internal/crypto/sha2.h"
+#include "internal/platform/logging.h"
+
+namespace nearby {
+namespace fastpair {
+namespace {
+constexpr int kBitsInByte = 8;
+
+// SASS enabled peripherals create their Bloom filters with the first byte
+// equals to either `the account key in use`
+// or `the most recently used account key`
+// see this spec:
+// https://developers.google.com/nearby/fast-pair/early-access/specifications/extensions/sass#SassInUseAccountKey
+constexpr uint8_t kRecentlyUsedByte = 0x05;
+constexpr uint8_t kInUseByte = 0x06;
+constexpr uint8_t kShowUi = 0b00110011;
+constexpr uint8_t kHideUi = 0b00110100;
+
+// Helper to AccountKeyFilter::IsAccountKeyInFilter().
+// Performs the test to see if |data| is in |bit_sets|, a Bloom filter.
+bool AccountKeyFilterChecker(const std::vector<uint8_t>& data,
+                             const std::vector<uint8_t>& bit_sets) {
+  std::array<uint8_t, 32> hashed = crypto::SHA256Hash(data);
+
+  // Iterate over the hashed input in 4 byte increments, combine those 4
+  // bytes into an unsigned int and use it as the index into our
+  // |bit_sets|.
+  for (size_t i = 0; i < hashed.size(); i += 4) {
+    uint32_t hash = uint32_t{hashed[i]} << 24 | uint32_t{hashed[i + 1]} << 16 |
+                    uint32_t{hashed[i + 2]} << 8 | hashed[i + 3];
+
+    size_t num_bits = bit_sets.size() * kBitsInByte;
+    size_t n = hash % num_bits;
+    size_t byte_index = floor(n / kBitsInByte);
+    size_t bit_index = n % kBitsInByte;
+    bool is_set = (bit_sets[byte_index] >> bit_index) & 0x01;
+
+    if (!is_set) return false;
+  }
+  NEARBY_LOGS(INFO) << __func__ << " The accountkey is possibly in set.";
+  return true;
+}
+
+}  // namespace
+
+AccountKeyFilter::AccountKeyFilter(
+    const NonDiscoverableAdvertisement& advertisement)
+    : bit_sets_(advertisement.account_key_filter) {
+  salt_values_.resize(advertisement.salt.size());
+  std::copy(advertisement.salt.begin(), advertisement.salt.end(),
+            salt_values_.begin());
+
+  // If the advertisement contains battery information, then that information
+  // was also appended to the account keys to generate the filter. We need to
+  // do the same when checking for matches, so save the values in salt_values_
+  // for that purpose later.
+  if (advertisement.battery_notification) {
+    salt_values_.push_back(advertisement.battery_notification->type ==
+                                   BatteryNotification::Type::kShowUi
+                               ? kShowUi
+                               : kHideUi);
+    for (auto battery_info :
+         advertisement.battery_notification->battery_infos) {
+      salt_values_.push_back(battery_info.ToByte());
+    }
+  }
+}
+
+AccountKeyFilter::AccountKeyFilter(
+    const std::vector<uint8_t>& account_key_filter_bytes,
+    const std::vector<uint8_t>& salt_values)
+    : bit_sets_(account_key_filter_bytes), salt_values_(salt_values) {}
+
+bool AccountKeyFilter::IsPossiblyInSet(const AccountKey& account_key) {
+  if (!account_key.Ok()) {
+    NEARBY_LOGS(INFO) << __func__ << " Invalid account key.";
+    return false;
+  }
+  if (bit_sets_.empty()) return false;
+  // We first need to append the salt value to the input (see
+  // https://developers.google.com/nearby/fast-pair/spec#AccountKeyFilter).
+  std::vector<uint8_t> data(account_key.GetAsBytes().begin(),
+                            account_key.GetAsBytes().end());
+  for (auto& byte : salt_values_) data.push_back(byte);
+
+  // We need to try account keys with different first bytes in case
+  // the peripheral is SASS per
+  // https://developers.google.com/nearby/fast-pair/early-access/specifications/extensions/sass#SassAdvertisingPayload
+  if (AccountKeyFilterChecker(data, bit_sets_)) {
+    return true;
+  }
+  data[0] = kRecentlyUsedByte;
+  if (AccountKeyFilterChecker(data, bit_sets_)) {
+    return true;
+  }
+  data[0] = kInUseByte;
+  return AccountKeyFilterChecker(data, bit_sets_);
+}
+
+}  // namespace fastpair
+}  // namespace nearby

fastpair/common/account_key_filter.h

@@ -0,0 +1,49 @@
+// Copyright 2023 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_FASTPAIR_COMMON_ACCOUNT_KEY_FILTER_H_
+#define THIRD_PARTY_NEARBY_FASTPAIR_COMMON_ACCOUNT_KEY_FILTER_H_
+
+#include <cstdint>
+#include <vector>
+
+#include "fastpair/common/account_key.h"
+#include "fastpair/common/non_discoverable_advertisement.h"
+
+namespace nearby {
+namespace fastpair {
+
+// Class which represents a Fast Pair Account Key account_key_filter
+// https://developers.google.com/nearby/fast-pair/specifications/service/provider#AccountKeyFilter
+class AccountKeyFilter {
+ public:
+  explicit AccountKeyFilter(const NonDiscoverableAdvertisement& advertisement);
+  AccountKeyFilter(const std::vector<uint8_t>& account_key_filter_bytes,
+                   const std::vector<uint8_t>& salt_values);
+  ~AccountKeyFilter() = default;
+
+  // Returns true if the `account_key` is possibly in the account key set
+  // defined by the filter.
+  // Return false if `account_key` is definitely not in set.
+  bool IsPossiblyInSet(const AccountKey& account_key);
+
+ private:
+  std::vector<uint8_t> bit_sets_;
+  std::vector<uint8_t> salt_values_;
+};
+
+}  // namespace fastpair
+}  // namespace nearby
+
+#endif  // THIRD_PARTY_NEARBY_FASTPAIR_COMMON_ACCOUNT_KEY_FILTER_H_

fastpair/common/account_key_filter_test.cc

@@ -0,0 +1,135 @@
+// Copyright 2023 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 "fastpair/common/account_key_filter.h"
+
+#include <optional>
+#include <vector>
+
+#include "gtest/gtest.h"
+#include "fastpair/common/battery_notification.h"
+#include "fastpair/common/non_discoverable_advertisement.h"
+
+namespace nearby {
+namespace fastpair {
+namespace {
+// Test data comes from:
+// https://developers.google.com/nearby/fast-pair/specifications/appendix/testcases#test_cases
+
+class AccountKeyFilterTest : public testing::Test {
+ protected:
+  const std::vector<uint8_t> salt_{0xC7, 0xC8};
+  const std::vector<uint8_t> account_key_1_{0x11, 0x22, 0x33, 0x44, 0x55, 0x66,
+                                            0x77, 0x88, 0x99, 0x00, 0xAA, 0xBB,
+                                            0xCC, 0xDD, 0xEE, 0xFF};
+  const std::vector<uint8_t> account_key_2_{0x11, 0x11, 0x22, 0x22, 0x33, 0x33,
+                                            0x44, 0x44, 0x55, 0x55, 0x66, 0x66,
+                                            0x77, 0x77, 0x88, 0x88};
+
+  const std::vector<uint8_t> filter_1_{0x02, 0x0C, 0x80, 0x2A};
+  const std::vector<uint8_t> filter_2_{0x84, 0x4A, 0x62, 0x20, 0x8B};
+  const std::vector<uint8_t> filter_1_and_2_{0x84, 0x4A, 0x62, 0x20, 0x8B};
+
+  const std::vector<uint8_t> battery_data_{0b00110011, 0b01000000, 0b01000000,
+                                           0b01000000};
+
+  const std::vector<uint8_t> filter_1_with_battery_{0x01, 0x01, 0x46, 0x0A};
+  const std::vector<uint8_t> filter_2_with_battery_{0x46, 0x15, 0x24, 0xD0,
+                                                    0x08};
+};
+
+TEST_F(AccountKeyFilterTest, ConstrutorWithNonDiscoverableAdvertisement) {
+  NonDiscoverableAdvertisement non_discoverable_advertisement(
+      filter_1_, NonDiscoverableAdvertisement::Type::kShowUi, salt_, {});
+  EXPECT_TRUE(AccountKeyFilter(non_discoverable_advertisement)
+                  .IsPossiblyInSet(AccountKey(account_key_1_)));
+  NonDiscoverableAdvertisement non_discoverable_advertisement_with_battery(
+      filter_1_with_battery_, NonDiscoverableAdvertisement::Type::kShowUi,
+      salt_,
+      BatteryNotification::FromBytes({0b01000000, 0b01000000, 0b01000000},
+                                     BatteryNotification::Type::kShowUi));
+  EXPECT_TRUE(AccountKeyFilter(non_discoverable_advertisement_with_battery)
+                  .IsPossiblyInSet(AccountKey(account_key_1_)));
+}
+
+TEST_F(AccountKeyFilterTest, EmptyAccountKeyFilter) {
+  AccountKeyFilter filter({}, {});
+
+  EXPECT_FALSE(filter.IsPossiblyInSet(AccountKey(account_key_1_)));
+  EXPECT_FALSE(filter.IsPossiblyInSet(AccountKey(account_key_2_)));
+}
+
+TEST_F(AccountKeyFilterTest, EmptyAccountKey) {
+  EXPECT_FALSE(
+      AccountKeyFilter(filter_1_, salt_).IsPossiblyInSet(AccountKey("")));
+}
+
+TEST_F(AccountKeyFilterTest, SingleAccountKey) {
+  EXPECT_TRUE(AccountKeyFilter(filter_1_, salt_)
+                  .IsPossiblyInSet(AccountKey(AccountKey(account_key_1_))));
+  EXPECT_TRUE(AccountKeyFilter(filter_2_, salt_)
+                  .IsPossiblyInSet(AccountKey(account_key_2_)));
+}
+
+TEST_F(AccountKeyFilterTest, TwoAccountKeys) {
+  AccountKeyFilter filter(filter_1_and_2_, salt_);
+  EXPECT_TRUE(filter.IsPossiblyInSet(AccountKey(account_key_1_)));
+  EXPECT_TRUE(filter.IsPossiblyInSet(AccountKey(account_key_2_)));
+}
+
+TEST_F(AccountKeyFilterTest, MissingAccountKey) {
+  const std::vector<uint8_t> bytes{0x12, 0x22, 0x33, 0x44, 0x55, 0x66,
+                                   0x77, 0x88, 0x99, 0x00, 0xAA, 0xBB,
+                                   0xCC, 0xDD, 0xEE, 0xFF};
+  AccountKey account_key(bytes);
+  EXPECT_FALSE(AccountKeyFilter(filter_1_, salt_).IsPossiblyInSet(account_key));
+  EXPECT_FALSE(
+      AccountKeyFilter(filter_1_and_2_, salt_).IsPossiblyInSet(account_key));
+}
+
+TEST_F(AccountKeyFilterTest, AccountKeyWithBatteryData) {
+  std::vector<uint8_t> salt_1 = salt_;
+  for (auto& byte : battery_data_) salt_1.push_back(byte);
+
+  EXPECT_TRUE(AccountKeyFilter(filter_1_with_battery_, salt_1)
+                  .IsPossiblyInSet(AccountKey(account_key_1_)));
+
+  std::vector<uint8_t> salt_2 = salt_;
+  for (auto& byte : battery_data_) salt_2.push_back(byte);
+
+  EXPECT_TRUE(AccountKeyFilter(filter_2_with_battery_, salt_2)
+                  .IsPossiblyInSet(AccountKey(account_key_2_)));
+}
+
+TEST_F(AccountKeyFilterTest, SassEnabledPeripheral) {
+  // Value source: b/243855406#comment24
+  const std::vector<uint8_t> bytes{0x06, 0x3F, 0xC1, 0x8C, 0x63, 0xDC,
+                                   0x75, 0x1A, 0xE8, 0x1A, 0xCF, 0x65,
+                                   0x10, 0x15, 0x1D, 0xB0};
+  AccountKey account_key_3(bytes);
+  const std::vector<uint8_t> filter4{0x19, 0x23, 0x50, 0xE8, 0x37,
+                                     0x68, 0xF0, 0x65, 0x22};
+  const std::vector<uint8_t> salt4{0xD7, 0xDE};
+  const std::vector<uint8_t> batteryData4{0x33, 0xE4, 0xE4, 0x64};
+  std::vector<uint8_t> salt_values{};
+  salt_values.insert(salt_values.end(), salt4.begin(), salt4.end());
+  salt_values.insert(salt_values.end(), batteryData4.begin(),
+                     batteryData4.end());
+  EXPECT_TRUE(
+      AccountKeyFilter(filter4, salt_values).IsPossiblyInSet(account_key_3));
+}
+
+}  // namespace
+}  // namespace fastpair
+}  // namespace nearby

fastpair/common/fast_pair_device.cc

@@ -26,8 +26,8 @@ namespace fastpair {
 std::ostream& operator<<(std::ostream& stream, const FastPairDevice& device) {
   stream << "[Device: model_id = " << device.GetModelId()
          << ", ble_address = " << device.GetBleAddress()
-         << ", public_address = " << device.public_address().value_or("null")
-         << ", display_name = " << device.display_name().value_or("null")
+         << ", public_address = " << device.GetPublicAddress().value_or("null")
+         << ", display_name = " << device.GetDisplayName().value_or("null")
          << ", " << device.GetAccountKey()
          << ", protocol = " << device.GetProtocol() << "]";
 

fastpair/common/fast_pair_device.h

@@ -23,6 +23,7 @@
 #include <vector>
 
 #include "absl/container/flat_hash_map.h"
+#include "absl/strings/string_view.h"
 #include "fastpair/common/account_key.h"
 #include "fastpair/common/protocol.h"
 
@@ -48,25 +49,23 @@ class FastPairDevice {
   FastPairDevice& operator=(FastPairDevice&&) = delete;
   ~FastPairDevice() = default;
 
-  const std::optional<std::string>& public_address() const {
+  std::optional<std::string> GetPublicAddress() const {
     return public_address_;
   }
 
-  void set_public_address(absl::string_view address) {
+  void SetPublicAddress(absl::string_view address) {
     public_address_ = std::string(address);
   }
 
-  const std::optional<std::string>& display_name() const {
-    return display_name_;
+  std::optional<std::string> GetDisplayName() const { return display_name_; }
+
+  void SetDisplayName(absl::string_view display_name) {
+    display_name_ = std::string(display_name);
   }
 
-  void set_display_name(const std::optional<std::string>& display_name) {
-    display_name_ = display_name;
-  }
+  std::optional<DeviceFastPairVersion> GetVersion() { return version_; }
 
-  std::optional<DeviceFastPairVersion> version() { return version_; }
-
-  void set_version(std::optional<DeviceFastPairVersion> version) {
+  void SetVersion(std::optional<DeviceFastPairVersion> version) {
     version_ = version;
   }
 
@@ -74,7 +73,10 @@ class FastPairDevice {
 
   void SetAccountKey(AccountKey account_key) { account_key_ = account_key; }
 
-  void SetModelId(absl::string_view model_id) { model_id_ = model_id; }
+  void SetModelId(absl::string_view model_id) {
+    model_id_ = std::string(model_id);
+  }
+
   absl::string_view GetModelId() const { return model_id_; }
 
   void SetBleAddress(absl::string_view address) {

fastpair/common/fast_pair_device_test.cc

@@ -46,44 +46,34 @@ TEST(FastPairDevice, GetAndSetName) {
   FastPairDevice device("model_id", "ble_address",
                         Protocol::kFastPairInitialPairing);
   // Test that name returns null before any sets.
-  std::optional<std::string> name = device.display_name();
-  EXPECT_FALSE(name.has_value());
+  EXPECT_FALSE(device.GetDisplayName().has_value());
 
   // Test that name returns the set value.
   std::string test_name = "test_name";
-  device.set_display_name(test_name);
-  name = device.display_name();
-  EXPECT_TRUE(name.has_value());
-  EXPECT_EQ(name.value(), test_name);
+  device.SetDisplayName(test_name);
+  EXPECT_EQ(device.GetDisplayName().value(), test_name);
 
   // Test that overriding works.
   std::string new_test_name = "new_test_name";
-  device.set_display_name(new_test_name);
-  name = device.display_name();
-  EXPECT_TRUE(name.has_value());
-  EXPECT_EQ(name.value(), new_test_name);
+  device.SetDisplayName(new_test_name);
+  EXPECT_EQ(device.GetDisplayName().value(), new_test_name);
 }
 
 TEST(FastPairDevice, GetAndPublicAddress) {
   FastPairDevice device("model_id", "ble_address",
                         Protocol::kFastPairInitialPairing);
   // Test that public address returns null before any sets.
-  std::optional<std::string> public_address = device.public_address();
-  EXPECT_FALSE(public_address.has_value());
+  EXPECT_FALSE(device.GetPublicAddress().has_value());
 
   // Test that name returns the set value.
-  std::string test_public_address = "test_public_address ";
-  device.set_public_address(test_public_address);
-  public_address = device.public_address();
-  EXPECT_TRUE(public_address.has_value());
-  EXPECT_EQ(public_address.value(), test_public_address);
+  std::string test_GetPublicAddress = "test_GetPublicAddress ";
+  device.SetPublicAddress(test_GetPublicAddress);
+  EXPECT_EQ(device.GetPublicAddress().value(), test_GetPublicAddress);
 
   // Test that overriding works.
-  std::string new_test_public_address = "new_test_public_address ";
-  device.set_public_address(new_test_public_address);
-  public_address = device.public_address();
-  EXPECT_TRUE(public_address.has_value());
-  EXPECT_EQ(public_address.value(), new_test_public_address);
+  std::string new_test_GetPublicAddress = "new_test_GetPublicAddress ";
+  device.SetPublicAddress(new_test_GetPublicAddress);
+  EXPECT_EQ(device.GetPublicAddress().value(), new_test_GetPublicAddress);
 }
 
 }  // namespace

fastpair/fast_pair_controller.cc

@@ -33,7 +33,7 @@ namespace fastpair {
 FastPairController::FastPairController(Mediums* mediums,
                                        const BluetoothDevice& device)
     : mediums_(mediums), device_(Protocol::kFastPairRetroactivePairing) {
-  device_.set_public_address(device.GetMacAddress());
+  device_.SetPublicAddress(device.GetMacAddress());
 }
 
 absl::Status FastPairController::OpenMessageStream() {

fastpair/handshake/fast_pair_handshake_impl.cc

@@ -126,7 +126,7 @@ void FastPairHandshakeImpl::OnParseDecryptedResponse(
   NEARBY_LOGS(INFO) << __func__
                     << ": Successfully decrypted and parsed response.";
 
-  device.set_public_address(
+  device.SetPublicAddress(
       device::CanonicalizeBluetoothAddress(response->address_bytes));
   completed_successfully_ = true;
   std::move(on_complete_callback_)(device, absl::nullopt);

fastpair/handshake/fast_pair_handshake_impl_test.cc

@@ -225,7 +225,7 @@ TEST_F(FastPairHandshakeImplTest, Success) {
       device, mediums,
       [&](FastPairDevice& callback_device, std::optional<PairFailure> failure) {
         EXPECT_EQ(&device, &callback_device);
-        EXPECT_EQ(device.public_address(), kPublicAddress);
+        EXPECT_EQ(device.GetPublicAddress(), kPublicAddress);
         EXPECT_FALSE(failure.has_value());
         latch.CountDown();
       });

fastpair/handshake/fast_pair_handshake_lookup.cc

@@ -46,7 +46,7 @@ FastPairHandshake* FastPairHandshakeLookup::Get(FastPairDevice* device) {
 FastPairHandshake* FastPairHandshakeLookup::Get(absl::string_view address) {
   absl::MutexLock lock(&mutex_);
   for (const auto& pair : fast_pair_handshakes_) {
-    if (pair.first->public_address() == address ||
+    if (pair.first->GetPublicAddress() == address ||
         pair.first->GetBleAddress() == address) {
       return pair.second.get();
     }
@@ -62,7 +62,7 @@ bool FastPairHandshakeLookup::Erase(FastPairDevice* device) {
 bool FastPairHandshakeLookup::Erase(absl::string_view address) {
   absl::MutexLock lock(&mutex_);
   for (const auto& pair : fast_pair_handshakes_) {
-    if (pair.first->public_address() == address ||
+    if (pair.first->GetPublicAddress() == address ||
         pair.first->GetBleAddress() == address) {
       fast_pair_handshakes_.erase(pair.first);
       return true;

fastpair/handshake/fast_pair_handshake_lookup_test.cc

@@ -45,7 +45,7 @@ class FastPairHandshakeLookupTest : public ::testing::Test {
     provider_address_ = adapter_.GetMacAddress();
     device_ = new FastPairDevice(kValidModelId, provider_address_,
                                  Protocol::kFastPairInitialPairing);
-    device_->set_public_address(kPubliceAddress);
+    device_->SetPublicAddress(kPubliceAddress);
   }
 
   ~FastPairHandshakeLookupTest() override { delete device_; }

fastpair/message_stream/medium.cc

@@ -37,7 +37,7 @@ absl::Status Medium::OpenRfcomm() {
   if (!bt_classic_medium_.has_value()) {
     return absl::FailedPreconditionError("BT classic unsupported");
   }
-  if (!device_.public_address().has_value()) {
+  if (!device_.GetPublicAddress().has_value()) {
     return absl::FailedPreconditionError(
         "Connect open RFCOMM without public BT address");
   }
@@ -45,10 +45,11 @@ absl::Status Medium::OpenRfcomm() {
   executor_.Execute("open-rfcomm", [this, classic_medium]() {
     if (cancellation_flag_.Cancelled()) return;
     BluetoothDevice device =
-        classic_medium->GetRemoteDevice(device_.public_address().value());
+        classic_medium->GetRemoteDevice(device_.GetPublicAddress().value());
     if (!device.IsValid()) {
-      observer_.OnConnectionResult(absl::UnavailableError(absl::StrFormat(
-          "Remote BT device %s not found", device_.public_address().value())));
+      observer_.OnConnectionResult(absl::UnavailableError(
+          absl::StrFormat("Remote BT device %s not found",
+                          device_.GetPublicAddress().value())));
       return;
     }
     SetSocket(classic_medium->ConnectToService(device, kRfcommUuid,
@@ -58,7 +59,7 @@ absl::Status Medium::OpenRfcomm() {
                               ? absl::OkStatus()
                               : absl::UnavailableError(absl::StrFormat(
                                     "Failed to open RFCOMM with %s",
-                                    device_.public_address().value()));
+                                    device_.GetPublicAddress().value()));
     observer_.OnConnectionResult(status);
     if (status.ok()) {
       RunLoop(std::move(socket));

fastpair/message_stream/medium_test.cc

@@ -72,7 +72,7 @@ class MediumTest : public testing::Test {
 TEST_F(MediumTest, ConnectWithNonExistingDeviceFails) {
   FastPairDevice fp_device("model id", "ble address",
                            Protocol::kFastPairRetroactivePairing);
-  fp_device.set_public_address("11:22:33:44:55:66");
+  fp_device.SetPublicAddress("11:22:33:44:55:66");
   Medium medium =
       Medium(fp_device, std::optional<BluetoothClassicMedium*>(&seeker_medium_),
              observer_);
@@ -86,7 +86,7 @@ TEST_F(MediumTest, ConnectWithNonExistingDeviceFails) {
 TEST_F(MediumTest, Connect) {
   FastPairDevice fp_device("model id", "ble address",
                            Protocol::kFastPairRetroactivePairing);
-  fp_device.set_public_address(provider_.GetMacAddress());
+  fp_device.SetPublicAddress(provider_.GetMacAddress());
   provider_.DiscoverProvider(seeker_medium_);
   provider_.EnableProviderRfcomm();
   Medium medium =
@@ -101,7 +101,7 @@ TEST_F(MediumTest, Connect) {
 TEST_F(MediumTest, ProviderDisconnectsCallsOnDisconnectCallback) {
   FastPairDevice fp_device("model id", "ble address",
                            Protocol::kFastPairRetroactivePairing);
-  fp_device.set_public_address(provider_.GetMacAddress());
+  fp_device.SetPublicAddress(provider_.GetMacAddress());
   provider_.DiscoverProvider(seeker_medium_);
   provider_.EnableProviderRfcomm();
   Medium medium =
@@ -120,7 +120,7 @@ TEST_F(MediumTest, ProviderDisconnectsCallsOnDisconnectCallback) {
 TEST_F(MediumTest, DisconnectSendFails) {
   FastPairDevice fp_device("model id", "ble address",
                            Protocol::kFastPairRetroactivePairing);
-  fp_device.set_public_address(provider_.GetMacAddress());
+  fp_device.SetPublicAddress(provider_.GetMacAddress());
   provider_.DiscoverProvider(seeker_medium_);
   provider_.EnableProviderRfcomm();
   Medium medium =
@@ -145,7 +145,7 @@ TEST_F(MediumTest, SendMessage) {
   std::string expected_result = absl::HexStringToBytes("030A0003ABCDEF");
   FastPairDevice fp_device("model id", "ble address",
                            Protocol::kFastPairRetroactivePairing);
-  fp_device.set_public_address(provider_.GetMacAddress());
+  fp_device.SetPublicAddress(provider_.GetMacAddress());
   provider_.DiscoverProvider(seeker_medium_);
   provider_.EnableProviderRfcomm();
   Medium medium =
@@ -170,7 +170,7 @@ TEST_F(MediumTest, ReceiveMessage) {
   std::string input = absl::HexStringToBytes("030A0003ABCDEF");
   FastPairDevice fp_device("model id", "ble address",
                            Protocol::kFastPairRetroactivePairing);
-  fp_device.set_public_address(provider_.GetMacAddress());
+  fp_device.SetPublicAddress(provider_.GetMacAddress());
   provider_.DiscoverProvider(seeker_medium_);
   provider_.EnableProviderRfcomm();
   Medium medium =
@@ -192,7 +192,7 @@ class MediumFuzzTest : public fuzztest::PerIterationFixtureAdapter<MediumTest> {
   void HandlesAnyInput(absl::string_view input) {
     FastPairDevice fp_device("model id", "ble address",
                              Protocol::kFastPairRetroactivePairing);
-    fp_device.set_public_address(provider_.GetMacAddress());
+    fp_device.SetPublicAddress(provider_.GetMacAddress());
     provider_.DiscoverProvider(seeker_medium_);
     provider_.EnableProviderRfcomm();
     Medium medium = Medium(
@@ -213,7 +213,7 @@ class MediumFuzzTest : public fuzztest::PerIterationFixtureAdapter<MediumTest> {
         .payload = std::string(payload)};
     FastPairDevice fp_device("model id", "ble address",
                              Protocol::kFastPairRetroactivePairing);
-    fp_device.set_public_address(provider_.GetMacAddress());
+    fp_device.SetPublicAddress(provider_.GetMacAddress());
     provider_.DiscoverProvider(seeker_medium_);
     provider_.EnableProviderRfcomm();
     Medium medium = Medium(

fastpair/message_stream/message_stream_test.cc

@@ -111,7 +111,7 @@ class MessageStreamTest : public testing::Test {
   void SetUp() override {
     MediumEnvironment::Instance().Start();
 
-    fp_device_.set_public_address(provider_.GetMacAddress());
+    fp_device_.SetPublicAddress(provider_.GetMacAddress());
     provider_.DiscoverProvider(seeker_medium_);
     provider_.EnableProviderRfcomm();
   }

fastpair/ui/fast_pair/fast_pair_presenter_impl.cc

@@ -64,7 +64,7 @@ void FastPairPresenterImpl::ShowDiscovery(
 void FastPairPresenterImpl::OnDiscoveryMetadataRetrieved(
     FastPairDevice& device, const DeviceMetadata& device_metadata,
     FastPairNotificationController& notification_controller) {
-  device.set_version(device_metadata.GetFastPairVersion());
+  device.SetVersion(device_metadata.GetFastPairVersion());
   notification_controller.ShowGuestDiscoveryNotification(device_metadata,
                                                          std::move(callback_));
 }

fastpair/ui/fast_pair/fast_pair_presenter_impl_test.cc

@@ -58,7 +58,7 @@ TEST(FastPairPresenterImplTest, ShowDiscoveryForV1Version) {
                                     });
   latch_1->Await();
   EXPECT_EQ(notification_controller_observer.on_update_device_count(), 1);
-  EXPECT_EQ(device.version(), DeviceFastPairVersion::kV1);
+  EXPECT_EQ(device.GetVersion(), DeviceFastPairVersion::kV1);
   controller.OnDiscoveryClicked(DiscoveryAction::kPairToDevice);
   latch_2.Await();
   EXPECT_EQ(discovery_action, DiscoveryAction::kPairToDevice);
@@ -92,7 +92,7 @@ TEST(FastPairPresenterImplTest, ShowDiscoveryForHigherThanV1Version) {
                                     });
   latch_1->Await();
   EXPECT_EQ(notification_controller_observer.on_update_device_count(), 1);
-  EXPECT_EQ(device.version(), DeviceFastPairVersion::kHigherThanV1);
+  EXPECT_EQ(device.GetVersion(), DeviceFastPairVersion::kHigherThanV1);
   controller.OnDiscoveryClicked(DiscoveryAction::kDismissedByUser);
   latch_2.Await();
   EXPECT_EQ(discovery_action, DiscoveryAction::kDismissedByUser);

internal/platform/implementation/apple/platform.mm

@@ -233,8 +233,8 @@ absl::StatusOr<WebResponse> ImplementationPlatform::SendRequest(const WebRequest
     webResponse.headers.insert({[key UTF8String], [value UTF8String]});
   }
   if (blockData != nil) {
-    webResponse.body = [[[NSString alloc] initWithData:blockData
-                                              encoding:NSUTF8StringEncoding] UTF8String];
+    // Body is not a UTF-8 encoded string and is just using `std::string` as a container for data.
+    webResponse.body = std::string((char *)blockData.bytes, blockData.length);
   }
   return webResponse;
 }

internal/platform/implementation/windows/preferences_manager.cc

@@ -194,6 +194,12 @@ bool PreferencesManager::Commit() {
 }
 
 bool PreferencesManager::SetValue(absl::string_view key, const json& value) {
+  if (!value_.is_object()) {
+    NEARBY_LOGS(ERROR) << "Preferences is no longer an object! value_="
+                       << value_.dump(4);
+    value_ = json::object();
+  }
+
   if (value_[absl::StrCat(key)] == value) {
     return false;
   }
@@ -205,6 +211,12 @@ bool PreferencesManager::SetValue(absl::string_view key, const json& value) {
 template <typename T>
 T PreferencesManager::GetValue(absl::string_view key,
                                const T& default_value) const {
+  if (!value_.is_object()) {
+    NEARBY_LOGS(ERROR) << "Preferences is no longer an object! value_="
+                       << value_.dump(4);
+    return default_value;
+  }
+
   auto it = value_.find(absl::StrCat(key));
   if (it == value_.end()) {
     return default_value;
@@ -215,6 +227,12 @@ T PreferencesManager::GetValue(absl::string_view key,
 template <typename T>
 bool PreferencesManager::SetArrayValue(absl::string_view key,
                                        absl::Span<const T> value) {
+  if (!value_.is_object()) {
+    NEARBY_LOGS(ERROR) << "Preferences is no longer an object! value_="
+                       << value_.dump(4);
+    value_ = json::object();
+  }
+
   json array_value = json::array();
   for (const T& item_value : value) {
     array_value.push_back(item_value);
@@ -233,6 +251,16 @@ std::vector<T> PreferencesManager::GetArrayValue(
     absl::string_view key, absl::Span<const T> default_value) const {
   std::vector<T> result;
 
+  if (!value_.is_object()) {
+    NEARBY_LOGS(ERROR) << "Preferences is no longer an object! value_="
+                       << value_.dump(4);
+
+    for (const T& value : default_value) {
+      result.push_back(value);
+    }
+    return result;
+  }
+
   auto array_value = value_.find(absl::StrCat(key));
   if (array_value == value_.end() || !array_value->is_array()) {
     for (const T& value : default_value) {

internal/platform/implementation/windows/preferences_manager_test.cc

@@ -16,8 +16,10 @@
 
 #include <stdint.h>
 
+#include <codecvt>
 #include <filesystem>  // NOLINT(build/c++17)
 #include <fstream>
+#include <locale>
 #include <ostream>
 #include <string>
 #include <vector>
@@ -29,6 +31,7 @@
 #include "absl/types/span.h"
 #include "nlohmann/json.hpp"
 #include "nlohmann/json_fwd.hpp"
+#include "internal/platform/logging.h"
 
 namespace nearby {
 namespace windows {
@@ -38,27 +41,27 @@ constexpr absl::Duration kTimeOut = absl::Milliseconds(200);
 constexpr char kPreferencesFilePath[] = "Google/Nearby/Sharing";
 }  // namespace
 
+
 TEST(PreferencesManager, CorruptedConfigFile) {
   std::filesystem::path settingsPath =
-      std::filesystem::temp_directory_path() / "settings.json";
-  std::ofstream output_stream{settingsPath};
-  output_stream << "{\"data\":8, \"names\": [\"In valid\"}" << std::endl;
-  output_stream.close();
+      std::filesystem::temp_directory_path();
+  std::ofstream output_stream{settingsPath / "preferences.json"};
+  output_stream << "CORRUPTED" << std::endl;
 
-  // Should use an empty setting for a corrupted configuration file.
-  EXPECT_EQ(PreferencesManager(kPreferencesFilePath).GetInteger("data", 100),
+  NEARBY_LOGS(INFO) << "Loading preferences from: " << settingsPath.string();
+  EXPECT_EQ(PreferencesManager(settingsPath.string()).GetInteger("data", 100),
             100);
 }
 
 TEST(PreferencesManager, ValidConfigFile) {
   std::filesystem::path settingsPath =
-      std::filesystem::temp_directory_path() / "settings.json";
-  std::ofstream output_stream{settingsPath};
+      std::filesystem::temp_directory_path();
+  std::ofstream output_stream{settingsPath / "preferences.json"};
   output_stream << "{\"data\":8, \"name\": \"Valid\"}" << std::endl;
   output_stream.close();
 
-  // Should use an empty setting for a corrupted configuration file.
-  EXPECT_EQ(PreferencesManager(kPreferencesFilePath).GetInteger("data", 100),
+  NEARBY_LOGS(INFO) << "Loading preferences from: " << settingsPath.string();
+  EXPECT_EQ(PreferencesManager(settingsPath.string()).GetInteger("data", 100),
             8);
 }
 

internal/platform/implementation/windows/preferences_repository.cc

@@ -37,6 +37,15 @@ json PreferencesRepository::LoadPreferences() {
   absl::MutexLock lock(&mutex_);
   std::optional<json> preferences = AttemptLoad();
   if (preferences.has_value()) {
+    // The top level root should be an object, if it's not then something went
+    // wrong or the file was corrupted.
+    if (!preferences.value().is_object()) {
+      NEARBY_LOGS(ERROR) << "Preferences loaded was not a valid object: "
+                         << preferences.value().dump(4);
+
+      return json::object();
+    }
+
     return preferences.value();
   }
 

internal/platform/implementation/windows/preferences_repository_test.cc

@@ -40,6 +40,25 @@ TEST(PreferencesRepository, LoadWithBadPath) {
   EXPECT_TRUE(result.empty());
 }
 
+TEST(PreferencesRepository, RecoverFromBadPreferences) {
+  std::optional<std::filesystem::path> app_data_path =
+      api::ImplementationPlatform::CreateDeviceInfo()->GetLocalAppDataPath();
+  ASSERT_TRUE(app_data_path.has_value());
+  std::filesystem::path full_path = *app_data_path / kPreferencesPath;
+  std::filesystem::path full_name = full_path / kPreferencesFileName;
+
+  if (std::filesystem::exists(full_name)) {
+    std::filesystem::remove(full_name);
+  }
+
+  std::ofstream pref_file(full_name.c_str());
+  pref_file << "\"Bad top level object\"";
+  pref_file.close();
+
+  PreferencesRepository preferences_repository{full_path.string()};
+  EXPECT_EQ(preferences_repository.LoadPreferences(), json::object());
+}
+
 TEST(PreferencesRepository, SaveAndLoadPreferences) {
   std::optional<std::filesystem::path> app_data_path =
       api::ImplementationPlatform::CreateDeviceInfo()->GetLocalAppDataPath();

internal/platform/implementation/windows/scheduled_executor.h

@@ -62,7 +62,7 @@ class ScheduledExecutor : public api::ScheduledExecutor {
         : task_(std::move(task)), duration_(duration) {}
 
     bool Cancel() override {
-      if (is_executed_) {
+      if (is_executed_ || is_cancelled_) {
         return false;
       }
 

internal/proto/BUILD

@@ -42,3 +42,22 @@ cc_proto_library(
     ],
     deps = [":metadata_proto"],
 )
+
+proto_library(
+    name = "tachyon_proto",
+    srcs = [
+        "duration.proto",
+        "ice.proto",
+        "tachyon.proto",
+        "tachyon_common.proto",
+        "tachyon_enums.proto",
+    ],
+    compatible_with = ["//buildenv/target:non_prod"],
+)
+
+cc_proto_library(
+    name = "tachyon_cc_proto",
+    compatible_with = ["//buildenv/target:non_prod"],
+    visibility = ["//visibility:public"],
+    deps = [":tachyon_proto"],
+)

internal/proto/duration.proto

@@ -0,0 +1,24 @@
+// Copyright 2021 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.internal.tachyon_proto;
+
+option optimize_for = LITE_RUNTIME;
+
+message Duration {
+  int64 seconds = 1;
+  int32 nanos = 2;
+}

internal/proto/ice.proto

@@ -0,0 +1,46 @@
+// Copyright 2021 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.internal.tachyon_proto;
+
+import "internal/proto/duration.proto";
+
+option optimize_for = LITE_RUNTIME;
+
+message ICEConfiguration {
+  // Duration the config is valid for.
+  Duration lifetime_duration = 1;
+
+  // ICE servers to be used by the client to establish a connection.
+  // E.g.:
+  // [ { "urls": "stun:stun1.example.net" }, { "urls": "turn:turn.example.org",
+  // "username": "user", "credential": "myPassword" } ]
+  repeated ICEServerList ice_servers = 2;
+}
+
+message ICEServerList {
+  // STUN or TURN URI(s) as defined in [rfc7064] and [rfc7065] or other URI
+  // types.
+  repeated string urls = 1;
+
+  // If this IceServer object represents a TURN server, then this attribute
+  // specifies the credential to use with that TURN server.
+  string username = 2;
+
+  // If this IceServer object represents a TURN server, then this attribute
+  // specifies the credential to use with that TURN server.
+  string credential = 3;
+}

internal/proto/tachyon.proto

@@ -0,0 +1,55 @@
+// Copyright 2021 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.internal.tachyon_proto;
+
+import "internal/proto/ice.proto";
+import "internal/proto/tachyon_common.proto";
+
+option optimize_for = LITE_RUNTIME;
+
+message GetICEServerRequest {
+  // header is the request header
+  RequestHeader header = 1;
+  // ICE config preference.
+  string ice_config_preference = 3;
+}
+
+message GetICEServerResponse {
+  // The ice server configuration to use.
+  ICEConfiguration ice_config = 4;
+}
+
+message SendMessageExpressRequest {
+  RequestHeader header = 1;
+  Id dest_id = 3;
+  InboxMessage message = 4;
+}
+
+message ReceiveMessagesExpressRequest {
+  RequestHeader header = 1;
+}
+
+message ReceiveMessagesResponse {
+  message Header {}
+  Header header = 1;
+  message FastPathReady {}
+
+  oneof body {
+    InboxMessage inbox_message = 2;
+    FastPathReady fast_path_ready = 7;
+  }
+}

internal/proto/tachyon_common.proto

@@ -0,0 +1,119 @@
+// Copyright 2021 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.internal.tachyon_proto;
+
+import "internal/proto/tachyon_enums.proto";
+
+option optimize_for = LITE_RUNTIME;
+
+message Id {
+  // type defines what the id field contains, e.g. phone number, Fi-number, Gaia
+  // ID etc.
+  IdType.Type type = 1;
+  // id is a unique (for this type and app) identifier of a message source or
+  // recipient.
+  string id = 2;
+
+  // app is the tachyon client application that generated or is to receive a
+  // message.
+  string app = 3;
+
+  // location_hint is used as a hint for the user's region.
+  LocationHint location_hint = 5;
+}
+
+// LocationHint is used to specify a location as well as format.
+message LocationHint {
+  // Location is the location, provided in the format specified by format.
+  string location = 1;
+
+  // the format of location.
+  LocationStandard.Format format = 2;
+}
+
+// RequestHeader must be included in all request messages with the field name
+// `header`.
+// This will make the generated Go types of requests have GetHeader and
+// SetHeader funcs, which makes it simple to create a common interface to access
+// the header values.
+message RequestHeader {
+  // request_id identifies this request and its responses, must be unique, and
+  // is generated by the request creator.
+  // This does not need to be set if request_id_binary is set.
+  string request_id = 1;
+
+  // app identifies the application; this is used to isolate different
+  // applications in the backend.
+  string app = 3;
+
+  // client_info holds information about the calling client application.
+  ClientInfo client_info = 7;
+
+  // requester_id is the user ID of the requester.
+  Id requester_id = 10;
+}
+
+message ClientInfo {
+  // major, minor, point and details carry version information from client.
+  int32 major = 3;
+  int32 minor = 4;
+  int32 point = 5;
+
+  // api_version identifies what api_version the client was built against.
+  // This is used by server to:
+  // - push warning messages to clients during bind
+  // - fail RPCs if client is using a too old version
+  // - add backwards compatible code
+  ApiVersion.Value api_version = 7;
+
+  // platform_type is the type of platform, used to construct user agent string
+  // and to determine client node type in logging.
+  Platform.Type platform_type = 9;
+
+  // The APK version name (e.g. "4.0.006_RC2").
+  // Fireball Android sends down the version code in the above major field and
+  // leaves minor and point empty. The code results in a version that is tough
+  // to decipher like "20011296.0.0". The version name here is formatted to make
+  // versioning easier and safer on the server. Currently, this field is only
+  // populated by Fireball Android. See go/fireball-gbot-version.
+  string app_version = 10;
+}
+
+message InboxMessage {
+  string message_id = 1;
+
+  enum MessageType {
+    UNKNOWN = 0;
+    BASIC = 4;
+  }
+  MessageType message_type = 2;
+
+  bytes message = 12;
+
+  enum MessageClass {
+    USER = 0;
+    EPHEMERAL = 2;
+  }
+  MessageClass message_class = 5;
+}
+
+// Matches StreamBody definition from the server:
+// google3/google/rpc/stream_body.proto
+message StreamBody {
+  repeated bytes messages = 1;
+  repeated bytes noop = 15;
+}

internal/proto/tachyon_enums.proto

@@ -0,0 +1,86 @@
+// Copyright 2021 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.internal.tachyon_proto;
+
+option optimize_for = LITE_RUNTIME;
+
+message ApiVersion {
+  enum Value {
+    UNKNOWN = 0;
+
+    // Initial version.
+    V1 = 1;
+
+    // RequestHeaders and AuthToken only.
+    V2 = 2;
+
+    // Block leaked playstore APKs.
+    V3 = 3;
+
+    // InboxMessage.message bytes are used.
+    V4 = 4;
+  }
+}
+
+message Platform {
+  enum Type {
+    UNKNOWN = 0;
+    TEST = 4;
+    DESKTOP = 6;
+  }
+}
+
+message LocationStandard {
+  enum Format {
+    UNKNOWN = 0;
+    // E164 country codes:
+    // https://en.wikipedia.org/wiki/List_of_country_calling_codes
+    // e.g. +1 for USA
+    E164_CALLING = 1;
+
+    // ISO 3166-1 alpha-2 country codes:
+    // https://en.wikipedia.org/wiki/ISO_3166-1_alpha-2
+    ISO_3166_1_ALPHA_2 = 2;
+  }
+}
+
+message IdType {
+  enum Type {
+    UNSET = 0;
+    NOT_KNOWN = 25;  // client does not know the id type.
+    NEARBY_ID = 27;  // Used by Nearby devices. go/tachyon-express-nearby.
+  }
+}
+
+// Connectivity status from the client network to ice_servers.
+//
+// This message is intended to be binary compatible and should be kept up to
+// date with BlockStatus in
+// google3/google/communications/networktraversal/v1alpha/networktraversal.proto
+message ConnectivityStatus {
+  // Status enumerations.
+  enum Type {
+    // Unspecified.
+    UNKNOWN = 0;
+
+    // ICE connectivity is not blocked.
+    ICE_UNBLOCKED = 1;
+
+    // ICE connectivity is possibly blocked.
+    ICE_POSSIBLY_BLOCKED = 2;
+  }
+}