ConsumerDispatcher.h

Go to the documentation of this file.

 
#pragma once
#include "ConsumerDispatcherOptions.h"
#include "Disruptor.h"
#include "DisruptorConsumer.h"
#include "DisruptorInspector.h"
#include "catapult/utils/NamedObject.h"
#include <boost/thread.hpp>
#include <atomic>
 
namespace catapult { namespace disruptor { class ConsumerEntry; } }
 
namespace catapult { namespace disruptor {
 
    class ConsumerDispatcher final : public utils::NamedObjectMixin {
    public:
        ConsumerDispatcher(
                const ConsumerDispatcherOptions& options,
                const std::vector<DisruptorConsumer>& consumers,
                const DisruptorInspector& inspector);
 
        explicit ConsumerDispatcher(const ConsumerDispatcherOptions& options, const std::vector<DisruptorConsumer>& consumers);
 
        ~ConsumerDispatcher();
 
    public:
        void shutdown();
 
        bool isRunning() const;
 
        size_t size() const;
 
        DisruptorElementId processElement(ConsumerInput&& input, const ProcessingCompleteFunc& processingComplete);
 
        DisruptorElementId processElement(ConsumerInput&& input);
 
        size_t numAddedElements() const;
 
        size_t numActiveElements() const;
 
    private:
        DisruptorElement* tryNext(ConsumerEntry& consumerEntry);
 
        void advance(ConsumerEntry& consumerEntry);
 
        bool canProcessNextElement() const;
 
        ProcessingCompleteFunc wrap(const ProcessingCompleteFunc& processingComplete);
 
    private:
        size_t m_elementTraceInterval;
        bool m_shouldThrowIfFull;
        std::atomic_bool m_keepRunning;
        DisruptorBarriers m_barriers;
        Disruptor m_disruptor;
        DisruptorInspector m_inspector;
        boost::thread_group m_threads;
        std::atomic<size_t> m_numActiveElements;
 
        utils::SpinLock m_addSpinLock; // lock to serialize access to Disruptor::add
    };
}}