#pragma once

#include "trygvis/sensor.h"

#include <mutex>
#include <boost/asio/buffer.hpp>

namespace trygvis {
namespace sensor {
namespace io {

using namespace std;
using namespace boost::asio;

struct sample_output_stream_option {
    virtual ~sample_output_stream_option() {
    };
};

struct output_fields_option : sample_output_stream_option {
    ~output_fields_option() {
    }

    vector<string> fields;
};

struct timestamp_field_option : sample_output_stream_option {
    timestamp_field_option(const string name) : name(name) {
    }

    ~timestamp_field_option() {
    }

    const string name;
};

class table_name_option : public sample_output_stream_option {
public:
    table_name_option(const string name) : name(name) {
    }

    ~table_name_option() {
    }

    const string name;
};

class sample_output_stream_options : public vector<sample_output_stream_option *> {
public:
    ~sample_output_stream_options() {
    }

    template<typename T>
    o<T *> find_option() const {
        for (auto it = begin(); it != end(); ++it) {
            T *x = dynamic_cast<T *>(*it);

            if (x != nullptr) {
                return o<T *>(x);
            }
        }

        return o<T *>();
    }
};

struct missing_required_option_error : runtime_error {
    missing_required_option_error(string what) : runtime_error(what) {
    }

    ~missing_required_option_error() {
    }
};

class SampleStreamParser;

class SampleOutputStream;

/**
 * Throws missing_required_option_error
 */
unique_ptr<SampleStreamParser> open_sample_stream_parser(
        shared_ptr<SampleOutputStream> output,
        KeyDictionary &dict,
        sample_format_type type = sample_format_type::AUTO);

/**
 * Throws missing_required_option_error
 */
unique_ptr<SampleOutputStream> open_sample_output_stream(
        shared_ptr<ostream> output,
        KeyDictionary &dict,
        sample_format_type type,
        sample_output_stream_options options);

static inline
unique_ptr<SampleOutputStream> open_sample_output_stream(
        shared_ptr<ostream> output,
        KeyDictionary &dict,
        sample_format_type type) {
    sample_output_stream_options options;
    return open_sample_output_stream(output, dict, type, options);
}

class SampleOutputStream {
public:
    virtual void write(SampleRecord const &sample) = 0;
};

class VectorSampleOutputStream : public SampleOutputStream {

public:
    virtual void write(SampleRecord const &sample) override;

public:
    vector<SampleRecord> samples;
};

class ThreadSafeSampleOutputStream : public SampleOutputStream {
public:
    ThreadSafeSampleOutputStream(unique_ptr<SampleOutputStream> underlying);

    ~ThreadSafeSampleOutputStream() {
    }

    void write(SampleRecord const &sample) override;

private:
    unique_ptr<SampleOutputStream> underlying;
    std::mutex mutex;
};

class AddTimestampSampleOutputStream : public SampleOutputStream {
public:
    AddTimestampSampleOutputStream(unique_ptr<SampleOutputStream> underlying, KeyDictionary &dict, const string &timestamp_name);

    ~AddTimestampSampleOutputStream() {
    }

    void write(SampleRecord const &sample) override;

private:
    unique_ptr<SampleOutputStream> underlying_;
    const SampleKey* timestamp_key;
};

class CsvSampleOutputStream : public SampleOutputStream {
public:
    CsvSampleOutputStream(shared_ptr<ostream> stream, KeyDictionary &dict);

    void write(SampleRecord const &sample);

private:
    void writeHeader();

    KeyDictionary &dict;
    shared_ptr<ostream> stream;
    bool headerWritten;
};

class JsonSampleOutputStream : public SampleOutputStream {
public:
    JsonSampleOutputStream(shared_ptr<ostream> stream, KeyDictionary &dict);

    void write(SampleRecord const &sample) override;

private:
    KeyDictionary &dict;
    shared_ptr<ostream> stream;
};

class KeyValueSampleOutputStream : public SampleOutputStream {
public:
    KeyValueSampleOutputStream(shared_ptr<ostream> stream, KeyDictionary &dict);

    void write(SampleRecord const &sample) override;

private:
    KeyDictionary &dict;
    shared_ptr<ostream> stream;
};

class RrdSampleOutputStream : public SampleOutputStream {
public:
    RrdSampleOutputStream(shared_ptr<ostream> stream, KeyDictionary &dict, const SampleKey *timestamp_key, o<output_fields_option *> output_fields);

    void write(SampleRecord const &sample) override;

private:
    vector<SampleKey *> keys;
    shared_ptr<ostream> stream;
    const SampleKey *timestamp_key;
};

class SqlSampleOutputStream : public SampleOutputStream {
public:
    SqlSampleOutputStream(shared_ptr<ostream> stream, KeyDictionary &dict, string table_name);

    void write(SampleRecord const &sample) override;

private:
    KeyDictionary &dict;
    shared_ptr<ostream> stream;
    const string table_name;
};

class SampleStreamParser {
public:
    virtual int process(mutable_buffers_1 &buffer) = 0;

    virtual int finish() = 0;

    virtual sample_format_type type() {
        return type_;
    }

    virtual KeyDictionary &dict() {
        return dict_;
    }

protected:
    sample_format_type type_;
    KeyDictionary &dict_;

    SampleStreamParser(const sample_format_type type, KeyDictionary &dict) : type_(type), dict_(dict) {
    }
};

class KeyValueSampleStreamParser : public SampleStreamParser {

public:
    KeyValueSampleStreamParser(shared_ptr<SampleOutputStream> output, KeyDictionary &dict) :
            SampleStreamParser(sample_format_type::CSV, dict), output(output),
            line(make_shared<vector<uint8_t>>()) {
    }

    int process(mutable_buffers_1 &buffer) override;

    int finish() override;

private:
    void process_line(shared_ptr<vector<uint8_t>> &packet);

    static const uint8_t packet_delimiter = '\n';
    shared_ptr<SampleOutputStream> output;
    shared_ptr<vector<uint8_t>> line;
};

class AutoSampleParser : public SampleStreamParser {
public:
    AutoSampleParser(shared_ptr<SampleOutputStream> output, KeyDictionary &dict);

    virtual int process(mutable_buffers_1 &buffer) override;

    virtual int finish() override;

private:
    unique_ptr<SampleStreamParser> parser;
    unique_ptr<KeyValueSampleStreamParser> keyValueParser;
};

static inline
unique_ptr<ThreadSafeSampleOutputStream> thread_safe_sample_output_stream(unique_ptr<SampleOutputStream> underlying) {
    return make_unique<ThreadSafeSampleOutputStream>(move(underlying));
};

}
}
}