diff --git a/.github/workflows/cmake.yml b/.github/workflows/cmake.yml index 5dbbfcb..776691f 100644 --- a/.github/workflows/cmake.yml +++ b/.github/workflows/cmake.yml @@ -28,6 +28,3 @@ jobs: - name: Build run: cmake --build build - - - name: Test - run: ctest --test-dir build --no-tests=error -C debug diff --git a/CMakeLists.txt b/CMakeLists.txt index ab1453b..dbcb0f5 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -19,11 +19,6 @@ add_library(BaseModuleLib ${SOURCES} ) -# Test executable -add_executable(BaseModuleTest - ${SOURCE_TESTS_DIR}/BaseModuleTest.cpp -) - add_subdirectory(components/Chunk_Types) add_subdirectory(components/plog) @@ -36,16 +31,6 @@ target_include_directories(BaseModuleLib PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/components/plog/include ) -target_include_directories(BaseModuleTest - PRIVATE ${INCLUDE_DIR} - PRIVATE ${INCLUDE_TESTS_DIR} - PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/components/doctest/doctest - PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/components/Chunk_Types/components/Nlohmann_JSON/include - PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/components/Chunk_Types/components/Nlohmann_JSON/include/nlohmann - PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/components/Chunk_Types/include - PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/components/plog/include -) - # Adding threads target_link_libraries(BaseModuleLib PRIVATE pthread) target_link_libraries(BaseModuleLib PRIVATE ${THREADS_LIBRARIES}) @@ -53,9 +38,6 @@ target_link_libraries(BaseModuleLib PRIVATE ${THREADS_LIBRARIES}) # Link with required libraries for the main executable target_link_libraries(BaseModuleLib PRIVATE ChunkTypesLib doctest::doctest) -# Link with required libraries for the test executable -target_link_libraries(BaseModuleTest PRIVATE ChunkTypesLib BaseModuleLib doctest::doctest) - # Enable testing enable_testing() diff --git a/include/BaseModule.h b/include/BaseModule.h index 4c356b0..bc45642 100644 --- a/include/BaseModule.h +++ b/include/BaseModule.h @@ -17,6 +17,7 @@ #include #include #include +#include /* Custom Includes */ #include "BaseChunk.h" @@ -83,59 +84,51 @@ class BaseModule void TrackProcessTime(bool bTrackTime, std::string sTrackerMessage); /* - * @brief used to test the processing of a particular module - * @param[in] pBaseChunk Pointer to base chuk - */ - void TestProcess(std::shared_ptr pBaseChunk); - - /* - * @brief Puts the module into a test mode where a single chunk will be processed and returned when requested - * @param[in] Boolean as to whethere the module should be in test mode + * @brief Function to start the reporting thread */ - void SetTestMode(bool bTestModeState); + void StartReporting(); - /* - * @brief If the mo - * @param[in] Boolean as to whethere the module should be in test mode + /** + * @brief Registers which function should be called when processing a chunk + * @param[in] eChunkType pointer to chunk to be processed + * @param[in] function class function (&,ClassName::FunctionName) */ - std::shared_ptr GetTestOutput(); + template + void RegisterChunkCallbackFunction(ChunkType eChunkType,T function); - /* - * @brief Function to start the reporting thread + /** + * @brief Calls processing function for particular chunk type + * @param[in] pBaseChunk pointer to chunk to be processed */ - void StartReporting(); + void CallChunkCallbackFunction(std::shared_ptr pBaseChunk); private: size_t m_uMaxInputBufferSize; ///< Max size of the class input buffer + // Contolling Timing For Debugging std::atomic m_bTrackProcessTime = false; ///< Boolean as to whether to track and log the processing time std::string m_sTrackerMessage = ""; ///< Log message printed when logging chunk processing time std::chrono::high_resolution_clock::time_point m_CurrentTrackingTime; ///< Initial time used to track time between consecutive chunk passes std::chrono::high_resolution_clock::time_point m_PreviousTimeTracking; ///< Final time used to track time between consecutive chunk passes - bool m_bTestMode; ///< Boolean as to whether the module is doing doing normal processing or test processing - std::shared_ptr m_pTestChunkOutput; ///< Member used to store test outputs - protected: + // Controlling Queues std::condition_variable m_cvDataInBuffer; ///< Conditional variable to control data in circulat buffer - std::queue> m_cbBaseChunkBuffer; ///< Input buffer of module + std::queue> m_cbBaseChunkBuffer; ///< Input buffer of module std::shared_ptr m_pNextModule; ///< Shared pointer to next module into which messages are passed std::atomic m_bShutDown; ///< Whether to try continuously process std::mutex m_BufferStateMutex; ///< Mutex to facilitate multi module buffer size checking std::thread m_thread; ///< Thread object for module processing bool m_bAlreadyLoggedBufferFull; ///< Boolean State machine tracking if we have logged whether queue is full + // Controlling Reporting std::thread m_QueueSizeReportingThread; std::chrono::high_resolution_clock::time_point m_CurrentQueueReportTime; std::chrono::high_resolution_clock::time_point m_PreviousQueueReportTime; - - /** - * @brief Returns true if a message pointer had been retrieved an passed on to next module. - * If no pointer in queue then returns false - * @param[in] pBaseChunk pointer to base chunk - */ - virtual void Process(std::shared_ptr pBaseChunk); + // Controling Function Calls + std::mutex m_FunctionCallbackMapMutex; + std::map)>> m_FunctionCallbackMap; /** * @brief Passes base chunk pointer to next module @@ -153,7 +146,7 @@ class BaseModule */ bool TakeFromBuffer(std::shared_ptr &pBaseChunk); - /* + /** * @brief Calls the infinite loop to report */ void StartReportingLoop(); diff --git a/include_tests/BaseModuleTests.h b/include_tests/BaseModuleTests.h deleted file mode 100644 index 87fc783..0000000 --- a/include_tests/BaseModuleTests.h +++ /dev/null @@ -1,35 +0,0 @@ -#ifndef BASE_MODULE_TEST -#define BASE_MODULE_TEST - -#include "doctest.h" -#include "BaseModule.h" - -TEST_CASE("BaseModule Test") -{ - // First we set up logging so we can see what is going on - plog::init(plog::info, "BaseModuleTestLogs.txt", 10000); - - unsigned uBufferLength = 10; - BaseModule baseModule(uBufferLength); - - // This should avoid fail fast mode - baseModule.SetTestMode(true); - - PLOG_INFO << __FUNCTION__ << "Starting Basic Type Check"; - SUBCASE("Basic Type Check") - { - CHECK(baseModule.GetModuleType() == "BaseModule"); - } - - PLOG_INFO << __FUNCTION__ << "Stating to check if one can try repeatedly start module and stop it"; - SUBCASE("Checking if one can try repeatedly start module and stop it") - { - // If this fails I will never get to the check - baseModule.StartProcessing(); - baseModule.StartProcessing(); - baseModule.StopProcessing(); - CHECK(1 == 1); - } -} - -#endif \ No newline at end of file diff --git a/source/BaseModule.cpp b/source/BaseModule.cpp index eed2ca5..0a0946c 100644 --- a/source/BaseModule.cpp +++ b/source/BaseModule.cpp @@ -7,7 +7,6 @@ BaseModule::BaseModule(unsigned uMaxInputBufferSize) : m_pNextModule(nullptr), m_bShutDown() { - } BaseModule::~BaseModule() @@ -20,18 +19,13 @@ BaseModule::~BaseModule() m_thread.join(); } -void BaseModule::Process(std::shared_ptr pBaseChunk) -{ - TryPassChunk(pBaseChunk); -} - void BaseModule::ContinuouslyTryProcess() { while (!m_bShutDown) { std::shared_ptr pBaseChunk; if (TakeFromBuffer(pBaseChunk)) - Process(pBaseChunk); + CallChunkCallbackFunction(pBaseChunk); else { // Wait to be notified that there is data available @@ -56,8 +50,6 @@ void BaseModule::StartProcessing() // Log warning std::string strWarning = std::string(__FUNCTION__) + ": Processing thread already started"; PLOG_WARNING << strWarning; - // And force a crash if in debug and not release - assert(true && m_bTestMode); } } @@ -89,8 +81,6 @@ bool BaseModule::TryPassChunk(const std::shared_ptr &pBaseChunk) if (m_pNextModule != nullptr) bReturnSuccessful = m_pNextModule->TakeChunkFromModule(std::move(pBaseChunk)); - else if (m_bTestMode) - m_pTestChunkOutput = pBaseChunk; return bReturnSuccessful; } @@ -152,28 +142,6 @@ void BaseModule::TrackProcessTime(bool bTrackTime, std::string sTrackerMessage) m_CurrentTrackingTime = std::chrono::high_resolution_clock::now(); } -void BaseModule::TestProcess(std::shared_ptr pBaseChunk) -{ - Process(pBaseChunk); -} - -std::shared_ptr BaseModule::GetTestOutput() -{ - if (m_pTestChunkOutput != nullptr) - return std::move(m_pTestChunkOutput); - else - { - std::string strError = std::string(__FUNCTION__) + " No test chunk output to return"; - PLOG_ERROR << strError; - throw; - } -} - -void BaseModule::SetTestMode(bool bTestModeState) -{ - m_bTestMode = bTestModeState; -} - void BaseModule::StartReporting() { m_QueueSizeReportingThread = std::thread([this]() @@ -192,12 +160,30 @@ void BaseModule::StartReportingLoop() // Then transmit auto pQueueChunk = std::make_shared(strModuleName, u16CurrentBufferSize); - auto bSuccessfullyPassed = TryPassChunk(pQueueChunk); - - if (!bSuccessfullyPassed) - PLOG_WARNING << GetModuleType() + " Failed to pass chunk" ; + CallChunkCallbackFunction(pQueueChunk); // And sleep as not to send too many std::this_thread::sleep_for(std::chrono::milliseconds(1000)); } +} + +template +void BaseModule::RegisterChunkCallbackFunction(ChunkType eChunkType,T function) +{ + std::unique_lock BufferAccessLock(m_FunctionCallbackMapMutex); + m_FunctionCallbackMap[eChunkType] = std::bind(function, this, std::placeholders::_1); +} + +void BaseModule::CallChunkCallbackFunction(std::shared_ptr pBaseChunk) +{ + std::unique_lock BufferAccessLock(m_FunctionCallbackMapMutex); + auto eChunkType = pBaseChunk->GetChunkType(); + + // Check if we have registered a function to process the chunk + if (m_FunctionCallbackMap.find(eChunkType) != m_FunctionCallbackMap.end()) + m_FunctionCallbackMap[eChunkType](pBaseChunk); + else + // Otherwise pass on + TryPassChunk(pBaseChunk); + } \ No newline at end of file