Implemented the GetData with CancellableFuture in versioned client

Implemented the GetData method which returns CancellableFuture in the dataservice-read VersionedLayerClient class.
Also, added integration tests.

Relates to: OLPEDGE-958

Signed-off-by: Bohdan Kurylovych <[email protected]>

Author: Bohdan Kurylovych

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

@@ -120,6 +120,21 @@ class DATASERVICE_READ_API VersionedLayerClient final {
   client::CancellationToken GetData(DataRequest data_request,
                                     DataResponseCallback 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 will be invoked with an empty DataResponse (nullptr for result and
+   * error). If neither Partition Id or Data Handle were set in the request, the
+   * callback will be invoked with an error with ErrorCode::InvalidRequest.
+   * @param data_request contains the complete set of request parameters.
+   * \note \c DataRequest's GetLayerId value will be ignored and the parameter
+   * from the constructor will be used instead.
+   * @return \c CancellableFuture of type \c DataResponse, which when
+   * complete will contain the data or an error. Alternatively, the
+   * \c CancellableFuture can be used to cancel this request.
+   */
+  client::CancellableFuture<DataResponse> GetData(DataRequest data_request);
+
   /**
    * @brief Fetches a list of partitions for the given generic layer
    * asynchronously.

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

@@ -39,6 +39,11 @@ client::CancellationToken VersionedLayerClient::GetData(
   return impl_->GetData(std::move(data_request), std::move(callback));
 }
 
+client::CancellableFuture<DataResponse> VersionedLayerClient::GetData(
+    DataRequest data_request) {
+  return impl_->GetData(std::move(data_request));
+}
+
 client::CancellationToken VersionedLayerClient::GetPartitions(
     PartitionsRequest partitions_request, PartitionsResponseCallback callback) {
   return impl_->GetPartitions(std::move(partitions_request),

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

@@ -147,6 +147,17 @@ client::CancellationToken VersionedLayerClientImpl::GetData(
   }
 }
 
+client::CancellableFuture<DataResponse> VersionedLayerClientImpl::GetData(
+    DataRequest data_request) {
+  auto promise = std::make_shared<std::promise<DataResponse>>();
+  auto cancel_token =
+      GetData(std::move(data_request), [promise](DataResponse response) {
+        promise->set_value(std::move(response));
+      });
+  return client::CancellableFuture<DataResponse>(std::move(cancel_token),
+                                                 std::move(promise));
+}
+
 client::CancellationToken VersionedLayerClientImpl::PrefetchTiles(
     PrefetchTilesRequest request, PrefetchTilesResponseCallback callback) {
   const int64_t request_key = pending_requests_->GenerateRequestPlaceholder();

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

@@ -55,6 +55,9 @@ class VersionedLayerClientImpl {
   virtual client::CancellationToken GetData(DataRequest data_request,
                                             DataResponseCallback callback);
 
+  virtual client::CancellableFuture<DataResponse> GetData(
+      DataRequest data_request);
+
   virtual client::CancellationToken GetPartitions(
       PartitionsRequest partitions_request,
       PartitionsResponseCallback callback);

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

@@ -366,6 +366,45 @@ TEST_F(DataserviceReadVersionedLayerClientTest, GetDataFromPartitionAsync) {
   ASSERT_NE(response.GetResult()->size(), 0u);
 }
 
+TEST_F(DataserviceReadVersionedLayerClientTest,
+       GetDataFromPartitionAsyncWithCancellableFuture) {
+  EXPECT_CALL(*network_mock_, Send(_, _, _, _, _))
+      .WillOnce(ReturnHttpResponse(olp::http::NetworkResponse().WithStatus(
+                                       olp::http::HttpStatusCode::OK),
+                                   kHttpResponseLookupQuery))
+      .WillOnce(ReturnHttpResponse(olp::http::NetworkResponse().WithStatus(
+                                       olp::http::HttpStatusCode::OK),
+                                   kHttpResponsePartition_269))
+      .WillOnce(ReturnHttpResponse(olp::http::NetworkResponse().WithStatus(
+                                       olp::http::HttpStatusCode::OK),
+                                   kHttpResponseLookupBlob))
+      .WillOnce(ReturnHttpResponse(olp::http::NetworkResponse().WithStatus(
+                                       olp::http::HttpStatusCode::OK),
+                                   kHttpResponseBlobData_269));
+
+  auto catalog = olp::client::HRN::FromString(
+      GetArgument("dataservice_read_test_catalog"));
+  auto layer = GetArgument("dataservice_read_test_layer");
+  auto version = std::stoi(GetArgument("dataservice_read_test_layer_version"));
+  auto catalog_client =
+      std::make_unique<olp::dataservice::read::VersionedLayerClient>(
+          catalog, layer, *settings_);
+
+  auto partition = GetArgument("dataservice_read_test_partition");
+  auto data_request = olp::dataservice::read::DataRequest()
+                          .WithVersion(version)
+                          .WithPartitionId(partition);
+  auto cancellable_future = catalog_client->GetData(std::move(data_request));
+
+  auto raw_future = cancellable_future.GetFuture();
+  ASSERT_NE(raw_future.wait_for(kWaitTimeout), std::future_status::timeout);
+  DataResponse response = raw_future.get();
+
+  ASSERT_TRUE(response.IsSuccessful()) << response.GetError().GetMessage();
+  ASSERT_NE(response.GetResult(), nullptr);
+  ASSERT_NE(response.GetResult()->size(), 0u);
+}
+
 TEST_F(DataserviceReadVersionedLayerClientTest, GetDataFromPartitionSync) {
   EXPECT_CALL(*network_mock_, Send(_, _, _, _, _))
       .WillOnce(ReturnHttpResponse(olp::http::NetworkResponse().WithStatus(
@@ -406,6 +445,49 @@ TEST_F(DataserviceReadVersionedLayerClientTest, GetDataFromPartitionSync) {
   ASSERT_NE(response.GetResult()->size(), 0u);
 }
 
+TEST_F(DataserviceReadVersionedLayerClientTest,
+       GetDataFromPartitionSyncWithCancellableFuture) {
+  EXPECT_CALL(*network_mock_, Send(_, _, _, _, _))
+      .WillOnce(ReturnHttpResponse(olp::http::NetworkResponse().WithStatus(
+                                       olp::http::HttpStatusCode::OK),
+                                   kHttpResponseLookupQuery))
+      .WillOnce(ReturnHttpResponse(olp::http::NetworkResponse().WithStatus(
+                                       olp::http::HttpStatusCode::OK),
+                                   kHttpResponsePartition_269))
+      .WillOnce(ReturnHttpResponse(olp::http::NetworkResponse().WithStatus(
+                                       olp::http::HttpStatusCode::OK),
+                                   kHttpResponseLookupBlob))
+      .WillOnce(ReturnHttpResponse(olp::http::NetworkResponse().WithStatus(
+                                       olp::http::HttpStatusCode::OK),
+                                   kHttpResponseBlobData_269));
+
+  auto catalog = olp::client::HRN::FromString(
+      GetArgument("dataservice_read_test_catalog"));
+  auto layer = GetArgument("dataservice_read_test_layer");
+  auto version = 0;
+
+  auto sync_settings = *settings_;
+  sync_settings.task_scheduler.reset();
+  auto catalog_client =
+      std::make_unique<olp::dataservice::read::VersionedLayerClient>(
+          catalog, layer, sync_settings);
+  ASSERT_TRUE(catalog_client);
+
+  auto partition = GetArgument("dataservice_read_test_partition");
+  auto data_request = olp::dataservice::read::DataRequest()
+                          .WithVersion(version)
+                          .WithPartitionId(partition);
+  auto cancellable_future = catalog_client->GetData(std::move(data_request));
+
+  auto raw_future = cancellable_future.GetFuture();
+  ASSERT_NE(raw_future.wait_for(kWaitTimeout), std::future_status::timeout);
+  DataResponse response = raw_future.get();
+
+  ASSERT_TRUE(response.IsSuccessful()) << response.GetError().GetMessage();
+  ASSERT_NE(response.GetResult(), nullptr);
+  ASSERT_NE(response.GetResult()->size(), 0u);
+}
+
 TEST_F(DataserviceReadVersionedLayerClientTest,
        GetDataFromPartitionSyncLatestVersionOk) {
   EXPECT_CALL(*network_mock_, Send(_, _, _, _, _))
@@ -617,6 +699,53 @@ TEST_F(DataserviceReadVersionedLayerClientTest,
   ASSERT_TRUE(response.GetResult() == nullptr);
 }
 
+TEST_F(DataserviceReadVersionedLayerClientTest,
+       GetDataFromPartitionCancelLookupWithCancellableFuture) {
+  auto wait_for_cancel = std::make_shared<std::promise<void>>();
+  auto pause_for_cancel = std::make_shared<std::promise<void>>();
+
+  olp::http::RequestId request_id;
+  NetworkCallback send_mock;
+  CancelCallback cancel_mock;
+  std::tie(request_id, send_mock, cancel_mock) = GenerateNetworkMockActions(
+      wait_for_cancel, pause_for_cancel,
+      {olp::http::HttpStatusCode::OK, kHttpResponseLookupQuery});
+
+  EXPECT_CALL(*network_mock_, Send(_, _, _, _, _))
+      .WillOnce(testing::Invoke(std::move(send_mock)));
+
+  EXPECT_CALL(*network_mock_, Cancel(_))
+      .WillOnce(testing::Invoke(std::move(cancel_mock)));
+
+  auto catalog = olp::client::HRN::FromString(
+      GetArgument("dataservice_read_test_catalog"));
+  auto layer = GetArgument("dataservice_read_test_layer");
+  auto version = std::stoi(GetArgument("dataservice_read_test_layer_version"));
+
+  auto catalog_client =
+      std::make_unique<olp::dataservice::read::VersionedLayerClient>(
+          catalog, layer, *settings_);
+  ASSERT_TRUE(catalog_client);
+
+  auto partition = GetArgument("dataservice_read_test_partition");
+  auto data_request = olp::dataservice::read::DataRequest()
+                          .WithVersion(version)
+                          .WithPartitionId(partition);
+
+  auto cancellable_future = catalog_client->GetData(std::move(data_request));
+
+  wait_for_cancel->get_future().get();
+  cancellable_future.GetCancellationToken().cancel();
+  pause_for_cancel->set_value();
+
+  auto raw_future = cancellable_future.GetFuture();
+  ASSERT_NE(raw_future.wait_for(kWaitTimeout), std::future_status::timeout);
+  DataResponse response = raw_future.get();
+
+  ASSERT_FALSE(response.IsSuccessful()) << response.GetError().GetMessage();
+  ASSERT_TRUE(response.GetResult() == nullptr);
+}
+
 TEST_F(DataserviceReadVersionedLayerClientTest,
        GetDataFromPartitionCancelPartition) {
   auto wait_for_cancel = std::make_shared<std::promise<void>>();