report_queue.h 7.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172
  1. // Copyright 2021 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_H_
  5. #define COMPONENTS_REPORTING_CLIENT_REPORT_QUEUE_H_
  6. #include <memory>
  7. #include <queue>
  8. #include <string>
  9. #include <utility>
  10. #include "base/callback.h"
  11. #include "base/values.h"
  12. #include "components/reporting/proto/synced/record.pb.h"
  13. #include "components/reporting/proto/synced/record_constants.pb.h"
  14. #include "components/reporting/storage/storage_uploader_interface.h"
  15. #include "components/reporting/util/status.h"
  16. #include "components/reporting/util/statusor.h"
  17. #include "third_party/protobuf/src/google/protobuf/message_lite.h"
  18. namespace reporting {
  19. // A |ReportQueue| is not meant to be created directly, instead it is
  20. // instantiated by |ReportQueueProvider|. |ReportQueue| allows a user
  21. // to |Enqueue| a message for delivery to a handler specified by the
  22. // |Destination| held by the provided |ReportQueueConfiguration|.
  23. // |ReportQueue| implementation handles scheduling storage and
  24. // delivery.
  25. // Enqueue can also be used with a |base::Value| or |std::string|.
  26. //
  27. // Example Usage:
  28. // void SendMessage(google::protobuf::ImportantMessage important_message,
  29. // reporting::ReportQueue::EnqueueCallback done_cb) {
  30. // // Create configuration.
  31. // auto config_result = reporting::ReportQueueConfiguration::Create(...);
  32. // // Bail out if configuration failed to create.
  33. // if (!config_result.ok()) {
  34. // std::move(done_cb).Run(config_result.status());
  35. // return;
  36. // }
  37. // // Asynchronously instantiate ReportQueue.
  38. // base::ThreadPool::PostTask(
  39. // FROM_HERE,
  40. // base::BindOnce(
  41. // [](google::protobuf::ImportantMessage important_message,
  42. // reporting::ReportQueue::EnqueueCallback done_cb,
  43. // std::unique_ptr<reporting::ReportQueueConfiguration> config) {
  44. // reporting::ReportQueueProvider::CreateQueue(
  45. // std::move(config),
  46. // base::BindOnce(
  47. // [](google::protobuf::ImportantMessage important_message,
  48. // reporting::ReportQueue::EnqueueCallback done_cb,
  49. // reporting::StatusOr<std::unique_ptr<
  50. // reporting::ReportQueue>> report_queue_result) {
  51. // // Bail out if queue failed to create.
  52. // if (!report_queue_result.ok()) {
  53. // std::move(done_cb).Run(report_queue_result.status());
  54. // return;
  55. // }
  56. // // Queue created successfully, enqueue the message.
  57. // report_queue_result.ValueOrDie()->Enqueue(
  58. // std::move(important_message), std::move(done_cb));
  59. // },
  60. // std::move(important_message), std::move(done_cb)));
  61. // },
  62. // std::move(important_message), std::move(done_cb),
  63. // std::move(config_result.ValueOrDie())));
  64. // }
  65. //
  66. // |SpeculativeReportQueueImpl| is an extension to |ReportQueue| which allows
  67. // to speculatively enqueue records before the actual |ReportQueue| is created
  68. // (which may be delayed by inability to initialize |ReportClient|).
  69. // Instantiated by |ReportQueueProvider| and can be used anywhere |ReportQueue|
  70. // fits. Note however, that records enqueued before actual |ReportQueue|
  71. // is ready may be lost, e.g. if the machine reboots, so for the records
  72. // that need to be definiately recorded |ReportQueue| is preferable.
  73. //
  74. // Example Usage:
  75. // void SendMessage(google::protobuf::LessImportantMessage
  76. // less_important_message,
  77. // reporting::ReportQueue::EnqueueCallback done_cb) {
  78. // // Create configuration.
  79. // auto config_result = reporting::ReportQueueConfiguration::Create(...);
  80. // // Bail out if configuration failed to create.
  81. // if (!config_result.ok()) {
  82. // std::move(done_cb).Run(config_result.status());
  83. // return;
  84. // }
  85. // // Synchronously instantiate SpeculativeReportQueueImpl, returning it as
  86. // // ReportQueue still.
  87. // auto report_queue_result =
  88. // reporting::ReportQueueProvider::CreateSpeculativeQueue(
  89. // std::move(config));
  90. // if (!report_queue_result.ok()) {
  91. // std::move(done_cb).Run(config_result.status());
  92. // return;
  93. // }
  94. // // Enqueue event (store it in memory only until the actual queue is
  95. // // created).
  96. // report_queue_result.ValueOrDie()->Enqueue(
  97. // std::move(less_important_message), std::move(done_cb));
  98. // }
  99. class ReportQueue {
  100. public:
  101. // A callback to asynchronously generate data to be added to |Storage|.
  102. using RecordProducer = base::OnceCallback<StatusOr<std::string>()>;
  103. // An EnqueueCallback is called on the completion of any |Enqueue| call.
  104. using EnqueueCallback = base::OnceCallback<void(Status)>;
  105. // A FlushCallback is called on the completion of |Flush| call.
  106. using FlushCallback = base::OnceCallback<void(Status)>;
  107. virtual ~ReportQueue();
  108. // Enqueue asynchronously stores and delivers a record. The |callback| will
  109. // be called on any errors. If storage is successful |callback| will be called
  110. // with an OK status.
  111. //
  112. // |priority| will Enqueue the record to the specified Priority queue.
  113. //
  114. // The current destinations have the following data requirements:
  115. // (destination : requirement)
  116. // UPLOAD_EVENTS : UploadEventsRequest
  117. //
  118. // |record| string (owned) will be sent with no conversion.
  119. void Enqueue(std::string record,
  120. Priority priority,
  121. EnqueueCallback callback) const;
  122. // |record| as a dictionary (owned) will be converted to a JSON string with
  123. // base::JsonWriter::Write.
  124. void Enqueue(base::Value::Dict record,
  125. Priority priority,
  126. EnqueueCallback callback) const;
  127. // |record| as a protobuf (owned) will be converted to a string with
  128. // SerializeToString(). The handler is responsible for converting the record
  129. // back to a proto with a ParseFromString() call.
  130. void Enqueue(std::unique_ptr<const google::protobuf::MessageLite> record,
  131. Priority priority,
  132. EnqueueCallback callback) const;
  133. // Initiates upload of collected records according to the priority.
  134. // Called usually for a queue with an infinite or very large upload period.
  135. // Multiple |Flush| calls can safely run in parallel.
  136. // Returns error if cannot start upload.
  137. virtual void Flush(Priority priority, FlushCallback callback) = 0;
  138. // Prepares a callback to attach actual queue to the speculative.
  139. // Implemented only in SpeculativeReportQueue, CHECKs in a regular one.
  140. [[nodiscard]] virtual base::OnceCallback<
  141. void(StatusOr<std::unique_ptr<ReportQueue>>)>
  142. PrepareToAttachActualQueue() const = 0;
  143. private:
  144. // Allow SpeculativeReportQueue access to |AddProducedRecord|.
  145. friend class SpeculativeReportQueueImpl;
  146. // Invokes |record_producer| and posts resulting data to the queue storage.
  147. // |record_producer| is expected to be called asynchronously.
  148. // Should only be used within ReportQueue implementation and its derivatives.
  149. virtual void AddProducedRecord(RecordProducer record_producer,
  150. Priority priority,
  151. EnqueueCallback callback) const = 0;
  152. };
  153. } // namespace reporting
  154. #endif // COMPONENTS_REPORTING_CLIENT_REPORT_QUEUE_H_