Add TaskContinuation tests/change cancellation (#1307)

Add unit tests that check asynchronous behaviour of
TaskContinuation.
Add clearing a callback and tasks on cancellation.
Fix calling SetError in case a continuation is cancelled.
Change the mechanism to cover cancellation cases(cancel before/
after run) in case of calling SetError.
Simplify usage of promise/future.

Relates-To: OLPEDGE-2721

Signed-off-by: Yevhenii Dudnyk <[email protected]>

Author: dudnyk

olp-cpp-sdk-core/include/olp/core/thread/Continuation.h

@@ -99,13 +99,6 @@ class CORE_API ContinuationImpl final {
    */
   const ExecutionContext& GetExecutionContext() const;
 
-  /**
-   * @brief Gets the `CancellationContext` object.
-   *
-   * @return The `CancellationContext` instance.
-   */
-  const client::CancellationContext& GetContext() const;
-
   /**
    * @brief Checks whether the `CancellationContext` instance is cancelled.
    *

olp-cpp-sdk-core/include/olp/core/thread/Continuation.inl

@@ -29,29 +29,48 @@ class TaskScheduler;
 
 template <typename ResultType>
 void Continuation<ResultType>::Run() {
+  // Run should not start an execution if the `Finally` method is not called
+  // for the task continuation.
   if (!finally_callback_) {
     impl_.Clear();
     return;
   }
 
-  impl_.SetFailedCallback(
-      [this](client::ApiError error) { finally_callback_(std::move(error)); });
-
+  // In case of the continuation is cancelled before calling the `Run` method,
+  // return the `Cancelled` error.
   if (impl_.Cancelled()) {
-    finally_callback_(client::ApiError::Cancelled());
+    const auto finally_callback = std::move(finally_callback_);
+    if (finally_callback) {
+      finally_callback(client::ApiError::Cancelled());
+    }
+
     impl_.Clear();
-    finally_callback_ = nullptr;
     return;
   }
 
-  impl_.Run([this](void* input, bool cancelled) {
-    if (cancelled) {
-      finally_callback_(client::ApiError::Cancelled());
-    } else {
-      auto value = std::move(*static_cast<ResultType*>(input));
-      finally_callback_(std::move(value));
+  // Use std::move to set the `finally_callback_` member to nullptr.
+  // It's needed to check if the execution is done, and prevent calling
+  // the `Cancel` method of the `Continuation` object.
+  // The `SetError` method of the `ExecutionContext` should not execute a
+  // callback as well.
+  impl_.SetFailedCallback([=](client::ApiError error) {
+    const auto finally_callback = std::move(finally_callback_);
+    if (finally_callback) {
+      finally_callback(std::move(error));
+    }
+  });
+
+  impl_.Run([=](void* input, bool cancelled) {
+    impl_.Clear();
+
+    const auto finally_callback = std::move(finally_callback_);
+    if (finally_callback) {
+      if (cancelled) {
+        finally_callback(client::ApiError::Cancelled());
+      } else {
+        finally_callback(std::move(*static_cast<ResultType*>(input)));
+      }
     }
-    finally_callback_ = nullptr;
   });
 }
 
@@ -104,21 +123,19 @@ Continuation<ResultType>::Then(std::function<void(ExecutionContext, ResultType,
                                                   std::function<void(NewType)>)>
                                    task) {
   using NewResultType = internal::RemoveRefAndConst<NewType>;
-  auto context = impl_.GetExecutionContext();
-
-  return impl_.Then({[=](void* input, CallbackType callback) {
-                       auto in = *static_cast<ResultType*>(input);
-                       task(std::move(context), std::move(in),
-                            [callback, context](NewResultType arg) {
-                              callback(static_cast<void*>(&arg));
-                            });
-                     },
-                     [](void* input) {
-                       auto in = *static_cast<NewResultType*>(input);
-                       auto result =
-                           std::make_unique<NewResultType>(std::move(in));
-                       return internal::MakeUntypedUnique(std::move(result));
-                     }});
+  const auto context = impl_.GetExecutionContext();
+
+  return impl_.Then(
+      {[=](void* input, CallbackType callback) {
+         auto in = *static_cast<ResultType*>(input);
+         task(std::move(context), std::move(in),
+              [=](NewResultType arg) { callback(static_cast<void*>(&arg)); });
+       },
+       [](void* input) {
+         auto in = *static_cast<NewResultType*>(input);
+         auto result = std::make_unique<NewResultType>(std::move(in));
+         return internal::MakeUntypedUnique(std::move(result));
+       }});
 }
 
 template <typename ResultType>
@@ -127,9 +144,9 @@ client::CancellationToken Continuation<ResultType>::CancelToken() {
     return client::CancellationToken();
   }
 
-  auto context = impl_.GetContext();
+  auto context = impl_.GetExecutionContext();
   return client::CancellationToken(
-      [context]() mutable { context.CancelOperation(); });
+      [=]() mutable { context.CancelOperation(); });
 }
 
 template <typename ResultType>

olp-cpp-sdk-core/src/thread/Continuation.cpp

@@ -114,7 +114,7 @@ class Processor {
 
   // creates a task for execution by TaskContinuation
   Task CreateFollowingTask(const std::shared_ptr<ProcessorInternal>& context) {
-    return [this, context] { ProcessNextTask(context); };
+    return [=] { ProcessNextTask(context); };
   }
 
   // starts the first task in a chain
@@ -148,10 +148,6 @@ class Processor {
     // checks whether the Continuation is cancelled
     bool IsCancelled() const { return public_execution_context_.Cancelled(); }
 
-    const ExecutionContext& GetExecutionContext() {
-      return public_execution_context_;
-    }
-
     // An input argument used by the current task in the queue, it's the result
     // of the previous task
     void* LastOutput() const {
@@ -193,10 +189,6 @@ const ExecutionContext& ContinuationImpl::GetExecutionContext() const {
   return execution_context_;
 }
 
-const client::CancellationContext& ContinuationImpl::GetContext() const {
-  return execution_context_.GetContext();
-}
-
 bool ContinuationImpl::Cancelled() const {
   return execution_context_.Cancelled();
 }