#include <rados/librados.hpp>
#include <rados/buffer.h>
#include <rbd/librbd.hpp>
#include <string>
#include <vector>
#include <iostream>
#include <chrono>
#include <thread>
#include <atomic>
#include <regex>
#include <csignal>
#include <string.h>
// https://github.com/ricab/scope_guard.git
#include "./scope_guard.hpp"
// https://github.com/CLIUtils/CLI11.git
#include "./CLI11.hpp"

using namespace librados;
using namespace librbd;
using namespace sg;
using std::cerr;
using std::cout;
using std::endl;
using std::string;
using std::atomic;
using std::signal;

// export CEPH_ROOT=/home/runsisi/src/ceph
// g++ -o write write.cc -std=c++11 -I$CEPH_ROOT/src/include -L$CEPH_ROOT/build/lib -lrbd -lrados

atomic<bool> g_stop = {false};
void signal_handler(int s) {
    cout << "stopping.." << endl;
    g_stop = true;
};

int main(int argc, const char **argv) {
    string str_name = "client.admin";
    string str_conf = "./ceph.conf";
    string str_keyring = "./keyring";
    string str_pool = "rbd";
    string str_image = "";
    string str_image_spec = "";
    int int_msleep = 1000;

    CLI::App app;
    app.add_option("-n,--name", str_name, "client name used to connect to rados cluster");
    app.add_option("-c,--conf", str_conf, "ceph conf file");
    app.add_option("-k,--keyring", str_keyring, "cephx keyring file");
    app.add_option("-p,--pool", str_pool, "pool name");
    app.add_option("--image", str_image, "image name");
    app.add_option("image-spec", str_image_spec, "image spec");
    app.add_option("--sleep", int_msleep, "sleep between writes(in millisecond)");
    app.allow_extras(true);
    CLI11_PARSE(app, argc, argv);

    string str_tmp_pool = "";
    string str_tmp_image = "";
    if (!str_image_spec.empty()) {
        std::regex regex{R"([/]+)"}; // split on slash
        std::sregex_token_iterator it{str_image_spec.begin(), str_image_spec.end(), regex, -1};
        std::vector<string> words{it, {}};

        if (words.size() > 2) {
            cerr << "invalid image spec" << endl;
            return 1;
        }

        if (words.size() == 2) {
            str_tmp_pool = words[0];
            str_tmp_image = words[1];
        } else {
            str_tmp_pool = "rbd";
            str_tmp_image = words[0];
        }
    }

    if (str_tmp_image.empty() && str_image.empty()) {
        cerr << "image name / image spec must be provided" << endl;
        return 1;
    }

    if (!str_tmp_pool.empty() && !str_pool.empty() && str_tmp_pool != str_pool) {
        cerr << "incompatible pool name and image spec" << endl;
        return 1;
    }
    if (!str_tmp_image.empty() && !str_image.empty() && str_tmp_image != str_image) {
        cerr << "incompatible image name and image spec" << endl;
        return 1;
    }

    if (!str_tmp_pool.empty()) {
        str_pool = str_tmp_pool;
    }
    if (!str_tmp_image.empty()) {
        str_image = str_tmp_image;
    }

    if (int_msleep < 0) {
        cerr << "invalid sleep millisecond" << endl;
        return 1;
    }

    Rados client;
    int r = client.init2(str_name.c_str(), "ceph", 0);
    if (r < 0) {
        cerr << "librados.init2 failed: " << r << endl;
        return r;
    }
    r = client.conf_read_file(str_conf.c_str());
    if (r < 0) {
        cerr << "librados.conf_read_file failed: " << r << endl;
        return r;
    }

    std::vector<string> args = std::move(app.remaining());
    if (args.size()) {
        // rados.conf_parse_argv ignores the first argument
        int argc = args.size() + 1;
        const char **argv = (const char **)new const char*[argc];
        int i = 0;
        for (auto &a : args) {
            i++;
            argv[i] = a.c_str();
        }
        auto free_argv = make_scope_guard([argc, &argv]() {
            delete[] argv;
        });

        r = client.conf_parse_argv(argc, argv);
        if (r < 0) {
            cerr << "rados.conf_parse_argv failed: " << r << endl;
            return r;
        }
    }
    r = client.conf_set("keyring", str_keyring.c_str());
    if (r < 0) {
        cerr << "librados.conf_set failed: " << r << endl;
        return r;
    }

    r = client.connect();
    if (r < 0) {
        cerr << "librados.connect failed: " << r << endl;
        return r;
    }
    auto free_rados = make_scope_guard([&client]() {
        cout << "rados.shutdown" << endl;
        client.shutdown();
    });

    IoCtx ioctx;
    r = client.ioctx_create(str_pool.c_str(), ioctx);
    if (r < 0) {
        cerr << "librados.ioctx_create failed: " << r << endl;
        return r;
    }
    auto free_ioctx = make_scope_guard([&ioctx]() {
        cout << "ioctx.close" << endl;
        ioctx.close();
    });

    RBD rbd;
    Image img;
    r = rbd.open(ioctx, img, str_image.c_str());
    if (r < 0) {
        cerr << "librbd.open failed: " << r << endl;
        return r;
    }
    auto free_image = make_scope_guard([&img]() {
        cout << "image.close" << endl;
        img.close();
    });


    signal(SIGINT, signal_handler);
    signal(SIGTERM, signal_handler);

    while (!g_stop.load()) {
        // TODO: io pattern likes fio / rbd-bench
        int *arg = nullptr;
        RBD::AioCompletion *c = new RBD::AioCompletion(arg, [](void *pc, void *parg) {
            RBD::AioCompletion *c = (RBD::AioCompletion *)pc;
            int *arg = (int *)parg;

            int r = c->get_return_value();
            cout << "aio completed with: " << r << endl;

            c->release();
        });

        string data1("runsisi");
        bufferlist bl1;
        bl1.append(data1);
        r = img.aio_write2(0, bl1.length(), bl1, c, 0);
        if (r < 0) {
            cerr << "image.aio_write2 failed: " << r << endl;
            c->release();
        }

        std::this_thread::sleep_for(std::chrono::milliseconds(int_msleep));
    }

    return 0;
}