<commit_before>#pragma once

#include <boost/filesystem.hpp>
#include <boost/filesystem/fstream.hpp>
#include <boost/version.hpp>

#include "blackhole/error.hpp"
#include "blackhole/factory.hpp"
#include "blackhole/sink/files/rotator.hpp"

namespace blackhole {

namespace sink {

class boost_backend_t {
    boost::filesystem::path m_path;
    boost::filesystem::ofstream m_file;
public:
    boost_backend_t(const std::string& path) :
        m_path(path)
    {
    }

    bool opened() const {
        return m_file.is_open();
    }

    bool open() {
        if (!create_directories(m_path.parent_path())) {
            return false;
        }

        m_file.open(m_path, std::ios_base::out | std::ios_base::app);
        return m_file.is_open();
    }

    std::string path() const {
        return m_path.string();
    }

    void write(const std::string& message) {
        m_file.write(message.data(), static_cast<std::streamsize>(message.size()));
        m_file.put('\n');
    }

    void flush() {
        m_file.flush();
    }

private:
    template<typename Path>
    bool create_directories(const Path& path) {
        try {
            boost::filesystem::create_directories(path);
        } catch (const boost::filesystem::filesystem_error&) {
            return false;
        }

        return true;
    }
};

namespace file {

template<class Rotator = void>
struct config_t {
    std::string path;
    bool autoflush;

    config_t(const std::string& path = "/dev/stdout", bool autoflush = true) :
        path(path),
        autoflush(autoflush)
    {}
};

template<class Backend>
struct config_t<rotator_t<Backend>> {
    std::string path;
    bool autoflush;
    rotator::config_t rotator;

    config_t(const std::string& path = "/dev/stdout", bool autoflush = true) :
        path(path),
        autoflush(autoflush)
    {}
};

} // namespace file

template<class Backend>
class writer_t {
    Backend& backend;
public:
    writer_t(Backend& backend) :
        backend(backend)
    {}

    void write(const std::string& message) {
        if (!backend.opened()) {
            if (!backend.open()) {
                throw error_t("failed to open file '%s' for writing", backend.path());
            }
        }
        backend.write(message);
    }
};

template<class Backend>
class flusher_t {
    bool autoflush;
    Backend& backend;
public:
    flusher_t(bool autoflush, Backend& backend) :
        autoflush(autoflush),
        backend(backend)
    {}

    void flush() {
        if (autoflush) {
            backend.flush();
        }
    }
};

template<class Backend = boost_backend_t, class Rotator = void>
class file_t {
    Backend m_backend;
    writer_t<Backend> m_writer;
    flusher_t<Backend> m_flusher;
public:
    typedef file::config_t<Rotator> config_type;

    static const char* name() {
        return "files";
    }

    file_t(const config_type& config) :
        m_backend(config.path),
        m_writer(m_backend),
        m_flusher(config.autoflush, m_backend)
    {}

    void consume(const std::string& message) {
        m_writer.write(message);
        m_flusher.flush();
        // if (rotator.necessary())
        //    rotator.rotate();
    }

    Backend& backend() {
        return m_backend;
    }
};

template<class Backend>
class file_t<Backend, void> {
    Backend m_backend;
    writer_t<Backend> m_writer;
    flusher_t<Backend> m_flusher;
public:
    typedef file::config_t<void> config_type;

    static const char* name() {
        return "files";
    }

    file_t(const config_type& config) :
        m_backend(config.path),
        m_writer(m_backend),
        m_flusher(config.autoflush, m_backend)
    {}

    void consume(const std::string& message) {
        m_writer.write(message);
        m_flusher.flush();
        // if (rotator.necessary())
        //    rotator.rotate();
    }

