Change PrefetchHelper to template class (#1045)

Refactor PrefetchHelper, QueryMetadataJob, DownloadItemsJob
to work with different root types.
With PrefetchHelper we can start query data handles for
different types. User should only specify such functions as
'query' - to request metadata needed to download data,
'filter' - to manage results received after query,
'download' - download data.

Relates-To: OLPEDGE-1900

Signed-off-by: Liubov Didkivska <ext-liubov.didkivska@here.com>

Author: Liubov Didkivska

olp-cpp-sdk-dataservice-read/src/DownloadItemsJob.h

@@ -0,0 +1,124 @@
+/*
+ * Copyright (C) 2020 HERE Europe B.V.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * License-Filename: LICENSE
+ */
+
+#pragma once
+
+#include <olp/core/client/CancellationContext.h>
+#include <olp/core/logging/Log.h>
+#include <olp/dataservice/read/Types.h>
+
+#include "Common.h"
+#include "ExtendedApiResponse.h"
+#include "ExtendedApiResponseHelpers.h"
+
+namespace olp {
+namespace dataservice {
+namespace read {
+
+using ExtendedDataResponse = ExtendedApiResponse<model::Data, client::ApiError,
+                                                 client::NetworkStatistics>;
+
+// Prototype of function used to download data using data handle.
+using DownloadFunc = std::function<ExtendedDataResponse(
+    std::string, client::CancellationContext)>;
+
+template <typename ItemType, typename PrefetchResult>
+using AppendResultFunc =
+    std::function<void(ExtendedDataResponse response, ItemType item,
+                       PrefetchResult& prefetch_result)>;
+
+template <typename ItemType, typename PrefetchResult>
+class DownloadItemsJob {
+ public:
+  DownloadItemsJob(DownloadFunc download,
+                   AppendResultFunc<ItemType, PrefetchResult> append_result,
+                   Callback<PrefetchResult> user_callback,
+                   PrefetchStatusCallback status_callback)
+      : download_(std::move(download)),
+        append_result_(std::move(append_result)),
+        user_callback_(std::move(user_callback)),
+        status_callback_(std::move(status_callback)) {}
+
+  void Initialize(size_t items_count, client::NetworkStatistics statistics) {
+    download_task_count_ = total_download_task_count_ = items_count;
+    accumulated_statistics_ = statistics;
+  }
+
+  ExtendedDataResponse Download(const std::string& data_handle,
+                                client::CancellationContext context) {
+    return download_(data_handle, context);
+  }
+  size_t GetAccumulatedBytes(const olp::client::NetworkStatistics& statistics) {
+    // This narrow cast is necessary to avoid narrowing compiler errors like
+    // -Wc++11-narrowing when building for 32bit targets.
+    const auto bytes_transferred =
+        statistics.GetBytesDownloaded() + statistics.GetBytesUploaded();
+    return static_cast<size_t>(bytes_transferred &
+                               std::numeric_limits<size_t>::max());
+  }
+
+  void CompleteItem(ItemType item, ExtendedDataResponse response) {
+    std::lock_guard<std::mutex> lock(mutex_);
+    accumulated_statistics_ += GetNetworkStatistics(response);
+
+    if (response.IsSuccessful()) {
+      requests_succeeded_++;
+    } else {
+      requests_failed_++;
+    }
+
+    append_result_(response, item, prefetch_result_);
+
+    if (status_callback_) {
+      status_callback_(PrefetchStatus{
+          requests_succeeded_ + requests_failed_, total_download_task_count_,
+          GetAccumulatedBytes(accumulated_statistics_)});
+    }
+
+    if (!--download_task_count_) {
+      OLP_SDK_LOG_DEBUG_F("DownloadItemsJob",
+                          "Download complete, succeeded=%zu, failed=%zu",
+                          requests_succeeded_, requests_failed_);
+
+      OnPrefetchCompleted(std::move(prefetch_result_));
+    }
+  }
+
+  void OnPrefetchCompleted(Response<PrefetchResult> result) {
+    auto prefetch_callback = std::move(user_callback_);
+    prefetch_callback(std::move(result));
+  }
+
+ private:
+  DownloadFunc download_;
+  AppendResultFunc<ItemType, PrefetchResult> append_result_;
+  Callback<PrefetchResult> user_callback_;
+  PrefetchStatusCallback status_callback_;
+  size_t download_task_count_{0};
+  size_t total_download_task_count_{0};
+  size_t requests_succeeded_{0};
+  size_t requests_failed_{0};
+  client::NetworkStatistics accumulated_statistics_;
+  PrefetchResult prefetch_result_;
+  std::mutex mutex_;
+};
+
+}  // namespace read
+}  // namespace dataservice
+}  // namespace olp

olp-cpp-sdk-dataservice-read/src/PrefetchHelper.h

@@ -0,0 +1,92 @@
+/*
+ * Copyright (C) 2020 HERE Europe B.V.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * License-Filename: LICENSE
+ */
+
+#pragma once
+
+#include <olp/core/client/CancellationContext.h>
+#include <olp/core/client/PendingRequests.h>
+#include <olp/core/logging/Log.h>
+#include <olp/dataservice/read/Types.h>
+
+#include "Common.h"
+#include "DownloadItemsJob.h"
+#include "ExtendedApiResponse.h"
+#include "ExtendedApiResponseHelpers.h"
+#include "QueryMetadataJob.h"
+
+namespace olp {
+namespace dataservice {
+namespace read {
+
+template <typename PrefetchItemsResult>
+using PrefetchItemsResponseCallback = Callback<PrefetchItemsResult>;
+
+class PrefetchHelper {
+ public:
+  template <typename ItemType, typename QueryType, typename PrefetchResult>
+  static client::CancellationToken Prefetch(
+      const std::vector<QueryType>& roots,
+      QueryItemsFunc<ItemType, QueryType> query,
+      FilterItemsFunc<ItemType> filter, DownloadFunc download,
+      AppendResultFunc<ItemType, PrefetchResult> append_result,
+      Callback<PrefetchResult> user_callback,
+      PrefetchStatusCallback status_callback,
+      std::shared_ptr<thread::TaskScheduler> task_scheduler,
+      std::shared_ptr<client::PendingRequests> pending_requests) {
+    client::CancellationContext execution_context;
+
+    auto download_job =
+        std::make_shared<DownloadItemsJob<ItemType, PrefetchResult>>(
+            std::move(download), std::move(append_result),
+            std::move(user_callback), std::move(status_callback));
+
+    auto query_job =
+        std::make_shared<QueryMetadataJob<ItemType, QueryType, PrefetchResult>>(
+            std::move(query), std::move(filter), download_job, task_scheduler,
+            pending_requests, execution_context);
+
+    query_job->Initialize(roots.size());
+
+    OLP_SDK_LOG_DEBUG_F("PrefetchJob", "Starting queries, requests=%zu",
+                        roots.size());
+
+    execution_context.ExecuteOrCancelled([&]() {
+      VectorOfTokens tokens;
+      std::transform(std::begin(roots), std::end(roots),
+                     std::back_inserter(tokens), [&](QueryType root) {
+                       return AddTask(
+                           task_scheduler, pending_requests,
+                           [=](client::CancellationContext context) {
+                             return query_job->Query(root, context);
+                           },
+                           [=](QueryItemsResponse<ItemType> response) {
+                             query_job->CompleteQuery(std::move(response));
+                           });
+                     });
+      return CreateToken(std::move(tokens));
+    });
+
+    return client::CancellationToken(
+        [execution_context]() mutable { execution_context.CancelOperation(); });
+  }
+};
+
+}  // namespace read
+}  // namespace dataservice
+}  // namespace olp