report_queue_impl.cc | Explore in Territory

// Copyright 2020 The Chromium Authors
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#include "components/reporting/client/report_queue_impl.h"

#include <memory>
#include <optional>
#include <queue>
#include <string>
#include <utility>

#include "base/check.h"
#include "base/check_op.h"
#include "base/functional/bind.h"
#include "base/functional/callback.h"
#include "base/location.h"
#include "base/logging.h"
#include "base/memory/ptr_util.h"
#include "base/memory/scoped_refptr.h"
#include "base/memory/weak_ptr.h"
#include "base/metrics/histogram_functions.h"
#include "base/notreached.h"
#include "base/sequence_checker.h"
#include "base/strings/strcat.h"
#include "base/strings/string_number_conversions.h"
#include "base/task/bind_post_task.h"
#include "base/task/sequenced_task_runner.h"
#include "base/task/task_traits.h"
#include "base/task/thread_pool.h"
#include "base/time/time.h"
#include "base/types/expected.h"
#include "components/reporting/client/report_queue_configuration.h"
#include "components/reporting/proto/synced/metric_data.pb.h"
#include "components/reporting/proto/synced/record.pb.h"
#include "components/reporting/proto/synced/record_constants.pb.h"
#include "components/reporting/storage/storage_module_interface.h"
#include "components/reporting/util/reporting_errors.h"
#include "components/reporting/util/status.h"
#include "components/reporting/util/statusor.h"

namespace reporting {
namespace {

// UTC time of 2122-01-01T00:00:00Z since Unix epoch 1970-01-01T00:00:00Z in
// microseconds.
constexpr int64_t kTime2122 = …;

// Returns true if record is allowed to go to `destination`. Returns false
// otherwise.
static bool RecordMayGoToDestination(const std::string& record_data,
                                     Destination destination) { … }

StatusOr<Record> ProduceRecord(std::string dm_token,
                               Destination destination,
                               int64_t reserved_space,
                               std::optional<SourceInfo> source_info,
                               ReportQueue::RecordProducer record_producer) { … }

// Calls |record_producer|, checks the result and in case of success, forwards
// it to the storage. In production code should be invoked asynchronously, on
// a thread pool (no synchronization expected).
void AddRecordToStorage(scoped_refptr<StorageModuleInterface> storage,
                        Priority priority,
                        WrappedRateLimiter::AsyncAcquireCb acquire_cb,
                        std::string dm_token,
                        Destination destination,
                        int64_t reserved_space,
                        std::optional<SourceInfo> source_info,
                        ReportQueue::RecordProducer record_producer,
                        StorageModuleInterface::EnqueueCallback callback) { … }
}  // namespace

void ReportQueueImpl::Create(
    std::unique_ptr<ReportQueueConfiguration> config,
    scoped_refptr<StorageModuleInterface> storage,
    base::OnceCallback<void(StatusOr<std::unique_ptr<ReportQueue>>)> cb) { … }

ReportQueueImpl::ReportQueueImpl(
    std::unique_ptr<ReportQueueConfiguration> config,
    scoped_refptr<StorageModuleInterface> storage)
    : … { … }

ReportQueueImpl::~ReportQueueImpl() = default;

void ReportQueueImpl::AddProducedRecord(RecordProducer record_producer,
                                        Priority priority,
                                        EnqueueCallback callback) const { … }

void ReportQueueImpl::Flush(Priority priority, FlushCallback callback) { … }

base::OnceCallback<void(StatusOr<std::unique_ptr<ReportQueue>>)>
ReportQueueImpl::PrepareToAttachActualQueue() const { … }

Destination ReportQueueImpl::GetDestination() const { … }

// Implementation of SpeculativeReportQueueImpl::PendingRecordProducer

SpeculativeReportQueueImpl::PendingRecordProducer::PendingRecordProducer(
    RecordProducer producer,
    EnqueueCallback callback,
    Priority priority)
    : … { … }

SpeculativeReportQueueImpl::PendingRecordProducer::PendingRecordProducer(
    PendingRecordProducer&& other)
    : … { … }

SpeculativeReportQueueImpl::PendingRecordProducer::~PendingRecordProducer() =
    default;

SpeculativeReportQueueImpl::PendingRecordProducer&
SpeculativeReportQueueImpl::PendingRecordProducer::operator=(
    PendingRecordProducer&& other) { … }

// static
std::unique_ptr<SpeculativeReportQueueImpl, base::OnTaskRunnerDeleter>
SpeculativeReportQueueImpl::Create(
    const SpeculativeConfigSettings& config_settings) { … }

SpeculativeReportQueueImpl::SpeculativeReportQueueImpl(
    const SpeculativeConfigSettings& config_settings,
    scoped_refptr<base::SequencedTaskRunner> sequenced_task_runner)
    : … { … }

SpeculativeReportQueueImpl::~SpeculativeReportQueueImpl() { … }

void SpeculativeReportQueueImpl::Flush(Priority priority,
                                       FlushCallback callback) { … }

void SpeculativeReportQueueImpl::AddProducedRecord(
    RecordProducer record_producer,
    Priority priority,
    EnqueueCallback callback) const { … }

void SpeculativeReportQueueImpl::MaybeEnqueueRecordProducer(
    Priority priority,
    EnqueueCallback callback,
    RecordProducer record_producer) const { … }

void SpeculativeReportQueueImpl::EnqueuePendingRecordProducers() const { … }

base::OnceCallback<void(StatusOr<std::unique_ptr<ReportQueue>>)>
SpeculativeReportQueueImpl::PrepareToAttachActualQueue() const { … }

Destination SpeculativeReportQueueImpl::GetDestination() const { … }

void SpeculativeReportQueueImpl::AttachActualQueue(
    StatusOr<std::unique_ptr<ReportQueue>> status_or_actual_queue) { … }

void SpeculativeReportQueueImpl::PurgePendingProducers(Status status) const { … }
}  // namespace reporting
chromium/components/reporting/client/report_queue_impl.cc
