Add a task context class.

Class take care of users callbacks, and make sure that client
callback is called exactly 1 time.
VolatileLayerClientImpl::GetData adapted to TaskContext usage
VersionedLayerClientImpl::GetData adapted to TaskContext usage

Resolves: OLPEDGE-909

Signed-off-by: Mykhailo Kuchma <[email protected]>
Signed-off-by: Diachenko Mykahilo <[email protected]>

Author: mykhailo-kuchma

olp-cpp-sdk-core/include/olp/core/client/ApiResponse.h

@@ -36,6 +36,9 @@ namespace client {
 template <typename Result, typename Error>
 class ApiResponse {
  public:
+  using ResultType = Result;
+  using ErrorType = Error;
+
   /**
    * @brief ApiResponse Default constructor.
    */
@@ -44,12 +47,12 @@ class ApiResponse {
   /**
    * @brief ApiResponse Constructor for a successfully executed request.
    */
-  ApiResponse(const Result& result) : result_(result), success_(true) {}
+  ApiResponse(const ResultType& result) : result_(result), success_(true) {}
 
   /**
    * @brief ApiResponse Constructor if request unsuccessfully executed
    */
-  ApiResponse(const Error& error) : error_(error), success_(false) {}
+  ApiResponse(const ErrorType& error) : error_(error), success_(false) {}
 
   /**
    * @brief ApiResponse Copy constructor.
@@ -68,18 +71,18 @@ class ApiResponse {
    * @return A valid Result if IsSuccessful() returns true. Undefined,
    * otherwise.
    */
-  inline const Result& GetResult() const { return result_; }
+  inline const ResultType& GetResult() const { return result_; }
 
   /**
    * @brief GetError Gets the error for an unsucccessful request attempt.
    * @return A valid Error if IsSuccessful() returns false. Undefined,
    * otherwise.
    */
-  inline const Error& GetError() const { return error_; }
+  inline const ErrorType& GetError() const { return error_; }
 
  private:
-  Result result_ {};
-  Error error_ {};
+  ResultType result_{};
+  ErrorType error_{};
   bool success_ {false};
 };
 

olp-cpp-sdk-core/include/olp/core/client/Condition.h

@@ -33,8 +33,7 @@ namespace client {
  */
 class Condition final {
  public:
-  Condition(std::chrono::milliseconds timeout = std::chrono::seconds(60))
-      : timeout_{timeout} {}
+  Condition() = default;
 
   /**
    * @brief Should be called by task's callback to notify \c Wait to unblock the
@@ -51,18 +50,19 @@ class Condition final {
   /**
    * @brief Waits a task for a \c Notify or \c CancellationContext to be
    * cancelled by the user.
+   * @param timeout milliseconds to wait on condition
+   * @return True on notified wake, False on timeout
    */
-  bool Wait() {
+  bool Wait(std::chrono::milliseconds timeout = std::chrono::seconds(60)) {
     std::unique_lock<std::mutex> lock(mutex_);
-    bool triggered = condition_.wait_for(
-        lock, timeout_, [&] { return signaled_; });
+    bool triggered =
+        condition_.wait_for(lock, timeout, [&] { return signaled_; });
 
     signaled_ = false;
     return triggered;
   }
 
  private:
-  const std::chrono::milliseconds timeout_;
   std::condition_variable condition_;
   std::mutex mutex_;
   bool signaled_{false};

olp-cpp-sdk-core/tests/CMakeLists.txt

@@ -22,6 +22,7 @@ set(OLP_CPP_SDK_CORE_TESTS_SOURCES
     ./cache/InMemoryCacheTest.cpp
 
     ./client/CancellationContextTest.cpp
+    ./client/ConditionTest.cpp
 
     ./geo/coordinates/GeoCoordinates3dTest.cpp
     ./geo/coordinates/GeoCoordinatesTest.cpp

olp-cpp-sdk-core/tests/client/ConditionTest.cpp

@@ -0,0 +1,57 @@
+/*
+ * Copyright (C) 2019 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
+ */
+
+#include <olp/core/client/Condition.h>
+
+#include <gtest/gtest.h>
+#include <future>
+
+using olp::client::Condition;
+
+namespace {
+TEST(ConditionTest, NotifyBeforeWaitRespected) {
+  Condition condition{};
+  condition.Notify();
+  EXPECT_TRUE(condition.Wait());
+}
+
+TEST(ConditionTest, WaitClearsTriggered) {
+  Condition condition{};
+  condition.Notify();
+  EXPECT_TRUE(condition.Wait());
+  EXPECT_FALSE(condition.Wait(std::chrono::seconds(0)));
+}
+
+TEST(ConditionTest, Timeouted) {
+  Condition condition{};
+  EXPECT_FALSE(condition.Wait(std::chrono::seconds(0)));
+}
+
+TEST(ConditionTest, WakeUp) {
+  Condition condition{};
+  auto wait_future =
+      std::async(std::launch::async, [&]() { return condition.Wait(); });
+  EXPECT_EQ(wait_future.wait_for(std::chrono::milliseconds(10)),
+            std::future_status::timeout);
+  condition.Notify();
+  EXPECT_EQ(wait_future.wait_for(std::chrono::milliseconds(10)),
+            std::future_status::ready);
+  EXPECT_TRUE(wait_future.get());
+}
+}  // namespace

olp-cpp-sdk-dataservice-read/src/ApiClientLookup.cpp

@@ -152,8 +152,7 @@ ApiClientLookup::ApiClientResponse ApiClientLookup::LookupApi(
     }
   }
 
-  client::Condition condition(
-      std::chrono::seconds(settings.retry_settings.timeout));
+  client::Condition condition{};
 
   auto client = std::make_shared<client::OlpClient>();
   client->SetSettings(std::move(settings));
@@ -190,7 +189,7 @@ ApiClientLookup::ApiClientResponse ApiClientLookup::LookupApi(
         condition.Notify();
       });
 
-  if (!condition.Wait()) {
+  if (!condition.Wait(std::chrono::seconds(settings.retry_settings.timeout))) {
     cancellation_context.CancelOperation();
     return client::ApiError(client::ErrorCode::RequestTimeout,
                             "Network request timed out.");

olp-cpp-sdk-dataservice-read/src/PendingRequests.cpp

@@ -18,6 +18,7 @@
  */
 
 #include "PendingRequests.h"
+#include "TaskContext.h"
 
 namespace olp {
 namespace dataservice {
@@ -30,11 +31,18 @@ PendingRequests::~PendingRequests(){};
 bool PendingRequests::CancelPendingRequests() {
   requests_lock_.lock();
   // Create local copy of the requests to cancel
+  auto contexts = std::move(task_contexts_);
   auto requests_map = requests_map_;
+
   requests_lock_.unlock();
   for (auto& pair : requests_map) {
     pair.second.cancel();
   }
+
+  for (auto context : contexts) {
+    context.BlockingCancel();
+  }
+
   return true;
 }
 
@@ -55,6 +63,11 @@ bool PendingRequests::Insert(const client::CancellationToken& request,
   return true;
 }
 
+void PendingRequests::Insert(TaskContext task_context) {
+  std::lock_guard<std::mutex> lk(requests_lock_);
+  task_contexts_.insert(task_context);
+}
+
 bool PendingRequests::Remove(int64_t key) {
   std::lock_guard<std::mutex> lk(requests_lock_);
   auto request = requests_map_.find(key);
@@ -65,6 +78,11 @@ bool PendingRequests::Remove(int64_t key) {
   return false;
 }
 
+void PendingRequests::Remove(TaskContext task_context) {
+  std::lock_guard<std::mutex> lk(requests_lock_);
+  task_contexts_.erase(task_context);
+}
+
 }  // namespace read
 }  // namespace dataservice
-}  // namespace olp
+}  // namespace olp

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

@@ -19,11 +19,14 @@
 
 #pragma once
 
-#include <unordered_map>
+#include <memory>
 #include <mutex>
+#include <unordered_map>
+#include <unordered_set>
 #include <vector>
 
 #include <olp/core/client/CancellationToken.h>
+#include "TaskContext.h"
 
 namespace olp {
 namespace dataservice {
@@ -56,16 +59,29 @@ class PendingRequests final {
    */
   bool Insert(const client::CancellationToken& token, int64_t key);
 
+  /**
+   * @brief Inserts task context into the requests container.
+   * @param task_context Task context.
+   */
+  void Insert(TaskContext task_context);
+
   /**
    * @brief Removes a pending request and placholder.
    * @param key Internal request key to remove
    * @return True on success
    */
   bool Remove(int64_t key);
 
+  /**
+   * @brief Removes a task context.
+   * @param task_context Task context.
+   */
+  void Remove(TaskContext task_context);
+
  private:
   int64_t key_ = 0;
   std::unordered_map<int64_t, client::CancellationToken> requests_map_;
+  std::unordered_set<TaskContext, TaskContextHash> task_contexts_;
   std::mutex requests_lock_;
 };