analytics: 3p NC: Implement internal class PendingPayload and LogicalConnection. This is pre-work for EstablishedConnections.

PiperOrigin-RevId: 394404345

cpp/analytics/analytics_recorder.cc

@@ -29,36 +29,29 @@ namespace nearby {
 namespace analytics {
 
 using ::location::nearby::analytics::proto::ConnectionsLog;
-using AdvertisingPhase =
-    ::location::nearby::analytics::proto::ConnectionsLog::AdvertisingPhase;
-using ClientSession =
-    ::location::nearby::analytics::proto::ConnectionsLog::ClientSession;
-using ConnectionRequest =
-    ::location::nearby::analytics::proto::ConnectionsLog::ConnectionRequest;
-using DiscoveredEndpoint =
-    ::location::nearby::analytics::proto::ConnectionsLog::DiscoveredEndpoint;
-using DiscoveryPhase =
-    ::location::nearby::analytics::proto::ConnectionsLog::DiscoveryPhase;
-using StrategySession =
-    ::location::nearby::analytics::proto::ConnectionsLog::StrategySession;
 using ::location::nearby::proto::connections::ACCEPTED;
 using ::location::nearby::proto::connections::ADVERTISER;
 using ::location::nearby::proto::connections::CLIENT_SESSION;
+using ::location::nearby::proto::connections::CONNECTION_CLOSED;
 using ::location::nearby::proto::connections::ConnectionAttemptResult;
 using ::location::nearby::proto::connections::ConnectionAttemptType;
 using ::location::nearby::proto::connections::ConnectionRequestResponse;
 using ::location::nearby::proto::connections::ConnectionsStrategy;
+using ::location::nearby::proto::connections::DisconnectionReason;
 using ::location::nearby::proto::connections::DISCOVERER;
 using ::location::nearby::proto::connections::EventType;
 using ::location::nearby::proto::connections::IGNORED;
 using ::location::nearby::proto::connections::INCOMING;
 using ::location::nearby::proto::connections::INITIAL;
 using ::location::nearby::proto::connections::Medium;
+using ::location::nearby::proto::connections::MOVED_TO_NEW_MEDIUM;
 using ::location::nearby::proto::connections::NOT_SENT;
 using ::location::nearby::proto::connections::OUTGOING;
 using ::location::nearby::proto::connections::P2P_CLUSTER;
 using ::location::nearby::proto::connections::P2P_POINT_TO_POINT;
 using ::location::nearby::proto::connections::P2P_STAR;
+using ::location::nearby::proto::connections::PayloadStatus;
+using ::location::nearby::proto::connections::PayloadType;
 using ::location::nearby::proto::connections::REJECTED;
 using ::location::nearby::proto::connections::RESULT_SUCCESS;
 using ::location::nearby::proto::connections::SessionRole;
@@ -66,7 +59,10 @@ using ::location::nearby::proto::connections::START_CLIENT_SESSION;
 using ::location::nearby::proto::connections::START_STRATEGY_SESSION;
 using ::location::nearby::proto::connections::STOP_CLIENT_SESSION;
 using ::location::nearby::proto::connections::STOP_STRATEGY_SESSION;
+using ::location::nearby::proto::connections::UNFINISHED;
+using ::location::nearby::proto::connections::UNKNOWN_MEDIUM;
 using ::location::nearby::proto::connections::UNKNOWN_STRATEGY;
+using ::location::nearby::proto::connections::UPGRADED;
 
 // These definitions are necessary before C++17.
 constexpr absl::string_view AnalyticsRecorder::kVersion;
@@ -104,7 +100,8 @@ void AnalyticsRecorder::OnStartAdvertising(connections::Strategy strategy,
 
   // Initialize and set a AdvertisingPhase.
   started_advertising_phase_time_ = SystemClock::ElapsedRealtime();
-  current_advertising_phase_ = std::make_unique<AdvertisingPhase>();
+  current_advertising_phase_ =
+      absl::make_unique<proto::ConnectionsLog::AdvertisingPhase>();
   absl::c_copy(mediums, RepeatedFieldBackInserter(
                             current_advertising_phase_->mutable_medium()));
 }
@@ -134,7 +131,8 @@ void AnalyticsRecorder::OnStartDiscovery(connections::Strategy strategy,
 
   // Initialize and set a DiscoveryPhase.
   started_discovery_phase_time_ = SystemClock::ElapsedRealtime();
-  current_discovery_phase_ = std::make_unique<DiscoveryPhase>();
+  current_discovery_phase_ =
+      absl::make_unique<ConnectionsLog::DiscoveryPhase>();
   for (auto medium : mediums) {
     current_discovery_phase_->add_medium(medium);
   }
@@ -158,7 +156,7 @@ void AnalyticsRecorder::OnEndpointFound(Medium medium) {
                          "current_discovery_phase_";
     return;
   }
-  DiscoveredEndpoint *discovered_endpoint =
+  ConnectionsLog::DiscoveredEndpoint *discovered_endpoint =
       current_discovery_phase_->add_discovered_endpoint();
   discovered_endpoint->set_medium(medium);
   discovered_endpoint->set_latency_millis(absl::ToInt64Milliseconds(
@@ -172,12 +170,13 @@ void AnalyticsRecorder::OnConnectionRequestReceived(
     return;
   }
   absl::Time current_time = SystemClock::ElapsedRealtime();
-  auto connection_request(std::make_unique<ConnectionRequest>());
+  auto connection_request(
+      absl::make_unique<ConnectionsLog::ConnectionRequest>());
   connection_request->set_duration_millis(absl::ToUnixMillis(current_time));
   connection_request->set_request_delay_millis(absl::ToInt64Milliseconds(
       current_time - started_advertising_phase_time_));
-  incoming_connection_requests_.emplace(remote_endpoint_id,
-                                        std::move(connection_request));
+  incoming_connection_requests_.insert(
+      {remote_endpoint_id, std::move(connection_request)});
 }
 
 void AnalyticsRecorder::OnConnectionRequestSent(
@@ -187,12 +186,13 @@ void AnalyticsRecorder::OnConnectionRequestSent(
     return;
   }
   absl::Time current_time = SystemClock::ElapsedRealtime();
-  auto connection_request(std::make_unique<ConnectionRequest>());
+  auto connection_request(
+      absl::make_unique<ConnectionsLog::ConnectionRequest>());
   connection_request->set_duration_millis(absl::ToUnixMillis(current_time));
   connection_request->set_request_delay_millis(
       absl::ToInt64Milliseconds(current_time - started_discovery_phase_time_));
-  outgoing_connection_requests_.emplace(remote_endpoint_id,
-                                        std::move(connection_request));
+  outgoing_connection_requests_.insert(
+      {remote_endpoint_id, std::move(connection_request)});
 }
 
 void AnalyticsRecorder::OnRemoteEndpointAccepted(
@@ -281,7 +281,8 @@ void AnalyticsRecorder::OnOutgoingConnectionAttempt(
       // ConnectionRequest that, since the ConnectionAttempt has failed, will
       // never be delivered to the advertiser.
       auto pair = outgoing_connection_requests_.extract(it);
-      std::unique_ptr<ConnectionRequest> &connection_request = pair.mapped();
+      std::unique_ptr<ConnectionsLog::ConnectionRequest> &connection_request =
+          pair.mapped();
       connection_request->set_local_response(NOT_SENT);
       connection_request->set_remote_response(NOT_SENT);
       UpdateDiscovererConnectionRequestLocked(connection_request.get());
@@ -323,19 +324,17 @@ bool AnalyticsRecorder::CanRecordAnalyticsLocked(
 }
 
 void AnalyticsRecorder::LogClientSession() {
-  serial_executor_.Execute(
-      "analytics-recorder", [this]() {
-        ConnectionsLog connections_log;
-        connections_log.set_event_type(CLIENT_SESSION);
-        connections_log.set_allocated_client_session(client_session_.release());
-        connections_log.set_version(kVersion);
+  serial_executor_.Execute("analytics-recorder", [this]() {
+    ConnectionsLog connections_log;
+    connections_log.set_event_type(CLIENT_SESSION);
+    connections_log.set_allocated_client_session(client_session_.release());
+    connections_log.set_version(kVersion);
 
-        NEARBY_LOGS(INFO)
-            << "AnalyticsRecorder LogClientSession connections_log="
-            << connections_log.DebugString();
+    NEARBY_LOGS(INFO) << "AnalyticsRecorder LogClientSession connections_log="
+                      << connections_log.DebugString();
 
-        event_logger_->Log(connections_log);
-      });
+    event_logger_->Log(connections_log);
+  });
 }
 
 void AnalyticsRecorder::LogEvent(EventType event_type) {
@@ -377,7 +376,8 @@ void AnalyticsRecorder::UpdateStrategySessionLocked(
     current_strategy_ = strategy;
     FinishStrategySessionLocked();
     LogEvent(START_STRATEGY_SESSION);
-    current_strategy_session_ = std::make_unique<StrategySession>();
+    current_strategy_session_ =
+        absl::make_unique<ConnectionsLog::StrategySession>();
     started_strategy_session_time_ = SystemClock::ElapsedRealtime();
     current_strategy_session_->set_strategy(
         StrategyToConnectionStrategy(strategy));
@@ -402,7 +402,7 @@ void AnalyticsRecorder::FinishAdvertisingPhaseLocked() {
     for (const auto &item : incoming_connection_requests_) {
       // ConnectionRequests still pending have been ignored by the local or
       // remote (or both) endpoints.
-      auto &connection_request = item.second;
+      const auto &connection_request = item.second;
       MarkConnectionRequestIgnoredLocked(connection_request.get());
       UpdateAdvertiserConnectionRequestLocked(connection_request.get());
     }
@@ -430,7 +430,7 @@ void AnalyticsRecorder::FinishDiscoveryPhaseLocked() {
     for (const auto &item : outgoing_connection_requests_) {
       // ConnectionRequests still pending have been ignored by the local or
       // remote (or both) endpoints.
-      auto &connection_request = item.second;
+      const auto &connection_request = item.second;
       MarkConnectionRequestIgnoredLocked(connection_request.get());
       UpdateDiscovererConnectionRequestLocked(connection_request.get());
     }
@@ -442,7 +442,7 @@ void AnalyticsRecorder::FinishDiscoveryPhaseLocked() {
 }
 
 bool AnalyticsRecorder::UpdateAdvertiserConnectionRequestLocked(
-    ConnectionRequest *request) {
+    ConnectionsLog::ConnectionRequest *request) {
   if (current_advertising_phase_ == nullptr) {
     NEARBY_LOGS(INFO)
         << "Unable to record advertiser connection request due to null "
@@ -460,7 +460,7 @@ bool AnalyticsRecorder::UpdateAdvertiserConnectionRequestLocked(
 }
 
 bool AnalyticsRecorder::UpdateDiscovererConnectionRequestLocked(
-    ConnectionRequest *request) {
+    ConnectionsLog::ConnectionRequest *request) {
   if (current_discovery_phase_ == nullptr) {
     NEARBY_LOGS(INFO) << "Unable to record discoverer connection request due "
                          "to null current_discovery_phase_.";
@@ -478,7 +478,7 @@ bool AnalyticsRecorder::UpdateDiscovererConnectionRequestLocked(
 }
 
 bool AnalyticsRecorder::BothEndpointsRespondedLocked(
-    ConnectionRequest *request) {
+    ConnectionsLog::ConnectionRequest *request) {
   return request->has_local_response() && request->has_remote_response();
 }
 
@@ -486,7 +486,7 @@ void AnalyticsRecorder::LocalEndpointRespondedLocked(
     const std::string &remote_endpoint_id, ConnectionRequestResponse response) {
   auto out = outgoing_connection_requests_.find(remote_endpoint_id);
   if (out != outgoing_connection_requests_.end()) {
-    ConnectionRequest *connection_request = out->second.get();
+    ConnectionsLog::ConnectionRequest *connection_request = out->second.get();
     connection_request->set_local_response(response);
     if (UpdateDiscovererConnectionRequestLocked(connection_request)) {
       outgoing_connection_requests_.erase(out);
@@ -494,7 +494,7 @@ void AnalyticsRecorder::LocalEndpointRespondedLocked(
   }
   auto in = incoming_connection_requests_.find(remote_endpoint_id);
   if (in != incoming_connection_requests_.end()) {
-    ConnectionRequest *connection_request = in->second.get();
+    ConnectionsLog::ConnectionRequest *connection_request = in->second.get();
     connection_request->set_local_response(response);
     if (UpdateAdvertiserConnectionRequestLocked(connection_request)) {
       incoming_connection_requests_.erase(in);
@@ -506,7 +506,7 @@ void AnalyticsRecorder::RemoteEndpointRespondedLocked(
     const std::string &remote_endpoint_id, ConnectionRequestResponse response) {
   auto out = outgoing_connection_requests_.find(remote_endpoint_id);
   if (out != outgoing_connection_requests_.end()) {
-    ConnectionRequest *connection_request = out->second.get();
+    ConnectionsLog::ConnectionRequest *connection_request = out->second.get();
     connection_request->set_remote_response(response);
     if (UpdateDiscovererConnectionRequestLocked(connection_request)) {
       outgoing_connection_requests_.erase(out);
@@ -514,7 +514,7 @@ void AnalyticsRecorder::RemoteEndpointRespondedLocked(
   }
   auto in = incoming_connection_requests_.find(remote_endpoint_id);
   if (in != incoming_connection_requests_.end()) {
-    ConnectionRequest *connection_request = in->second.get();
+    ConnectionsLog::ConnectionRequest *connection_request = in->second.get();
     connection_request->set_remote_response(response);
     if (UpdateAdvertiserConnectionRequestLocked(connection_request)) {
       incoming_connection_requests_.erase(in);
@@ -523,7 +523,7 @@ void AnalyticsRecorder::RemoteEndpointRespondedLocked(
 }
 
 void AnalyticsRecorder::MarkConnectionRequestIgnoredLocked(
-    ConnectionRequest *request) {
+    ConnectionsLog::ConnectionRequest *request) {
   if (!request->has_local_response()) {
     request->set_local_response(IGNORED);
   }
@@ -539,7 +539,7 @@ void AnalyticsRecorder::FinishStrategySessionLocked() {
 
     // Add the StrategySession in ClientSession
     current_strategy_session_->set_duration_millis(absl::ToInt64Milliseconds(
-        SystemClock::ElapsedRealtime() - SystemClock::ElapsedRealtime()));
+        started_strategy_session_time_ - SystemClock::ElapsedRealtime()));
     *client_session_->add_strategy_session() =
         *std::move(current_strategy_session_);
     LogEvent(STOP_STRATEGY_SESSION);
@@ -560,6 +560,236 @@ ConnectionsStrategy AnalyticsRecorder::StrategyToConnectionStrategy(
   return UNKNOWN_STRATEGY;
 }
 
+void AnalyticsRecorder::PendingPayload::AddChunk(
+    std::int64_t chunk_size_bytes) {
+  num_bytes_transferred_ += chunk_size_bytes;
+  num_chunks_++;
+}
+
+ConnectionsLog::Payload AnalyticsRecorder::PendingPayload::GetProtoPayload(
+    PayloadStatus status) {
+  ConnectionsLog::Payload payload;
+  payload.set_duration_millis(
+      absl::ToInt64Milliseconds(SystemClock::ElapsedRealtime() - start_time_));
+  payload.set_type(type_);
+  payload.set_total_size_bytes(total_size_bytes_);
+  payload.set_num_bytes_transferred(num_bytes_transferred_);
+  payload.set_num_chunks(num_chunks_);
+  payload.set_status(status);
+
+  return payload;
+}
+
+void AnalyticsRecorder::LogicalConnection::PhysicalConnectionEstablished(
+    Medium medium, const std::string &connection_token) {
+  if (current_medium_ != UNKNOWN_MEDIUM) {
+    NEARBY_LOGS(WARNING)
+        << "Unexpected call to PhysicalConnectionEstablished while "
+           "AnalyticsRecorder still has an active current medium.";
+  }
+
+  auto established_connection =
+      absl::make_unique<ConnectionsLog::EstablishedConnection>();
+  established_connection->set_medium(medium);
+  established_connection->set_duration_millis(
+      absl::ToUnixMillis(SystemClock::ElapsedRealtime()));
+  established_connection->set_connection_token(connection_token);
+  physical_connections_.insert({medium, std::move(established_connection)});
+  current_medium_ = medium;
+}
+
+void AnalyticsRecorder::LogicalConnection::PhysicalConnectionClosed(
+    Medium medium, DisconnectionReason reason) {
+  if (current_medium_ == UNKNOWN_MEDIUM) {
+    NEARBY_LOGS(WARNING)
+        << "Unexpected call to PhysicalConnectionClosed() for medium  "
+        << Medium_Name(medium)
+        << " while AnalyticsRecorder has no active current medium";
+  } else if (current_medium_ != medium) {
+    NEARBY_LOGS(WARNING)
+        << "Unexpected call to PhysicalConnectionClosed() for medium "
+        << Medium_Name(medium) << "while AnalyticsRecorder has active medium "
+        << Medium_Name(current_medium_);
+  }
+
+  auto it = physical_connections_.find(medium);
+  if (it == physical_connections_.end()) {
+    NEARBY_LOGS(WARNING)
+        << "Unexpected call to physicalConnectionClosed() for medium "
+        << Medium_Name(medium)
+        << " with no corresponding EstablishedConnection that was previously"
+           " opened.";
+    return;
+  }
+  auto *established_connection = it->second.get();
+  if (established_connection->has_disconnection_reason()) {
+    NEARBY_LOGS(WARNING)
+        << "Unexpected call to physicalConnectionClosed() for medium "
+        << Medium_Name(medium) << " which already has disconnection reason "
+        << DisconnectionReason_Name(
+               established_connection->disconnection_reason());
+    return;
+  }
+  FinishPhysicalConnection(established_connection, reason);
+
+  if (medium == current_medium_) {
+    // If the EstablishedConnection we just closed was the one that we have
+    // marked as current, unset currentMedium.
+    current_medium_ = UNKNOWN_MEDIUM;
+  }
+}
+
+void AnalyticsRecorder::LogicalConnection::CloseAllPhysicalConnections() {
+  for (const auto &physical_connection : physical_connections_) {
+    auto *established_connection = physical_connection.second.get();
+    if (!established_connection->has_disconnection_reason()) {
+      FinishPhysicalConnection(established_connection, UNFINISHED);
+    }
+  }
+  current_medium_ = UNKNOWN_MEDIUM;
+}
+
+std::vector<ConnectionsLog::EstablishedConnection>
+AnalyticsRecorder::LogicalConnection::GetEstablisedConnections() {
+  std::vector<ConnectionsLog::EstablishedConnection> established_connections;
+  if (current_medium_ != UNKNOWN_MEDIUM) {
+    NEARBY_LOGS(WARNING)
+        << "AnalyticsRecorder expected no more active physical connections "
+           "before logging this endpoint connection.";
+    return established_connections;
+  }
+  std::transform(physical_connections_.begin(), physical_connections_.end(),
+                 std::back_inserter(established_connections),
+                 [](auto &kv) { return *kv.second; });
+  physical_connections_.clear();
+  return established_connections;
+}
+
+void AnalyticsRecorder::LogicalConnection::IncomingPayloadStarted(
+    std::int64_t payload_id, PayloadType type, std::int64_t total_size_bytes) {
+  incoming_payloads_.insert(
+      {payload_id, absl::make_unique<PendingPayload>(type, total_size_bytes)});
+}
+
+void AnalyticsRecorder::LogicalConnection::ChunkReceived(
+    std::int64_t payload_id, std::int64_t size_bytes) {
+  auto it = incoming_payloads_.find(payload_id);
+  if (it == incoming_payloads_.end()) {
+    return;
+  }
+  auto *pending_payload = it->second.get();
+  pending_payload->AddChunk(size_bytes);
+}
+
+void AnalyticsRecorder::LogicalConnection::IncomingPayloadDone(
+    std::int64_t payload_id, PayloadStatus status) {
+  if (current_medium_ == UNKNOWN_MEDIUM) {
+    NEARBY_LOGS(WARNING) << "Unexpected call to incomingPayloadDone() while "
+                            "AnalyticsRecorder has no active current medium.";
+    return;
+  }
+  auto it = physical_connections_.find(current_medium_);
+  if (it != physical_connections_.end()) {
+    const auto &established_connection = it->second;
+    auto it = incoming_payloads_.find(payload_id);
+    if (it != incoming_payloads_.end()) {
+      *established_connection->add_received_payload() =
+          it->second->GetProtoPayload(status);
+      incoming_payloads_.erase(it);
+    }
+  }
+}
+
+void AnalyticsRecorder::LogicalConnection::OutgoingPayloadStarted(
+    std::int64_t payload_id, PayloadType type, std::int64_t total_size_bytes) {
+  outgoing_payloads_.insert(
+      {payload_id, absl::make_unique<PendingPayload>(type, total_size_bytes)});
+}
+
+void AnalyticsRecorder::LogicalConnection::ChunkSent(std::int64_t payload_id,
+                                                     std::int64_t size_bytes) {
+  auto it = outgoing_payloads_.find(payload_id);
+  if (it == outgoing_payloads_.end()) {
+    return;
+  }
+  auto *payload = it->second.get();
+  payload->AddChunk(size_bytes);
+}
+
+void AnalyticsRecorder::LogicalConnection::OutgoingPayloadDone(
+    std::int64_t payload_id, PayloadStatus status) {
+  if (current_medium_ == UNKNOWN_MEDIUM) {
+    NEARBY_LOGS(WARNING) << "Unexpected call to outgoingPayloadDone() while "
+                            "AnalyticsRecorder has no active current medium.";
+    return;
+  }
+  auto it = physical_connections_.find(current_medium_);
+  if (it != physical_connections_.end()) {
+    const auto &established_connection = it->second;
+    auto it = outgoing_payloads_.find(payload_id);
+    if (it != outgoing_payloads_.end()) {
+      *established_connection->add_sent_payload() =
+          it->second->GetProtoPayload(status);
+      outgoing_payloads_.erase(it);
+    }
+  }
+}
+
+void AnalyticsRecorder::LogicalConnection::FinishPhysicalConnection(
+    ConnectionsLog::EstablishedConnection *established_connection,
+    DisconnectionReason reason) {
+  established_connection->set_disconnection_reason(reason);
+  established_connection->set_duration_millis(
+      absl::ToUnixMillis(SystemClock::ElapsedRealtime()) -
+      established_connection->duration_millis());
+
+  // Add any not-yet-finished payloads to this EstablishedConnection.
+  std::vector<ConnectionsLog::Payload> in_payloads =
+      ResolvePendingPayloads(incoming_payloads_, reason);
+  absl::c_move(in_payloads,
+               RepeatedFieldBackInserter(
+                   established_connection->mutable_received_payload()));
+  std::vector<ConnectionsLog::Payload> out_payloads =
+      ResolvePendingPayloads(outgoing_payloads_, reason);
+  absl::c_move(out_payloads,
+               RepeatedFieldBackInserter(
+                   established_connection->mutable_sent_payload()));
+}
+
+std::vector<ConnectionsLog::Payload>
+AnalyticsRecorder::LogicalConnection::ResolvePendingPayloads(
+    absl::btree_map<std::int64_t, std::unique_ptr<PendingPayload>>
+        &pending_payloads,
+    DisconnectionReason reason) {
+  std::vector<ConnectionsLog::Payload> completed_payloads;
+  absl::btree_map<std::int64_t, std::unique_ptr<PendingPayload>>
+      upgraded_payloads;
+  PayloadStatus status =
+      reason == UPGRADED ? MOVED_TO_NEW_MEDIUM : CONNECTION_CLOSED;
+  for (const auto &item : pending_payloads) {
+    const auto &pending_payload = item.second;
+    ConnectionsLog::Payload proto_payload =
+        pending_payload->GetProtoPayload(status);
+    completed_payloads.push_back(proto_payload);
+    if (reason == UPGRADED) {
+      upgraded_payloads.insert(
+          {item.first,
+           absl::make_unique<PendingPayload>(
+               pending_payload->type(), pending_payload->total_size_bytes())});
+    }
+  }
+  pending_payloads.clear();
+
+  if (reason == UPGRADED) {
+    // Re-populate the map with a new PendingPayload for each pending payload,
+    // since we expect them to be completed on the next EstablishedConnection.
+    pending_payloads = std::move(upgraded_payloads);
+  }
+  // Return the list of completed payloads to be added to the current
+  // EstablishedConnection.
+  return completed_payloads;
+}
+
 }  // namespace analytics
 }  // namespace nearby
 }  // namespace location

cpp/analytics/analytics_recorder.h

@@ -20,6 +20,7 @@
 #include "absl/container/btree_map.h"
 #include "absl/time/time.h"
 #include "core/event_logger.h"
+#include "core/payload.h"
 #include "core/strategy.h"
 #include "platform/public/mutex.h"
 #include "platform/public/single_thread_executor.h"
@@ -40,17 +41,17 @@ class AnalyticsRecorder {
   // Advertising phase
   void OnStartAdvertising(
       connections::Strategy strategy,
-      const std::vector<::location::nearby::proto::connections::Medium>
-          &mediums) ABSL_LOCKS_EXCLUDED(mutex_);
+      const std::vector<location::nearby::proto::connections::Medium> &mediums)
+      ABSL_LOCKS_EXCLUDED(mutex_);
   void OnStopAdvertising() ABSL_LOCKS_EXCLUDED(mutex_);
 
   // Discovery phase
   void OnStartDiscovery(
       connections::Strategy strategy,
-      const std::vector<::location::nearby::proto::connections::Medium>
-          &mediums) ABSL_LOCKS_EXCLUDED(mutex_);
+      const std::vector<location::nearby::proto::connections::Medium> &mediums)
+      ABSL_LOCKS_EXCLUDED(mutex_);
   void OnStopDiscovery() ABSL_LOCKS_EXCLUDED(mutex_);
-  void OnEndpointFound(::location::nearby::proto::connections::Medium medium)
+  void OnEndpointFound(location::nearby::proto::connections::Medium medium)
       ABSL_LOCKS_EXCLUDED(mutex_);
 
   // Connection request
@@ -69,16 +70,16 @@ class AnalyticsRecorder {
 
   // Connection attempt
   void OnIncomingConnectionAttempt(
-      ::location::nearby::proto::connections::ConnectionAttemptType type,
-      ::location::nearby::proto::connections::Medium medium,
-      ::location::nearby::proto::connections::ConnectionAttemptResult result,
+      location::nearby::proto::connections::ConnectionAttemptType type,
+      location::nearby::proto::connections::Medium medium,
+      location::nearby::proto::connections::ConnectionAttemptResult result,
       absl::Duration duration, const std::string &connection_token)
       ABSL_LOCKS_EXCLUDED(mutex_);
   void OnOutgoingConnectionAttempt(
       const std::string &remote_endpoint_id,
-      ::location::nearby::proto::connections::ConnectionAttemptType type,
-      ::location::nearby::proto::connections::Medium medium,
-      ::location::nearby::proto::connections::ConnectionAttemptResult result,
+      location::nearby::proto::connections::ConnectionAttemptType type,
+      location::nearby::proto::connections::Medium medium,
+      location::nearby::proto::connections::ConnectionAttemptResult result,
       absl::Duration duration, const std::string &connection_token)
       ABSL_LOCKS_EXCLUDED(mutex_);
 
@@ -88,6 +89,103 @@ class AnalyticsRecorder {
   void LogSession() ABSL_LOCKS_EXCLUDED(mutex_);
 
  private:
+  // Tracks the chunks and duration of a Payload on a particular medium.
+  class PendingPayload {
+   public:
+    PendingPayload(location::nearby::proto::connections::PayloadType type,
+                   std::int64_t total_size_bytes)
+        : start_time_(SystemClock::ElapsedRealtime()),
+          type_(type),
+          total_size_bytes_(total_size_bytes),
+          num_bytes_transferred_(0),
+          num_chunks_(0) {}
+    ~PendingPayload() = default;
+
+    void AddChunk(std::int64_t chunk_size_bytes);
+
+    proto::ConnectionsLog::Payload GetProtoPayload(
+        location::nearby::proto::connections::PayloadStatus status);
+
+    location::nearby::proto::connections::PayloadType type() const {
+      return type_;
+    }
+
+    std::int64_t total_size_bytes() const { return total_size_bytes_; }
+
+   private:
+    absl::Time start_time_;
+    location::nearby::proto::connections::PayloadType type_;
+    std::int64_t total_size_bytes_;
+    std::int64_t num_bytes_transferred_;
+    int num_chunks_;
+  };
+
+  class LogicalConnection {
+   public:
+    LogicalConnection(
+        location::nearby::proto::connections::Medium initial_medium,
+        const std::string &connection_token) {
+      PhysicalConnectionEstablished(initial_medium, connection_token);
+    }
+    LogicalConnection(const LogicalConnection &) = delete;
+    LogicalConnection(LogicalConnection &&other)
+        : current_medium_(std::move(other.current_medium_)),
+          physical_connections_{std::move(other.physical_connections_)},
+          incoming_payloads_{std::move(other.incoming_payloads_)},
+          outgoing_payloads_{std::move(other.outgoing_payloads_)} {}
+    LogicalConnection &operator=(const LogicalConnection &) = delete;
+    LogicalConnection &&operator=(LogicalConnection &&) = delete;
+    ~LogicalConnection() = default;
+
+    void PhysicalConnectionEstablished(
+        location::nearby::proto::connections::Medium medium,
+        const std::string &connection_token);
+    void PhysicalConnectionClosed(
+        location::nearby::proto::connections::Medium medium,
+        location::nearby::proto::connections::DisconnectionReason reason);
+    void CloseAllPhysicalConnections();
+
+    void IncomingPayloadStarted(
+        std::int64_t payload_id,
+        location::nearby::proto::connections::PayloadType type,
+        std::int64_t total_size_bytes);
+    void ChunkReceived(std::int64_t payload_id, std::int64_t size_bytes);
+    void IncomingPayloadDone(
+        std::int64_t payload_id,
+        location::nearby::proto::connections::PayloadStatus status);
+    void OutgoingPayloadStarted(
+        std::int64_t payload_id,
+        location::nearby::proto::connections::PayloadType type,
+        std::int64_t total_size_bytes);
+    void ChunkSent(std::int64_t payload_id, std::int64_t size_bytes);
+    void OutgoingPayloadDone(
+        std::int64_t payload_id,
+        location::nearby::proto::connections::PayloadStatus status);
+
+    std::vector<proto::ConnectionsLog::EstablishedConnection>
+    GetEstablisedConnections();
+
+   private:
+    void FinishPhysicalConnection(
+        proto::ConnectionsLog::EstablishedConnection *established_connection,
+        location::nearby::proto::connections::DisconnectionReason reason);
+    std::vector<proto::ConnectionsLog::Payload> ResolvePendingPayloads(
+        absl::btree_map<std::int64_t, std::unique_ptr<PendingPayload>>
+            &pending_payloads,
+        location::nearby::proto::connections::DisconnectionReason reason);
+
+    location::nearby::proto::connections::Medium current_medium_ =
+        location::nearby::proto::connections::UNKNOWN_MEDIUM;
+    absl::btree_map<
+        location::nearby::proto::connections::Medium,
+        std::unique_ptr<proto::ConnectionsLog::EstablishedConnection>>
+        physical_connections_;
+    absl::btree_map<std::int64_t, std::unique_ptr<PendingPayload>>
+        incoming_payloads_;
+    absl::btree_map<std::int64_t, std::unique_ptr<PendingPayload>>
+        outgoing_payloads_;
+  };
+
   bool CanRecordAnalyticsLocked(const std::string &method_name)
       ABSL_SHARED_LOCKS_REQUIRED(mutex_);
 
@@ -95,11 +193,11 @@ class AnalyticsRecorder {
   // ClientSession sub-proto.
   void LogClientSession();
   // Callbacks the ConnectionsLog proto byte array data to the EventLogger.
-  void LogEvent(::location::nearby::proto::connections::EventType event_type);
+  void LogEvent(location::nearby::proto::connections::EventType event_type);
 
   void UpdateStrategySessionLocked(
       connections::Strategy strategy,
-      ::location::nearby::proto::connections::SessionRole role)
+      location::nearby::proto::connections::SessionRole role)
       ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
   void RecordAdvertisingPhaseDurationLocked() const
       ABSL_SHARED_LOCKS_REQUIRED(mutex_);
@@ -108,28 +206,28 @@ class AnalyticsRecorder {
       ABSL_SHARED_LOCKS_REQUIRED(mutex_);
   void FinishDiscoveryPhaseLocked() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
   bool UpdateAdvertiserConnectionRequestLocked(
-      ::location::nearby::analytics::proto::ConnectionsLog::ConnectionRequest
-          *request) ABSL_SHARED_LOCKS_REQUIRED(mutex_);
+      proto::ConnectionsLog::ConnectionRequest *request)
+      ABSL_SHARED_LOCKS_REQUIRED(mutex_);
   bool UpdateDiscovererConnectionRequestLocked(
-      ::location::nearby::analytics::proto::ConnectionsLog::ConnectionRequest
-          *request) ABSL_SHARED_LOCKS_REQUIRED(mutex_);
+      proto::ConnectionsLog::ConnectionRequest *request)
+      ABSL_SHARED_LOCKS_REQUIRED(mutex_);
   bool BothEndpointsRespondedLocked(
-      ::location::nearby::analytics::proto::ConnectionsLog::ConnectionRequest
-          *request) ABSL_SHARED_LOCKS_REQUIRED(mutex_);
+      proto::ConnectionsLog::ConnectionRequest *request)
+      ABSL_SHARED_LOCKS_REQUIRED(mutex_);
   void LocalEndpointRespondedLocked(
       const std::string &remote_endpoint_id,
-      ::location::nearby::proto::connections::ConnectionRequestResponse
-          response) ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+      location::nearby::proto::connections::ConnectionRequestResponse response)
+      ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
   void RemoteEndpointRespondedLocked(
       const std::string &remote_endpoint_id,
-      ::location::nearby::proto::connections::ConnectionRequestResponse
-          response) ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+      location::nearby::proto::connections::ConnectionRequestResponse response)
+      ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
   void MarkConnectionRequestIgnoredLocked(
-      ::location::nearby::analytics::proto::ConnectionsLog::ConnectionRequest
-          *request) ABSL_SHARED_LOCKS_REQUIRED(mutex_);
+      proto::ConnectionsLog::ConnectionRequest *request)
+      ABSL_SHARED_LOCKS_REQUIRED(mutex_);
   void FinishStrategySessionLocked() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
 
-  ::location::nearby::proto::connections::ConnectionsStrategy
+  location::nearby::proto::connections::ConnectionsStrategy
   StrategyToConnectionStrategy(connections::Strategy strategy);
 
   // Not owned by AnalyticsRecorder. Pointer must refer to a valid object
@@ -141,40 +239,33 @@ class AnalyticsRecorder {
   Mutex mutex_;
 
   // ClientSession
-  std::unique_ptr<
-      ::location::nearby::analytics::proto::ConnectionsLog::ClientSession>
-      client_session_ = std::make_unique<::location::nearby::analytics::proto::
-                                             ConnectionsLog::ClientSession>();
+  std::unique_ptr<proto::ConnectionsLog::ClientSession> client_session_ =
+      std::make_unique<proto::ConnectionsLog::ClientSession>();
   absl::Time started_client_session_time_;
   bool session_was_logged_ ABSL_GUARDED_BY(mutex_) = false;
 
   // Current StrategySession
   connections::Strategy current_strategy_ ABSL_GUARDED_BY(mutex_) =
       connections::Strategy::kNone;
-  std::unique_ptr<
-      ::location::nearby::analytics::proto::ConnectionsLog::StrategySession>
+  std::unique_ptr<proto::ConnectionsLog::StrategySession>
       current_strategy_session_ ABSL_GUARDED_BY(mutex_);
   absl::Time started_strategy_session_time_ ABSL_GUARDED_BY(mutex_);
 
   // Current AdvertisingPhase
-  std::unique_ptr<
-      ::location::nearby::analytics::proto::ConnectionsLog::AdvertisingPhase>
+  std::unique_ptr<proto::ConnectionsLog::AdvertisingPhase>
       current_advertising_phase_;
   absl::Time started_advertising_phase_time_;
 
   // Current DiscoveryPhase
-  std::unique_ptr<
-      ::location::nearby::analytics::proto::ConnectionsLog::DiscoveryPhase>
+  std::unique_ptr<proto::ConnectionsLog::DiscoveryPhase>
       current_discovery_phase_;
   absl::Time started_discovery_phase_time_;
 
   absl::btree_map<std::string,
-                  std::unique_ptr<::location::nearby::analytics::proto::
-                                      ConnectionsLog::ConnectionRequest>>
+                  std::unique_ptr<proto::ConnectionsLog::ConnectionRequest>>
       incoming_connection_requests_ ABSL_GUARDED_BY(mutex_);
   absl::btree_map<std::string,
-                  std::unique_ptr<::location::nearby::analytics::proto::
-                                      ConnectionsLog::ConnectionRequest>>
+                  std::unique_ptr<proto::ConnectionsLog::ConnectionRequest>>
       outgoing_connection_requests_ ABSL_GUARDED_BY(mutex_);
 };
 

cpp/analytics/analytics_recorder_test.cc

@@ -45,20 +45,6 @@ using ::testing::Contains;
 using ::testing::EqualsProto;
 using ::testing::proto::Partially;
 
-using ::location::nearby::analytics::proto::ConnectionsLog;
-using ClientSession =
-    ::location::nearby::analytics::proto::ConnectionsLog::ClientSession;
-using StrategySession =
-    ::location::nearby::analytics::proto::ConnectionsLog::StrategySession;
-using AdvertisingPhase =
-    ::location::nearby::analytics::proto::ConnectionsLog::AdvertisingPhase;
-using DiscoveryPhase =
-    ::location::nearby::analytics::proto::ConnectionsLog::DiscoveryPhase;
-using DiscoveredEndpoint =
-    ::location::nearby::analytics::proto::ConnectionsLog::DiscoveredEndpoint;
-using ConnectionRequest =
-    ::location::nearby::analytics::proto::ConnectionsLog::ConnectionRequest;
-
 constexpr absl::Duration kDefaultTimeout = absl::Milliseconds(1000);
 
 class FakeEventLogger : public EventLogger {
@@ -66,7 +52,7 @@ class FakeEventLogger : public EventLogger {
   explicit FakeEventLogger(CountDownLatch& client_session_done_latch)
       : client_session_done_latch_(client_session_done_latch) {}
 
-  void Log(const ConnectionsLog& connections_log) override {
+  void Log(const proto::ConnectionsLog& connections_log) override {
     EventType event_type = connections_log.event_type();
     logged_event_types_.push_back(event_type);
     if (event_type == CLIENT_SESSION) {
@@ -82,7 +68,7 @@ class FakeEventLogger : public EventLogger {
     return logged_client_session_count_;
   }
 
-  const ClientSession& GetLoggedClientSession() {
+  const proto::ConnectionsLog::ClientSession& GetLoggedClientSession() {
     return logged_client_session_;
   }
 
@@ -91,7 +77,7 @@ class FakeEventLogger : public EventLogger {
  private:
   int logged_client_session_count_ = 0;
   CountDownLatch& client_session_done_latch_;
-  ClientSession logged_client_session_;
+  proto::ConnectionsLog::ClientSession logged_client_session_;
   std::vector<EventType> logged_event_types_;
 };
 
@@ -125,18 +111,13 @@ TEST(AnalyticsRecorderTest, SetFieldsCorrectlyForNestedAdvertisingCalls) {
   analytics_recorder.LogSession();
   ASSERT_TRUE(client_session_done_latch.Await(kDefaultTimeout).result());
 
-  constexpr char kExpected[] =
-      R"pb(
-    strategy_session <
-      strategy: P2P_STAR
-      role: ADVERTISER
-      advertising_phase < medium: BLE medium: BLUETOOTH >
-      advertising_phase < medium: BLUETOOTH >
-    >)pb";
-
-  const ClientSession& client_session = event_logger.GetLoggedClientSession();
-
-  EXPECT_THAT(client_session, Partially(EqualsProto(kExpected)));
+  EXPECT_THAT(event_logger.GetLoggedClientSession(), Partially(EqualsProto(R"pb(
+                strategy_session <
+                  strategy: P2P_STAR
+                  role: ADVERTISER
+                  advertising_phase < medium: BLE medium: BLUETOOTH >
+                  advertising_phase < medium: BLUETOOTH >
+                >)pb")));
 }
 
 TEST(AnalyticsRecorderTest, SetFieldsCorrectlyForNestedDiscoveryCalls) {
@@ -476,7 +457,7 @@ TEST(AnalyticsRecorderTest,
 
   CountDownLatch client_session_done_latch(1);
   FakeEventLogger event_logger(client_session_done_latch);
-  AnalyticsRecorder analytics_recorder{&event_logger};
+  AnalyticsRecorder analytics_recorder(&event_logger);
 
   analytics_recorder.OnStartDiscovery(strategy, mediums);
   analytics_recorder.OnConnectionRequestSent(endpoint_id);