Reporting: Implement delivery agent.

net/reporting/reporting_delivery_agent.cc

Index: net/reporting/reporting_delivery_agent.cc
diff --git a/net/reporting/reporting_delivery_agent.cc b/net/reporting/reporting_delivery_agent.cc
new file mode 100644
index 0000000000000000000000000000000000000000..46ff38e028a457891b0bf1f21a9754917cd1a275
--- /dev/null
+++ b/net/reporting/reporting_delivery_agent.cc
@@ -0,0 +1,148 @@
+// 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_report.h"
+#include "net/reporting/reporting_uploader.h"
+#include "url/gurl.h"
+#include "url/origin.h"
+
+namespace net {
+
+namespace {
+
+std::string SerializeReports(const std::vector<const ReportingReport*>& reports,
+                             base::TimeTicks now) {
+  base::ListValue reports_value;
+
+  for (const ReportingReport* 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 ReportingReport*>& reports)
+      : endpoint(endpoint), reports(reports) {}
+
+  ~Delivery() {}
+
+  const GURL endpoint;
+  const std::vector<const ReportingReport*> reports;
+};
+
+void ReportingDeliveryAgent::SendReports() {
+  std::vector<const ReportingReport*> reports;
+  cache_->GetReports(&reports);
+
+  // Sort reports into (origin, group) buckets.

[shivanisha, 2017/03/30 20:16:11]

Since we want to key the reports with <origin, group> pair, how about using the
pair as the key. That should help in removing the nested loops in using this map
below.
std::map < OriginGroup, std::vector<const ReportingReport*> >

[Julia Tuttle, 2017/03/31 14:49:48]

Good call -- I wrote this before I added pending_origin_groups_, which is when I
added OriginGroup.

+  std::map<url::Origin,
+           std::map<std::string, std::vector<const ReportingReport*>>>
+      origin_group_reports;
+  for (const ReportingReport* 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 ReportingReport*>> endpoint_reports;

[shivanisha, 2017/03/30 20:16:11]

Since its not allowed for multiple reports for the same endpoint to be sent at
the same time, why do we have a vector of reports here instead of a single
report?

[Julia Tuttle, 2017/03/31 14:49:48]

What is disallowed is multiple *uploads*, not multiple *reports*. An upload can
contain an infinite number of reports. (Actually, that might be something we
want to cap at some point.)

I've added a note below the concurrency explanation to clarify that an upload
can contain multiple reports, in case anyone else has this question.

+  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) {
+  if (outcome == ReportingUploader::SUCCESS) {
+    cache_->RemoveReports(delivery->reports);
+    endpoint_manager_.InformOfEndpointRequest(delivery->endpoint, true);
+  } else {
+    cache_->IncrementReportsAttempts(delivery->reports);
+    endpoint_manager_.InformOfEndpointRequest(delivery->endpoint, false);
+  }
+
+  if (outcome == ReportingUploader::REMOVE_ENDPOINT)
+    cache_->RemoveClientsForEndpoint(delivery->endpoint);
+
+  for (const ReportingReport* report : delivery->reports) {
+    url::Origin origin(report->url);
+    pending_origin_groups_.erase(std::make_pair(origin, report->group));
+  }
+
+  cache_->ClearReportsPending(delivery->reports);
+  endpoint_manager_.ClearEndpointPending(delivery->endpoint);
+}
+
+}  // namespace net