mirror of
https://github.com/kidfromjupiter/nearby.git
synced 2026-09-14 14:46:12 -04:00
Move OutgoingTargetsManager to separate files.
PiperOrigin-RevId: 811122178
This commit is contained in:
Francis Tsui
committed by
Copybara-Service
Copybara-Service
parent
3786a3784f
commit
eb3ae063f5
@@ -260,6 +260,33 @@ cc_library(
|
||||
],
|
||||
)
|
||||
|
||||
cc_library(
|
||||
name = "outgoing_targets_manager",
|
||||
srcs = ["outgoing_targets_manager.cc"],
|
||||
hdrs = ["outgoing_targets_manager.h"],
|
||||
deps = [
|
||||
":share_session",
|
||||
":thread_timer",
|
||||
":transfer_metadata",
|
||||
":types",
|
||||
"//internal/flags:nearby_flags",
|
||||
"//internal/platform:types",
|
||||
"//proto:sharing_enums_cc_proto",
|
||||
"//sharing/analytics",
|
||||
"//sharing/certificates",
|
||||
"//sharing/flags/generated:generated_flags",
|
||||
"//sharing/internal/public:logging",
|
||||
"//sharing/proto:enums_cc_proto",
|
||||
"//sharing/proto:share_cc_proto",
|
||||
"//sharing/proto:wire_format_cc_proto",
|
||||
"@com_google_absl//absl/container:flat_hash_map",
|
||||
"@com_google_absl//absl/functional:any_invocable",
|
||||
"@com_google_absl//absl/strings",
|
||||
"@com_google_absl//absl/strings:string_view",
|
||||
"@com_google_absl//absl/time",
|
||||
],
|
||||
)
|
||||
|
||||
cc_library(
|
||||
name = "nearby_sharing_service",
|
||||
srcs = [
|
||||
@@ -314,6 +341,7 @@ cc_library(
|
||||
":incoming_frame_reader",
|
||||
":nearby_connection_impl",
|
||||
":nearby_sharing_decoder",
|
||||
":outgoing_targets_manager",
|
||||
":paired_key_verification_runner",
|
||||
":share_session",
|
||||
":thread_timer",
|
||||
|
||||
@@ -37,7 +37,6 @@
|
||||
#include "absl/random/random.h"
|
||||
#include "absl/status/statusor.h"
|
||||
#include "absl/strings/escaping.h"
|
||||
#include "absl/strings/str_cat.h"
|
||||
#include "absl/strings/str_format.h"
|
||||
#include "absl/strings/string_view.h"
|
||||
#include "absl/time/time.h"
|
||||
@@ -3036,204 +3035,6 @@ IncomingShareSession& NearbySharingServiceImpl::CreateIncomingShareSession(
|
||||
return it->second;
|
||||
}
|
||||
|
||||
void OutgoingTargetsManager::DeduplicateInOutgoingShareTarget(
|
||||
const ShareTarget& share_target, absl::string_view endpoint_id,
|
||||
std::optional<NearbyShareDecryptedPublicCertificate> certificate) {
|
||||
// TODO(b/343764269): may need to update last_outgoing_metadata_ if the
|
||||
// deduped target id matches the one in last_outgoing_metadata_.
|
||||
// But since we do not modify the share target of a connected session, it may
|
||||
// not happen.
|
||||
|
||||
auto session_it = outgoing_share_session_map_.find(share_target.id);
|
||||
if (session_it == outgoing_share_session_map_.end()) {
|
||||
LOG(WARNING) << __func__ << ": share_target.id=" << share_target.id
|
||||
<< " not found in outgoing share session map.";
|
||||
return;
|
||||
}
|
||||
if (session_it->second.IsConnected()) {
|
||||
LOG(INFO) << __func__ << ": share_target.id=" << share_target.id
|
||||
<< " is connected, not updating outgoing_share_session_map_.";
|
||||
return;
|
||||
}
|
||||
session_it->second.UpdateSessionForDedup(share_target, std::move(certificate),
|
||||
endpoint_id);
|
||||
|
||||
share_target_updated_callback_(share_target);
|
||||
|
||||
LOG(INFO) << __func__
|
||||
<< ": [Dedupped] NotifyShareTargetUpdated to all surfaces "
|
||||
"for share_target: "
|
||||
<< share_target.ToString();
|
||||
}
|
||||
|
||||
bool OutgoingTargetsManager::FindDuplicateInDiscoveryCache(
|
||||
absl::string_view endpoint_id, ShareTarget& share_target) {
|
||||
auto it = discovery_cache_.find(endpoint_id);
|
||||
if (it != discovery_cache_.end()) {
|
||||
// If endpoint info changes for an endpoint ID, NC will send a rediscovery
|
||||
// event for the same endpoint id.
|
||||
LOG(INFO) << __func__
|
||||
<< ": [Dedupped] Found duplicate endpoint_id: " << endpoint_id
|
||||
<< ", share_target.id changed from: " << share_target.id << " to "
|
||||
<< it->second.share_target.id;
|
||||
share_target.id = it->second.share_target.id;
|
||||
discovery_cache_.erase(it);
|
||||
return true;
|
||||
}
|
||||
|
||||
for (auto it = discovery_cache_.begin(); it != discovery_cache_.end(); ++it) {
|
||||
if (it->second.share_target.device_id == share_target.device_id) {
|
||||
LOG(INFO) << __func__
|
||||
<< ": [Dedupped] Found duplicate device_id, share_target.id "
|
||||
"changed from: "
|
||||
<< share_target.id << " to " << it->second.share_target.id
|
||||
<< ". New endpoint_id: " << endpoint_id;
|
||||
// Share targets in discovery cache have receive_disabled set to true.
|
||||
// Copy only the id field from cache entry,
|
||||
share_target.id = it->second.share_target.id;
|
||||
discovery_cache_.erase(it);
|
||||
return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
bool OutgoingTargetsManager::FindDuplicateInOutgoingShareTargets(
|
||||
absl::string_view endpoint_id, ShareTarget& share_target) {
|
||||
// If the duplicate is found, share_target.id needs to be updated to the old
|
||||
// "discovered" share_target_id so NotifyShareTargetUpdated matches a target
|
||||
// that was discovered before.
|
||||
|
||||
auto it = outgoing_share_target_map_.find(endpoint_id);
|
||||
if (it != outgoing_share_target_map_.end()) {
|
||||
// If endpoint info changes for an endpoint ID, NC will send a rediscovery
|
||||
// event for the same endpoint id.
|
||||
LOG(INFO) << __func__
|
||||
<< ": [Dedupped] Found duplicate endpoint_id: " << endpoint_id
|
||||
<< " in outgoing_share_target_map, share_target.id changed from: "
|
||||
<< share_target.id << " to " << it->second.id;
|
||||
share_target.id = it->second.id;
|
||||
it->second = share_target;
|
||||
return true;
|
||||
}
|
||||
|
||||
for (auto it = outgoing_share_target_map_.begin();
|
||||
it != outgoing_share_target_map_.end(); ++it) {
|
||||
if (it->second.device_id == share_target.device_id) {
|
||||
LOG(INFO)
|
||||
<< __func__
|
||||
<< ": [Dedupped] Found duplicate device_id, endpoint ID "
|
||||
"changed from: "
|
||||
<< it->first << " to " << endpoint_id
|
||||
<< " in outgoing_share_target_map, share_target.id changed from: "
|
||||
<< share_target.id << " to " << it->second.id;
|
||||
share_target.id = it->second.id;
|
||||
outgoing_share_target_map_.erase(it);
|
||||
outgoing_share_target_map_.insert_or_assign(endpoint_id, share_target);
|
||||
return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
std::optional<ShareTarget>
|
||||
OutgoingTargetsManager::RemoveOutgoingShareTargetWithEndpointId(
|
||||
absl::string_view endpoint_id) {
|
||||
VLOG(1) << __func__ << ":Outgoing connection to " << endpoint_id
|
||||
<< " disconnected";
|
||||
auto target_node = outgoing_share_target_map_.extract(endpoint_id);
|
||||
if (target_node.empty()) {
|
||||
LOG(WARNING) << __func__ << ": endpoint_id=" << endpoint_id
|
||||
<< " not found in outgoing share target map.";
|
||||
return std::nullopt;
|
||||
}
|
||||
ShareTarget& share_target = target_node.mapped();
|
||||
VLOG(1) << __func__ << ": Removing (endpoint_id=" << endpoint_id
|
||||
<< ", share_target.id=" << target_node.mapped().id
|
||||
<< ") from outgoing share target map";
|
||||
|
||||
// Do not destroy the session until it has been removed from the map.
|
||||
// Session destruction can trigger callbacks that traverses the map and it
|
||||
// cannot access the map while it is being modified.
|
||||
auto session_node =
|
||||
outgoing_share_session_map_.extract(target_node.mapped().id);
|
||||
if (!session_node.empty()) {
|
||||
session_node.mapped().OnDisconnect();
|
||||
} else {
|
||||
LOG(WARNING) << __func__ << ": share_target.id=" << target_node.mapped().id
|
||||
<< " not found in outgoing share session map.";
|
||||
}
|
||||
return share_target;
|
||||
}
|
||||
|
||||
// Pass endpoint_id by value here since we remove entries from the
|
||||
// outgoing_share_target_map_ in this function, and some callers like
|
||||
// DisableAllOutgoingShareTargets pass the map item key as the endpoint_id.
|
||||
// This prevents the endpoint_id from being invalidated in this function.
|
||||
void OutgoingTargetsManager::MoveToDiscoveryCache(std::string endpoint_id,
|
||||
uint64_t expiry_ms) {
|
||||
std::optional<ShareTarget> share_target_opt =
|
||||
RemoveOutgoingShareTargetWithEndpointId(endpoint_id);
|
||||
if (!share_target_opt.has_value()) {
|
||||
return;
|
||||
}
|
||||
DiscoveryCacheEntry cache_entry;
|
||||
cache_entry.share_target = std::move(share_target_opt.value());
|
||||
// Entries in Discovery Cache are all receive disabled.
|
||||
cache_entry.share_target.receive_disabled = true;
|
||||
cache_entry.expiry_timer = std::make_unique<ThreadTimer>(
|
||||
service_thread_, absl::StrCat("discovery_cache_timeout_", endpoint_id),
|
||||
absl::Milliseconds(expiry_ms),
|
||||
[this, expiry_ms, endpoint_id = std::string(endpoint_id)]() {
|
||||
auto cache_node = discovery_cache_.extract(endpoint_id);
|
||||
if (cache_node.empty()) {
|
||||
LOG(WARNING) << "Trying to remove endpoint_id: " << endpoint_id
|
||||
<< " from discovery_cache, but cannot find it";
|
||||
return;
|
||||
}
|
||||
ShareTarget& share_target = cache_node.mapped().share_target;
|
||||
LOG(INFO) << ": Removing (endpoint_id=" << endpoint_id
|
||||
<< ", share_target.id=" << share_target.id
|
||||
<< ") from discovery_cache after " << expiry_ms << "ms";
|
||||
|
||||
share_target_lost_callback_(share_target);
|
||||
|
||||
VLOG(1) << "discovery_cache entry: " << endpoint_id << " timeout after "
|
||||
<< expiry_ms << "ms"
|
||||
<< ": [Dedupped] NotifyShareTargetLost to all surfaces for "
|
||||
<< "share_target: " << share_target.ToString();
|
||||
});
|
||||
// Send ShareTarget update to set receive disabled to true.
|
||||
share_target_updated_callback_(cache_entry.share_target);
|
||||
auto [it, inserted] =
|
||||
discovery_cache_.insert_or_assign(endpoint_id, std::move(cache_entry));
|
||||
LOG(INFO) << "[Dedupped] added to discovery_cache: " << endpoint_id << " by "
|
||||
<< (inserted ? "insert" : "assign");
|
||||
}
|
||||
|
||||
void OutgoingTargetsManager::CreateOutgoingShareSession(
|
||||
const ShareTarget& share_target, absl::string_view endpoint_id,
|
||||
std::optional<NearbyShareDecryptedPublicCertificate> certificate,
|
||||
absl::AnyInvocable<void(OutgoingShareSession& session,
|
||||
const TransferMetadata& metadata)>
|
||||
transfer_update_callback) {
|
||||
outgoing_share_target_map_.insert_or_assign(endpoint_id, share_target);
|
||||
auto [it_out, inserted] = outgoing_share_session_map_.try_emplace(
|
||||
share_target.id, &clock_, service_thread_, &connections_manager_,
|
||||
analytics_recorder_, std::string(endpoint_id), share_target,
|
||||
std::move(transfer_update_callback));
|
||||
if (!inserted) {
|
||||
LOG(WARNING) << __func__ << ": share_target.id=" << share_target.id
|
||||
<< " already exists in outgoing share session map. This "
|
||||
"should NOT happen";
|
||||
} else {
|
||||
auto& session = it_out->second;
|
||||
if (certificate.has_value()) {
|
||||
session.set_certificate(std::move(*certificate));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
ShareSession* NearbySharingServiceImpl::GetShareSession(
|
||||
int64_t share_target_id) {
|
||||
ShareSession* result = GetIncomingShareSession(share_target_id);
|
||||
@@ -3253,16 +3054,6 @@ IncomingShareSession* NearbySharingServiceImpl::GetIncomingShareSession(
|
||||
return &it->second;
|
||||
}
|
||||
|
||||
OutgoingShareSession* OutgoingTargetsManager::GetOutgoingShareSession(
|
||||
int64_t share_target_id) {
|
||||
auto it = outgoing_share_session_map_.find(share_target_id);
|
||||
if (it == outgoing_share_session_map_.end()) {
|
||||
return nullptr;
|
||||
}
|
||||
|
||||
return &it->second;
|
||||
}
|
||||
|
||||
std::optional<std::vector<uint8_t>>
|
||||
NearbySharingServiceImpl::GetBluetoothMacAddressForShareTarget(
|
||||
OutgoingShareSession& session) {
|
||||
@@ -3277,18 +3068,6 @@ NearbySharingServiceImpl::GetBluetoothMacAddressForShareTarget(
|
||||
return GetBluetoothMacAddressFromCertificate(*certificate);
|
||||
}
|
||||
|
||||
void OutgoingTargetsManager::DisableAllOutgoingShareTargets() {
|
||||
VLOG(1) << "Move all outgoing share targets to discovery cache.";
|
||||
while (!outgoing_share_target_map_.empty()) {
|
||||
MoveToDiscoveryCache(outgoing_share_target_map_.begin()->first,
|
||||
NearbyFlags::GetInstance().GetInt64Flag(
|
||||
config_package_nearby::nearby_sharing_feature::
|
||||
kUnregisterTargetDiscoveryCacheLostExpiryMs));
|
||||
}
|
||||
DCHECK(outgoing_share_target_map_.empty());
|
||||
DCHECK(outgoing_share_session_map_.empty());
|
||||
}
|
||||
|
||||
void NearbySharingServiceImpl::UnregisterShareTarget(int64_t share_target_id) {
|
||||
LOG(INFO) << __func__ << ": Unregister share target " << share_target_id;
|
||||
|
||||
@@ -3544,37 +3323,4 @@ void NearbySharingServiceImpl::UpdateFilePathsInProgress(
|
||||
<< ": Update file paths in progress: " << update_file_paths;
|
||||
}
|
||||
|
||||
OutgoingTargetsManager::OutgoingTargetsManager(
|
||||
Clock* clock, TaskRunner* service_thread,
|
||||
NearbyConnectionsManager* connections_manager,
|
||||
analytics::AnalyticsRecorder* analytics_recorder,
|
||||
absl::AnyInvocable<void(const ShareTarget&)> share_target_updated_callback,
|
||||
absl::AnyInvocable<void(const ShareTarget&)> share_target_lost_callback)
|
||||
: clock_(*clock),
|
||||
service_thread_(*service_thread),
|
||||
connections_manager_(*connections_manager),
|
||||
analytics_recorder_(*analytics_recorder),
|
||||
share_target_updated_callback_(std::move(share_target_updated_callback)),
|
||||
share_target_lost_callback_(std::move(share_target_lost_callback)) {}
|
||||
|
||||
void OutgoingTargetsManager::Cleanup() {
|
||||
while (!outgoing_share_target_map_.empty()) {
|
||||
RemoveOutgoingShareTargetWithEndpointId(
|
||||
outgoing_share_target_map_.begin()->first);
|
||||
}
|
||||
discovery_cache_.clear();
|
||||
}
|
||||
|
||||
void OutgoingTargetsManager::ForEachShareTarget(
|
||||
absl::AnyInvocable<void(const ShareTarget&)> callback) {
|
||||
// All share targets in discovery_cache have received_disabled set to true,
|
||||
// send them to new send surface in discovered events..
|
||||
for (const auto& [endpoint_id, discovery_cache_entry] : discovery_cache_) {
|
||||
callback(discovery_cache_entry.share_target);
|
||||
}
|
||||
for (const auto& [endpoint_id, share_target] : outgoing_share_target_map_) {
|
||||
callback(share_target);
|
||||
}
|
||||
}
|
||||
|
||||
} // namespace nearby::sharing
|
||||
|
||||
@@ -63,6 +63,7 @@
|
||||
#include "sharing/nearby_sharing_service_extension.h"
|
||||
#include "sharing/nearby_sharing_settings.h"
|
||||
#include "sharing/outgoing_share_session.h"
|
||||
#include "sharing/outgoing_targets_manager.h"
|
||||
#include "sharing/paired_key_verification_runner.h"
|
||||
#include "sharing/proto/enums.pb.h"
|
||||
#include "sharing/proto/wire_format.pb.h"
|
||||
@@ -83,89 +84,6 @@ class NearbySharingServiceImplTest_CreateShareTarget_Test;
|
||||
class NearbySharingServiceImplTest_RemoveIncomingPayloads_Test;
|
||||
}; // namespace NearbySharingServiceUnitTests
|
||||
|
||||
class OutgoingTargetsManager {
|
||||
public:
|
||||
OutgoingTargetsManager(
|
||||
Clock* clock, TaskRunner* service_thread,
|
||||
NearbyConnectionsManager* connections_manager,
|
||||
analytics::AnalyticsRecorder* analytics_recorder,
|
||||
absl::AnyInvocable<void(const ShareTarget&)>
|
||||
share_target_updated_callback,
|
||||
absl::AnyInvocable<void(const ShareTarget&)> share_target_lost_callback);
|
||||
|
||||
void Cleanup();
|
||||
|
||||
OutgoingShareSession* GetOutgoingShareSession(int64_t share_target_id);
|
||||
|
||||
// Update the entry in outgoing_share_session_map_ with the new share target
|
||||
// and OnShareTargetUpdated is called.
|
||||
void DeduplicateInOutgoingShareTarget(
|
||||
const ShareTarget& share_target, absl::string_view endpoint_id,
|
||||
std::optional<NearbyShareDecryptedPublicCertificate> certificate);
|
||||
|
||||
// Looks for a duplicate of the share target in the discovery cache.
|
||||
// If found, the share target is removed from the discovery cache and its
|
||||
// id is copied into `share_target`.
|
||||
// Returns true if the duplicate is found.
|
||||
bool FindDuplicateInDiscoveryCache(absl::string_view endpoint_id,
|
||||
ShareTarget& share_target);
|
||||
|
||||
// Looks for a duplicate of the share target in the outgoing share
|
||||
// target map. The share target's id is changed to match an existing target if
|
||||
// available. Returns true if the duplicate is found.
|
||||
bool FindDuplicateInOutgoingShareTargets(absl::string_view endpoint_id,
|
||||
ShareTarget& share_target);
|
||||
|
||||
// Returns the share target if it has been removed, std::nullopt otherwise.
|
||||
std::optional<ShareTarget> RemoveOutgoingShareTargetWithEndpointId(
|
||||
absl::string_view endpoint_id);
|
||||
|
||||
// Move the endpoint to the discovery cache with the given expiry time.
|
||||
void MoveToDiscoveryCache(std::string endpoint_id, uint64_t expiry_ms);
|
||||
|
||||
// Move all outgoing share targets to the discovery cache so that they will be
|
||||
// reported as receive_disabled.
|
||||
void DisableAllOutgoingShareTargets();
|
||||
|
||||
void CreateOutgoingShareSession(
|
||||
const ShareTarget& share_target, absl::string_view endpoint_id,
|
||||
std::optional<NearbyShareDecryptedPublicCertificate> certificate,
|
||||
absl::AnyInvocable<void(OutgoingShareSession& session,
|
||||
const TransferMetadata& metadata)>
|
||||
transfer_update_callback);
|
||||
|
||||
void ForEachShareTarget(
|
||||
absl::AnyInvocable<void(const ShareTarget&)> callback);
|
||||
|
||||
private:
|
||||
// Cache a recently lost share target to be re-discovered.
|
||||
// Purged after expiry_timer.
|
||||
struct DiscoveryCacheEntry {
|
||||
// If needed, we can add "state" field to model "Tomb" state.
|
||||
std::unique_ptr<ThreadTimer> expiry_timer;
|
||||
ShareTarget share_target;
|
||||
};
|
||||
|
||||
Clock& clock_;
|
||||
TaskRunner& service_thread_;
|
||||
NearbyConnectionsManager& connections_manager_;
|
||||
analytics::AnalyticsRecorder& analytics_recorder_;
|
||||
absl::AnyInvocable<void(const ShareTarget&)> share_target_updated_callback_;
|
||||
absl::AnyInvocable<void(const ShareTarget&)> share_target_lost_callback_;
|
||||
|
||||
// A map of endpoint id to ShareTarget, where each ShareTarget entry
|
||||
// directly corresponds to a OutgoingShareSession entry in
|
||||
// outgoing_share_target_info_map_;
|
||||
absl::flat_hash_map<std::string, ShareTarget> outgoing_share_target_map_;
|
||||
// A map of ShareTarget id to OutgoingShareSession. This lets us know which
|
||||
// endpoint and public certificate are related to the outgoing share target.
|
||||
absl::flat_hash_map<int64_t, OutgoingShareSession>
|
||||
outgoing_share_session_map_;
|
||||
// A map of Endpoint id to DiscoveryCacheEntry.
|
||||
// All ShareTargets in discovery cache have received_disabled set to true.
|
||||
absl::flat_hash_map<std::string, DiscoveryCacheEntry> discovery_cache_;
|
||||
};
|
||||
|
||||
// All methods should be called from the same sequence that created the service.
|
||||
class NearbySharingServiceImpl
|
||||
: public NearbySharingService,
|
||||
|
||||
@@ -0,0 +1,301 @@
|
||||
// Copyright 2022-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 "sharing/outgoing_targets_manager.h"
|
||||
|
||||
#include <stdint.h>
|
||||
|
||||
#include <memory>
|
||||
#include <optional>
|
||||
#include <string>
|
||||
#include <utility>
|
||||
|
||||
#include "absl/functional/any_invocable.h"
|
||||
#include "absl/strings/str_cat.h"
|
||||
#include "absl/strings/string_view.h"
|
||||
#include "absl/time/time.h"
|
||||
#include "internal/flags/nearby_flags.h"
|
||||
#include "internal/platform/clock.h"
|
||||
#include "internal/platform/task_runner.h"
|
||||
#include "proto/sharing_enums.pb.h"
|
||||
#include "sharing/analytics/analytics_recorder.h"
|
||||
#include "sharing/certificates/nearby_share_decrypted_public_certificate.h"
|
||||
#include "sharing/flags/generated/nearby_sharing_feature_flags.h"
|
||||
#include "sharing/internal/public/logging.h"
|
||||
#include "sharing/nearby_connections_manager.h"
|
||||
#include "sharing/outgoing_share_session.h"
|
||||
#include "sharing/proto/encrypted_metadata.pb.h"
|
||||
#include "sharing/proto/enums.pb.h"
|
||||
#include "sharing/proto/wire_format.pb.h"
|
||||
#include "sharing/share_target.h"
|
||||
#include "sharing/thread_timer.h"
|
||||
#include "sharing/transfer_metadata.h"
|
||||
|
||||
namespace nearby::sharing {
|
||||
|
||||
OutgoingTargetsManager::OutgoingTargetsManager(
|
||||
Clock* clock, TaskRunner* service_thread,
|
||||
NearbyConnectionsManager* connections_manager,
|
||||
analytics::AnalyticsRecorder* analytics_recorder,
|
||||
absl::AnyInvocable<void(const ShareTarget&)> share_target_updated_callback,
|
||||
absl::AnyInvocable<void(const ShareTarget&)> share_target_lost_callback)
|
||||
: clock_(*clock),
|
||||
service_thread_(*service_thread),
|
||||
connections_manager_(*connections_manager),
|
||||
analytics_recorder_(*analytics_recorder),
|
||||
share_target_updated_callback_(std::move(share_target_updated_callback)),
|
||||
share_target_lost_callback_(std::move(share_target_lost_callback)) {}
|
||||
|
||||
|
||||
void OutgoingTargetsManager::DeduplicateInOutgoingShareTarget(
|
||||
const ShareTarget& share_target, absl::string_view endpoint_id,
|
||||
std::optional<NearbyShareDecryptedPublicCertificate> certificate) {
|
||||
// TODO(b/343764269): may need to update last_outgoing_metadata_ if the
|
||||
// deduped target id matches the one in last_outgoing_metadata_.
|
||||
// But since we do not modify the share target of a connected session, it may
|
||||
// not happen.
|
||||
|
||||
auto session_it = outgoing_share_session_map_.find(share_target.id);
|
||||
if (session_it == outgoing_share_session_map_.end()) {
|
||||
LOG(WARNING) << __func__ << ": share_target.id=" << share_target.id
|
||||
<< " not found in outgoing share session map.";
|
||||
return;
|
||||
}
|
||||
if (session_it->second.IsConnected()) {
|
||||
LOG(INFO) << __func__ << ": share_target.id=" << share_target.id
|
||||
<< " is connected, not updating outgoing_share_session_map_.";
|
||||
return;
|
||||
}
|
||||
session_it->second.UpdateSessionForDedup(share_target, std::move(certificate),
|
||||
endpoint_id);
|
||||
|
||||
share_target_updated_callback_(share_target);
|
||||
|
||||
LOG(INFO) << __func__
|
||||
<< ": [Dedupped] NotifyShareTargetUpdated to all surfaces "
|
||||
"for share_target: "
|
||||
<< share_target.ToString();
|
||||
}
|
||||
|
||||
bool OutgoingTargetsManager::FindDuplicateInDiscoveryCache(
|
||||
absl::string_view endpoint_id, ShareTarget& share_target) {
|
||||
auto it = discovery_cache_.find(endpoint_id);
|
||||
if (it != discovery_cache_.end()) {
|
||||
// If endpoint info changes for an endpoint ID, NC will send a rediscovery
|
||||
// event for the same endpoint id.
|
||||
LOG(INFO) << __func__
|
||||
<< ": [Dedupped] Found duplicate endpoint_id: " << endpoint_id
|
||||
<< ", share_target.id changed from: " << share_target.id << " to "
|
||||
<< it->second.share_target.id;
|
||||
share_target.id = it->second.share_target.id;
|
||||
discovery_cache_.erase(it);
|
||||
return true;
|
||||
}
|
||||
|
||||
for (auto it = discovery_cache_.begin(); it != discovery_cache_.end(); ++it) {
|
||||
if (it->second.share_target.device_id == share_target.device_id) {
|
||||
LOG(INFO) << __func__
|
||||
<< ": [Dedupped] Found duplicate device_id, share_target.id "
|
||||
"changed from: "
|
||||
<< share_target.id << " to " << it->second.share_target.id
|
||||
<< ". New endpoint_id: " << endpoint_id;
|
||||
// Share targets in discovery cache have receive_disabled set to true.
|
||||
// Copy only the id field from cache entry,
|
||||
share_target.id = it->second.share_target.id;
|
||||
discovery_cache_.erase(it);
|
||||
return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
bool OutgoingTargetsManager::FindDuplicateInOutgoingShareTargets(
|
||||
absl::string_view endpoint_id, ShareTarget& share_target) {
|
||||
// If the duplicate is found, share_target.id needs to be updated to the old
|
||||
// "discovered" share_target_id so NotifyShareTargetUpdated matches a target
|
||||
// that was discovered before.
|
||||
|
||||
auto it = outgoing_share_target_map_.find(endpoint_id);
|
||||
if (it != outgoing_share_target_map_.end()) {
|
||||
// If endpoint info changes for an endpoint ID, NC will send a rediscovery
|
||||
// event for the same endpoint id.
|
||||
LOG(INFO) << __func__
|
||||
<< ": [Dedupped] Found duplicate endpoint_id: " << endpoint_id
|
||||
<< " in outgoing_share_target_map, share_target.id changed from: "
|
||||
<< share_target.id << " to " << it->second.id;
|
||||
share_target.id = it->second.id;
|
||||
it->second = share_target;
|
||||
return true;
|
||||
}
|
||||
|
||||
for (auto it = outgoing_share_target_map_.begin();
|
||||
it != outgoing_share_target_map_.end(); ++it) {
|
||||
if (it->second.device_id == share_target.device_id) {
|
||||
LOG(INFO)
|
||||
<< __func__
|
||||
<< ": [Dedupped] Found duplicate device_id, endpoint ID "
|
||||
"changed from: "
|
||||
<< it->first << " to " << endpoint_id
|
||||
<< " in outgoing_share_target_map, share_target.id changed from: "
|
||||
<< share_target.id << " to " << it->second.id;
|
||||
share_target.id = it->second.id;
|
||||
outgoing_share_target_map_.erase(it);
|
||||
outgoing_share_target_map_.insert_or_assign(endpoint_id, share_target);
|
||||
return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
std::optional<ShareTarget>
|
||||
OutgoingTargetsManager::RemoveOutgoingShareTargetWithEndpointId(
|
||||
absl::string_view endpoint_id) {
|
||||
VLOG(1) << __func__ << ":Outgoing connection to " << endpoint_id
|
||||
<< " disconnected";
|
||||
auto target_node = outgoing_share_target_map_.extract(endpoint_id);
|
||||
if (target_node.empty()) {
|
||||
LOG(WARNING) << __func__ << ": endpoint_id=" << endpoint_id
|
||||
<< " not found in outgoing share target map.";
|
||||
return std::nullopt;
|
||||
}
|
||||
ShareTarget& share_target = target_node.mapped();
|
||||
VLOG(1) << __func__ << ": Removing (endpoint_id=" << endpoint_id
|
||||
<< ", share_target.id=" << target_node.mapped().id
|
||||
<< ") from outgoing share target map";
|
||||
|
||||
// Do not destroy the session until it has been removed from the map.
|
||||
// Session destruction can trigger callbacks that traverses the map and it
|
||||
// cannot access the map while it is being modified.
|
||||
auto session_node =
|
||||
outgoing_share_session_map_.extract(target_node.mapped().id);
|
||||
if (!session_node.empty()) {
|
||||
session_node.mapped().OnDisconnect();
|
||||
} else {
|
||||
LOG(WARNING) << __func__ << ": share_target.id=" << target_node.mapped().id
|
||||
<< " not found in outgoing share session map.";
|
||||
}
|
||||
return share_target;
|
||||
}
|
||||
|
||||
// Pass endpoint_id by value here since we remove entries from the
|
||||
// outgoing_share_target_map_ in this function, and some callers like
|
||||
// DisableAllOutgoingShareTargets pass the map item key as the endpoint_id.
|
||||
// This prevents the endpoint_id from being invalidated in this function.
|
||||
void OutgoingTargetsManager::MoveToDiscoveryCache(std::string endpoint_id,
|
||||
uint64_t expiry_ms) {
|
||||
std::optional<ShareTarget> share_target_opt =
|
||||
RemoveOutgoingShareTargetWithEndpointId(endpoint_id);
|
||||
if (!share_target_opt.has_value()) {
|
||||
return;
|
||||
}
|
||||
DiscoveryCacheEntry cache_entry;
|
||||
cache_entry.share_target = std::move(share_target_opt.value());
|
||||
// Entries in Discovery Cache are all receive disabled.
|
||||
cache_entry.share_target.receive_disabled = true;
|
||||
cache_entry.expiry_timer = std::make_unique<ThreadTimer>(
|
||||
service_thread_, absl::StrCat("discovery_cache_timeout_", endpoint_id),
|
||||
absl::Milliseconds(expiry_ms),
|
||||
[this, expiry_ms, endpoint_id = std::string(endpoint_id)]() {
|
||||
auto cache_node = discovery_cache_.extract(endpoint_id);
|
||||
if (cache_node.empty()) {
|
||||
LOG(WARNING) << "Trying to remove endpoint_id: " << endpoint_id
|
||||
<< " from discovery_cache, but cannot find it";
|
||||
return;
|
||||
}
|
||||
ShareTarget& share_target = cache_node.mapped().share_target;
|
||||
LOG(INFO) << ": Removing (endpoint_id=" << endpoint_id
|
||||
<< ", share_target.id=" << share_target.id
|
||||
<< ") from discovery_cache after " << expiry_ms << "ms";
|
||||
|
||||
share_target_lost_callback_(share_target);
|
||||
|
||||
VLOG(1) << "discovery_cache entry: " << endpoint_id << " timeout after "
|
||||
<< expiry_ms << "ms"
|
||||
<< ": [Dedupped] NotifyShareTargetLost to all surfaces for "
|
||||
<< "share_target: " << share_target.ToString();
|
||||
});
|
||||
// Send ShareTarget update to set receive disabled to true.
|
||||
share_target_updated_callback_(cache_entry.share_target);
|
||||
auto [it, inserted] =
|
||||
discovery_cache_.insert_or_assign(endpoint_id, std::move(cache_entry));
|
||||
LOG(INFO) << "[Dedupped] added to discovery_cache: " << endpoint_id << " by "
|
||||
<< (inserted ? "insert" : "assign");
|
||||
}
|
||||
|
||||
void OutgoingTargetsManager::CreateOutgoingShareSession(
|
||||
const ShareTarget& share_target, absl::string_view endpoint_id,
|
||||
std::optional<NearbyShareDecryptedPublicCertificate> certificate,
|
||||
absl::AnyInvocable<void(OutgoingShareSession& session,
|
||||
const TransferMetadata& metadata)>
|
||||
transfer_update_callback) {
|
||||
outgoing_share_target_map_.insert_or_assign(endpoint_id, share_target);
|
||||
auto [it_out, inserted] = outgoing_share_session_map_.try_emplace(
|
||||
share_target.id, &clock_, service_thread_, &connections_manager_,
|
||||
analytics_recorder_, std::string(endpoint_id), share_target,
|
||||
std::move(transfer_update_callback));
|
||||
if (!inserted) {
|
||||
LOG(WARNING) << __func__ << ": share_target.id=" << share_target.id
|
||||
<< " already exists in outgoing share session map. This "
|
||||
"should NOT happen";
|
||||
} else {
|
||||
auto& session = it_out->second;
|
||||
if (certificate.has_value()) {
|
||||
session.set_certificate(std::move(*certificate));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
OutgoingShareSession* OutgoingTargetsManager::GetOutgoingShareSession(
|
||||
int64_t share_target_id) {
|
||||
auto it = outgoing_share_session_map_.find(share_target_id);
|
||||
if (it == outgoing_share_session_map_.end()) {
|
||||
return nullptr;
|
||||
}
|
||||
|
||||
return &it->second;
|
||||
}
|
||||
|
||||
void OutgoingTargetsManager::DisableAllOutgoingShareTargets() {
|
||||
VLOG(1) << "Move all outgoing share targets to discovery cache.";
|
||||
while (!outgoing_share_target_map_.empty()) {
|
||||
MoveToDiscoveryCache(outgoing_share_target_map_.begin()->first,
|
||||
NearbyFlags::GetInstance().GetInt64Flag(
|
||||
config_package_nearby::nearby_sharing_feature::
|
||||
kUnregisterTargetDiscoveryCacheLostExpiryMs));
|
||||
}
|
||||
DCHECK(outgoing_share_target_map_.empty());
|
||||
DCHECK(outgoing_share_session_map_.empty());
|
||||
}
|
||||
|
||||
void OutgoingTargetsManager::Cleanup() {
|
||||
while (!outgoing_share_target_map_.empty()) {
|
||||
RemoveOutgoingShareTargetWithEndpointId(
|
||||
outgoing_share_target_map_.begin()->first);
|
||||
}
|
||||
discovery_cache_.clear();
|
||||
}
|
||||
|
||||
void OutgoingTargetsManager::ForEachShareTarget(
|
||||
absl::AnyInvocable<void(const ShareTarget&)> callback) {
|
||||
// All share targets in discovery_cache have received_disabled set to true,
|
||||
// send them to new send surface in discovered events..
|
||||
for (const auto& [endpoint_id, discovery_cache_entry] : discovery_cache_) {
|
||||
callback(discovery_cache_entry.share_target);
|
||||
}
|
||||
for (const auto& [endpoint_id, share_target] : outgoing_share_target_map_) {
|
||||
callback(share_target);
|
||||
}
|
||||
}
|
||||
|
||||
} // namespace nearby::sharing
|
||||
@@ -0,0 +1,134 @@
|
||||
// Copyright 2022-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_SHARING_OUTGOING_TARGETS_MANAGER_H_
|
||||
#define THIRD_PARTY_NEARBY_SHARING_OUTGOING_TARGETS_MANAGER_H_
|
||||
|
||||
#include <stddef.h>
|
||||
#include <stdint.h>
|
||||
|
||||
#include <memory>
|
||||
#include <optional>
|
||||
#include <string>
|
||||
|
||||
#include "absl/container/flat_hash_map.h"
|
||||
#include "absl/functional/any_invocable.h"
|
||||
#include "absl/strings/string_view.h"
|
||||
#include "internal/platform/clock.h"
|
||||
#include "internal/platform/task_runner.h"
|
||||
#include "proto/sharing_enums.pb.h"
|
||||
#include "sharing/analytics/analytics_recorder.h"
|
||||
#include "sharing/certificates/nearby_share_decrypted_public_certificate.h"
|
||||
#include "sharing/nearby_connections_manager.h"
|
||||
#include "sharing/outgoing_share_session.h"
|
||||
#include "sharing/proto/enums.pb.h"
|
||||
#include "sharing/proto/wire_format.pb.h"
|
||||
#include "sharing/share_target.h"
|
||||
#include "sharing/thread_timer.h"
|
||||
#include "sharing/transfer_metadata.h"
|
||||
|
||||
namespace nearby::sharing {
|
||||
class NearbyShareContactManager;
|
||||
|
||||
namespace NearbySharingServiceUnitTests {
|
||||
class NearbySharingServiceImplTest_CreateShareTarget_Test;
|
||||
class NearbySharingServiceImplTest_RemoveIncomingPayloads_Test;
|
||||
}; // namespace NearbySharingServiceUnitTests
|
||||
|
||||
class OutgoingTargetsManager {
|
||||
public:
|
||||
OutgoingTargetsManager(
|
||||
Clock* clock, TaskRunner* service_thread,
|
||||
NearbyConnectionsManager* connections_manager,
|
||||
analytics::AnalyticsRecorder* analytics_recorder,
|
||||
absl::AnyInvocable<void(const ShareTarget&)>
|
||||
share_target_updated_callback,
|
||||
absl::AnyInvocable<void(const ShareTarget&)> share_target_lost_callback);
|
||||
|
||||
void Cleanup();
|
||||
|
||||
OutgoingShareSession* GetOutgoingShareSession(int64_t share_target_id);
|
||||
|
||||
// Update the entry in outgoing_share_session_map_ with the new share target
|
||||
// and OnShareTargetUpdated is called.
|
||||
void DeduplicateInOutgoingShareTarget(
|
||||
const ShareTarget& share_target, absl::string_view endpoint_id,
|
||||
std::optional<NearbyShareDecryptedPublicCertificate> certificate);
|
||||
|
||||
// Looks for a duplicate of the share target in the discovery cache.
|
||||
// If found, the share target is removed from the discovery cache and its
|
||||
// id is copied into `share_target`.
|
||||
// Returns true if the duplicate is found.
|
||||
bool FindDuplicateInDiscoveryCache(absl::string_view endpoint_id,
|
||||
ShareTarget& share_target);
|
||||
|
||||
// Looks for a duplicate of the share target in the outgoing share
|
||||
// target map. The share target's id is changed to match an existing target if
|
||||
// available. Returns true if the duplicate is found.
|
||||
bool FindDuplicateInOutgoingShareTargets(absl::string_view endpoint_id,
|
||||
ShareTarget& share_target);
|
||||
|
||||
// Move the endpoint to the discovery cache with the given expiry time.
|
||||
void MoveToDiscoveryCache(std::string endpoint_id, uint64_t expiry_ms);
|
||||
|
||||
// Move all outgoing share targets to the discovery cache so that they will be
|
||||
// reported as receive_disabled.
|
||||
void DisableAllOutgoingShareTargets();
|
||||
|
||||
void CreateOutgoingShareSession(
|
||||
const ShareTarget& share_target, absl::string_view endpoint_id,
|
||||
std::optional<NearbyShareDecryptedPublicCertificate> certificate,
|
||||
absl::AnyInvocable<void(OutgoingShareSession& session,
|
||||
const TransferMetadata& metadata)>
|
||||
transfer_update_callback);
|
||||
|
||||
void ForEachShareTarget(
|
||||
absl::AnyInvocable<void(const ShareTarget&)> callback);
|
||||
|
||||
private:
|
||||
// Returns the share target if it has been removed, std::nullopt otherwise.
|
||||
std::optional<ShareTarget> RemoveOutgoingShareTargetWithEndpointId(
|
||||
absl::string_view endpoint_id);
|
||||
|
||||
// Cache a recently lost share target to be re-discovered.
|
||||
// Purged after expiry_timer.
|
||||
struct DiscoveryCacheEntry {
|
||||
// If needed, we can add "state" field to model "Tomb" state.
|
||||
std::unique_ptr<ThreadTimer> expiry_timer;
|
||||
ShareTarget share_target;
|
||||
};
|
||||
|
||||
Clock& clock_;
|
||||
TaskRunner& service_thread_;
|
||||
NearbyConnectionsManager& connections_manager_;
|
||||
analytics::AnalyticsRecorder& analytics_recorder_;
|
||||
absl::AnyInvocable<void(const ShareTarget&)> share_target_updated_callback_;
|
||||
absl::AnyInvocable<void(const ShareTarget&)> share_target_lost_callback_;
|
||||
|
||||
// A map of endpoint id to ShareTarget, where each ShareTarget entry
|
||||
// directly corresponds to a OutgoingShareSession entry in
|
||||
// outgoing_share_target_info_map_;
|
||||
absl::flat_hash_map<std::string, ShareTarget> outgoing_share_target_map_;
|
||||
// A map of ShareTarget id to OutgoingShareSession. This lets us know which
|
||||
// endpoint and public certificate are related to the outgoing share target.
|
||||
absl::flat_hash_map<int64_t, OutgoingShareSession>
|
||||
outgoing_share_session_map_;
|
||||
// A map of Endpoint id to DiscoveryCacheEntry.
|
||||
// All ShareTargets in discovery cache have received_disabled set to true.
|
||||
absl::flat_hash_map<std::string, DiscoveryCacheEntry> discovery_cache_;
|
||||
};
|
||||
|
||||
} // namespace nearby::sharing
|
||||
|
||||
#endif // THIRD_PARTY_NEARBY_SHARING_OUTGOING_TARGETS_MANAGER_H_
|
||||
Reference in New Issue
Block a user