blob: 1b5eb555945cc72ef19b2104b7d85666303cf982 [file] [log] [blame]
// Copyright 2021 The Chromium Authors
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
#include <memory>
#include <queue>
#include <utility>
#include "base/feature_list.h"
#include "base/functional/callback.h"
#include "base/memory/ref_counted.h"
#include "base/memory/scoped_refptr.h"
#include "base/memory/weak_ptr.h"
#include "base/task/sequenced_task_runner.h"
#include "components/reporting/client/report_queue.h"
#include "components/reporting/client/report_queue_configuration.h"
#include "components/reporting/proto/synced/record_constants.pb.h"
#include "components/reporting/storage/storage_module_interface.h"
#include "components/reporting/util/status.h"
#include "components/reporting/util/status_macros.h"
#include "components/reporting/util/statusor.h"
namespace reporting {
// ReportQueueProvider acts a single point for instantiating
// |reporting::ReportQueue|s. By performing initialization atomically it ensures
// that all ReportQueues are created with the same global settings.
// In order to utilize the ReportQueueProvider the EncryptedReportingPipeline
// feature must be turned on using --enable-features=EncryptedReportingPipeline.
// `ReportQueueProvider` must be created for a specific configuration - Chrome
// (for ChromeOS and for other OSes), other ChromeOS executables. It is then
// registered and can be accessed through static method
// `ReportQueueProvider::GetInstance()`.
// Example Usage:
// void SendMessage(google::protobuf::ImportantMessage important_message,
// reporting::ReportQueue::EnqueueCallback done_cb) {
// // Create configuration.
// StatusOr<reporting::ReportQueueConfiguration> config_result =
// reporting::ReportQueueConfiguration::Create({...}).Set...().Build();
// // Bail out if configuration failed to create.
// if (!config_result.has_value()) {
// std::move(done_cb).Run(config_result.error());
// return;
// }
// // Asynchronously create ReportingQueue.
// base::ThreadPool::PostTask(
// base::BindOnce(
// [](google::protobuf::ImportantMessage important_message,
// reporting::ReportQueue::EnqueueCallback done_cb,
// std::unique_ptr<reporting::ReportQueueConfiguration> config) {
// // Asynchronously create ReportingQueue.
// reporting::ReportQueueProvider::CreateQueue(
// std::move(config),
// base::BindOnce(
// [](std::string_view data,
// reporting::ReportQueue::EnqueueCallback
// done_cb, reporting::StatusOr<std::unique_ptr<
// reporting::ReportQueue>>
// report_queue_result) {
// // Bail out if queue failed to create.
// if (!report_queue_result.has_value()) {
// std::move(done_cb).Run(report_queue_result.error());
// return;
// }
// // Queue created successfully, enqueue the message.
// report_queue_result.value()->Enqueue(
// important_message, std::move(done_cb));
// },
// important_message, std::move(done_cb)));
// },
// important_message, std::move(done_cb),
// std::move(config_result.value())))
// }
class ReportQueueProvider {
// `ReportQueueProvider` and its descendants need to be destructed on
// sequenced task runners; to facilitate that the following flavor of smart
// pointer is declared.
template <typename T>
using SmartPtr = std::unique_ptr<T, base::OnTaskRunnerDeleter>;
using CreateReportQueueResponse = StatusOr<std::unique_ptr<ReportQueue>>;
// The response will come back utilizing the ReportQueueProvider's thread. It
// is likely that within Chromium you will want to the response to come back
// on your own thread. The simplest way to achieve that is to pass a
// base::BindPostTask rather than a base::OnceCallback. Another way to achieve
// the same result is to utilize base::OnceCallback, and capture the response
// and forward it to your own thread. We maintain base::OnceCallback here for
// use in ChromiumOS.
using CreateReportQueueCallback =
using OnStorageModuleCreatedCallback =
using StorageModuleCreateCallback =
// Callback triggered with updated report queue config after it has been
// configured with appropriate DM tokens after it is retrieved. Typically,
// this is when we go ahead and create the report queue.
using ReportQueueConfiguredCallback = base::OnceCallback<void(
ReportQueueProvider(const ReportQueueProvider& other) = delete;
ReportQueueProvider& operator=(const ReportQueueProvider& other) = delete;
virtual ~ReportQueueProvider();
// Asynchronously creates a queue based on the configuration. In the process
// `ReportQueueProvider` is expected to exist and could still be initializing
// (if initialization fails, `CreateQueue` will return with error, but next
// attempt may succeed).
// Returns with the callback handing ownership to the caller (unless there is
// an error, and then it gets the error status).
static void CreateQueue(std::unique_ptr<ReportQueueConfiguration> config,
CreateReportQueueCallback queue_cb);
// Synchronously creates a speculative queue based on the configuration.
// Returns result as a smart pointer to ReportQueue with the on-thread
// deleter.
static StatusOr<std::unique_ptr<ReportQueue, base::OnTaskRunnerDeleter>>
CreateSpeculativeQueue(std::unique_ptr<ReportQueueConfiguration> config);
// Retrieves current `ReportQueueProvider` instance (created before and
// referring to `StorageModuleInterface` and optional `Uploader`).
static ReportQueueProvider* GetInstance();
static bool IsEncryptedReportingPipelineEnabled();
// Accessors.
base::WeakPtr<ReportQueueProvider> GetWeakPtr();
scoped_refptr<StorageModuleInterface> storage() const;
scoped_refptr<base::SequencedTaskRunner> sequenced_task_runner() const;
// Storage module creator (can be substituted for testing purposes).
StorageModuleCreateCallback storage_create_cb_;
ReportQueueProvider(StorageModuleCreateCallback storage_create_cb,
scoped_refptr<base::SequencedTaskRunner> seq_task_runner);
// Holds the creation request for a ReportQueue.
class CreateReportQueueRequest;
// Finalizes provider, if the initialization process succeeded.
// May to be overridden by subclass to make more updates to the provider.
virtual void OnInitCompleted();
// Creates and initializes queue implementation. Returns status in case of
// error.
virtual void CreateNewQueue(std::unique_ptr<ReportQueueConfiguration> config,
CreateReportQueueCallback cb);
virtual StatusOr<std::unique_ptr<ReportQueue, base::OnTaskRunnerDeleter>>
const ReportQueue::SpeculativeConfigSettings& config_settings);
// Configures a given report queue config with appropriate DM tokens after its
// retrieval so it can be used for downstream processing while building a
// report queue, and triggers the corresponding completion callback with the
// updated config.
virtual void ConfigureReportQueue(
std::unique_ptr<ReportQueueConfiguration> report_queue_config,
ReportQueueConfiguredCallback completion_cb) = 0;
// Checks whether the provider has been initialized, and if so, processes all
// pending queue creation requests.
void CheckInitializationState();
// Processes storage or error returned by async call to `storage_create_cb_`.
void OnStorageModuleConfigured(
StatusOr<scoped_refptr<StorageModuleInterface>> storage_result);
// Task runner used for guarding the provider elements.
scoped_refptr<base::SequencedTaskRunner> sequenced_task_runner_;
// Queue for storing creation requests while the provider is
// initializing.
std::queue<std::unique_ptr<CreateReportQueueRequest>> create_request_queue_
// Storage module associated with the provider. It serves all queues created
// by it. `storage_` is null initially and when it becomes non-null, the
// provider is ready to create actual queues (speculative queues can be
// created before that as well).
scoped_refptr<StorageModuleInterface> storage_
// Weak pointer factory.
base::WeakPtrFactory<ReportQueueProvider> weak_ptr_factory_{this};
} // namespace reporting