inotify.cpp

// #include "inotify.hpp"

#include <boost/asio/buffer.hpp>
#include <boost/asio/detached.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/posix/stream_descriptor.hpp>
#include <boost/asio/stream_file.hpp>
#include <iostream>

#include <boost/asio/awaitable.hpp>
#include <boost/asio/co_spawn.hpp>
#include <boost/asio/experimental/as_tuple.hpp>
#include <boost/asio/experimental/awaitable_operators.hpp>
#include <boost/asio/redirect_error.hpp>
#include <boost/asio/steady_timer.hpp>
#include <boost/asio/stream_file.hpp>
#include <boost/asio/use_awaitable.hpp>
#include <boost/asio/experimental/awaitable_operators.hpp>

#include <sys/inotify.h>

using namespace std::chrono_literals;

using boost::asio::awaitable;
using boost::asio::co_spawn;
using boost::asio::redirect_error;
using boost::asio::use_awaitable;

// https://github.com/chriskohlhoff/talking-async/blob/master/episode1/step_7.cpp
constexpr auto use_nothrow_awaitable =
    boost::asio::experimental::as_tuple(use_awaitable);

using namespace boost::asio::experimental::awaitable_operators;

awaitable<size_t> readAll(std::string filename)
{
    auto executor = co_await boost::asio::this_coro::executor;
    
    boost::asio::steady_timer timer(executor);
    boost::asio::stream_file stream(executor);
    boost::system::error_code ec;
    stream.open(filename, boost::asio::stream_file::read_only, ec);

    size_t total = 0, count = 0;
    char data[256];
    for (;;)
    {
        timer.expires_after(250ms);
        auto result = co_await (timer.async_wait(use_awaitable) || 
                                stream.async_read_some(boost::asio::buffer(data), use_awaitable));
        if (result.index() == 0)
        {
            std::cout << "timeout, read " << total << " bytes" << std::endl;
            break;
        }

        total += std::get<size_t>(result);
    }

    co_return total;
}

awaitable<void> watch(boost::asio::posix::stream_descriptor &&desc) {
  auto executor = co_await boost::asio::this_coro::executor;

  boost::system::error_code ec;
  while (true) {
    std::array<char, sizeof(struct inotify_event) + NAME_MAX + 1> buf;
    auto n =
        co_await desc.async_read_some(boost::asio::buffer(buf), use_awaitable);

    size_t ofs = 0;
    while (n > 0) {
      inotify_event *ev = reinterpret_cast<inotify_event *>(buf.data() + ofs);
      size_t len = sizeof(inotify_event) + ev->len;
      assert(len <= buf.size() - ofs);
      std::cout << "event: " << n << " bytes, len=" << ev->len
                << " file: " << ev->name << std::endl;
      co_spawn(executor, readAll(std::string(ev->name)), boost::asio::detached);
      // co_await co_spawn(executor, readAll(std::string(ev->name)), use_awaitable);
      ofs += len;
      n -= len;
    }
  }
}

int main(int argc, char **argv) {
  if (argc < 2) {
    std::cerr << "Usage: " << argv[0] << " FILES...\n";
    return 1;
  }

  try {
    boost::asio::io_context ioc;

    auto fd = inotify_init();
    inotify_add_watch(fd, ".", IN_CLOSE_WRITE);

    boost::asio::posix::stream_descriptor desc(ioc, fd);

    co_spawn(ioc, watch(std::move(desc)), [&](std::exception_ptr pex) {
      try {
        if (pex)
          std::rethrow_exception(pex);
      } catch (const std::exception &ex) {
        std::cerr << "exception: " << ex.what() << std::endl;
      }
    });    
    ioc.run();
  } catch (std::exception const &exc) {
    std::cerr << exc.what() << std::endl;
  }
}