Queue.h

 
#pragma once
 
#include <aws/core/client/ClientConfiguration.h>
#include <aws/queues/Queues_EXPORTS.h>
#include <chrono>
#include <thread>
#include <atomic>
#include <functional>
 
namespace Aws
{
    namespace Queues
    {
        static const char* MEM_TAG = "Aws::Queues::Queue";
 
        template<typename MESSAGE_TYPE>
        class Queue
        {
            typedef std::function<void(const Queue*, const MESSAGE_TYPE&, bool&)> MessageReceivedEventHandler;
            typedef std::function<void(const Queue*, const MESSAGE_TYPE&)> MessageDeleteFailedEventHandler;
            typedef std::function<void(const Queue*, const MESSAGE_TYPE&)> MessageDeleteSuccessEventHandler;
            typedef std::function<void(const Queue*, const MESSAGE_TYPE&)> MessageSendFailedEventHandler;
            typedef std::function<void(const Queue*, const MESSAGE_TYPE&)> MessageSendSuccessEventHandler;
 
        public:
            Queue(unsigned pollingFrequency) :
                    m_continue(true), m_pollingFrequencyMs(pollingFrequency), m_pollingThread(nullptr)
            {
            }
 
            virtual ~Queue()
            {
                StopPolling();
            }
 
            virtual MESSAGE_TYPE Top() const = 0;
            virtual void Delete(const MESSAGE_TYPE&) = 0;
            virtual void Push(const MESSAGE_TYPE&) = 0;
 
            void StartPolling()
            {
                if(!m_pollingThread)
                {
                    m_continue = true;
                    m_pollingThread = Aws::MakeUnique<std::thread>(MEM_TAG, &Queue::Main, this);
                }
            }
 
            void StopPolling()
            {
                m_continue = false;
                if(m_pollingThread)
                {
                    m_pollingThread->join();
                    m_pollingThread = nullptr;
                }
            }
 
            inline void SetMessageReceivedEventHandler(const MessageReceivedEventHandler& messageHandler) { m_messageReceivedHandler = messageHandler; }
            inline void SetMessageDeleteFailedEventHandler(const MessageDeleteFailedEventHandler& messageHandler) { m_messageDeleteFailedHandler = messageHandler; }
            inline void SetMessageDeleteSuccessEventHandler(const MessageDeleteSuccessEventHandler& messageHandler) { m_messageDeleteSuccessHandler = messageHandler; }
            inline void SetMessageSendFailedEventHandler(const MessageSendFailedEventHandler& messageHandler) { m_messageSendFailedHandler = messageHandler; }
            inline void SetMessageSendSuccessEventHandler(const MessageSendSuccessEventHandler& messageHandler) { m_messageSendSuccessHandler = messageHandler; }
 
            inline void SetMessageReceivedEventHandler(MessageReceivedEventHandler&& messageHandler) { m_messageReceivedHandler = messageHandler; }
            inline void SetMessageDeleteFailedEventHandler(MessageDeleteFailedEventHandler&& messageHandler) { m_messageDeleteFailedHandler = messageHandler; }
            inline void SetMessageDeleteSuccessEventHandler(MessageDeleteSuccessEventHandler&& messageHandler) { m_messageDeleteSuccessHandler = messageHandler; }
            inline void SetMessageSendFailedEventHandler(MessageSendFailedEventHandler&& messageHandler) { m_messageSendFailedHandler = messageHandler; }
            inline void SetMessageSendSuccessEventHandler(MessageSendSuccessEventHandler&& messageHandler) { m_messageSendSuccessHandler = messageHandler; }
 
            inline const MessageReceivedEventHandler& GetMessageReceivedEventHandler() const { return m_messageReceivedHandler; }
            inline const MessageDeleteFailedEventHandler& GetMessageDeleteFailedEventHandler() const { return m_messageDeleteFailedHandler; }
            inline const MessageDeleteSuccessEventHandler& GetMessageDeleteSuccessEventHandler() const { return m_messageDeleteSuccessHandler; }
            inline const MessageSendFailedEventHandler& GetMessageSendFailedEventHandler() const { return m_messageSendFailedHandler; }
            inline const MessageSendSuccessEventHandler& GetMessageSendSuccessEventHandler() const { return m_messageSendSuccessHandler; }
 
        protected:
            std::atomic<bool> m_continue;
 
        private:
            void Main()
            {
                while(m_continue)
                {
                    auto start = std::chrono::system_clock::now();
                    MESSAGE_TYPE topMessage = Top();
                    bool deleteMessage = false;
                    
                    auto& receivedHandler = GetMessageReceivedEventHandler();
                    if (receivedHandler)
                    {
                        receivedHandler(this, topMessage, deleteMessage);
                    }
 
                    if (deleteMessage)
                    {
                        Delete(topMessage);
                    }
 
                    if(m_continue)
                    {
                        auto stop = std::chrono::system_clock::now();
                        auto timeTaken = std::chrono::duration_cast<std::chrono::milliseconds>(stop - start);
 
                        if (m_pollingFrequencyMs >= timeTaken.count())
                        {
                            std::this_thread::sleep_for(std::chrono::milliseconds(m_pollingFrequencyMs - timeTaken.count()));
                        }
                    }
                }
            }
 
            unsigned m_pollingFrequencyMs;
            Aws::UniquePtr<std::thread> m_pollingThread;
 
            // Handlers
            MessageReceivedEventHandler m_messageReceivedHandler;
            MessageDeleteFailedEventHandler m_messageDeleteFailedHandler;
            MessageDeleteSuccessEventHandler m_messageDeleteSuccessHandler;
            MessageSendFailedEventHandler m_messageSendFailedHandler;
            MessageSendSuccessEventHandler m_messageSendSuccessHandler;
        };
    }
}