report_queue_impl.h 6.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151
  1. // Copyright 2020 The Chromium Authors. All rights reserved.
  2. // Use of this source code is governed by a BSD-style license that can be
  3. // found in the LICENSE file.
  4. #ifndef COMPONENTS_REPORTING_CLIENT_REPORT_QUEUE_IMPL_H_
  5. #define COMPONENTS_REPORTING_CLIENT_REPORT_QUEUE_IMPL_H_
  6. #include <memory>
  7. #include <queue>
  8. #include <string>
  9. #include <utility>
  10. #include "base/callback.h"
  11. #include "base/memory/scoped_refptr.h"
  12. #include "base/memory/weak_ptr.h"
  13. #include "base/sequence_checker.h"
  14. #include "base/task/sequenced_task_runner.h"
  15. #include "components/reporting/client/report_queue.h"
  16. #include "components/reporting/client/report_queue_configuration.h"
  17. #include "components/reporting/proto/synced/record.pb.h"
  18. #include "components/reporting/proto/synced/record_constants.pb.h"
  19. #include "components/reporting/storage/storage_module_interface.h"
  20. #include "components/reporting/util/status.h"
  21. #include "components/reporting/util/statusor.h"
  22. #include "third_party/abseil-cpp/absl/types/optional.h"
  23. namespace reporting {
  24. // A |ReportQueueImpl| is configured with a |ReportQueueConfiguration|. A
  25. // |ReportQueueImpl| allows a user to |Enqueue| a message for delivery to a
  26. // handler specified by the |Destination| held by the provided
  27. // |ReportQueueConfiguration|. |ReportQueueImpl| handles scheduling storage and
  28. // delivery.
  29. //
  30. // ReportQueues are not meant to be created directly, instead use the
  31. // reporting::ReportQueueProvider::CreateQueue(...) function. See the
  32. // comments for reporting::ReportingClient for example usage.
  33. //
  34. // Enqueue can also be used with a |base::Value| or |std::string|.
  35. class ReportQueueImpl : public ReportQueue {
  36. public:
  37. // Factory
  38. static void Create(
  39. std::unique_ptr<ReportQueueConfiguration> config,
  40. scoped_refptr<StorageModuleInterface> storage,
  41. base::OnceCallback<void(StatusOr<std::unique_ptr<ReportQueue>>)> cb);
  42. ReportQueueImpl(const ReportQueueImpl& other) = delete;
  43. ReportQueueImpl& operator=(const ReportQueueImpl& other) = delete;
  44. ~ReportQueueImpl() override;
  45. void Flush(Priority priority, FlushCallback callback) override;
  46. // Dummy implementation for a regular queue.
  47. [[nodiscard]] base::OnceCallback<void(StatusOr<std::unique_ptr<ReportQueue>>)>
  48. PrepareToAttachActualQueue() const override;
  49. protected:
  50. ReportQueueImpl(std::unique_ptr<ReportQueueConfiguration> config,
  51. scoped_refptr<StorageModuleInterface> storage);
  52. private:
  53. void AddProducedRecord(RecordProducer record_producer,
  54. Priority priority,
  55. EnqueueCallback callback) const override;
  56. const std::unique_ptr<ReportQueueConfiguration> config_;
  57. const scoped_refptr<StorageModuleInterface> storage_;
  58. };
  59. class SpeculativeReportQueueImpl : public ReportQueue {
  60. public:
  61. // Factory method returns a smart pointer with on-thread deleter.
  62. static std::unique_ptr<SpeculativeReportQueueImpl, base::OnTaskRunnerDeleter>
  63. Create();
  64. SpeculativeReportQueueImpl(const SpeculativeReportQueueImpl& other) = delete;
  65. SpeculativeReportQueueImpl& operator=(
  66. const SpeculativeReportQueueImpl& other) = delete;
  67. ~SpeculativeReportQueueImpl() override;
  68. // Forwards |Flush| to |ReportQueue|, if already created.
  69. // Returns with failure otherwise.
  70. void Flush(Priority priority, FlushCallback callback) override;
  71. // Provides a callback to attach initialized actual queue to the speculative
  72. // queue.
  73. [[nodiscard]] base::OnceCallback<void(StatusOr<std::unique_ptr<ReportQueue>>)>
  74. PrepareToAttachActualQueue() const override;
  75. // Substitutes actual queue to the speculative, when ready.
  76. // Initiates processesing of all pending records.
  77. void AttachActualQueue(
  78. StatusOr<std::unique_ptr<ReportQueue>> status_or_actual_queue);
  79. private:
  80. // Moveable, non-copyable struct holding a pending record producer for the
  81. // |pending_record_producers_| queue below.
  82. struct PendingRecordProducer {
  83. PendingRecordProducer(RecordProducer producer, Priority priority);
  84. PendingRecordProducer(PendingRecordProducer&& other);
  85. PendingRecordProducer& operator=(PendingRecordProducer&& other);
  86. ~PendingRecordProducer();
  87. RecordProducer record_producer;
  88. Priority record_priority;
  89. };
  90. // Private constructor, used by the factory method only.
  91. explicit SpeculativeReportQueueImpl(
  92. scoped_refptr<base::SequencedTaskRunner> sequenced_task_runner);
  93. // Forwards |AddProducedRecord| to |ReportQueue|, if already created.
  94. // Records the record internally otherwise.
  95. void AddProducedRecord(RecordProducer record_producer,
  96. Priority priority,
  97. EnqueueCallback callback) const override;
  98. // Enqueues head of the |pending_record_producers_| and reapplies for the rest
  99. // of it.
  100. void EnqueuePendingRecordProducers(EnqueueCallback callback) const;
  101. // Optionally enqueues |record_producer| (owned) to actual queue, if ready.
  102. // Otherwise adds it to the end of |pending_record_producers_|.
  103. void MaybeEnqueueRecordProducer(Priority priority,
  104. EnqueueCallback callback,
  105. RecordProducer record_producer) const;
  106. // Task runner that protects |report_queue_| and |pending_record_producers_|
  107. // and allows to synchronize the initialization.
  108. const scoped_refptr<base::SequencedTaskRunner> sequenced_task_runner_;
  109. SEQUENCE_CHECKER(sequence_checker_);
  110. // Creation status of |ReportQueue| and actual |ReportQueue| if successfully
  111. // created.
  112. absl::optional<StatusOr<std::unique_ptr<ReportQueue>>> status_or_report_queue_
  113. GUARDED_BY_CONTEXT(sequence_checker_);
  114. // Queue of the pending record producers, collected before actual queue has
  115. // been created. Declared 'mutable', because it is accessed by 'const'
  116. // methods.
  117. mutable std::queue<PendingRecordProducer> pending_record_producers_
  118. GUARDED_BY_CONTEXT(sequence_checker_);
  119. // Weak pointer factory.
  120. base::WeakPtrFactory<SpeculativeReportQueueImpl> weak_ptr_factory_{this};
  121. };
  122. } // namespace reporting
  123. #endif // COMPONENTS_REPORTING_CLIENT_REPORT_QUEUE_IMPL_H_