open_source / transqueue

Browse Code »

Commit 32b0e19aaa1fec06d3452228290e944e9e7cd28a

Authored by Steven de Ridder 2022-01-24 15:40:53 +0100

0 parents

Initial commit. dependencies not resolved yet.

Inline Side-by-side

Showing 7 changed files with 403 additions and 0 deletions

.gitignore 0 → 100644

View file @32b0e19

	1	+++ a/.gitignore
	1	+build/
	2	+CMakeLists.txt.user
...	...

CMakeLists.txt 0 → 100644

View file @32b0e19

	1	+++ a/CMakeLists.txt
	1	+cmake_minimum_required(VERSION 3.0)
	2	+
	3	+# Check to see where cmake is located.
	4	+if( IS_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/cmake )
	5	+ LIST(APPEND CMAKE_MODULE_PATH ${CMAKE_CURRENT_SOURCE_DIR}/cmake)
	6	+elseif( IS_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/../cmake )
	7	+ LIST(APPEND CMAKE_MODULE_PATH ${CMAKE_CURRENT_SOURCE_DIR}/../cmake)
	8	+else()
	9	+ return()
	10	+endif()
	11	+
	12	+# Check to see if there is versioning information available
	13	+if(IS_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/osdev_versioning/cmake)
	14	+ LIST(APPEND CMAKE_MODULE_PATH ${CMAKE_CURRENT_SOURCE_DIR}/osdev_versioning/cmake)
	15	+ include(osdevversion)
	16	+endif()
	17	+
	18	+include(projectheader)
	19	+project_header(osdev_transqueue)
	20	+
	21	+add_subdirectory(src)
	22	+add_subdirectory(tests)
	23	+
	24	+# include(packaging)
	25	+# package_component()
...	...

README.md 0 → 100644

View file @32b0e19

	1	+++ a/README.md
...	...

src/CMakeLists.txt 0 → 100644

View file @32b0e19

	1	+++ a/src/CMakeLists.txt
	1	+cmake_minimum_required(VERSION 3.0)
	2	+LIST(APPEND CMAKE_MODULE_PATH ${CMAKE_CURRENT_SOURCE_DIR}/../../../cmake)
	3	+include(projectheader)
	4	+project_header(transqueue)
	5	+
	6	+find_package( Qt5Core REQUIRED )
	7	+find_package( Qt5Sql REQUIRED )
	8	+
	9	+include_directories( SYSTEM
	10	+ ${Qt5Core_INCLUDE_DIRS}
	11	+ ${CMAKE_CURRENT_SOURCE_DIR}/../datatypes
	12	+ ${CMAKE_CURRENT_SOURCE_DIR}/../logutils
	13	+ ${CMAKE_CURRENT_SOURCE_DIR}/../config
	14	+ ${CMAKE_CURRENT_SOURCE_DIR}/../pugixml
	15	+)
	16	+
	17	+include(compiler)
	18	+
	19	+set(SRC_LIST
	20	+ ${CMAKE_CURRENT_SOURCE_DIR}/transqueue.cpp
	21	+)
	22	+
	23	+include(qtmoc)
	24	+create_mocs( SRC_LIST MOC_LIST
	25	+ ${CMAKE_CURRENT_SOURCE_DIR}/transqueue.h
	26	+)
	27	+
	28	+include(library)
	29	+add_libraries(
	30	+ ${Qt5Core_LIBRARIES}
	31	+ datatypes
	32	+ logutils
	33	+ config
	34	+ pugixml
	35	+)
	36	+
	37	+include(installation)
	38	+install_component()
...	...

src/transqueue.cpp 0 → 100644

