Add CancellableFuture methods in VolatileLayerClient

Add GetData and GetPartitions methods to VolatileLayerClient
that return CancellableFuture. Add tests.

Resolves: OLPEDGE-959

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

Author: diachenko-mischa

olp-cpp-sdk-dataservice-read/include/olp/dataservice/read/VolatileLayerClient.h

@@ -111,20 +111,43 @@ class DATASERVICE_READ_API VolatileLayerClient final {
   client::CancellationToken GetPartitions(
       PartitionsRequest request, Callback<PartitionsResponse> callback);
 
+  /**
+   * @brief Fetches a list partitions for given volatile layer asynchronously.
+   * @param request contains the complete set of request parameters.
+   * @return CancellableFuture, which when complete will contain the
+   * PartitionsResponse or an error. Alternatively, the CancellableFuture can be
+   * used to cancel this request.
+   */
+  olp::client::CancellableFuture<PartitionsResponse> GetPartitions(
+      PartitionsRequest request);
+
   /**
    * @brief GetData fetches data for a partition or data handle asynchronously.
    * If the specified partition or data handle cannot be found in the layer, the
    * callback is invoked with an empty DataResponse (a nullptr result and
    * error). If Partition Id or Data Handle are not set in the request, the
    * callback is invoked with the error ErrorCode::InvalidRequest.
-   * @param request contains a complete set of request parameters.
+   * @param request Contains a complete set of request parameters.
    * @param callback is invoked once the DataResult is available, or an
    * error is encountered.
-   * @return A token that can be used to cancel this request.
+   * @return Token that can be used to cancel this request.
    */
   olp::client::CancellationToken GetData(DataRequest request,
                                          Callback<DataResponse> callback);
 
+  /**
+   * @brief Fetches data for a partition or data handle asynchronously.
+   * If the specified partition or data handle cannot be found in the layer, the
+   * callback is invoked with an empty DataResponse (a nullptr result and
+   * error). If Partition Id or Data Handle are not set in the request, the
+   * callback is invoked with the error ErrorCode::InvalidRequest.
+   * @param request contains a complete set of request parameters.
+   * @return CancellableFuture, which when complete will contain the
+   * PartitionsResponse or an error. Alternatively, the CancellableFuture can be
+   * used to cancel this request.
+   */
+  olp::client::CancellableFuture<DataResponse> GetData(DataRequest request);
+
  private:
   std::unique_ptr<VolatileLayerClientImpl> impl_;
 };

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

@@ -39,10 +39,20 @@ client::CancellationToken VolatileLayerClient::GetPartitions(
   return impl_->GetPartitions(std::move(request), std::move(callback));
 }
 
+olp::client::CancellableFuture<VolatileLayerClient::PartitionsResponse>
+VolatileLayerClient::GetPartitions(PartitionsRequest request) {
+  return impl_->GetPartitions(std::move(request));
+}
+
 client::CancellationToken VolatileLayerClient::GetData(
     DataRequest request, Callback<DataResponse> callback) {
   return impl_->GetData(std::move(request), std::move(callback));
 }
+
+olp::client::CancellableFuture<VolatileLayerClient::DataResponse>
+VolatileLayerClient::GetData(DataRequest request) {
+  return impl_->GetData(std::move(request));
+}
 }  // namespace read
 }  // namespace dataservice
 }  // namespace olp

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

@@ -22,7 +22,6 @@
 #include <olp/core/cache/DefaultCache.h>
 #include <olp/core/client/CancellationContext.h>
 #include <olp/core/client/OlpClientSettingsFactory.h>
-#include <olp/dataservice/read/PartitionsRequest.h>
 
 #include "TaskContext.h"
 #include "repositories/ApiRepository.h"
@@ -100,6 +99,16 @@ client::CancellationToken VolatileLayerClientImpl::GetPartitions(
   return token;
 }
 
+client::CancellableFuture<PartitionsResponse>
+VolatileLayerClientImpl::GetPartitions(PartitionsRequest request) {
+  auto promise = std::make_shared<std::promise<PartitionsResponse> >();
+  auto callback = [=](PartitionsResponse resp) {
+    promise->set_value(std::move(resp));
+  };
+  auto token = GetPartitions(std::move(request), std::move(callback));
+  return olp::client::CancellableFuture<PartitionsResponse>(token, promise);
+}
+
 client::CancellationToken VolatileLayerClientImpl::GetData(
     DataRequest request, Callback<DataResponse> callback) {
   auto add_task = [&](DataRequest& request, Callback<DataResponse> callback) {
@@ -141,6 +150,16 @@ client::CancellationToken VolatileLayerClientImpl::GetData(
   }
 }
 
+client::CancellableFuture<DataResponse> VolatileLayerClientImpl::GetData(
+    DataRequest request) {
+  auto promise = std::make_shared<std::promise<DataResponse> >();
+  auto callback = [=](DataResponse resp) {
+    promise->set_value(std::move(resp));
+  };
+  auto token = GetData(std::move(request), std::move(callback));
+  return olp::client::CancellableFuture<DataResponse>(token, promise);
+}
+
 }  // namespace read
 }  // namespace dataservice
 }  // namespace olp

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

