Simple log collector using libuv

#include <string>
#include <cstdio>
#include <unordered_map>
#if __has_include(<optional>)
#   include <optional>
#else
#   include <experimental/optional>
namespace std {
    using std::experimental::optional;
}
#endif
#include <utility>
#include <cassert>
#include <boost/coroutine2/all.hpp>
#include <set>
#include "uvw.hpp"

struct async {
    using coro_t = boost::coroutines2::coroutine<void>;

    template <typename EventType, typename EmitterType>
    EventType await(std::shared_ptr<EmitterType>& emitter) {
        int errorCode = 0;
        std::optional<EventType> okEvent;

        typename EmitterType::template Connection<EventType> ok;

        auto err = emitter->template once<uvw::ErrorEvent>([&](uvw::ErrorEvent &error, auto&) {
            emitter->template erase(ok);
            errorCode = error.code();
            (*this->pull)();
        });
        ok = emitter->template once<EventType>([&](EventType& event, auto&) {
            // These events are emitted in main routine
            emitter->template erase(err);
            okEvent = std::move(event);

            // Continue async code
            (*this->pull)();
        });

        // Goes back to the main routine, which eventually gives the control back to libuv
        (*this->push)();

        // `(*this->pull)()` inside the event goes here
        assert((okEvent && !errorCode) || (!okEvent && errorCode));
        return errorCode ? throw uvw::ErrorEvent(errorCode) : std::move(*okEvent);
    }

    // The callback should be noexcept
    template <typename Callback>
    static void run(Callback&& cb) {
        // We cannot use std::make_unique here or we must make the constructor public
        auto runner = std::unique_ptr<async>(new async(std::move(cb)));
        runner->instanceHolder = std::move(runner);
    }

private:
    async(std::function<void (async&)>&& _cb): callback(std::move(_cb)) {
        // Go into the subroutine here
        this->pull = coro_t::pull_type([&](coro_t::push_type& _push) {
            this->push = coro_t::push_type(std::move(_push));

            // All async code runs in the subroutine
            this->callback(*this);

            // async instance is destructed here
            this->instanceHolder = nullptr;
        });
    }

    async() = delete;
    async(const async&) = delete;

private:
    std::optional<coro_t::push_type> push;
    std::optional<coro_t::pull_type> pull;

private:
    // callback holder, to prevent lambda object from being destructed unexpectly
    std::function<void (async&)> callback;
    // instance holder, to prevent async instance itself from being destructed unexpectly
    std::unique_ptr<async> instanceHolder;
};

using namespace std::literals;

int main(int argc, char *argv[]) {
    using namespace uvw;

    if (argc != 2) {
        fputs("Usage: ./main FOLDER_TO_LISTEN\n", stderr);
        return 1;
    }
    const std::string path2listen = argv[1];

    auto loop = Loop::getDefault();
    auto fsEventHandle = loop->resource<FsEventHandle>();
    std::unordered_map<std::string, size_t> progresses;

    fsEventHandle->on<FsEventEvent>([&](FsEventEvent& event, FsEventHandle&) {
        // This function will return at the first await call.
        // It's expected behavior and unavoidable. Do NOT use its local variables after that.
        async::run([&, filename = std::string(event.filename)](async& runner) noexcept {
            // We should carefully hold other "closure variables" because the variables outside may have been destructed.
            // Local variables are safe to use
            const char *str;
            switch (event.flags) {
                case (int)FsEventHandle::Watch::RENAME:
                    str = "RENAME";
                    break;

                case (int)FsEventHandle::Watch::CHANGE:
                    str = "CHANGE";
                    break;

                case (int)FsEventHandle::Watch::RENAME | (int)FsEventHandle::Watch::CHANGE:
                    str = "RENAME | CHANGE";
                    break;

                default:
                    str = "";
                    break;
            }
            std::printf("[Detected changes] %s: %s\n", filename.c_str(), str);

            auto fileReq = loop->resource<FileReq>();

            try {
                fileReq->open(path2listen + "/" + filename, Flags<FileReq::FileOpen>(FileReq::FileOpen::RDONLY), S_IRUSR);
                runner.await<FsEvent<FsReq::Type::OPEN>>(fileReq); // Do NOT use `this->event` after the first await!

                fileReq->stat();
                const auto fileSize = runner.await<FsEvent<FsReq::Type::FSTAT>>(fileReq).stat.st_size;

                const auto iter = progresses.try_emplace(filename, 0).first;
                const auto progress = std::exchange(iter->second, fileSize);

                const auto appendedSize = static_cast<unsigned int>(fileSize - progress);
                if (appendedSize) { // The changed contents may have been handled by other event.
                    fileReq->read(progress, appendedSize);
                    const auto readEvent = runner.await<FsEvent<FsReq::Type::READ>>(fileReq);
                    const auto fileContent = std::string(readEvent.data.get(), readEvent.size);
                    std::printf("[Contents appended]: %s\n", fileContent.c_str());
                }

                fileReq->close();
                runner.await<FsEvent<FsReq::Type::CLOSE>>(fileReq);
            } catch (ErrorEvent& e) {
                std::fprintf(stderr, "Error happens: %s\n", e.what());
                progresses.erase(filename);
                fileReq->closeSync();
            }
        });
    });

    fsEventHandle->start(path2listen);
    loop->run();
}