JournalQueuingStrategy.h

#pragma once

#include <functional>
#include <list>
#include <map>

#include <Poco/DigestEngine.h>
#include <Poco/File.h>
#include <Poco/Path.h>
#include <Poco/Timespan.h>
#include <Poco/Timestamp.h>

#include "exporters/QueuingStrategy.h"
#include "util/Journal.h"
#include "util/Loggable.h"

namespace BeeeOn {

class JournalQueuingStrategy : public QueuingStrategy, protected Loggable {
public:
        JournalQueuingStrategy();

        void setRootDir(const std::string &path);

        Poco::Path rootDir() const;

        void setDisableGC(bool disable);

        void setNeverDropOldest(bool neverDrop);

        void setBytesLimit(int bytes);

        bool overLimit(size_t bytes) const;

        void setIgnoreIndexErrors(bool ignore);

        virtual void setup();

        bool empty() override;

        void push(const std::vector<SensorData> &data) override;

        size_t peek(std::vector<SensorData> &data, size_t count) override;

        void pop(size_t count) override;

protected:
        typedef std::function<void(
                const std::string &name,
                size_t offset,
                Poco::Timestamp &newest)> BrokenHandler;
        Poco::Timestamp initIndexAndScan(BrokenHandler broken);

        void initIndex(const Poco::Path &index);

        void prescanBuffers(Poco::Timestamp &newest, BrokenHandler broken);

        void reportStats(const Poco::Timestamp &newest) const;

        void inspectAndRegisterBuffer(
                const std::string &name,
                size_t offset,
                Poco::Timestamp &newest);

        Journal::Ptr index() const;

        std::string writeData(
                const std::string &data,
                bool force = true) const;

        bool whipeFile(Poco::File file, bool usuallyFails = false) const;

        void collectReferenced(std::set<std::string> &referenced) const;

        bool garbageCollect(const size_t bytes);

        void dropOldestBuffers(const size_t bytes);

        size_t bytesUsed() const;

        size_t bytesUsedAll() const;

        Poco::Path pathTo(const std::string &name) const;

        static std::string tsString(const Poco::Timestamp &t);

        class Entry {
        public:
                Entry(
                        const SensorData &data,
                        const std::string &buffer,
                        const size_t nextOffset);

                const SensorData &data() const;
                std::string buffer() const;
                size_t nextOffset() const;

        private:
                SensorData m_data;
                std::string m_buffer;
                size_t m_nextOffset;
        };

        size_t precacheEntries(size_t count);

        size_t readEntries(
                std::function<void(const Entry &entry)> proc,
                size_t count);

        struct FileBufferStat {
                Poco::Timestamp oldest = Poco::Timestamp::TIMEVAL_MAX;
                Poco::Timestamp newest = Poco::Timestamp::TIMEVAL_MIN;
                size_t bytes = 0;
                size_t offset = 0;
                size_t broken = 0;
                size_t count = 0;

                void update(const Poco::Timestamp &timestamp);
        };

        class FileBuffer {
        public:
                FileBuffer(
                        const Poco::Path &path,
                        size_t offset,
                        size_t size);

                std::string name() const;
                Poco::Path path() const;
                size_t offset() const;
                size_t size() const;

                bool exhausted() const;

                size_t readEntries(
                        std::function<void(const Entry &entry)> proc);

                size_t readEntries(
                        std::function<void(const Entry &entry)> proc,
                        const size_t count);

                void inspectAndVerify(
                        const Poco::DigestEngine::Digest &digest,
                        FileBufferStat &stat) const;

                static std::string formatEntries(
                        const std::vector<SensorData> &data);

        protected:
                size_t scanEntries(
                        size_t offset,
                        std::function<void(const Entry &entry)> proc,
                        size_t &bytes,
                        const size_t count) const;

                size_t scanEntries(
                        std::istream &in,
                        std::function<void(const Entry &entry)> proc,
                        size_t &bytes,
                        const size_t count) const;

        private:
                Poco::Path m_path;
                size_t m_offset;
                size_t m_size;
        };

        void registerBuffer(
                const FileBuffer &buffer,
                const FileBufferStat &stat);

private:
        Poco::Path m_rootDir;
        bool m_gcDisabled;
        bool m_neverDropOldest;
        ssize_t m_bytesLimit;
        bool m_ignoreIndexErrors;
        Journal::Ptr m_index;

        std::list<FileBuffer> m_buffers;

        std::map<std::string, FileBuffer> m_exhausted;

        std::list<Entry> m_entryCache;
};

}