View file @32b0e19

	1	+++ a/src/transqueue.cpp
	1	+/* ****************************************************************************
	2	+ * Copyright 2019 Open Systems Development BV *
	3	+ * *
	4	+ * Permission is hereby granted, free of charge, to any person obtaining a *
	5	+ * copy of this software and associated documentation files (the "Software"), *
	6	+ * to deal in the Software without restriction, including without limitation *
	7	+ * the rights to use, copy, modify, merge, publish, distribute, sublicense, *
	8	+ * and/or sell copies of the Software, and to permit persons to whom the *
	9	+ * Software is furnished to do so, subject to the following conditions: *
	10	+ * *
	11	+ * The above copyright notice and this permission notice shall be included in *
	12	+ * all copies or substantial portions of the Software. *
	13	+ * *
	14	+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR *
	15	+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, *
	16	+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL *
	17	+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER *
	18	+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING *
	19	+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER *
	20	+ * DEALINGS IN THE SOFTWARE. *
	21	+ * ***************************************************************************/
	22	+
	23	+#include "transqueue.h"
	24	+
	25	+#include "dcxmlconfig.h"
	26	+#include "log.h"
	27	+#include "threadcontext.h"
	28	+#include "ormxmlwriter.h"
	29	+
	30	+#include <QFile>
	31	+#include <QDateTime>
	32	+#include <QTextStream>
	33	+
	34	+using namespace osdev::components;
	35	+
	36	+TransQueue::TransQueue( QObject *_parent )
	37	+ : QObject( _parent )
	38	+ , m_TTL( DCXmlConfig::Instance().timeToLive() )
	39	+ , m_queueTimer()
	40	+ , m_queueMutex()
	41	+ , m_queue()
	42	+{
	43	+ // Connect the timer to the internal timeout-slot.
	44	+ connect( &m_queueTimer, &QTimer::timeout, this, &TransQueue::slotProcessQueue );
	45	+}
	46	+
	47	+void TransQueue::setTimeOut( int milli_seconds )
	48	+{
	49	+ m_queueTimer.setSingleShot( true );
	50	+ m_queueTimer.setInterval( milli_seconds );
	51	+}
	52	+
	53	+void TransQueue::setTransaction( const QSharedPointer<ORMRelData>& pData )
	54	+{
	55	+ QMutexLocker mutLock( &m_queueMutex );
	56	+ if( pData )
	57	+ {
	58	+ if( pData->TTL() == -1 )
	59	+ {
	60	+ pData->setTTL( m_TTL );
	61	+ m_queue.enqueue( pData );
	62	+ }
	63	+ else if( pData->TTL() == 0 )
	64	+ {
	65	+ if( !this->dumpToDisk( pData ) )
	66	+ {
	67	+ LogError( "[TransQueue::setTransaction]",
	68	+ QString( "There was a problem dumping the data disk. That is why we dump it to the LogFile : %1" ).arg( pData->asString() ) );
	69	+ }
	70	+
	71	+ }
	72	+ else if( pData->TTL() > 0 )
	73	+ {
	74	+ pData->decTTL();
	75	+ m_queue.enqueue( pData );
	76	+ }
	77	+
	78	+
	79	+ this->startProcessing();
	80	+ }
	81	+}
	82	+
	83	+bool TransQueue::processing() const
	84	+{
	85	+ return m_queueTimer.isActive();
	86	+}
	87	+
	88	+void TransQueue::startProcessing( bool force )
	89	+{
	90	+ if( force )
	91	+ {
	92	+ m_queueTimer.stop();
	93	+ /*
	94	+ * The timer will be started after slotProcessQueue is done processing.
	95	+ * Timer is implemented as a singleshot, preventing timeout() signals
	96	+ * to pile up due to the processingtime of the slot.
	97	+ */
	98	+ this->slotProcessQueue();
	99	+ }
	100	+ else
	101	+ {
	102	+ if( !m_queueTimer.isActive() )
	103	+ {
	104	+ m_queueTimer.start();
	105	+ }
	106	+ else
	107	+ {
	108	+ // Do nothing. timeout will trigger...
	109	+ }
	110	+ }
	111	+}
	112	+
	113	+void TransQueue::stopProcessing( bool force )
	114	+{
	115	+ if( force && m_queueTimer.isActive() )
	116	+ {
	117	+ m_queueTimer.stop();
	118	+ }
	119	+}
	120	+
	121	+void TransQueue::slotProcessQueue()
	122	+{
	123	+ LogDebug( "[TransQueue::slotProcessQueue()]", QString("Starting processing the queue.") );
	124	+ if( !m_queue.isEmpty() )
	125	+ {
	126	+ LogDebug( "[TransQueue::slotProcessQueue()]", QString("Number of entries in the queue before : %1").arg( m_queue.size() ) );
	127	+ auto pData = m_queue.dequeue();
	128	+ ThreadContextScope tcs(pData->traceId());
	129	+ emit signalProcessData( pData );
	130	+ LogDebug( "[TransQueue::slotProcessQueue()]", QString("Number of entries in the queue after : %1").arg( m_queue.size() ) );
	131	+ LogInfo( "[TransQueue::slotProcessQueue()]", QString("TTL Of package : %1").arg( pData->TTL() ) );
	132	+ }
	133	+
	134	+ // Timer is intentionally set as a singleshot. We have to restart it
	135	+ // every time this slot is called.
	136	+ if( m_queue.size() > 0 )
	137	+ {
	138	+ LogInfo( "[TransQueue::slotProcessQueue()]", QString( "Number of transactions in Queue : %1" ).arg( m_queue.size() ) );
	139	+ m_queueTimer.start();
	140	+ }
	141	+ else
	142	+ {
	143	+ LogInfo( "[TransQueue::slotProcessQueue()]", "Transaction Queue is Empty." );
	144	+ // Just to be sure.....
	145	+ m_queueTimer.stop();
	146	+ }
	147	+}
	148	+
	149	+bool TransQueue::dumpToDisk( const QSharedPointer<ORMRelData>& data ) const
	150	+{
	151	+ if( data )
	152	+ {
	153	+ QString strLogFile = DCXmlConfig::Instance().transactionPath();
	154	+ OrmXmlWriter oWriter( strLogFile.toStdString() );
	155	+ if( oWriter.writeToDisk() )
	156	+ {
	157	+ LogWarning( "[TransQueue::dumpToDisk]",
	158	+ QString( "Datafile failed to save to disk. Content : <ormreldata_%1>%2" ).arg( data->getMainTableContainer()->timeStamp() ).arg( QString( oWriter.asString().c_str() ) ) );
	159	+ }
	160	+
	161	+ }
	162	+ return false;
	163	+}
