// 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 "sharing/transfer_manager.h" #include #include #include #include #include "absl/strings/string_view.h" #include "absl/synchronization/mutex.h" #include "absl/time/time.h" #include "sharing/internal/public/context.h" #include "sharing/internal/public/logging.h" #include "sharing/nearby_connections_types.h" #include "sharing/thread_timer.h" namespace nearby { namespace sharing { namespace { bool IsHighQualityMedium(Medium medium) { if (medium == Medium::kWifiLan || medium == Medium::kWifiAware || medium == Medium::kWifiDirect || medium == Medium::kWifiHotspot || medium == Medium::kWebRtc) { return true; } return false; } } // namespace TransferManager::TransferManager(Context* context, absl::string_view endpoint_id) : context_(context), endpoint_id_(endpoint_id) {} TransferManager::~TransferManager() { absl::MutexLock lock(&mutex_); timeout_timer_.reset(); pending_tasks_.clear(); } void TransferManager::Send(std::function task) { absl::MutexLock lock(&mutex_); if (is_waiting_for_high_quality_medium_) { LOG(INFO) << "Connection to endpoint " << endpoint_id_ << " is waiting for a high quality medium, delaying payload transfer."; pending_tasks_.push_back(task); return; } task(); } void TransferManager::OnMediumQualityChanged(Medium current_medium) { absl::MutexLock lock(&mutex_); if (!is_waiting_for_high_quality_medium_) { LOG(WARNING) << "It is not waiting for high quality medium."; return; } if (!IsHighQualityMedium(current_medium)) { LOG(WARNING) << "medium switched to low quality Medium: " << static_cast(current_medium); return; } LOG(INFO) << "Connection to endpoint " << endpoint_id_ << " has changed to a high quality medium: " << static_cast(current_medium); StopWaitingForHighQualityMedium(); } bool TransferManager::StartTransfer() { absl::MutexLock lock(&mutex_); if (!is_waiting_for_high_quality_medium_) { VLOG(1) << "No need to wait for high quality medium."; return false; } if (timeout_timer_ != nullptr) { LOG(WARNING) << "transfer already started."; return false; } timeout_timer_ = std::make_unique( *context_->GetTaskRunner(), "transfer_manager_timeout_timer", kMediumUpgradeTimeout, [this]() { absl::MutexLock lock(&mutex_); LOG(INFO) << "Timed out for endpoint " << endpoint_id_ << " after " << kMediumUpgradeTimeout; StopWaitingForHighQualityMedium(); }); LOG(INFO) << "Attempting to upgrade the bandwidth for endpoint " + endpoint_id_ + ". Large payloads will be delayed" + " until either bandwidth is upgraded or a timeout of " << (kMediumUpgradeTimeout / absl::Milliseconds(1)) << " milliseconds is reached"; return true; } bool TransferManager::CancelTransfer() { absl::MutexLock lock(&mutex_); if (timeout_timer_ == nullptr) { LOG(WARNING) << "No running transfer."; return false; } timeout_timer_.reset(); LOG(INFO) << __func__ << "Transfer is canceled"; return true; } void TransferManager::StopWaitingForHighQualityMedium() { is_waiting_for_high_quality_medium_ = false; for (const auto& task : pending_tasks_) { LOG(INFO) << "Sending delayed payload to endpoint " << endpoint_id_; task(); } pending_tasks_.clear(); timeout_timer_.reset(); } } // namespace sharing } // namespace nearby