    Backend& backend() {
        return m_backend;<commit_msg>Added file sink specialization with rotation.<commit_after>#pragma once

#include <boost/filesystem.hpp>
#include <boost/filesystem/fstream.hpp>
#include <boost/version.hpp>

#include "blackhole/error.hpp"
#include "blackhole/factory.hpp"
#include "blackhole/sink/files/rotator.hpp"

namespace blackhole {

namespace sink {

class boost_backend_t {
    boost::filesystem::path m_path;
    boost::filesystem::ofstream m_file;
public:
    boost_backend_t(const std::string& path) :
        m_path(path)
    {
    }

    bool opened() const {
        return m_file.is_open();
    }

    bool open() {
        if (!create_directories(m_path.parent_path())) {
            return false;
        }

        m_file.open(m_path, std::ios_base::out | std::ios_base::app);
        return m_file.is_open();
    }

    std::string path() const {
        return m_path.string();
    }

    void write(const std::string& message) {
        m_file.write(message.data(), static_cast<std::streamsize>(message.size()));
        m_file.put('\n');
    }

    void flush() {
        m_file.flush();
    }

private:
    template<typename Path>
    bool create_directories(const Path& path) {
        try {
            boost::filesystem::create_directories(path);
        } catch (const boost::filesystem::filesystem_error&) {
            return false;
        }

        return true;
    }
};

namespace file {

template<class Rotator = void>
struct config_t {
    std::string path;
    bool autoflush;

    config_t(const std::string& path = "/dev/stdout", bool autoflush = true) :
        path(path),
        autoflush(autoflush)
    {}
};

template<class Backend>
struct config_t<rotator_t<Backend>> {
    std::string path;
    bool autoflush;
    rotator::config_t rotator;

    config_t(const std::string& path = "/dev/stdout", bool autoflush = true) :
        path(path),
        autoflush(autoflush)
    {}
};

} // namespace file

template<class Backend>
class writer_t {
    Backend& backend;
public:
    writer_t(Backend& backend) :
        backend(backend)
    {}

    void write(const std::string& message) {
        if (!backend.opened()) {
            if (!backend.open()) {
                throw error_t("failed to open file '%s' for writing", backend.path());
            }
        }
        backend.write(message);
    }
};

template<class Backend>
class flusher_t {
    bool autoflush;
    Backend& backend;
public:
    flusher_t(bool autoflush, Backend& backend) :
        autoflush(autoflush),
        backend(backend)
    {}

    void flush() {
        if (autoflush) {
            backend.flush();
        }
    }
};

template<class Backend = boost_backend_t, class Rotator = void, typename = void>
class file_t;

template<class Backend, class Rotator>
class file_t<Backend, Rotator, typename std::enable_if<!std::is_void<Rotator>::value>::type> {
    Backend m_backend;
    writer_t<Backend> m_writer;
    flusher_t<Backend> m_flusher;
    Rotator m_rotator;
public:
    typedef file::config_t<Rotator> config_type;

    static const char* name() {
        return "files";
    }

    file_t(const config_type& config) :
        m_backend(config.path),
        m_writer(m_backend),
        m_flusher(config.autoflush, m_backend),
        m_rotator(config.rotator, m_backend)
    {}

    void consume(const std::string& message) {
        m_writer.write(message);
        m_flusher.flush();
        if (m_rotator.necessary()) {
            m_rotator.rotate();
        }
    }

    Backend& backend() {
        return m_backend;
    }
};

template<class Backend>
class file_t<Backend, void, void> {
    Backend m_backend;
    writer_t<Backend> m_writer;
    flusher_t<Backend> m_flusher;
public:
    typedef file::config_t<void> config_type;

    static const char* name() {
        return "files";
    }

    file_t(const config_type& config) :
        m_backend(config.path),
        m_writer(m_backend),
        m_flusher(config.autoflush, m_backend)
    {}

    void consume(const std::string& message) {
        m_writer.write(message);
        m_flusher.flush();
        // if (rotator.necessary())
        //    rotator.rotate();
    }

    Backend& backend() {
        return m_backend;
{'fraction_non_alphanumeric': 0.12584740539874276, 'fraction_numerical': 0.0, 'mean_word_length': 2.5760246804759808, 'pattern_counts': {'":': 0, '<': 47, '<?xml version=': 0, '>': 47, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '333767', 'n_tokens_mistral': 2682, 'n_tokens_neox': 2643, 'n_words': 520}