...	...

src/transqueue.h 0 → 100644

View file @32b0e19

	1	+++ a/src/transqueue.h
	1	+/* ****************************************************************************
	2	+ * Copyright 2019 Open Systems Development BV *
	3	+ * *
	4	+ * Permission is hereby granted, free of charge, to any person obtaining a *
	5	+ * copy of this software and associated documentation files (the "Software"), *
	6	+ * to deal in the Software without restriction, including without limitation *
	7	+ * the rights to use, copy, modify, merge, publish, distribute, sublicense, *
	8	+ * and/or sell copies of the Software, and to permit persons to whom the *
	9	+ * Software is furnished to do so, subject to the following conditions: *
	10	+ * *
	11	+ * The above copyright notice and this permission notice shall be included in *
	12	+ * all copies or substantial portions of the Software. *
	13	+ * *
	14	+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR *
	15	+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, *
	16	+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL *
	17	+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER *
	18	+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING *
	19	+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER *
	20	+ * DEALINGS IN THE SOFTWARE. *
	21	+ * ***************************************************************************/
	22	+
	23	+#ifndef OSDEV_COMPONENTS_TRANSQUEUE_H
	24	+#define OSDEV_COMPONENTS_TRANSQUEUE_H
	25	+
	26	+#include "ormreldata.h"
	27	+
	28	+#include <QObject>
	29	+#include <QQueue>
	30	+#include <QTimer>
	31	+#include <QMutex>
	32	+#include <QMutexLocker>
	33	+
	34	+/*
	35	+ * _________________________________________
	36	+ * / If voting could change the system, it \
	37	+ * \| would be illegal. If not voting could \|
	38	+ * \ change the system, it would be illegal. /
	39	+ * -----------------------------------------
	40	+ * \
	41	+ * \
	42	+ * .--.
	43	+ * \|o_o \|
	44	+ * \|:_/ \|
	45	+ * // \ \
	46	+ * (\| \| )
	47	+ * /'\_ _/`\
	48	+ * \___)=(___/
	49	+ *
	50	+ */
	51	+
	52	+namespace osdev {
	53	+namespace components {
	54	+/*!
	55	+ * \class TransQueue
	56	+ * \brief The TransQueue class implements a transaction queue which will
	57	+ * catch rejected transactions from the ORM layer and offers them
	58	+ * back through the ETL-layer. An ORMRelPackage is "tagged" with
	59	+ * a transacion-id, which will keep track on how many times an
	60	+ * ORMRelPackage enters the queue. If the "TTL_iD (Time_To_live)"
	61	+ * reaches zero, the package is dropped and the ID is removed from
	62	+ * the administration.
	63	+ */
	64	+class TransQueue : public QObject
	65	+{
	66	+ Q_OBJECT
	67	+
	68	+public:
	69	+ /*!
	70	+ * \brief Represents a queue of transactions.
	71	+ * \param parent The parent QObject instance.
	72	+ */
	73	+ explicit TransQueue( QObject *parent = nullptr );
	74	+
	75	+ // Deleted copy- and move constructors
	76	+ TransQueue( const TransQueue& ) = delete;
	77	+ TransQueue( const TransQueue&& ) = delete;
	78	+ TransQueue& operator=( const TransQueue& ) = delete;
	79	+ TransQueue& operator=( const TransQueue&& ) = delete;
	80	+
	81	+ /*!
	82	+ * \brief Sets the TimeOut.
	83	+ * \param milliseconds The timeout to set in milliseconds.
	84	+ */
	85	+ void setTimeOut( int milliseconds = 1 );
	86	+
	87	+ /*!
	88	+ * \return Whether this instance is currently processing transactions.
	89	+ */
	90	+ bool processing() const;
	91	+
	92	+ /*!
	93	+ * \brief Starts processing of transactions.
	94	+ * \param force Forces start of processing.
	95	+ */
	96	+ void startProcessing( bool force = false );
	97	+
	98	+ /*!
	99	+ * \brief Stops processing of transactions.
	100	+ * \param force Forces stop of processing.
	101	+ */
	102	+ void stopProcessing( bool force = false );
	103	+
	104	+ /*!
	105	+ * \return The number of currently outstanding transactions.
	106	+ */
	107	+ int transactions() const { return m_queue.size(); }
	108	+
	109	+private:
	110	+ /*!
	111	+ * \brief This method will write the Contents of the data structure
	112	+ * to a transaction file for archiving and traceability purposes.
	113	+ * Not really decided on the format yet, but this will evolve over time.
	114	+ * \param data - The datastructure we want to write to a transaction Logfile.
	115	+ * Send as pointer. After writing to disk, the object have to be deleted.
	116	+ */
	117	+ bool dumpToDisk( const QSharedPointer<ORMRelData>& data ) const;
	118	+
	119	+ int m_TTL; ///< Time to Live start value.
	120	+ QTimer m_queueTimer; ///< Timer controlling the processing of the queue.
	121	+ QMutex m_queueMutex; ///< Mutex to prevent race conditions. Although QQueue is thread safe, we have to make sure we are.
	122	+ QQueue<QSharedPointer<ORMRelData>> m_queue; ///< The actual FIFO taking an ORMRelData pointer as input.
	123	+
	124	+signals:
	125	+ void signalProcessData( QSharedPointer<ORMRelData> pData );
	126	+
	127	+public slots:
	128	+ /*!
	129	+ * \brief Sets (adds) a transaction to the queue.
	130	+ * \param pData The transaction data to set.
	131	+ */
	132	+ void setTransaction( const QSharedPointer<ORMRelData>& pData );
	133	+
	134	+
	135	+private slots:
	136	+ /*!
	137	+ * \brief slotProcessQueue
	138	+ */
	139	+ void slotProcessQueue();
	140	+};
	141	+
	142	+} /* End namespace components */
	143	+} /* End namespace osdev */
	144	+
	145	+#endif /* OSDEV_COMPONENTS_TRANSQUEUE_H */
...	...

tests/CMakeLists.txt 0 → 100644

View file @32b0e19

	1	+++ a/tests/CMakeLists.txt
	1	+cmake_minimum_required(VERSION 3.0)
	2	+LIST(APPEND CMAKE_MODULE_PATH ${CMAKE_CURRENT_SOURCE_DIR}/../cmake)
	3	+
	4	+include(projectheader)
	5	+project_header(test_logutils)
	6	+
	7	+include_directories( SYSTEM
	8	+ ${CMAKE_CURRENT_SOURCE_DIR}/../../src
	9	+)
	10	+
	11	+include(compiler)
	12	+set(SRC_LIST
	13	+)
	14	+
	15	+# add_executable( ${PROJECT_NAME}
	16	+# ${SRC_LIST}
	17	+# )
	18	+
	19	+# target_link_libraries(
	20	+# ${PROJECT_NAME}
	21	+# )
	22	+
	23	+# set_target_properties( ${PROJECT_NAME} PROPERTIES
	24	+# RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/bin
	25	+# LIBRARY_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/lib
	26	+# ARCHIVE_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/archive
	27	+# )
	28	+
	29	+# include(installation)
	30	+# install_application()
...	...