Core Library 1.7.0.0 Library containing core utilities and tools for threading, networking, logging, INI and CSV file management etc.

MessageQueueThread.h
Go to the documentation of this file.
1 // This file is part of CoreLibrary containing useful reusable utility
2 // classes.
3 //
4 // Copyright (C) 2014 to present, Duncan Crutchley
5 // Contact <dac1976github@outlook.com>
6 //
7 // This program is free software: you can redistribute it and/or modify
8 // it under the terms of the GNU Lesser General Public License as published
9 // by the Free Software Foundation, either version 3 of the License, or
10 // (at your option) any later version.
11 //
12 // This program is distributed in the hope that it will be useful,
13 // but WITHOUT ANY WARRANTY; without even the implied warranty of
14 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 // GNU General Public License and GNU Lesser General Public License
16 // for more details.
17 //
18 // You should have received a copy of the GNU General Public License
19 // and GNU Lesser General Public License along with this program. If
20 // not, see <http://www.gnu.org/licenses/>.
21
27 #ifndef MESSAGEQUEUETHREAD
28 #define MESSAGEQUEUETHREAD
29
30 #include "Platform/PlatformDefines.h"
31 #include <functional>
32 #include <map>
33 #include <stdexcept>
34 #include <boost/throw_exception.hpp>
35 #include "ThreadBase.h"
36 #include "ConcurrentQueue.h"
37
39 namespace core_lib
40 {
42 namespace threads
43 {
44
46 enum class eOnDestroyOptions
47 {
49 ignoreRemainingItems,
51 processRemainingItems
52 };
67 template <typename MessageId, typename MessageType>
68 class MessageQueueThread final : public ThreadBase
69 {
77 using msg_id_decoder_t = std::function<MessageId(const MessageType&)>;
84 using msg_deleter_t = std::function<void(MessageType&)>;
85
86 public:
95 explicit MessageQueueThread(
96 const msg_id_decoder_t& messageIdDecoder,
97 eOnDestroyOptions destroyOptions = eOnDestroyOptions::ignoreRemainingItems,
98 const msg_deleter_t& messageDeleter = msg_deleter_t())
99 : m_msgIdDecoder{messageIdDecoder}
100 , m_destroyOptions{destroyOptions}
101 , m_messageDeleter{messageDeleter}
102 {
103 if (!Start())
104 {
105 BOOST_THROW_EXCEPTION(std::runtime_error("ThreadBase::Start() returned false"));
106 }
107 }
109 MessageQueueThread(const MessageQueueThread&) = delete;
111 MessageQueueThread& operator=(const MessageQueueThread&) = delete;
113 MessageQueueThread(MessageQueueThread&&) = delete;
115 MessageQueueThread& operator=(MessageQueueThread&&) = delete;
117 ~MessageQueueThread() override
118 {
119 Stop();
120
121 while (!m_messageQueue.Empty())
122 {
123 switch (m_destroyOptions)
124 {
125 case eOnDestroyOptions::processRemainingItems:
126 ProcessNextMessage();
127 break;
128 case eOnDestroyOptions::ignoreRemainingItems:
129 default:
130 DeleteNextMessage();
131 break;
132 }
133 }
134 }
143 using msg_handler_t = std::function<bool(MessageType&)>;
152 void RegisterMessageHandler(MessageId const& messageID, msg_handler_t const& messageHandler)
153 {
154 std::lock_guard<std::mutex> lock{m_mutex};
155
156 if (m_msgHandlerMap.count(messageID) > 0)
157 {
158 BOOST_THROW_EXCEPTION(std::runtime_error("message handler already defined"));
159 }
160
161 m_msgHandlerMap.emplace(messageID, messageHandler);
162 }
167 void Push(MessageType&& msg)
168 {
169 m_messageQueue.Push(std::forward<MessageType>(msg));
170 }
171
172 private:
174 void ThreadIteration() NO_EXCEPT_ override
175 {
176 ProcessNextMessage();
177 }
179 void ProcessTerminationConditions() NO_EXCEPT_ override
180 {
181 // Make sure we break out of m_messageQueue.Pop();
182 m_messageQueue.BreakPopWait();
183 }
185 void ProcessNextMessage()
186 {
187 MessageType msg{};
188
189 if (!m_messageQueue.Pop(msg))
190 {
191 return;
192 }
193
194 bool canDeleteMsg;
195
196 try
197 {
198 MessageId messageId{m_msgIdDecoder(msg)};
199 std::lock_guard<std::mutex> lock{m_mutex};
200
201 if (m_msgHandlerMap.count(messageId) > 0)
202 {
203 canDeleteMsg = m_msgHandlerMap[messageId](msg);
204 }
205 else
206 {
207 canDeleteMsg = true;
208 }
209 }
210 catch (...)
211 {
212 canDeleteMsg = true;
213 }
214
215 if (canDeleteMsg && m_messageDeleter)
216 {
217 try
218 {
219 m_messageDeleter(msg);
220 }
221 catch (...)
222 {
223 // Do nothing.
224 }
225 }
226 }
228 void DeleteNextMessage()
229 {
230 MessageType msg{};
231
232 if (!m_messageQueue.Pop(msg))
233 {
234 return;
235 }
236
237 try
238 {
239 if (m_messageDeleter)
240 {
241 m_messageDeleter(msg);
242 }
243 }
244 catch (...)
245 {
246 // Do nothing.
247 }
248 }
249
250 private:
252 mutable std::mutex m_mutex;
254 msg_id_decoder_t m_msgIdDecoder{};
256 eOnDestroyOptions m_destroyOptions{};
258 msg_deleter_t m_messageDeleter{};
260 using msg_map_t = std::map<MessageId, msg_handler_t>;
262 msg_map_t m_msgHandlerMap{};
264 ConcurrentQueue<MessageType> m_messageQueue{};
265 };
266
267 } // namespace threads
268 } // namespace core_lib
269
270 #endif // MESSAGEQUEUETHREAD
core_lib::threads::MessageQueueThread::m_mutex
std::mutex m_mutex
Definition: MessageQueueThread.h:252
core_lib::threads::MessageQueueThread::msg_map_t
std::map< MessageId, msg_handler_t > msg_map_t
Typedef for message map type.
Definition: MessageQueueThread.h:260
core_lib::threads::ThreadBase
Thread base class.
Definition: ThreadBase.h:49
core_lib::threads::eOnDestroyOptions::processRemainingItems
core_lib::threads::eOnDestroyOptions::ignoreRemainingItems
core_lib::threads::MessageQueueThread::MessageQueueThread
MessageQueueThread(const msg_id_decoder_t &messageIdDecoder, eOnDestroyOptions destroyOptions=eOnDestroyOptions::ignoreRemainingItems, const msg_deleter_t &messageDeleter=msg_deleter_t())
Default constructor.
Definition: MessageQueueThread.h:95
ThreadBase.h
File containing declaration of ThreadBase class.
ConcurrentQueue.h
File containing concurrent queue declaration.
core_lib
The core_lib namespace.
Definition: AsioDefines.h:59
core_lib::threads::MessageQueueThread::msg_deleter_t
std::function< void(MessageType &)> msg_deleter_t
Typedef defining message deleter function.
Definition: MessageQueueThread.h:84
core_lib::threads::MessageQueueThread::ProcessTerminationConditions
void ProcessTerminationConditions() NO_EXCEPT_ override
Perform any special termination actions.
Definition: MessageQueueThread.h:179
core_lib::threads::MessageQueueThread::~MessageQueueThread
~MessageQueueThread() override
Destructor.
Definition: MessageQueueThread.h:117
core_lib::threads::MessageQueueThread::Push
void Push(MessageType &&msg)
Push a message as an array of items onto this thread&#39;s queue.
Definition: MessageQueueThread.h:167
core_lib::threads::eOnDestroyOptions
eOnDestroyOptions
Control how messages get destroyed in destructor.
Definition: MessageQueueThread.h:46
core_lib::threads::MessageQueueThread::ProcessNextMessage
void ProcessNextMessage()
Process next message.
Definition: MessageQueueThread.h:185
PlatformDefines.h
File containing platform specific definitions.
core_lib::threads::MessageQueueThread::msg_handler_t
std::function< bool(MessageType &)> msg_handler_t
Typedef defining message handler functor.
Definition: MessageQueueThread.h:143
core_lib::threads::ConcurrentQueue< MessageType >
core_lib::threads::MessageQueueThread::DeleteNextMessage
void DeleteNextMessage()
Delete next message.
Definition: MessageQueueThread.h:228
core_lib::threads::MessageQueueThread::ThreadIteration
void ThreadIteration() NO_EXCEPT_ override
Execute a single iteration of the thread.
Definition: MessageQueueThread.h:174
core_lib::threads::MessageQueueThread::RegisterMessageHandler
void RegisterMessageHandler(MessageId const &messageID, msg_handler_t const &messageHandler)
Register a function to handle a particular message.
Definition: MessageQueueThread.h:152
core_lib::threads::MessageQueueThread::msg_id_decoder_t
std::function< MessageId(const MessageType &)> msg_id_decoder_t
Typedef defining message ID decoder function.
Definition: MessageQueueThread.h:77
core_lib::threads::MessageQueueThread
Message Queue Thread.
Definition: MessageQueueThread.h:68
NO_EXCEPT_
#define NO_EXCEPT_
NO_EXCEPT_ definition mapping to noexcept.
Definition: PlatformDefines.h:57