Add CancelPendingRequests in VersionedLayerClient

Add Add CancelPendingRequests in VersionedLayerClient and provide it
implementation. Enable test DataserviceReadVersionedLayerClientTest.CancelPendingRequestsPartitions.

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

Author: diachenko-mischa

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

@@ -106,6 +106,12 @@ class DATASERVICE_READ_API VersionedLayerClient final {
 
   ~VersionedLayerClient();
 
+  /**
+   * @brief Cancels all active and pending requests.
+   * @return True on success.
+   */
+  bool CancelPendingRequests();
+
   /**
    * @brief Fetches data for a partition ID or data handle asynchronously.
    *

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

@@ -40,6 +40,10 @@ VersionedLayerClient& VersionedLayerClient::operator=(
 
 VersionedLayerClient::~VersionedLayerClient() = default;
 
+bool VersionedLayerClient::CancelPendingRequests() {
+  return impl_->CancelPendingRequests();
+}
+
 client::CancellationToken VersionedLayerClient::GetData(
     DataRequest data_request, DataResponseCallback callback) {
   return impl_->GetData(std::move(data_request), std::move(callback));

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

@@ -39,6 +39,10 @@ namespace olp {
 namespace dataservice {
 namespace read {
 
+namespace {
+constexpr auto kLogTag = "VersionedLayerClientImpl";
+}  // namespace
+
 VersionedLayerClientImpl::VersionedLayerClientImpl(
     client::HRN catalog, std::string layer_id,
     client::OlpClientSettings settings)
@@ -80,6 +84,11 @@ VersionedLayerClientImpl::~VersionedLayerClientImpl() {
   pending_requests_->CancelAllAndWait();
 }
 
+bool VersionedLayerClientImpl::CancelPendingRequests() {
+  OLP_SDK_LOG_TRACE(kLogTag, "CancelPendingRequests");
+  return pending_requests_->CancelAll();
+}
+
 client::CancellationToken VersionedLayerClientImpl::GetPartitions(
     PartitionsRequest request, PartitionsResponseCallback callback) {
   auto schedule_get_partitions = [&](PartitionsRequest request,

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

@@ -51,6 +51,8 @@ class VersionedLayerClientImpl {
 
   virtual ~VersionedLayerClientImpl();
 
+  virtual bool CancelPendingRequests();
+
   virtual client::CancellationToken GetData(DataRequest data_request,
                                             DataResponseCallback callback);
 

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

@@ -2033,48 +2033,25 @@ TEST_F(DataserviceReadVersionedLayerClientTest, GetDataCacheWithUpdate) {
 }
 
 TEST_F(DataserviceReadVersionedLayerClientTest,
-       DISABLED_CancelPendingRequestsPartitions) {
+       CancelPendingRequestsPartitions) {
   olp::client::HRN hrn(GetTestCatalog());
-  testing::InSequence s;
-  std::vector<std::shared_ptr<std::promise<void>>> waits;
-  std::vector<std::shared_ptr<std::promise<void>>> pauses;
 
   auto client =
       std::make_unique<VersionedLayerClient>(hrn, "testlayer", *settings_);
   auto partitions_request = PartitionsRequest().WithFetchOption(OnlineOnly);
   auto data_request =
       DataRequest().WithPartitionId("269").WithFetchOption(OnlineOnly);
 
-  // Make a few requests
-  auto wait_for_cancel1 = std::make_shared<std::promise<void>>();
-  auto pause_for_cancel1 = std::make_shared<std::promise<void>>();
-  auto wait_for_cancel2 = std::make_shared<std::promise<void>>();
-  auto pause_for_cancel2 = std::make_shared<std::promise<void>>();
+  auto request_started = std::make_shared<std::promise<void>>();
+  auto continue_request = 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_cancel1, pause_for_cancel1,
-        {olp::http::HttpStatusCode::OK, HTTP_RESPONSE_LAYER_VERSIONS});
-
-    EXPECT_CALL(*network_mock_,
-                Send(IsGetRequest(URL_LAYER_VERSIONS), _, _, _, _))
-        .Times(1)
-        .WillOnce(testing::Invoke(std::move(send_mock)));
-
-    EXPECT_CALL(*network_mock_, Cancel(request_id))
-        .WillOnce(testing::Invoke(std::move(cancel_mock)));
-  }
-  {
-    olp::http::RequestId request_id;
-    NetworkCallback send_mock;
-    CancelCallback cancel_mock;
-
-    std::tie(request_id, send_mock, cancel_mock) = GenerateNetworkMockActions(
-        wait_for_cancel2, pause_for_cancel2,
+        request_started, continue_request,
         {olp::http::HttpStatusCode::OK, HTTP_RESPONSE_BLOB_DATA_269});
 
     EXPECT_CALL(*network_mock_,
@@ -2086,26 +2063,13 @@ TEST_F(DataserviceReadVersionedLayerClientTest,
         .WillOnce(testing::Invoke(std::move(cancel_mock)));
   }
 
-  waits.push_back(wait_for_cancel1);
-  pauses.push_back(pause_for_cancel1);
+  auto data_future = client->GetData(data_request);
   auto partitions_future = client->GetPartitions(partitions_request);
 
-  waits.push_back(wait_for_cancel2);
-  pauses.push_back(pause_for_cancel2);
-  auto data_future = client->GetData(data_request);
-  std::cout << "waiting" << std::endl;
-  for (auto wait : waits) {
-    wait->get_future().get();
-  }
-  std::cout << "done waitingg" << std::endl;
-  // Cancel them all
-  client.reset();
-  std::cout << "done cancelling" << std::endl;
-  for (auto pause : pauses) {
-    pause->set_value();
-  }
+  request_started->get_future().get();
+  client->CancelPendingRequests();
+  continue_request->set_value();
 
-  // Verify they are all cancelled
   PartitionsResponse partitions_response = partitions_future.GetFuture().get();
 
   ASSERT_FALSE(partitions_response.IsSuccessful())