@@ -63,9 +63,14 @@ class VolatileLayerClientImpl {
   client::CancellationToken GetPartitions(
       PartitionsRequest request, Callback<PartitionsResponse> callback);
 
+  client::CancellableFuture<PartitionsResponse> GetPartitions(
+      PartitionsRequest request);
+
   client::CancellationToken GetData(DataRequest request,
                                     Callback<DataResponse> callback);
 
+  client::CancellableFuture<DataResponse> GetData(DataRequest request);
+
  private:
   client::HRN catalog_;
   std::string layer_id_;

olp-cpp-sdk-dataservice-read/tests/VolatileLayerClientImplTest.cpp

@@ -168,6 +168,90 @@ TEST(VolatileLayerClientImplTest, GetData) {
   }
 }
 
+TEST(VolatileLayerClientImplTest, GetDataCancellableFuture) {
+  std::shared_ptr<NetworkMock> network_mock = std::make_shared<NetworkMock>();
+  std::shared_ptr<CacheMock> cache_mock = std::make_shared<CacheMock>();
+  olp::client::OlpClientSettings settings;
+  settings.network_request_handler = network_mock;
+  settings.cache = cache_mock;
+
+  VolatileLayerClientImpl client(kHRN, kLayerId, settings);
+
+  {
+    SCOPED_TRACE("Get Data with DataHandle");
+
+    SetupNetworkExpectation(*network_mock, kUrlLookupVolatileBlob,
+                            kHttpResponseLookupVolatileBlob,
+                            olp::http::HttpStatusCode::OK);
+
+    SetupNetworkExpectation(*network_mock, kUrlVolatileBlobData, "someData",
+                            olp::http::HttpStatusCode::OK);
+
+    auto future = client.GetData(DataRequest().WithDataHandle(kBlobDataHandle))
+                      .GetFuture();
+
+    EXPECT_EQ(future.wait_for(kTimeout), std::future_status::ready);
+
+    const auto& response = future.get();
+    ASSERT_TRUE(response.IsSuccessful());
+
+    Mock::VerifyAndClearExpectations(network_mock.get());
+  }
+
+  {
+    SCOPED_TRACE("Get Data with PartitionId");
+
+    SetupNetworkExpectation(*network_mock, kUrlLookupQuery,
+                            kHttpResponseLookupQuery,
+                            olp::http::HttpStatusCode::OK);
+
+    SetupNetworkExpectation(*network_mock, kUrlQueryPartition269,
+                            kHttpResponsePartition269,
+                            olp::http::HttpStatusCode::OK);
+
+    SetupNetworkExpectation(*network_mock, kUrlLookupVolatileBlob,
+                            kHttpResponseLookupVolatileBlob,
+                            olp::http::HttpStatusCode::OK);
+
+    SetupNetworkExpectation(*network_mock, kUrlVolatileBlobData, "someData",
+                            olp::http::HttpStatusCode::OK);
+
+    auto future =
+        client.GetData(DataRequest().WithPartitionId(kPartitionId)).GetFuture();
+
+    EXPECT_EQ(future.wait_for(kTimeout), std::future_status::ready);
+
+    const auto& response = future.get();
+    ASSERT_TRUE(response.IsSuccessful());
+
+    Mock::VerifyAndClearExpectations(network_mock.get());
+  }
+
+  {
+    SCOPED_TRACE("Get Data from non existent partition");
+
+    SetupNetworkExpectation(*network_mock, kUrlLookupQuery,
+                            kHttpResponseLookupQuery,
+                            olp::http::HttpStatusCode::OK);
+
+    SetupNetworkExpectation(*network_mock, kUrlQueryPartition269,
+                            kHttpResponseNoPartition,
+                            olp::http::HttpStatusCode::OK);
+
+    auto future =
+        client.GetData(DataRequest().WithPartitionId(kPartitionId)).GetFuture();
+
+    EXPECT_EQ(future.wait_for(kTimeout), std::future_status::ready);
+
+    const auto& response = future.get();
+    ASSERT_FALSE(response.IsSuccessful());
+    EXPECT_EQ(response.GetError().GetErrorCode(),
+              olp::client::ErrorCode::NotFound);
+
+    Mock::VerifyAndClearExpectations(network_mock.get());
+  }
+}
+
 TEST(VolatileLayerClientImplTest, GetDataCancelOnClientDestroy) {
   std::shared_ptr<NetworkMock> network_mock = std::make_shared<NetworkMock>();
   std::shared_ptr<CacheMock> cache_mock = std::make_shared<CacheMock>();
@@ -201,4 +285,28 @@ TEST(VolatileLayerClientImplTest, GetDataCancelOnClientDestroy) {
   }
 }
 
