#include <cassert>
#include <condition_variable>
#include <curl/curl.h>
#include <thread>
#include <atomic>
#include <iostream>
#include <map>
#include <mutex>
#include <set>
#include <vector>
#include <memory>

class TransferItem2
{
    std::mutex lk;
    std::vector<char> buf1;
    std::condition_variable cv;
    bool done = false;
public:
    std::unique_ptr<CURL, decltype([](auto * c) { curl_easy_cleanup(c); })> req;

    static size_t writeCallbackWrapper(void * contents, size_t size, size_t nmemb, void * userp)
    {
        return static_cast<TransferItem2 *>(userp)->writeCallback(contents, size, nmemb);
    }

    size_t writeCallback(void *contents, size_t size, size_t nmemb)
    {
        const size_t realSize = size * nmemb;
        std::lock_guard<std::mutex> _l(lk);

        if (buf1.size() > 1024 * 1024) {
            std::cout << "pause " << req.get() << "\n";
            return CURL_WRITEFUNC_PAUSE;
        }

        // std::cout << "recv " << req.get() << " bytes: " << realSize << "\n";

        buf1.insert(buf1.end(), realSize, realSize + size);
        cv.notify_all();
        return realSize;
    }

    // In a real application this would actually do something with the data.
    bool consume()
    {
        std::unique_lock<std::mutex> g(lk);
        if (done) return true;
        if (buf1.empty()) {
            cv.wait(g);
        }
        auto _ = std::move(buf1);
        buf1 = {};
        return done;
    }

    void finish(CURLcode code)
    {
        std::cout << "Finish item " << req.get() << " with code " << code << " error: " << curl_easy_strerror(code) << "\n";
        done = true;
    }

    static int debugCallback(CURL * handle, curl_infotype type, char * data, size_t size, void * userptr)
    {
        if (type == CURLINFO_TEXT)
            std::cout << "curl: " << std::string(data, size) << "\n";
        return 0;
    }

    TransferItem2(std::string uri) : req(curl_easy_init()) {
        curl_easy_setopt(req.get(), CURLOPT_VERBOSE, 1);
        curl_easy_setopt(req.get(), CURLOPT_DEBUGFUNCTION, TransferItem2::debugCallback);
        curl_easy_setopt(req.get(), CURLOPT_URL, uri.c_str());
        curl_easy_setopt(req.get(), CURLOPT_FOLLOWLOCATION, 1L);
        curl_easy_setopt(req.get(), CURLOPT_ACCEPT_ENCODING, ""); // all of them!
        curl_easy_setopt(req.get(), CURLOPT_MAXREDIRS, 10);
        curl_easy_setopt(req.get(), CURLOPT_NOSIGNAL, 1);
        curl_easy_setopt(req.get(), CURLOPT_PIPEWAIT, 1);
        curl_easy_setopt(req.get(), CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_2TLS);
        curl_easy_setopt(req.get(), CURLOPT_WRITEFUNCTION, TransferItem2::writeCallbackWrapper);
        curl_easy_setopt(req.get(), CURLOPT_WRITEDATA, this);

        curl_easy_setopt(req.get(), CURLOPT_NOPROGRESS, 1);

        curl_easy_setopt(req.get(), CURLOPT_PROTOCOLS_STR, "http,https,ftp,ftps");

        curl_easy_setopt(req.get(), CURLOPT_CONNECTTIMEOUT, 5);

        curl_easy_setopt(req.get(), CURLOPT_LOW_SPEED_LIMIT, 1L);
        curl_easy_setopt(req.get(), CURLOPT_LOW_SPEED_TIME, 300);
    }
};

class Transferrer
{
    std::unique_ptr<CURLM, decltype([](auto * m) { curl_multi_cleanup(m); })> curlm;
    std::thread thread;
    std::atomic<bool> stop = false;
    std::map<CURL *, std::shared_ptr<TransferItem2>> items;

    std::vector<std::shared_ptr<TransferItem2>> incoming;
    std::set<std::shared_ptr<TransferItem2>> unpause;
    std::mutex lk;

    public:
    void add(std::shared_ptr<TransferItem2> item)
    {
        auto _l = std::lock_guard(lk);
        incoming.push_back(item);
    }

    void requestUnpause(std::shared_ptr<TransferItem2> item)
    {
        std::cout << "requestUnpause " << item->req.get() << "\n";
        std::lock_guard<std::mutex> _l(lk);
        unpause.insert(item);
    }

    bool consume(std::shared_ptr<TransferItem2> item)
    {
        requestUnpause(item);
        return item->consume();
    }

    void transferThread()
    {
        while (!stop) {
            int running;
            CURLMcode mc = curl_multi_perform(curlm.get(), &running);
            if (mc != CURLM_OK)
                throw std::runtime_error("curl_multi_perform");

            CURLMsg * msg;
            int left;
            while ((msg = curl_multi_info_read(curlm.get(), &left))) {
                if (msg->msg == CURLMSG_DONE) {
                    auto i = items.find(msg->easy_handle);
                    assert(i != items.end());
                    i->second->finish(msg->data.result);
                    curl_multi_remove_handle(curlm.get(), i->second->req.get());
                    items.erase(i);
                }
            }

            /* Wait for activity, including wakeup events. */
            mc = curl_multi_poll(curlm.get(), nullptr, 0, 64, nullptr);
            if (mc != CURLM_OK)
                throw std::runtime_error("unexpected error from curl_multi_poll()");

            {
                auto unpause_ = ([&]() {
                    std::lock_guard _l(lk);
                    auto ret = std::move(this->unpause);
                    this->unpause.clear();
                    return ret;
                })();

                for (auto & item : unpause_) {
                    std::cout << "unpause " << item->req.get() << "\n";
                    curl_easy_pause(item->req.get(), CURLPAUSE_CONT);
                }
            }


            {
                auto _l = std::lock_guard(lk);
                auto incoming_ = std::move(incoming);
                incoming.clear();

                for (auto && tf : incoming_) {
                    items.insert(std::make_pair(tf->req.get(), tf));
                    curl_multi_add_handle(curlm.get(), tf->req.get());
                }
            }
        }
    }

    Transferrer() : curlm(curl_multi_init()) {
        static std::once_flag globalInit;
        std::call_once(globalInit, curl_global_init, CURL_GLOBAL_ALL);

        curl_multi_setopt(curlm.get(), CURLMOPT_PIPELINING, CURLPIPE_MULTIPLEX);
        curl_multi_setopt(curlm.get(), CURLMOPT_MAX_TOTAL_CONNECTIONS, 10);

        thread = std::thread([&]() {
            transferThread();
        });
    }

    ~Transferrer()
    {
        stop = true;
        thread.join();
    }
};


int main()
{
    auto t1 = std::make_shared<TransferItem2>("https://jade.fyi/zeros.txt");
    auto t2 = std::make_shared<TransferItem2>("https://jade.fyi/zeros.txt");
    auto mgr = Transferrer{};
    std::cout << "t1 is " << t1->req.get() << ", t2 is " << t2->req.get() << "\n";

    // take a little byte out of t1 and t2 so that they are definitely in a data transfer phase
    mgr.add(t1);
    mgr.consume(t1);
    mgr.add(t2);
    mgr.consume(t2);

    // hungrily take more bytes out of t1
    while (!mgr.consume(t1)) {
        std::cout << "omnomnom\n";
    }
}