Processes batching and event threshold (#117)

* initial commit

* config tests

* checking if linting caused an error in windows

* logs

* name changes

* Update example_integration.cpp

* Update example_integration.cpp

---------

Co-authored-by: ArtursKadikis <[email protected]>

Author: turtledreams

CHANGELOG.md

@@ -1,61 +1,64 @@
+## xx.xx.x
+- Request queue processing now is limited to 100 requests at a time
+
 ## 22.09.1
-* Mitigated a problem that caused invalid pointer error if the database path was set wrong 
+- Mitigated a problem that caused invalid pointer error if the database path was set wrong 
 
 ## 22.09.0
-* ! Minor breaking change ! SDK configuration now can't be changed after initialization/start
-* Added a persistent requests queue when building the SDK with the 'COUNTLY_USE_SQLITE' flag
-* Fixed a bug where view's name was being overriden by segmentation provided.
+- ! Minor breaking change ! SDK configuration now can't be changed after initialization/start
+- Added a persistent requests queue when building the SDK with the 'COUNTLY_USE_SQLITE' flag
+- Fixed a bug where view's name was being overriden by segmentation provided.
 
 ## 22.06.4
-* Fixed a bug where the SDK 'mutex' was being locked twice when built with the 'COUNTLY_USE_SQLITE' flag.
+- Fixed a bug where the SDK 'mutex' was being locked twice when built with the 'COUNTLY_USE_SQLITE' flag.
 
 ## 22.06.3
-* Fixed a bug where empty metrics were sent with session begin request.
+- Fixed a bug where empty metrics were sent with session begin request.
 
 ## 22.06.2
-* Fixed a bug where metrics were not sent with session begin request.
+- Fixed a bug where metrics were not sent with session begin request.
 
 ## 22.06.1
-* !! Major breaking change !! We are removing the 'LogLevel' enum out of the 'Countly' class which will change how that enum can be referenced. 'Countly::LogLevel' will not work, you will have to use 'cly::LogLevel' instead.
-* Added functionality to record crash.
-* Added ability to record breadcrumbs for crash recording.
+- !! Major breaking change !! We are removing the 'LogLevel' enum out of the 'Countly' class which will change how that enum can be referenced. 'Countly::LogLevel' will not work, you will have to use 'cly::LogLevel' instead.
+- Added functionality to record crash.
+- Added ability to record breadcrumbs for crash recording.
 
 ## 22.06.0
-* !! Major breaking change !! We are adding the 'cly' namespace on 'Countly' class which will change how that class can be referenced. 'Countly::' will not work, you will have to use 'cly::Countly::' instead.
-* !! Major breaking change !! We are extracting the 'Event' class out of the 'Countly' class which will change how that class can be referenced. 'Countly::Event' will not work, you will have to use 'cly::Event' instead.
-* !! Major breaking change !! Increased the compiler version required to compile the SDK. It's increased from version C++11 to C++14.
-* Making network requests has been reworked. They will now be sent on a separate thread. Requests will also be added in an internal queue and will be sent one at a time.
-* When making network requests, the SDK will now unlock it's mutex.
-* Fixed a bug that caused an exception on windows when encoding data that contains special characters. 
+- !! Major breaking change !! We are adding the 'cly' namespace on 'Countly' class which will change how that class can be referenced. 'Countly::' will not work, you will have to use 'cly::Countly::' instead.
+- !! Major breaking change !! We are extracting the 'Event' class out of the 'Countly' class which will change how that class can be referenced. 'Countly::Event' will not work, you will have to use 'cly::Event' instead.
+- !! Major breaking change !! Increased the compiler version required to compile the SDK. It's increased from version C++11 to C++14.
+- Making network requests has been reworked. They will now be sent on a separate thread. Requests will also be added in an internal queue and will be sent one at a time.
+- When making network requests, the SDK will now unlock it's mutex.
+- Fixed a bug that caused an exception on windows when encoding data that contains special characters. 
 
 ## 22.02.0
-* Added 10-second time-outs for all windows HTTP transactions.
-* Added ability to record views.
+- Added 10-second time-outs for all windows HTTP transactions.
+- Added ability to record views.
 
 ## 21.11.3
-* Added functionality to set custom SHA256.
+- Added functionality to set custom SHA256.
 
 ## 21.11.2
-* Fixed a bug that occurred after trying to erase events from the SQLite database when there were none.
-* Fixed a bug with the checksum calculation.
+- Fixed a bug that occurred after trying to erase events from the SQLite database when there were none.
+- Fixed a bug with the checksum calculation.
 
 ## 21.11.1
-* !! Major breaking change !! Fixed a bug that triggered when providing segmentation to the "RecordEvent" call. Previously, by mistake, every segmentation value was parsed as a JSON and threw an exception when it wasn't a valid JSON string. 
+- !! Major breaking change !! Fixed a bug that triggered when providing segmentation to the "RecordEvent" call. Previously, by mistake, every segmentation value was parsed as a JSON and threw an exception when it wasn't a valid JSON string. 
 Now this will not be the case and every String value can be provided. This is marked as a "major breaking change" in case some integrations were adding workarounds to this issue.
-* ! Minor breaking change ! Default automatic session update duration changed to 60 seconds.
-* Added a call to change the automatic session update duration.
-* Fixed a bug where session duration was reported wrong.
-* Fixed a bug that caused an exception when the application quit. This was due to the SDK attempting to send an end session request.
-* Fixed an issue with the custom HTTP client function pointer by setting it's default value.
-* Fixed a bug that caused GET requests to fail on Linux.
-* Fixed bug when changing device id with server merge.
-* Fixed bug when device id was changed without server merge. Previously the new session was started with the old device ID and not the new one.
-* Fixed a bug that was a typo ('COUNTLY_CUSTOM_HTTP' instead of 'COUNTLY_USE_CUSTOM_HTTP') in the cmake file that cause the SDK to be misconfigured. 
-* Fixed a bug that caused POST requests to fail on Windows.
-* Fixed issues with location requests.
-* Deprecated old location calls and introduced a new location call
+- ! Minor breaking change ! Default automatic session update duration changed to 60 seconds.
+- Added a call to change the automatic session update duration.
+- Fixed a bug where session duration was reported wrong.
+- Fixed a bug that caused an exception when the application quit. This was due to the SDK attempting to send an end session request.
+- Fixed an issue with the custom HTTP client function pointer by setting it's default value.
+- Fixed a bug that caused GET requests to fail on Linux.
+- Fixed bug when changing device id with server merge.
+- Fixed bug when device id was changed without server merge. Previously the new session was started with the old device ID and not the new one.
+- Fixed a bug that was a typo ('COUNTLY_CUSTOM_HTTP' instead of 'COUNTLY_USE_CUSTOM_HTTP') in the cmake file that cause the SDK to be misconfigured. 
+- Fixed a bug that caused POST requests to fail on Windows.
+- Fixed issues with location requests.
+- Deprecated old location calls and introduced a new location call
 
 ## 21.11.0
-* Fixed session duration issue.
-* Added functionality to report event duration manually.
-* 'startOnCloud' in 'Countly' is deprecated and this is going to be removed in the future.
+- Fixed session duration issue.
+- Added functionality to report event duration manually.
+- 'startOnCloud' in 'Countly' is deprecated and this is going to be removed in the future.

examples/example_integration.cpp

@@ -33,17 +33,25 @@ void printLog(LogLevel level, const string &msg) {
 int main() {
   cout << "Sample App" << endl;
   Countly &ct = Countly::getInstance();
+  // All configurations below are put here as an example
+  // Your configuration in your app may be different
+  // Please refer to the documentation for more information:
+  // https://support.count.ly/hc/en-us/articles/4416163384857-C-
+
   ct.alwaysUsePost(true);
   ct.setLogger(printLog);
   ct.SetPath("databaseFileName.db"); // this will be only built into account if the correct configurations are set
   ct.setDeviceID("test-device-id");
   // ct.setSalt("test-salt");
   // OS, OS_version, device, resolution, carrier, app_version);
   ct.SetMetrics("Windows 10", "10.22", "Mac", "800x600", "Carrier", "1.0");
-  // Server and port
+
+  // start the SDK (initialize the SDK)
   ct.start("YOUR_APP_KEY", "https://try.count.ly", 443, true);
+
   ct.SetMaxEventsPerMessage(10);
-  ct.setAutomaticSessionUpdateInterval(5);
+  ct.setAutomaticSessionUpdateInterval(5);// The value is set so low just for internal validation.
+  ct.setMaxRQProcessingBatchSize(2); // in most cases not needed to be set. The value is set so low just for internal validation
 
   ct.crash().addBreadcrumb("start");
 
@@ -158,4 +166,4 @@ int main() {
   ct.stop();
 
   return 0;
-}
+}

include/countly.hpp

@@ -48,6 +48,8 @@ class Countly : public cly::CountlyDelegates {
 
   void setMaxRequestQueueSize(unsigned int requestQueueSize);
 
+  void setMaxRQProcessingBatchSize(unsigned int requestQueueProcessingSize);
+
   void setSalt(const std::string &value);
 
   void setLogger(void (*fun)(LogLevel level, const std::string &message));

include/countly/countly_configuration.hpp

@@ -50,6 +50,11 @@ struct CountlyConfiguration {
    */
   unsigned int requestQueueThreshold = 1000;
 
+  /**
+   * Set limit for the number of requests that can be processed at a time.
+   */
+  unsigned int maxProcessingBatchSize = 100;
+
   /**
    * Set the maximum amount of breadcrumbs.
    */

src/countly.cpp

@@ -62,6 +62,17 @@ void Countly::setMaxRequestQueueSize(unsigned int requestQueueSize) {
   mutex->unlock();
 }
 
+/**
+ * Set limit for the number of requests that can be processed at a time.
+ * If the limit is reached, the rest of the requests will be processed in the next cycle.
+ * @param batchSize: max size of requests to process at a time
+ */
+void Countly::setMaxRQProcessingBatchSize(unsigned int batchSize) {
+  mutex->lock();
+  configuration->maxProcessingBatchSize = batchSize;
+  mutex->unlock();
+}
+
 void Countly::alwaysUsePost(bool value) {
   if (is_sdk_initialized) {
     log(LogLevel::WARNING, "[Countly][alwaysUsePost] You can not set the http method after SDK initialization.");
@@ -603,6 +614,7 @@ void Countly::flushEvents(std::chrono::seconds timeout) {
     log(LogLevel::FATAL, log_message.str());
   }
 }
+
 #ifdef COUNTLY_BUILD_TESTS
 std::vector<std::string> Countly::debugReturnStateOfEQ() {
   try {
@@ -729,7 +741,7 @@ bool Countly::updateSession() {
 #else
     if (database_path.empty()) {
       mutex->unlock();
-      log(LogLevel::FATAL, "Cannot fetch events, sqlite database path is not set");
+      log(LogLevel::FATAL, "Cannot fetch events, sqlite database path is not set.");
       return false;
     }
 
@@ -757,14 +769,16 @@ bool Countly::updateSession() {
           events.push_back(nlohmann::json::parse(table[(event_index * column_count) + 1]));
         }
 
-        log(LogLevel::DEBUG, "[Countly][updateSession] events count = " + events.size());
+        log(LogLevel::DEBUG, "[Countly][updateSession] events count = " + std::to_string(events.size()));
 
         event_id_stream.seekp(-1, event_id_stream.cur);
         event_id_stream << ')';
         event_ids = event_id_stream.str();
       } else if (return_value != SQLITE_OK) {
         log(LogLevel::ERROR, error_message);
         sqlite3_free(error_message);
+      } else {
+        log(LogLevel::DEBUG, "[Countly][updateSession] no events detected at the storage.");
       }
       sqlite3_free_table(table);
     }

src/request_module.cpp

@@ -122,21 +122,27 @@ void RequestModule::processQueue(std::shared_ptr<std::mutex> mutex) {
   impl->is_queue_being_processed = true;
   mutex->unlock();
 
+  // this counter is used to make sure that we don't get stuck in an infinite/long loop of request processing
+  int processedRequestsCounter = 0;
+
   while (true) {
     mutex->lock();
+    impl->_logger->log(LogLevel::DEBUG, cly::utils::format_string("[RequestModule] processQueue: Processing the request queue."));
     if (impl->_storageModule->RQCount() == 0) {
+      impl->_logger->log(LogLevel::DEBUG, cly::utils::format_string("[RequestModule] processQueue: Queue is empty."));
+
       // stop sending requests once the queue is empty
       mutex->unlock();
       break;
     }
 
-
     std::shared_ptr<DataEntry> data = impl->_storageModule->RQPeekFront();
     mutex->unlock();
     HTTPResponse response = sendHTTP("/i", data->getData());
 
     mutex->lock();
     if (!response.success) {
+      impl->_logger->log(LogLevel::DEBUG, cly::utils::format_string("[RequestModule] processQueue: Failed to deliver to server, will try again later."));
       // if the request was not a success, abort sending and try again in the future
       mutex->unlock();
       break;
@@ -145,6 +151,13 @@ void RequestModule::processQueue(std::shared_ptr<std::mutex> mutex) {
     // we pop the front only if it is still the same request
     // the queue might have changed while we were sending the request
     impl->_storageModule->RQRemoveFront(data);
+    processedRequestsCounter++;
+
+    if (processedRequestsCounter > impl->_configuration->maxProcessingBatchSize) {
+      impl->_logger->log(LogLevel::DEBUG, cly::utils::format_string("[RequestModule] processQueue: Batch limit has been reached, will do next batch later."));
+      mutex->unlock();
+      break;
+    }
 
     mutex->unlock();
   }
@@ -328,4 +341,4 @@ HTTPResponse RequestModule::sendHTTP(std::string path, std::string data) {
   return response;
 #endif
 }
-} // namespace cly
+} // namespace cly

src/storage_module_memory.cpp

@@ -53,7 +53,7 @@ long long StorageModuleMemory::RQCount() {
   }
 
   long long size = request_queue.size();
-  _logger->log(LogLevel::DEBUG, "[Countly][StorageModuleMemory] RQCount size = " + size);
+  _logger->log(LogLevel::DEBUG, "[Countly][StorageModuleMemory] RQCount size = [" + std::to_string(size) + "]");
   return size;
 }
 

tests/config.cpp

@@ -47,6 +47,7 @@ TEST_CASE("Validate setting configuration values") {
     CHECK(config.sessionDuration == 60);
     CHECK(config.eventQueueThreshold == 100);
     CHECK(config.requestQueueThreshold == 1000);
+    CHECK(config.maxProcessingBatchSize == 100);
     CHECK(config.breadcrumbsThreshold == 100);
     CHECK(config.forcePost == false);
     CHECK(config.port == 443);
@@ -74,6 +75,7 @@ TEST_CASE("Validate setting configuration values") {
     ct.setSalt("salt");
     ct.setMaxRequestQueueSize(10);
     ct.SetPath(TEST_DATABASE_NAME);
+    ct.setMaxRQProcessingBatchSize(10);
     ct.start("YOUR_APP_KEY", "https://try.count.ly", 443, false);
 
     // Get configuration values using Countly getters
@@ -88,6 +90,7 @@ TEST_CASE("Validate setting configuration values") {
     CHECK(config.sessionDuration == 5);
     CHECK(config.eventQueueThreshold == 10);
     CHECK(config.requestQueueThreshold == 10);
+    CHECK(config.maxProcessingBatchSize == 10);
     CHECK(config.breadcrumbsThreshold == 100);
     CHECK(config.forcePost == true);
     CHECK(config.port == 443);
@@ -127,6 +130,7 @@ TEST_CASE("Validate setting configuration values") {
     ct.setAutomaticSessionUpdateInterval(5);
     ct.setSalt("salt");
     ct.setMaxRequestQueueSize(10);
+    ct.setMaxRQProcessingBatchSize(10);
     ct.SetPath(TEST_DATABASE_NAME);
 
     // Server and port
@@ -145,6 +149,7 @@ TEST_CASE("Validate setting configuration values") {
     CHECK(config.sessionDuration == 5);
     CHECK(config.eventQueueThreshold == 10);
     CHECK(config.requestQueueThreshold == 10);
+    CHECK(config.maxProcessingBatchSize == 10);
     CHECK(config.breadcrumbsThreshold == 100);
     CHECK(config.forcePost == true);
     CHECK(config.port == 443);
@@ -168,6 +173,7 @@ TEST_CASE("Validate setting configuration values") {
     ct.setHTTPClient(new_CustomClient);
     ct.SetMetrics("Windows 11", "10", "p", "800x800", "Car", "1.1");
 
+    ct.setMaxRQProcessingBatchSize(50);
     ct.SetMaxEventsPerMessage(100);
     ct.setAutomaticSessionUpdateInterval(50);
     ct.setSalt("new-salt");
@@ -186,6 +192,7 @@ TEST_CASE("Validate setting configuration values") {
     CHECK(config.sessionDuration == 5);
     CHECK(config.eventQueueThreshold == 10);
     CHECK(config.requestQueueThreshold == 10);
+    CHECK(config.maxProcessingBatchSize == 50); // this one should be changed after init
     CHECK(config.breadcrumbsThreshold == 100);
     CHECK(config.forcePost == true);
     CHECK(config.port == 443);