Merge pull request wazuh#24924 from wazuh/enhancement/24352_multithre…

…ad-event-dispacher Enhancement multithreading dispatching
youngsecurity · Aug 8, 2024 · 207873a · 207873a
2 parents edbb13c + 36dfc9c
commit 207873a
Show file tree

Hide file tree

Showing 4 changed files with 273 additions and 57 deletions.
diff --git a/src/shared_modules/utils/tests/threadEventDispatcher_test.cpp b/src/shared_modules/utils/tests/threadEventDispatcher_test.cpp
@@ -21,7 +21,7 @@ void ThreadEventDispatcherTest::TearDown() {
 };
 
 constexpr auto BULK_SIZE {50};
-TEST_F(ThreadEventDispatcherTest, Ctor)
+TEST_F(ThreadEventDispatcherTest, ConstructorTestSingleThread)
 {
     static const std::vector<int> MESSAGES_TO_SEND_LIST {120, 100};
 
@@ -60,7 +60,47 @@ TEST_F(ThreadEventDispatcherTest, Ctor)
     }
 }
 
-TEST_F(ThreadEventDispatcherTest, CtorNoWorker)
+TEST_F(ThreadEventDispatcherTest, ConstructorTestMultiThread)
+{
+    static const std::vector<int> MESSAGES_TO_SEND_LIST {120, 100};
+    static const auto NUM_THREADS = 4;
+
+    for (auto MESSAGES_TO_SEND : MESSAGES_TO_SEND_LIST)
+    {
+        std::atomic<size_t> counter {0};
+        std::promise<void> promise;
+        auto index {0};
+
+        ThreadEventDispatcher<std::string, std::function<void(std::queue<std::string>&)>, NUM_THREADS> dispatcher(
+            [&counter, &index, &MESSAGES_TO_SEND, &promise](std::queue<std::string>& data)
+            {
+                counter += data.size();
+                while (!data.empty())
+                {
+                    auto value = data.front();
+                    data.pop();
+                    EXPECT_EQ(std::to_string(index), value);
+                    ++index;
+                }
+
+                if (counter == MESSAGES_TO_SEND)
+                {
+                    promise.set_value();
+                }
+            },
+            "test.db",
+            BULK_SIZE);
+
+        for (int i = 0; i < MESSAGES_TO_SEND; ++i)
+        {
+            dispatcher.push(std::to_string(i));
+        }
+        promise.get_future().wait_for(std::chrono::seconds(10));
+        EXPECT_EQ(MESSAGES_TO_SEND, counter);
+    }
+}
+
+TEST_F(ThreadEventDispatcherTest, CtorNoWorkerSingleThread)
 {
     static const std::vector<int> MESSAGES_TO_SEND_LIST {120, 100};
 
@@ -101,7 +141,7 @@ TEST_F(ThreadEventDispatcherTest, CtorNoWorker)
     }
 }
 
-TEST_F(ThreadEventDispatcherTest, CtorPopFeature)
+TEST_F(ThreadEventDispatcherTest, CtorPopFeatureSingleThread)
 {
     constexpr auto MESSAGES_TO_SEND {1000};
 
@@ -144,4 +184,3 @@ TEST_F(ThreadEventDispatcherTest, CtorPopFeature)
     promise.get_future().wait_for(std::chrono::seconds(10));
     EXPECT_EQ(MESSAGES_TO_SEND, counter);
 }
-