Reporting: Implement delivery agent.

net/reporting/reporting_delivery_agent.cc

+// Copyright 2017 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "net/reporting/reporting_delivery_agent.h"
+
+#include <map>
+#include <string>
+#include <vector>
+
+#include "base/bind.h"
+#include "base/json/json_writer.h"
+#include "base/logging.h"
+#include "base/memory/ptr_util.h"
+#include "base/time/tick_clock.h"
+#include "base/values.h"
+#include "net/reporting/reporting_cache.h"
+#include "net/reporting/reporting_endpoint_manager.h"
+#include "net/reporting/reporting_uploader.h"
+#include "url/gurl.h"
+#include "url/origin.h"
+
+namespace net {
+
+using Client = ReportingCache::Client;
+using Report = ReportingCache::Report;
+
+namespace {
+
+std::string SerializeReports(const std::vector<const Report*>& reports,
+                             base::TimeTicks now) {
+  base::ListValue reports_value;
+
+  for (const Report* report : reports) {
+    std::unique_ptr<base::DictionaryValue> report_value =
+        base::MakeUnique<base::DictionaryValue>();
+
+    report_value->SetInteger("age", (now - report->queued).InMilliseconds());
+    report_value->SetString("type", report->type);
+    report_value->SetString("url", report->url.spec());
+    report_value->Set("report", report->body->DeepCopy());
+
+    reports_value.Append(std::move(report_value));
+  }
+
+  std::string reports_json = "";
+  bool json_written = base::JSONWriter::Write(reports_value, &reports_json);
+  DCHECK(json_written);
+
+  return reports_json;
+}
+
+}  // namespace
+
+ReportingDeliveryAgent::ReportingDeliveryAgent(
+    base::TickClock* clock,
+    ReportingCache* cache,
+    ReportingUploader* uploader,
+    const BackoffEntry::Policy* endpoint_backoff_policy)
+    : clock_(clock),
+      cache_(cache),
+      uploader_(uploader),
+      endpoint_manager_(clock, cache, endpoint_backoff_policy),
+      weak_factory_(this) {}
+ReportingDeliveryAgent::~ReportingDeliveryAgent() {}
+
+class ReportingDeliveryAgent::Delivery {
+ public:
+  Delivery(const GURL& endpoint, const std::vector<const Report*>& reports)
+      : endpoint(endpoint), reports(reports) {}
+
+  ~Delivery() {}
+
+  const GURL endpoint;
+  const std::vector<const Report*> reports;
+};
+
+void ReportingDeliveryAgent::SendReports() {
+  std::vector<const Report*> reports;
+  cache_->GetReports(&reports);
+
+  // Sort reports into (origin, group) buckets.
+  std::map<url::Origin, std::map<std::string, std::vector<const Report*>>>
+      origin_group_reports;
+  for (const Report* report : reports) {
+    url::Origin origin(report->url);
+    origin_group_reports[origin][report->group].push_back(report);
+  }
+
+  // Find endpoint for each (origin, group) bucket and sort reports into
+  // endpoint buckets. Don't allow concurrent uploads to the same (origin,
+  // group) bucket.
+  std::map<GURL, std::vector<const Report*>> endpoint_reports;
+  for (auto& it : origin_group_reports) {
+    for (auto& jt : it.second) {
+      OriginGroup origin_group(it.first, jt.first);
+      if (pending_origin_groups_.count(origin_group) > 0u)
+        continue;
+      GURL endpoint_url;
+      if (!endpoint_manager_.FindEndpointForOriginAndGroup(it.first, jt.first,
+                                                           &endpoint_url)) {
+        continue;
+      }
+
+      endpoint_reports[endpoint_url].insert(
+          endpoint_reports[endpoint_url].end(), jt.second.begin(),
+          jt.second.end());
+      pending_origin_groups_.insert(origin_group);
+    }
+  }
+
+  // Start an upload to each endpoint.
+  for (auto& it : endpoint_reports) {
+    endpoint_manager_.SetEndpointPending(it.first);
+    cache_->SetReportsPending(it.second);
+
+    uploader_->StartUpload(
+        it.first, SerializeReports(it.second, clock_->NowTicks()),
+        base::Bind(&ReportingDeliveryAgent::OnUploadComplete,
+                   weak_factory_.GetWeakPtr(),
+                   base::MakeUnique<Delivery>(it.first, it.second)));
+  }
+}
+
+void ReportingDeliveryAgent::OnUploadComplete(
+    const std::unique_ptr<Delivery>& delivery,
+    ReportingUploader::Outcome outcome) {
+  switch (outcome) {
+    case ReportingUploader::SUCCESS:
+      cache_->RemoveReports(delivery->reports);
+      endpoint_manager_.InformOfEndpointRequest(delivery->endpoint, true);
+      break;
+    case ReportingUploader::REMOVE_ENDPOINT:
+      cache_->RemoveClientsForEndpoint(delivery->endpoint);
+    // Fall through:
+    case ReportingUploader::FAILURE:
+      cache_->IncrementReportsAttempts(delivery->reports);
+      endpoint_manager_.InformOfEndpointRequest(delivery->endpoint, false);
+      break;
+  }
+
+  for (const Report* report : delivery->reports) {
+    url::Origin origin(report->url);
+    pending_origin_groups_.erase(std::make_pair(origin, report->group));
+  }
+
+  endpoint_manager_.ClearEndpointPending(delivery->endpoint);
+  cache_->ClearReportsPending(delivery->reports);
+}
+
+}  // namespace net