EQ batch tests + feature (#120)

* 1

* name changes and linting

* tests

* changes

* new additions

* tabb

* changes

* wordings

* c

* fetch all

* divided tests

* todo

* todos

Author: turtledreams

CHANGELOG.md

@@ -1,13 +1,14 @@
 ## xx.xx.x
 - Request queue processing now is limited to 100 requests at a time
+- Added 'setEventsToRQThreshold' method that sets the number of events after which all events will be sent to the RQ. Default value is set to 100.
 
 ## 22.09.1
 - 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.
+- Fixed a bug where view's name was being overridden 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.

CMakeLists.txt

@@ -108,6 +108,7 @@ if(COUNTLY_BUILD_TESTS)
     ${CMAKE_CURRENT_SOURCE_DIR}/tests/views.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/tests/session.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/tests/storage.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/tests/event_queue.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/tests/storage_db.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/tests/event.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/tests/crash.cpp

README.md

@@ -44,7 +44,6 @@ int main(int argc, char *argv[]) {
 	ct.setCustomUserDetails({{"Account Type", "Basic"}, {"Employer", "Company4"}});
 	// Server and port
 	ct.Start("abf2034f975393fa994d1cf8adf9a93e4a29ac29", "https://myserver.com", 403);
-	ct.SetMaxEventsPerMessage(40);
 	ct.SetMinUpdatePeriod(2000);
 
 	ct.RecordEvent("MyCustomEvent", 123);

examples/example_integration.cpp

@@ -49,7 +49,6 @@ int main() {
   // start the SDK (initialize the SDK)
   ct.start("YOUR_APP_KEY", "https://try.count.ly", 443, true);
 
-  ct.SetMaxEventsPerMessage(10);
   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
 

include/countly.hpp

@@ -101,10 +101,20 @@ class Countly : public cly::CountlyDelegates {
   void addEvent(const cly::Event &event);
 
   /*
+   * Checks and returns the size of the event queue in memory or persistent storage.
+   */
+  int checkEQSize();
+
+  /**
    * Checks and returns the size of the event queue in persistent storage.
    */
   int checkPersistentEQSize();
 
+  /**
+   * Checks and returns the size of the event queue in memory.
+   */
+  int checkMemoryEQSize();
+
   /*
    * Erases/clears the event queue in persistent storage.
    */
@@ -131,10 +141,34 @@ class Countly : public cly::CountlyDelegates {
    */
   void fillEventsIntoJson(nlohmann::json &events, std::string &event_ids);
 
+  /*
+   * Sends the given events to the request queue.
+   */
+  void sendEventsToRQ(const nlohmann::json &events);
+
+  /*
+   * Checks if the event queue is empty.
+   */
+  bool isEQEmpty();
+
+  /*
+   * Checks if the event queue size is over the threshold and send the events to the RQ if the threshold is reached.
+   */
+  void checkAndSendEventToRQ();
+
   void addEventToSqlite(const cly::Event &event);
 
+  /**
+   * @deprecated use setEventsToRQThreshold instead
+   */
   void setMaxEvents(size_t value);
 
+  /*
+   * Sets the number of events after which all events will be sent to the RQ.
+   * Minimum value is 1. Default value is 100. Maximum value is 10000.
+   */
+  void setEventsToRQThreshold(int value);
+
   void flushEvents(std::chrono::seconds timeout = std::chrono::seconds(30));
 
   bool beginSession();
@@ -169,6 +203,9 @@ class Countly : public cly::CountlyDelegates {
 
   void SetMetrics(const std::string &os, const std::string &os_version, const std::string &device, const std::string &resolution, const std::string &carrier, const std::string &app_version) { setMetrics(os, os_version, device, resolution, carrier, app_version); }
 
+  /*
+   * @deprecated use setEventsToRQThreshold instead
+   */
   void SetMaxEventsPerMessage(int maxEvents) { setMaxEvents(maxEvents); }
 
   void SetMinUpdatePeriod(int minUpdateMillis) { setUpdateInterval(minUpdateMillis); }

include/countly/countly_configuration.hpp

@@ -42,7 +42,7 @@ struct CountlyConfiguration {
   /**
    * Set threshold value for the number of events that can be stored locally.
    */
-  unsigned int eventQueueThreshold = 100;
+  int eventQueueThreshold = 100;
 
   /**
    * Set limit for the number of requests that can be stored locally.

src/countly.cpp

@@ -474,37 +474,63 @@ void Countly::setUpdateInterval(size_t milliseconds) {
 void Countly::addEvent(const cly::Event &event) {
   mutex->lock();
 #ifndef COUNTLY_USE_SQLITE
-  if (event_queue.size() == configuration->eventQueueThreshold) {
-    log(LogLevel::WARNING, "Event queue is full, dropping the oldest event to insert a new one");
-    event_queue.pop_front();
-  }
   event_queue.push_back(event.serialize());
 #else
   addEventToSqlite(event);
+#endif
+  checkAndSendEventToRQ();
+  mutex->unlock();
+}
+
+void Countly::checkAndSendEventToRQ() {
+  nlohmann::json events = nlohmann::json::array();
   mutex->unlock();
-  int queueSize = checkPersistentEQSize();
+  int queueSize = checkEQSize();
   mutex->lock();
+#ifdef COUNTLY_USE_SQLITE
   if (queueSize >= configuration->eventQueueThreshold) {
     log(LogLevel::DEBUG, "Event queue threshold is reached");
+    std::string event_ids;
+
+    // fetch events up to the threshold from the database
+    fillEventsIntoJson(events, event_ids);
+    // send them to request queue
+    sendEventsToRQ(events);
+    // remove them from database
+    removeEventWithId(event_ids);
+  }
+#else
+  if (queueSize >= configuration->eventQueueThreshold) {
+    log(LogLevel::WARNING, "Event queue is full, dropping the oldest event to insert a new one");
+    for (const auto &event_json : event_queue) {
+      events.push_back(nlohmann::json::parse(event_json));
+    }
+    sendEventsToRQ(events);
+    event_queue.clear();
   }
 #endif
-  mutex->unlock();
 }
 
 void Countly::setMaxEvents(size_t value) {
-  if (is_sdk_initialized) {
-    log(LogLevel::WARNING, "[Countly][setMaxEvents] You can not set the event queue size after SDK initialization.");
-    return;
-  }
+  log(LogLevel::WARNING, "[Countly][setMaxEvents/SetMaxEventsPerMessage] These calls are deprecated. Use 'setEventsToRQThreshold' instead.");
+  setEventsToRQThreshold(value);
+}
 
+void Countly::setEventsToRQThreshold(int value) {
+  log(LogLevel::DEBUG, "[Countly][setEventsToRQThreshold] Given threshold:[" + std::to_string(value) + "]");
   mutex->lock();
-  configuration->eventQueueThreshold = value;
-#ifndef COUNTLY_USE_SQLITE
-  if (event_queue.size() > configuration->eventQueueThreshold) {
-    log(LogLevel::WARNING, "New event queue size is smaller than the old one, dropping the oldest events to fit");
-    event_queue.resize(configuration->eventQueueThreshold);
+  if (value < 1) {
+    log(LogLevel::WARNING, "[Countly][setEventsToRQThreshold] Threshold can not be less than 1. Setting it to 1 instead of:[" + std::to_string(value) + "]");
+    value = 1;
+  } else if (value > 10000) {
+    log(LogLevel::WARNING, "[Countly][setEventsToRQThreshold] Threshold can not be greater than 10000. Setting it to 10000 instead of:[" + std::to_string(value) + "]");
+    value = 10000;
   }
-#endif
+
+  // set the value
+  configuration->eventQueueThreshold = value;
+  // if current queue size is greater than the new threshold, send events to RQ
+  checkAndSendEventToRQ();
   mutex->unlock();
 }
 
@@ -555,14 +581,13 @@ bool Countly::attemptSessionUpdateEQ() {
   }
   mutex->unlock();
 #else
-  int event_count = checkPersistentEQSize();
+  int event_count = checkEQSize();
   if (event_count <= 0) {
     return false;
   }
 #endif
 
-  bool update_failed = !updateSession();
-  return update_failed;
+  return !updateSession();
 }
 
 void Countly::clearEQInternal() {
@@ -689,30 +714,21 @@ bool Countly::updateSession() {
 
     // events array
     nlohmann::json events = nlohmann::json::array();
-    bool no_events;
+    std::string event_ids;
+    bool no_events = isEQEmpty();
 
-#ifndef COUNTLY_USE_SQLITE
-    no_events = event_queue.empty();
     if (!no_events) {
+#ifndef COUNTLY_USE_SQLITE
       for (const auto &event_json : event_queue) {
         events.push_back(nlohmann::json::parse(event_json));
       }
-    } else {
-      log(LogLevel::DEBUG, "[Countly][updateSession] EQ empty.");
-    }
 #else
-    mutex->unlock();
-    no_events = checkPersistentEQSize() > 0 ? false : true;
-    mutex->lock();
-
-    std::string event_ids;
-    if (!no_events) {
       // TODO: If database_path was empty there was return false here
       fillEventsIntoJson(events, event_ids);
+#endif
     } else {
       log(LogLevel::DEBUG, "[Countly][updateSession] EQ empty.");
     }
-#endif
     mutex->unlock();
     auto duration = std::chrono::duration_cast<std::chrono::seconds>(getSessionDuration());
     mutex->lock();
@@ -728,10 +744,7 @@ bool Countly::updateSession() {
 
     // report events if there are any to request queue
     if (!no_events) {
-      log(LogLevel::DEBUG, "[Countly][updateSession] sending event.");
-      std::map<std::string, std::string> data = {{"app_key", session_params["app_key"].get<std::string>()}, {"device_id", session_params["device_id"].get<std::string>()}, {"events", events.dump()}};
-
-      requestModule->addRequestToQueue(data);
+      sendEventsToRQ(events);
     }
 
 // clear event queue
@@ -740,6 +753,7 @@ bool Countly::updateSession() {
     event_queue.clear();
 #else
     if (!event_ids.empty()) {
+      // this is a partial clearance, we only remove the events that were sent
       removeEventWithId(event_ids);
     }
 #endif
@@ -752,6 +766,23 @@ bool Countly::updateSession() {
   return true;
 }
 
+void Countly::sendEventsToRQ(const nlohmann::json &events) {
+  log(LogLevel::DEBUG, "[Countly][sendEventsToRQ] Sending events to RQ.");
+  std::map<std::string, std::string> data = {{"app_key", session_params["app_key"].get<std::string>()}, {"device_id", session_params["device_id"].get<std::string>()}, {"events", events.dump()}};
+  requestModule->addRequestToQueue(data);
+}
+
+bool Countly::isEQEmpty() {
+  log(LogLevel::DEBUG, "[Countly][isEQEmpty] Checking if the event queue is empty.");
+#ifdef COUNTLY_USE_SQLITE
+  mutex->unlock();
+  return checkEQSize() > 0 ? false : true;
+  mutex->lock();
+#else
+  return event_queue.empty();
+#endif
+}
+
 bool Countly::endSession() {
   log(LogLevel::INFO, "[Countly][endSession]");
   const std::chrono::system_clock::time_point now = Countly::getTimestamp();
@@ -777,6 +808,29 @@ bool Countly::endSession() {
 
 std::chrono::system_clock::time_point Countly::getTimestamp() { return std::chrono::system_clock::now(); }
 
+int Countly::checkEQSize() {
+  log(LogLevel::DEBUG, "[Countly][checkEQSize]");
+  int event_count = -1;
+  if (!is_sdk_initialized) {
+    log(LogLevel::DEBUG, "[Countly][checkEQSize] SDK is not initialized.");
+    return event_count;
+  }
+
+#ifdef COUNTLY_USE_SQLITE
+  event_count = checkPersistentEQSize();
+#else
+  event_count = checkMemoryEQSize();
+#endif
+  return event_count;
+}
+
+#ifndef COUNTLY_USE_SQLITE
+int Countly::checkMemoryEQSize() {
+  log(LogLevel::DEBUG, "[Countly][checkMemoryEQSize] Checking event queue size in memory");
+  return event_queue.size();
+}
+#endif
+
 // Standalone Sqlite functions
 #ifdef COUNTLY_USE_SQLITE
 void Countly::removeEventWithId(const std::string &event_ids) {
@@ -827,9 +881,7 @@ void Countly::fillEventsIntoJson(nlohmann::json &events, std::string &event_ids)
 
     // create sql statement to fetch events as much as the event queue threshold
     // TODO: check if this is something we want to do
-    std::ostringstream sql_statement_stream;
-    sql_statement_stream << "SELECT evtid, event FROM events LIMIT " << std::dec << configuration->eventQueueThreshold << ';';
-    std::string sql_statement = sql_statement_stream.str();
+    std::string sql_statement = "SELECT evtid, event FROM events;";
 
     // execute sql statement
     return_value = sqlite3_get_table(database, sql_statement.c_str(), &table, &row_count, &column_count, &error_message);
@@ -862,13 +914,12 @@ void Countly::fillEventsIntoJson(nlohmann::json &events, std::string &event_ids)
 }
 
 int Countly::checkPersistentEQSize() {
-  log(LogLevel::DEBUG, "[Countly][checkEQSize]");
-  int event_count = -1;
+  int result = -1;
   mutex->lock();
   if (database_path.empty()) {
     mutex->unlock();
     log(LogLevel::FATAL, "[Countly][checkEQSize] Sqlite database path is not set");
-    return event_count;
+    return result;
   }
 
   sqlite3 *database;
@@ -881,8 +932,8 @@ int Countly::checkPersistentEQSize() {
     char **table;
     return_value = sqlite3_get_table(database, "SELECT COUNT(*) FROM events;", &table, &row_count, &column_count, &error_message);
     if (return_value == SQLITE_OK) {
-      event_count = atoi(table[1]);
-      log(LogLevel::DEBUG, "[Countly][checkEQSize] Fetched event count from database: " + std::to_string(event_count));
+      result = atoi(table[1]);
+      log(LogLevel::DEBUG, "[Countly][checkEQSize] Fetched event count from database: " + std::to_string(result));
     } else {
       log(LogLevel::ERROR, error_message);
       sqlite3_free(error_message);
@@ -891,9 +942,8 @@ int Countly::checkPersistentEQSize() {
   } else {
     log(LogLevel::WARNING, "[Countly][checkEQSize] Could not open database");
   }
-
   sqlite3_close(database);
-  return event_count;
+  return result;
 }
 
 void Countly::addEventToSqlite(const cly::Event &event) {

tests/config.cpp

@@ -174,7 +174,7 @@ TEST_CASE("Validate setting configuration values") {
     ct.SetMetrics("Windows 11", "10", "p", "800x800", "Car", "1.1");
 
     ct.setMaxRQProcessingBatchSize(50);
-    ct.SetMaxEventsPerMessage(100);
+    ct.SetMaxEventsPerMessage(100); // this calls setMaxEvents
     ct.setAutomaticSessionUpdateInterval(50);
     ct.setSalt("new-salt");
     ct.setMaxRequestQueueSize(100);
@@ -190,7 +190,7 @@ TEST_CASE("Validate setting configuration values") {
     CHECK(config.databasePath == TEST_DATABASE_NAME);
 #endif
     CHECK(config.sessionDuration == 5);
-    CHECK(config.eventQueueThreshold == 10);
+    CHECK(config.eventQueueThreshold == 100); // this one should be changed after init
     CHECK(config.requestQueueThreshold == 10);
     CHECK(config.maxProcessingBatchSize == 50); // this one should be changed after init
     CHECK(config.breadcrumbsThreshold == 100);