DisruptorElement.h

Go to the documentation of this file.

 
#pragma once
#include "ConsumerInput.h"
#include "catapult/utils/SpinLock.h"
 
namespace catapult { namespace disruptor {
 
    class DisruptorElement {
    public:
        DisruptorElement()
                : m_id(static_cast<uint64_t>(-1))
                , m_processingComplete([](auto, auto) {})
                , m_pSpinLock(std::make_unique<utils::SpinLock>())
        {}
 
        explicit DisruptorElement(ConsumerInput&& input, DisruptorElementId id, const ProcessingCompleteFunc& processingComplete)
                : m_input(std::move(input))
                , m_id(id)
                , m_processingComplete(processingComplete)
                , m_pSpinLock(std::make_unique<utils::SpinLock>())
        {}
 
    public:
        ConsumerInput& input() {
            return m_input;
        }
 
        const ConsumerInput& input() const {
            return m_input;
        }
 
        DisruptorElementId id() const {
            return m_id;
        }
 
        bool isSkipped() const {
            utils::SpinLockGuard guard(*m_pSpinLock);
            return CompletionStatus::Aborted == m_result.CompletionStatus;
        }
 
        ConsumerCompletionResult completionResult() const {
            utils::SpinLockGuard guard(*m_pSpinLock);
            return m_result;
        }
 
    public:
        void markSkipped(PositionType position, CompletionCode code) {
            utils::SpinLockGuard guard(*m_pSpinLock);
            m_result.CompletionStatus = CompletionStatus::Aborted;
            m_result.CompletionCode = code;
            m_result.FinalConsumerPosition = position;
        }
 
        void markProcessingComplete() {
            m_processingComplete(m_id, m_result);
        }
 
    private:
        ConsumerInput m_input;
        DisruptorElementId m_id;
        ProcessingCompleteFunc m_processingComplete;
        ConsumerCompletionResult m_result;
        std::unique_ptr<utils::SpinLock> m_pSpinLock; // unique_ptr to allow moving of element
    };
 
    std::ostream& operator<<(std::ostream& out, const DisruptorElement& element);
 
    constexpr bool IsIntervalElementId(DisruptorElementId id, size_t interval) {
        return 0 != interval && 0 == id % interval;
    }
}}