+TEST(VolatileLayerClientImplTest, GetDataCancellableFutureCancel) {
+  std::shared_ptr<NetworkMock> network_mock = std::make_shared<NetworkMock>();
+  std::shared_ptr<CacheMock> cache_mock = std::make_shared<CacheMock>();
+  olp::client::OlpClientSettings settings;
+  settings.network_request_handler = network_mock;
+  settings.cache = cache_mock;
+  settings.task_scheduler =
+      olp::client::OlpClientSettingsFactory::CreateDefaultTaskScheduler(1);
+  VolatileLayerClientImpl client(kHRN, kLayerId, std::move(settings));
+
+  auto cancellable =
+      client.GetData(DataRequest().WithPartitionId(kPartitionId));
+
+  auto data_future = cancellable.GetFuture();
+  cancellable.GetCancellationToken().cancel();
+  ASSERT_EQ(data_future.wait_for(kTimeout), std::future_status::ready);
+
+  auto data_response = data_future.get();
+
+  // Callback must be called during client destructor.
+  EXPECT_FALSE(data_response.IsSuccessful());
+  EXPECT_EQ(data_response.GetError().GetErrorCode(),
+            olp::client::ErrorCode::Cancelled);
+}
 }  // namespace

tests/integration/olp-cpp-sdk-dataservice-read/VolatileLayerClientTest.cpp

@@ -295,6 +295,52 @@ TEST_F(DataserviceReadVolatileLayerClientTest, GetPartitions) {
   ASSERT_EQ(4u, partitions_response.GetResult().GetPartitions().size());
 }
 
+TEST_F(DataserviceReadVolatileLayerClientTest, GetPartitionsCancellableFuture) {
+  olp::client::HRN hrn(GetTestCatalog());
+
+  EXPECT_CALL(*network_mock_, Send(IsGetRequest(URL_CONFIG), _, _, _, _))
+      .Times(1);
+
+  olp::dataservice::read::VolatileLayerClient client(hrn, "testlayer",
+                                                     settings_);
+
+  auto request = olp::dataservice::read::PartitionsRequest();
+  auto cancellable = client.GetPartitions(request);
+  auto future = cancellable.GetFuture();
+
+  ASSERT_EQ(std::future_status::ready, future.wait_for(kTimeout));
+
+  auto response = future.get();
+  ASSERT_TRUE(response.IsSuccessful()) << ApiErrorToString(response.GetError());
+  ASSERT_EQ(4u, response.GetResult().GetPartitions().size());
+}
+
+TEST_F(DataserviceReadVolatileLayerClientTest,
+       GetPartitionsCancellableFutureCancellation) {
+  olp::client::HRN hrn(GetTestCatalog());
+
+  settings_.task_scheduler =
+      olp::client::OlpClientSettingsFactory::CreateDefaultTaskScheduler(1);
+  // Simulate a loaded queue
+  settings_.task_scheduler->ScheduleTask(
+      []() { std::this_thread::sleep_for(std::chrono::seconds(1)); });
+
+  olp::dataservice::read::VolatileLayerClient client(hrn, "testlayer",
+                                                     settings_);
+
+  auto request = olp::dataservice::read::PartitionsRequest();
+  auto cancellable = client.GetPartitions(request);
+  auto future = cancellable.GetFuture();
+
+  cancellable.GetCancellationToken().cancel();
+  ASSERT_EQ(std::future_status::ready, future.wait_for(kTimeout));
+
+  auto response = future.get();
+  EXPECT_FALSE(response.IsSuccessful());
+  EXPECT_EQ(response.GetError().GetErrorCode(),
+            olp::client::ErrorCode::Cancelled);
+}
+
 TEST_F(DataserviceReadVolatileLayerClientTest, GetEmptyPartitions) {
   olp::client::HRN hrn(GetTestCatalog());