Added the FlushRequest class in StreamLayerClient's API

Added the model::FlushRequest class, which manages the parameters for the customization of StreamLayerClient::Flush logic.
+ re-enabled the StreamLayerClientCacheTest integration tests on PSV

Relates to: OLPEDGE-826

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

Author: Bohdan Kurylovych

examples/dataservice-write/example.cpp

@@ -112,7 +112,8 @@ int RunExample() {
     }
 
     // Flush and wait for uploading
-    auto future_response = client->Flush();
+    auto flush_request = FlushRequest();
+    auto future_response = client->Flush(std::move(flush_request));
     auto responses = future_response.GetFuture().get();
     if (responses.empty()) {
       OLP_SDK_LOG_ERROR_F(kLogTag, "Error on Flush()");

olp-cpp-sdk-dataservice-write/include/olp/dataservice/write/StreamLayerClient.h

@@ -24,11 +24,11 @@
 #include <olp/core/client/ApiError.h>
 #include <olp/core/client/ApiResponse.h>
 #include <olp/core/client/OlpClientSettings.h>
-
 #include <olp/dataservice/write/DataServiceWriteApi.h>
 #include <olp/dataservice/write/FlushSettings.h>
 #include <olp/dataservice/write/generated/model/ResponseOk.h>
 #include <olp/dataservice/write/generated/model/ResponseOkSingle.h>
+#include <olp/dataservice/write/model/FlushRequest.h>
 #include <olp/dataservice/write/model/PublishDataRequest.h>
 #include <olp/dataservice/write/model/PublishSdiiRequest.h>
 
@@ -119,17 +119,25 @@ class DATASERVICE_WRITE_API StreamLayerClient {
   /**
    * @brief Flush PublishDataRequests which have been queued via the Queue
    * API.
+   * @param request \c FlushRequest object representing the parameters for
+   * this \c Flush method call.
    * @return A CancellableFuture containing the FlushResponse.
    */
-  olp::client::CancellableFuture<FlushResponse> Flush();
+  olp::client::CancellableFuture<FlushResponse> Flush(
+      model::FlushRequest request);
 
   /**
    * @brief Flush PublishDataRequests which have been queued via the Queue
    * API.
+   * @param request \c FlushRequest object representing the parameters for
+   * this \c Flush method call.
+   * @param callback The callback which will be called when all the flush
+   * results (see \c FlushResponse) will be available.
    * @return A CancellationToken which can be used to cancel the ongoing
    * request.
    */
-  olp::client::CancellationToken Flush(FlushCallback callback);
+  olp::client::CancellationToken Flush(model::FlushRequest request,
+                                       FlushCallback callback);
 
   /**
    * @brief Call to send a list of SDII messages to an OLP Stream Layer.

olp-cpp-sdk-dataservice-write/include/olp/dataservice/write/model/FlushRequest.h

@@ -0,0 +1,63 @@
+/*
+ * 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
+ */
+
+#pragma once
+
+#include <olp/dataservice/write/DataServiceWriteApi.h>
+
+namespace olp {
+namespace dataservice {
+namespace write {
+namespace model {
+/**
+ * @brief FlushRequest used to flush requests into an OLP Stream Layer.
+ */
+class DATASERVICE_WRITE_API FlushRequest {
+ public:
+  FlushRequest() = default;
+
+  /**
+   * @return The number of \c PublishDataRequest requests, which will be
+   * flushed by \c StreamLayerCLient. By default this value is initilized to 0,
+   * thus, all queued \c StreamLayerClient's requests will be flushed.
+   */
+  inline int GetNumberOfRequestsToFlush() const {
+    return num_requests_per_flush_;
+  }
+
+  /**
+   * @param num_requests Maximum number of partitions (i.e. \c
+   * PublishDataRequest) to be flushed by \c StreamLayerClient. If this value is
+   * negative, nothing will be flushed. Set 0 to flush all requests.
+   * @note Required.
+   */
+  inline FlushRequest& WithNumberOfRequestsToFlush(int num_requests) {
+    num_requests_per_flush_ = num_requests;
+    return *this;
+  }
+
+ private:
+  /// Number of requests to flush
+  int num_requests_per_flush_ = 0;
+};
+
+}  // namespace model
+}  // namespace write
+}  // namespace dataservice
+}  // namespace olp

olp-cpp-sdk-dataservice-write/src/AutoFlushController.cpp

@@ -167,8 +167,16 @@ class EnabledAutoFlushControllerImpl
     NotifyFlushEventStart();
     auto flush_thread = std::thread([self, impl_pointer]() {
       auto id = self->background_task_col_.AddTask();
-      auto cancel_token =
-          impl_pointer->Flush([self, id](FlushResponse results) {
+
+      const auto num_requests_to_flush =
+          self->flush_settings_.events_per_single_flush
+              ? *(self->flush_settings_.events_per_single_flush)
+              : 0;
+      model::FlushRequest request =
+          model::FlushRequest().WithNumberOfRequestsToFlush(
+              num_requests_to_flush);
+      auto cancel_token = impl_pointer->Flush(
+          std::move(request), [self, id](FlushResponse results) {
             self->background_task_col_.ReleaseTask(id);
             self->RemoveCancelToken(id);
             self->NotifyFlushEventResults(results);

olp-cpp-sdk-dataservice-write/src/StreamLayerClient.cpp

@@ -61,13 +61,13 @@ boost::optional<std::string> StreamLayerClient::Queue(
 }
 
 olp::client::CancellableFuture<StreamLayerClient::FlushResponse>
-StreamLayerClient::Flush() {
-  return impl_->Flush();
+StreamLayerClient::Flush(model::FlushRequest request) {
+  return impl_->Flush(std::move(request));
 }
 
 olp::client::CancellationToken StreamLayerClient::Flush(
-    FlushCallback callback) {
-  return impl_->Flush(std::move(callback));
+    model::FlushRequest request, FlushCallback callback) {
+  return impl_->Flush(std::move(request), std::move(callback));
 }
 
 olp::client::CancellableFuture<PublishSdiiResponse>

olp-cpp-sdk-dataservice-write/src/StreamLayerClientImpl.cpp

@@ -358,34 +358,34 @@ StreamLayerClientImpl::PopFromQueue() {
 }
 
 olp::client::CancellableFuture<StreamLayerClient::FlushResponse>
-StreamLayerClientImpl::Flush() {
+StreamLayerClientImpl::Flush(model::FlushRequest request) {
   auto promise =
       std::make_shared<std::promise<StreamLayerClient::FlushResponse> >();
-  auto cancel_token =
-      Flush([promise](StreamLayerClient::FlushResponse response) {
+  auto cancel_token = Flush(
+      std::move(request), [promise](StreamLayerClient::FlushResponse response) {
         promise->set_value(std::move(response));
       });
   return CancellableFuture<StreamLayerClient::FlushResponse>(cancel_token,
                                                              promise);
 }
 
 olp::client::CancellationToken StreamLayerClientImpl::Flush(
-    StreamLayerClient::FlushCallback callback) {
+    model::FlushRequest request, StreamLayerClient::FlushCallback callback) {
   CancellationContext cancel_context;
 
   auto self = shared_from_this();
 
   auto func = [=]() mutable {
     StreamLayerClient::FlushResponse response;
 
-    auto maximum_events_number = self->flush_settings_.events_per_single_flush;
-    if (maximum_events_number && *maximum_events_number <= 0) {
+    const auto maximum_events_number = request.GetNumberOfRequestsToFlush();
+    if (maximum_events_number < 0) {
       callback(std::move(response));
       return;
     }
 
     int counter = 0;
-    while ((!maximum_events_number || counter < *maximum_events_number) &&
+    while ((!maximum_events_number || counter < maximum_events_number) &&
            (self->QueueSize() > 0) && !cancel_context.IsCancelled()) {
       auto publish_request = self->PopFromQueue();
       if (publish_request == boost::none) {

olp-cpp-sdk-dataservice-write/src/StreamLayerClientImpl.h

@@ -55,9 +55,10 @@ class StreamLayerClientImpl
       const model::PublishDataRequest& request, PublishDataCallback callback);
 
   boost::optional<std::string> Queue(const model::PublishDataRequest& request);
-  olp::client::CancellableFuture<StreamLayerClient::FlushResponse> Flush();
+  olp::client::CancellableFuture<StreamLayerClient::FlushResponse> Flush(
+      model::FlushRequest request);
   olp::client::CancellationToken Flush(
-      StreamLayerClient::FlushCallback callback);
+      model::FlushRequest request, StreamLayerClient::FlushCallback callback);
   size_t QueueSize() const;
   boost::optional<model::PublishDataRequest> PopFromQueue();
 

scripts/linux/psv/travis_test_psv.sh

@@ -19,7 +19,6 @@ $CPP_TEST_SOURCE_DARASERVICE_WRITE/olp-cpp-sdk-dataservice-write-tests \
     --gtest_output="xml:olp-cpp-sdk-dataservice-write-tests-report.xml"
 echo ">>> Integration Test ... >>>"
 $CPP_TEST_SOURCE_INTEGRATION/olp-cpp-sdk-integration-tests \
-    --gtest_output="xml:olp-cpp-sdk-integration-tests-report.xml" \
-    --gtest_filter=-"StreamLayerClientCacheTest*"
+    --gtest_output="xml:olp-cpp-sdk-integration-tests-report.xml"
 
 bash <(curl -s https://codecov.io/bash)

tests/functional/olp-cpp-sdk-dataservice-write/DataserviceWriteStreamLayerClientCacheTest.cpp

@@ -232,7 +232,7 @@ TEST_F(DataserviceWriteStreamLayerClientCacheTest, FlushDataSingle) {
 
   ASSERT_FALSE(error) << error.get();
 
-  auto response = client_->Flush().GetFuture().get();
+  auto response = client_->Flush(model::FlushRequest()).GetFuture().get();
 
   ASSERT_FALSE(response.empty());
   ASSERT_NO_FATAL_FAILURE(PublishDataSuccessAssertions(response[0]));
@@ -241,7 +241,7 @@ TEST_F(DataserviceWriteStreamLayerClientCacheTest, FlushDataSingle) {
 TEST_F(DataserviceWriteStreamLayerClientCacheTest, FlushDataMultiple) {
   ASSERT_NO_FATAL_FAILURE(QueueMultipleEvents(5));
 
-  auto response = client_->Flush().GetFuture().get();
+  auto response = client_->Flush(model::FlushRequest()).GetFuture().get();
 
   ASSERT_EQ(5, response.size());
   for (auto& single_response : response) {
@@ -257,8 +257,8 @@ TEST_F(DataserviceWriteStreamLayerClientCacheTest, FlushDataSingleAsync) {
 
   std::promise<StreamLayerClient::FlushResponse> response_promise;
   bool call_is_async = true;
-  auto cancel_token =
-      client_->Flush([&](StreamLayerClient::FlushResponse response) {
+  auto cancel_token = client_->Flush(
+      model::FlushRequest(), [&](StreamLayerClient::FlushResponse response) {
         call_is_async = false;
         response_promise.set_value(response);
       });
@@ -280,8 +280,8 @@ TEST_F(DataserviceWriteStreamLayerClientCacheTest, FlushDataMultipleAsync) {
 
   std::promise<StreamLayerClient::FlushResponse> response_promise;
   bool call_is_async = true;
-  auto cancel_token =
-      client_->Flush([&](StreamLayerClient::FlushResponse response) {
+  auto cancel_token = client_->Flush(
+      model::FlushRequest(), [&](StreamLayerClient::FlushResponse response) {
         call_is_async = false;
         response_promise.set_value(response);
       });
@@ -306,7 +306,7 @@ TEST_F(DataserviceWriteStreamLayerClientCacheTest, FlushDataCancel) {
 
   ASSERT_FALSE(error) << error.get();
 
-  auto cancel_future = client_->Flush();
+  auto cancel_future = client_->Flush(model::FlushRequest());
 
   std::thread([cancel_future]() {
     std::this_thread::sleep_for(std::chrono::milliseconds(200));

tests/integration/olp-cpp-sdk-dataservice-write/StreamLayerClientCacheTest.cpp

@@ -226,7 +226,13 @@ class StreamLayerClientCacheTest : public ::testing::Test {
           PublishDataRequest().WithData(data_).WithLayerId(GetTestLayer()));
       EXPECT_FALSE(error) << error.get();
     }
-    auto response = client_->Flush().GetFuture().get();
+
+    const auto num_requests_to_flush =
+        max_events_per_flush ? *max_events_per_flush : 0;
+    auto flush_request = model::FlushRequest().WithNumberOfRequestsToFlush(
+        num_requests_to_flush);
+
+    auto response = client_->Flush(std::move(flush_request)).GetFuture().get();
     if (!max_events_per_flush || *max_events_per_flush > 5) {
       EXPECT_EQ(5, response.size());
     } else if (*max_events_per_flush <= 0) {
@@ -302,7 +308,7 @@ TEST_F(StreamLayerClientCacheTest, FlushDataSingle) {
 
   ASSERT_FALSE(error) << error.get();
 
-  auto response = client_->Flush().GetFuture().get();
+  auto response = client_->Flush(model::FlushRequest()).GetFuture().get();
 
   ASSERT_FALSE(response.empty());
   ASSERT_NO_FATAL_FAILURE(PublishDataSuccessAssertions(response[0]));
@@ -324,7 +330,7 @@ TEST_F(StreamLayerClientCacheTest, FlushDataMultiple) {
 
   ASSERT_NO_FATAL_FAILURE(QueueMultipleEvents(5));
 
-  auto response = client_->Flush().GetFuture().get();
+  auto response = client_->Flush(model::FlushRequest()).GetFuture().get();
 
   ASSERT_EQ(5, response.size());
   for (auto& single_response : response) {
@@ -360,7 +366,7 @@ TEST_F(StreamLayerClientCacheTest, DISABLED_FlushDataCancel) {
 
   ASSERT_FALSE(error) << error.get();
 
-  auto promise = client_->Flush();
+  auto promise = client_->Flush(model::FlushRequest());
   wait_for_cancel->get_future().get();
   promise.GetCancellationToken().cancel();
   pause_for_cancel->set_value();
@@ -599,7 +605,6 @@ TEST_F(StreamLayerClientCacheTest, FlushDataMaxEventsDefaultSetting) {
 TEST_F(StreamLayerClientCacheTest, FlushDataMaxEventsValidCustomSetting) {
   const int max_events_per_flush = 3;
   disk_cache_->Close();
-  flush_settings_.events_per_single_flush = max_events_per_flush;
   client_ = CreateStreamLayerClient();
   {
     testing::InSequence dummy;
@@ -621,7 +626,6 @@ TEST_F(StreamLayerClientCacheTest, FlushDataMaxEventsValidCustomSetting) {
 TEST_F(StreamLayerClientCacheTest, FlushDataMaxEventsInvalidCustomSetting) {
   const int max_events_per_flush = -3;
   disk_cache_->Close();
-  flush_settings_.events_per_single_flush = max_events_per_flush;
   client_ = CreateStreamLayerClient();
   {
     testing::InSequence dummy;
@@ -693,7 +697,7 @@ TEST_F(StreamLayerClientCacheTest, FlushSettingsMaximumRequests) {
   }
 
   QueueMultipleEvents(15);
-  auto response = client_->Flush().GetFuture().get();
+  auto response = client_->Flush(model::FlushRequest()).GetFuture().get();
 
   ASSERT_EQ(15, response.size());
   for (auto& single_response : response) {