BrokerMessageReaders.h

Go to the documentation of this file.

 
#pragma once
#include "catapult/io/BufferInputStreamAdapter.h"
#include "catapult/io/FileQueue.h"
#include "catapult/utils/traits/Traits.h"
 
namespace catapult { namespace subscribers {
 
    // region Flusher
 
    namespace detail {
        template<typename TSubscriber, typename = void>
        struct Flusher {
            static void Flush(const TSubscriber&)
            {}
        };
 
        template<typename TSubscriber>
        struct Flusher<TSubscriber, utils::traits::is_type_expression_t<decltype(reinterpret_cast<TSubscriber*>(0)->flush())>> {
            static void Flush(TSubscriber& subscriber) {
                subscriber.flush();
            }
        };
    }
 
    // endregion
 
    template<typename TSubscriber, typename TMessageReader>
    void ReadAll(io::InputStream& inputStream, TSubscriber& subscriber, TMessageReader readNextMessage) {
        while (!inputStream.eof())
            readNextMessage(inputStream, subscriber);
 
        detail::Flusher<TSubscriber>::Flush(subscriber);
    }
 
    template<typename TSubscriber, typename TMessageReader>
    void ReadAll(io::FileQueueReader& reader, TSubscriber& subscriber, TMessageReader readNextMessage) {
        bool shouldContinue = true;
        while (shouldContinue) {
            shouldContinue = reader.tryReadNextMessage([&subscriber, readNextMessage](const auto& buffer) {
                io::BufferInputStreamAdapter<std::vector<uint8_t>> inputStream(buffer);
                ReadAll(inputStream, subscriber, readNextMessage);
            });
        }
    }
 
    struct MessageQueueDescriptor {
        std::string QueuePath;
 
        std::string IndexReaderFilename;
 
        std::string IndexWriterFilename;
    };
 
    template<typename TSubscriber, typename TMessageReader>
    void ReadAll(const MessageQueueDescriptor& descriptor, TSubscriber& subscriber, TMessageReader readNextMessage) {
        io::FileQueueReader reader(descriptor.QueuePath, descriptor.IndexReaderFilename, descriptor.IndexWriterFilename);
 
        auto numPendingMessages = reader.pending();
        if (0 == numPendingMessages)
            return;
 
        CATAPULT_LOG(debug) << "preparing to process " << numPendingMessages << " messages from " << descriptor.QueuePath;
        subscribers::ReadAll(reader, subscriber, readNextMessage);
    }
}}