Add AddTask method to remove duplicate code

mykhailo-kuchma · mykhailo-kuchma · commit 55978d842f7c · 2019-11-21T17:44:03.000+02:00
Remove CatalogClientImpl::AsFuture, align with layer clients.

Relates-To: OLPEDGE-1109

Signed-off-by: Mykhailo Kuchma &lt;ext-mykhailo.kuchma@here.com&gt;
diff --git a/olp-cpp-sdk-dataservice-read/src/CatalogClientImpl.cpp b/olp-cpp-sdk-dataservice-read/src/CatalogClientImpl.cpp
@@ -71,25 +71,15 @@ CancellationToken CatalogClientImpl::GetCatalog(
                                   CatalogResponseCallback callback) {
     auto catalog = catalog_;
     auto settings = settings_;
-    auto pending_requests = pending_requests_;
 
-    auto data_task = [=](client::CancellationContext context) {
+    auto get_catalog_task = [=](client::CancellationContext context) {
       return repository::CatalogRepository::GetCatalog(
           std::move(catalog), std::move(context), std::move(request),
           std::move(settings));
     };
 
-    auto context =
-        client::TaskContext::Create(std::move(data_task), std::move(callback));
-
-    pending_requests->Insert(context);
-
-    repository::ExecuteOrSchedule(task_scheduler_, [=]() {
-      context.Execute();
-      pending_requests->Remove(context);
-    });
-
-    return context.CancelToken();
+    return AddTask(task_scheduler_, pending_requests_,
+                   std::move(get_catalog_task), std::move(callback));
   };
 
   return ScheduleFetch(std::move(schedule_get_catalog), std::move(request),
@@ -98,11 +88,13 @@ CancellationToken CatalogClientImpl::GetCatalog(
 
 CancellableFuture<CatalogResponse> CatalogClientImpl::GetCatalog(
     CatalogRequest request) {
-  return AsFuture<CatalogRequest, CatalogResponse>(
-      std::move(request),
-      static_cast<client::CancellationToken (CatalogClientImpl::*)(
-          CatalogRequest, CatalogResponseCallback)>(
-          &CatalogClientImpl::GetCatalog));
+  auto promise = std::make_shared<std::promise<CatalogResponse>>();
+  auto cancel_token =
+      GetCatalog(std::move(request), [promise](CatalogResponse response) {
+        promise->set_value(std::move(response));
+      });
+  return client::CancellableFuture<CatalogResponse>(std::move(cancel_token),
+                                                    std::move(promise));
 }
 
 CancellationToken CatalogClientImpl::GetLatestVersion(
@@ -112,25 +104,15 @@ CancellationToken CatalogClientImpl::GetLatestVersion(
                                          CatalogVersionCallback callback) {
     auto catalog = catalog_;
     auto settings = settings_;
-    auto pending_requests = pending_requests_;
 
-    auto data_task = [=](client::CancellationContext context) {
+    auto get_latest_version_task = [=](client::CancellationContext context) {
       return repository::CatalogRepository::GetLatestVersion(
           std::move(catalog), std::move(context), std::move(request),
           std::move(settings));
     };
 
-    auto context =
-        client::TaskContext::Create(std::move(data_task), std::move(callback));
-
-    pending_requests->Insert(context);
-
-    repository::ExecuteOrSchedule(task_scheduler_, [=]() {
-      context.Execute();
-      pending_requests->Remove(context);
-    });
-
-    return context.CancelToken();
+    return AddTask(task_scheduler_, pending_requests_,
+                   std::move(get_latest_version_task), std::move(callback));
   };
 
   return ScheduleFetch(std::move(schedule_get_latest_version),
@@ -139,11 +121,13 @@ CancellationToken CatalogClientImpl::GetLatestVersion(
 
 CancellableFuture<CatalogVersionResponse> CatalogClientImpl::GetLatestVersion(
     CatalogVersionRequest request) {
-  return AsFuture<CatalogVersionRequest, CatalogVersionResponse>(
-      std::move(request),
-      static_cast<client::CancellationToken (CatalogClientImpl::*)(
-          CatalogVersionRequest, CatalogVersionCallback)>(
-          &CatalogClientImpl::GetLatestVersion));
+  auto promise = std::make_shared<std::promise<CatalogVersionResponse>>();
+  auto cancel_token = GetLatestVersion(
+      std::move(request), [promise](CatalogVersionResponse response) {
+        promise->set_value(std::move(response));
+      });
+  return client::CancellableFuture<CatalogVersionResponse>(
+      std::move(cancel_token), std::move(promise));
 }
 }  // namespace read
 }  // namespace dataservice
diff --git a/olp-cpp-sdk-dataservice-read/src/CatalogClientImpl.h b/olp-cpp-sdk-dataservice-read/src/CatalogClientImpl.h
@@ -72,21 +72,6 @@ class CatalogClientImpl final {
   client::OlpClientSettings settings_;
   std::shared_ptr<thread::TaskScheduler> task_scheduler_;
   std::shared_ptr<client::PendingRequests> pending_requests_;
-
-  template <typename Request, typename Response>
-  client::CancellableFuture<Response> AsFuture(
-      const Request& request, std::function<client::CancellationToken(
-                                  CatalogClientImpl*, const Request&,
-                                  const std::function<void(Response)>&)>
-                                  func) {
-    auto promise = std::make_shared<std::promise<Response> >();
-
-    auto cancel_token = func(this, request, [promise](Response response) {
-      promise->set_value(response);
-    });
-
-    return client::CancellableFuture<Response>(cancel_token, promise);
-  }
 };
 
 }  // namespace read
diff --git a/olp-cpp-sdk-dataservice-read/src/Common.h b/olp-cpp-sdk-dataservice-read/src/Common.h
@@ -20,7 +20,10 @@
 #pragma once
 
 #include <olp/core/client/CancellationToken.h>
+#include <olp/core/client/PendingRequests.h>
+#include <olp/core/thread/TaskScheduler.h>
 #include <olp/dataservice/read/FetchOptions.h>
+#include "repositories/ExecuteOrSchedule.inl"
 
 namespace olp {
 namespace dataservice {
@@ -58,6 +61,35 @@ inline client::CancellationToken ScheduleFetch(TaskScheduler&& schedule_task,
                        std::forward<Callback>(callback));
 }
 
+/*
+ * @brief Common function used to wrap a lambda function and a callback that
+ * consumes the function result with a TaskContext class and schedule this to a
+ * task scheduler.
+ * @param task_scheduler Task scheduler instance.
+ * @param pending_requests PendingRequests instance that tracks current
+ * requests.
+ * @param task Function that will be executed.
+ * @param callback Function that will consume task output.
+ * @param args Additional agrs to pass to TaskContext.
+ * @return CancellationToken used to cancel the operation.
+ */
+template <typename Function, typename Callback, typename... Args>
+inline client::CancellationToken AddTask(
+    const std::shared_ptr<thread::TaskScheduler>& task_scheduler,
+    const std::shared_ptr<client::PendingRequests>& pending_requests,
+    Function task, Callback callback, Args&&... args) {
+  auto context = client::TaskContext::Create(
+      std::move(task), std::move(callback), std::forward<Args>(args)...);
+  pending_requests->Insert(context);
+
+  repository::ExecuteOrSchedule(task_scheduler, [=] {
+    context.Execute();
+    pending_requests->Remove(context);
+  });
+
+  return context.CancelToken();
+}
+
 }  // namespace read
 }  // namespace dataservice
 }  // namespace olp
diff --git a/olp-cpp-sdk-dataservice-read/src/VersionedLayerClientImpl.cpp b/olp-cpp-sdk-dataservice-read/src/VersionedLayerClientImpl.cpp
@@ -96,24 +96,14 @@ client::CancellationToken VersionedLayerClientImpl::GetPartitions(
     auto catalog = catalog_;
     auto layer_id = layer_id_;
     auto settings = *settings_;
-    auto pending_requests = pending_requests_;
 
     auto partitions_task = [=](client::CancellationContext context) {
       return repository::PartitionsRepository::GetVersionedPartitions(
           catalog, layer_id, context, std::move(request), settings);
     };
 
-    auto context = client::TaskContext::Create(std::move(partitions_task),
-                                               std::move(callback));
-
-    pending_requests->Insert(context);
-
-    repository::ExecuteOrSchedule(task_scheduler_, [=]() {
-      context.Execute();
-      pending_requests->Remove(context);
-    });
-
-    return context.CancelToken();
+    return AddTask(task_scheduler_, pending_requests_,
+                   std::move(partitions_task), std::move(callback));
   };
 
   return ScheduleFetch(std::move(schedule_get_partitions), std::move(request),
@@ -138,25 +128,15 @@ client::CancellationToken VersionedLayerClientImpl::GetData(
     auto catalog = catalog_;
     auto layer_id = layer_id_;
     auto settings = *settings_;
-    auto pending_requests = pending_requests_;
-
+    
     auto data_task = [=](client::CancellationContext context) {
       return repository::DataRepository::GetVersionedData(
           std::move(catalog), std::move(layer_id), std::move(request), context,
           std::move(settings));
     };
 
-    auto context =
-        client::TaskContext::Create(std::move(data_task), std::move(callback));
-
-    pending_requests->Insert(context);
-
-    repository::ExecuteOrSchedule(task_scheduler_, [=]() {
-      context.Execute();
-      pending_requests->Remove(context);
-    });
-
-    return context.CancelToken();
+    return AddTask(task_scheduler_, pending_requests_, std::move(data_task),
+                   std::move(callback));
   };
   return ScheduleFetch(std::move(schedule_get_data), std::move(request),
                        std::move(callback));
diff --git a/olp-cpp-sdk-dataservice-read/src/VolatileLayerClientImpl.cpp b/olp-cpp-sdk-dataservice-read/src/VolatileLayerClientImpl.cpp
@@ -78,25 +78,15 @@ client::CancellationToken VolatileLayerClientImpl::GetPartitions(
     auto catalog = catalog_;
     auto layer_id = layer_id_;
     auto settings = *settings_;
-    auto pending_requests = pending_requests_;
 
     auto data_task = [=](client::CancellationContext context) {
       return repository::PartitionsRepository::GetVolatilePartitions(
           std::move(catalog), std::move(layer_id), std::move(context),
           std::move(request), std::move(settings));
     };
 
-    auto context =
-        client::TaskContext::Create(std::move(data_task), std::move(callback));
-
-    pending_requests->Insert(context);
-
-    repository::ExecuteOrSchedule(task_scheduler_, [=]() {
-      context.Execute();
-      pending_requests->Remove(context);
-    });
-
-    return context.CancelToken();
+    return AddTask(task_scheduler_, pending_requests_, std::move(data_task),
+                   std::move(callback));
   };
 
   return ScheduleFetch(std::move(schedule_get_partitions), std::move(request),
@@ -120,24 +110,14 @@ client::CancellationToken VolatileLayerClientImpl::GetData(
     auto catalog = catalog_;
     auto layer_id = layer_id_;
     auto settings = *settings_;
-    auto pending_requests = pending_requests_;
 
-    auto data_task = [=](client::CancellationContext context) {
+    auto partitions_task = [=](client::CancellationContext context) {
       return repository::DataRepository::GetVolatileData(
           catalog, layer_id, request, context, settings);
     };
 
-    auto context =
-        client::TaskContext::Create(std::move(data_task), std::move(callback));
-
-    pending_requests->Insert(context);
-
-    repository::ExecuteOrSchedule(task_scheduler_, [=]() {
-      context.Execute();
-      pending_requests->Remove(context);
-    });
-
-    return context.CancelToken();
+    return AddTask(task_scheduler_, pending_requests_,
+                   std::move(partitions_task), std::move(callback));
   };
 
   return ScheduleFetch(std::move(schedule_get_data), std::move(request),
diff --git a/olp-cpp-sdk-dataservice-read/src/repositories/ExecuteOrSchedule.inl b/olp-cpp-sdk-dataservice-read/src/repositories/ExecuteOrSchedule.inl
@@ -17,6 +17,8 @@
  * License-Filename: LICENSE
  */
 
+#pragma once
+
 #include <olp/core/client/OlpClientSettings.h>
 #include <olp/core/thread/TaskScheduler.h>
 
@@ -25,11 +27,9 @@ namespace dataservice {
 namespace read {
 namespace repository {
 
-namespace {
-
 using CallFuncType = thread::TaskScheduler::CallFuncType;
 
-void ExecuteOrSchedule(
+inline void ExecuteOrSchedule(
     const std::shared_ptr<thread::TaskScheduler>& task_scheduler,
     CallFuncType&& func) {
   if (!task_scheduler) {
@@ -40,14 +40,12 @@ void ExecuteOrSchedule(
   }
 }
 
-void ExecuteOrSchedule(const client::OlpClientSettings* settings,
-                       CallFuncType&& func) {
+inline void ExecuteOrSchedule(const client::OlpClientSettings* settings,
+                              CallFuncType&& func) {
   ExecuteOrSchedule(settings ? settings->task_scheduler : nullptr,
                     std::move(func));
 }
 
-}  // namespace
-
 }  // namespace repository
 }  // namespace read
 }  // namespace dataservice
