nghttp2_coro.cpp

//
// Test alternative NGHTTP2 ASIO implementation with coroutines.
//
#include "nghttp2/nghttp2.h"
#include <boost/asio/any_io_executor.hpp>
#include <boost/asio/as_tuple.hpp>
#include <boost/asio/associated_cancellation_slot.hpp>
#include <boost/asio/cancellation_type.hpp>
#include <boost/asio/deferred.hpp>
#include <boost/asio/steady_timer.hpp>
#include <boost/asio/this_coro.hpp>
#include <boost/asio/use_awaitable.hpp>
#include <boost/beast/core/tcp_stream.hpp>
#include <boost/system/detail/error_code.hpp>
#include <iostream>
#include <string>
#include <type_traits>
#include <span>

#include <boost/asio.hpp>
#include <boost/asio/buffer.hpp>
#include <boost/asio/experimental/co_composed.hpp>
#include <boost/asio/experimental/channel.hpp>
#include <boost/asio/experimental/awaitable_operators.hpp>
#include <boost/asio/experimental/cancellation_condition.hpp>
#include <boost/asio/experimental/coro.hpp>

#include <boost/beast.hpp>

#include <fmt/ostream.h>
#include <fmtlog.h>

using namespace std::chrono_literals;

namespace net = boost::asio;
namespace beast = boost::beast;

using tcp = net::ip::tcp;       // from <boost/asio/ip/tcp.hpp>
using net::awaitable;
using net::experimental::coro;
using net::co_spawn;
using net::use_awaitable;

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

#include <nghttp2/nghttp2.h>
#include <nghttp2/asio_http2.h>
#include <nghttp2/asio_http2_server.h>

#include "detect_http2.hpp"

// https://fmt.dev/latest/api.html#std-ostream-support
template <> struct fmt::formatter<tcp::endpoint> : ostream_formatter {};

//
// https://www.boost.org/doc/libs/1_81_0/doc/html/boost_asio/reference/experimental__co_composed.html
//
template <typename AsyncReadStream, typename DynamicBuffer, typename CompletionToken>
auto async_detect_http2_client_preface_coro(AsyncReadStream &stream,
                                            DynamicBuffer &buffer,
                                            CompletionToken &&token) {
    static_assert(net::is_dynamic_buffer<DynamicBuffer>::value,
                  "DynamicBuffer type requirements not met");

    using namespace boost::asio;
    return async_initiate<CompletionToken, void(boost::system::error_code, size_t)>(
        experimental::co_composed<void(boost::system::error_code, bool)>(
            [](auto state, AsyncReadStream &stream, DynamicBuffer& buffer) -> void {
              try {
                state.throw_if_cancelled(true);
                state.reset_cancellation_state(enable_terminal_cancellation());

                //
                // FIXME: As a coroutine, we don't even need to loop like this.
                //        Instead, we could just read the minimum number of bytes needed for detection
                //        in one async call to async_read(). But we keep it like this for now, leaving
                //        it up to the detection function (is_http2_client_preface) to request as much
                //        additional buffer as it needs, dynamically.
                //
                for (;;)
                {
                    boost::tribool result = boost::beast::detail::is_http2_client_preface(buffer.cdata());
                    if (!boost::indeterminate(result))
                        co_return {{}, static_cast<bool>(result)};

                    auto prepared = buffer.prepare(read_size(buffer, 1536));
                    size_t n = co_await stream.async_read_some(prepared, deferred);
                    buffer.commit(n);
                }
              }
              catch (const boost::system::system_error &e) {
                std::cout << "ERROR: " << e.what() << std::endl;
                co_return {e.code(), false};
              }
            },
            stream),
        token, std::ref(stream), std::ref(buffer));
}

// inspired by <http://blog.korfuri.fr/post/go-defer-in-cpp/>, but our
// template can take functions returning other than void.
template <typename F, typename... T> struct Defer {
  Defer(F &&f, T &&...t) : f(std::bind(std::forward<F>(f), std::forward<T>(t)...)) {}
  Defer(Defer &&o) noexcept : f(std::move(o.f)) {}
  ~Defer() { f(); }

  using ResultType = std::invoke_result_t<F, T...>;
  std::function<ResultType()> f;
};

template <typename F, typename... T> Defer<F, T...> defer(F &&f, T &&...t) {
  return Defer<F, T...>(std::forward<F>(f), std::forward<T>(t)...);
}

// https://stackoverflow.com/questions/75758817/boost-asio-channels-no-type-named-receive-cancelled-signature
// https://think-async.com/Asio/boost_asio_1_22_2/doc/html/boost_asio/overview/channels.html
using data_channel = net::experimental::channel<void(boost::system::error_code, std::vector<uint8_t>)>;

using send_channel = net::experimental::channel<void(boost::system::error_code ec)>;

// ========================================================================================================

//
// https://www.boost.org/doc/libs/1_82_0/doc/html/boost_asio/example/cpp20/operations/callback_wrapper.cpp
//
template <boost::asio::completion_token_for<void(std::string)> CompletionToken>
auto async_read_input(const std::string& prompt, CompletionToken&& token)
{
  // Define a function object that contains the code to launch the asynchronous
  // operation. This is passed the concrete completion handler, followed by any
  // additional arguments that were passed through the call to async_initiate.
  auto init = [](
      boost::asio::completion_handler_for<void(std::string)> auto handler,
      const std::string& prompt)
  {
    // According to the rules for asynchronous operations, we need to track
    // outstanding work against the handler's associated executor until the
    // asynchronous operation is complete.
    auto work = boost::asio::make_work_guard(handler);

    // Launch the operation with a callback that will receive the result and
    // pass it through to the asynchronous operation's completion handler.
    read_input(prompt,
        [
          handler = std::move(handler),
          work = std::move(work)
        ](std::string result) mutable
        {
          // Get the handler's associated allocator. If the handler does not
          // specify an allocator, use the recycling allocator as the default.
          auto alloc = boost::asio::get_associated_allocator(
              handler, boost::asio::recycling_allocator<void>());

          // Dispatch the completion handler through the handler's associated
          // executor, using the handler's associated allocator.
          boost::asio::dispatch(work.get_executor(),
              boost::asio::bind_allocator(alloc,
                [
                  handler = std::move(handler),
                  result = std::string(result)
                ]() mutable
                {
                  std::move(handler)(result);
                }));
        });
  };

  // The async_initiate function is used to transform the supplied completion
  // token to the completion handler. When calling this function we explicitly
  // specify the completion signature of the operation. We must also return the
  // result of the call since the completion token may produce a return value,
  // such as a future.
  return boost::asio::async_initiate<CompletionToken, void(std::string)>(
      init, // First, pass the function object that launches the operation,
      token, // then the completion token that will be transformed to a handler,
      prompt); // and, finally, any additional arguments to the function object.
}


//
// TEST: try to wrap nghttp2 into example above
//
template <boost::asio::completion_token_for<void(std::string)> CompletionToken>
auto async_read_stream(CompletionToken&& token)
{
    auto init = [](
      boost::asio::completion_handler_for<void(std::string)> auto handler)
    {
        auto work = boost::asio::make_work_guard(handler);
        // INCOMPLETE. A channel for forwarding from the callback based API seems to work just fine, see below.
    };

  return boost::asio::async_initiate<CompletionToken, void(std::string)>(
      init, // First, pass the function object that launches the operation,
      token // then the completion token that will be transformed to a handler,
      ); // and, finally, any additional arguments to the function object.
}

// -----------------------------------------------------------------------------------------------------


// Create nghttp2_nv from string literal |name| and std::string
// |value|.
template <size_t N>
nghttp2_nv make_nv_ls(const char (&name)[N], const std::string &value) {
  return {(uint8_t *)name, (uint8_t *)value.c_str(), N - 1, value.size(),
          NGHTTP2_NV_FLAG_NO_COPY_NAME};
}

// -----------------------------------------------------------------------------------------------------

using mystream = net::as_tuple_t<net::use_awaitable_t<>>::as_default_on_t<beast::tcp_stream>;

class Stream;
class Session
{
public:
    Session(tcp::socket&& socket) : m_stream(std::move(socket)), m_executor(m_stream.socket().get_executor()), m_send_channel(m_executor) {}
    ~Session() { logi("Session: dtor"); }

    awaitable<void> do_session();
    awaitable<void> send_loop(mystream& stream, nghttp2_session* session);
    awaitable<void> read_loop(mystream& stream, nghttp2_session* session);

    void create_stream(int stream_id)
    {
        m_streams.emplace(stream_id, std::make_shared<Stream>(*this, stream_id));
    }

    Stream *find_stream(int32_t stream_id)
    {
        if (auto it = m_streams.find(stream_id); it != std::end(m_streams))
            return it->second.get();
        else
            return nullptr;
    }

    void close_stream(int32_t stream_id)
    {
        m_streams.erase(stream_id);
    }

    void start_write()
    {
        std::ignore = m_send_channel.try_send(boost::system::error_code{});
    }

    const net::any_io_executor& executor() const { return m_executor; }

private:
    net::as_tuple_t<net::use_awaitable_t<>>::as_default_on_t<beast::tcp_stream> m_stream;
    nghttp2_session *m_session = nullptr;
    net::any_io_executor m_executor;
    std::map<int, std::shared_ptr<Stream>> m_streams;
    send_channel m_send_channel;
};


class Stream
{
public:
    size_t pending = 0;
    size_t unhandled = 0;

    Stream(Session& parent, int id_) : parent(parent), reader(parent.executor()), writer(parent.executor()), id(id_)
    {
        // std::cout << "[" << id << "] ctor" << std::endl;
    }

    void call_on_data(nghttp2_session *session, int32_t id_, const uint8_t* data, size_t len)
    {
        assert(id == id_);
        // std::cout << "[" << stream_id << "] <<< read callback: " << len << " bytes..." << std::endl;

        if (len)
            nghttp2_session_consume_connection(session, len);

        pending += len;
        unhandled += len;

        auto buffer = std::vector<uint8_t>(data, data + len);
        bool ok = reader.try_send(boost::system::error_code{}, std::move(buffer));
        if (ok)
        {
            // std::cout << "[" << stream_id << "] <<<   try_send: succeeded" << std::endl;
            auto ok = nghttp2_session_consume_stream(session, id, len);
            pending -= len;
            return;
        }

        assert(buffer.size() == len);
        // std::cout << "[" << stream_id << "] <<<   async_send..." << std::endl;
        reader.async_send(boost::system::error_code{}, std::move(buffer), [this, session, len]
                    (boost::system::error_code ec) {
            if (ec)
                std::cout << "[" << id << "] <<< reader.async_send... error: " << ec.what() << std::endl;
            else
            {
                // std::cout << "[" << stream_id << "] <<< reader.async_send... done, " << len << " bytes (pending=" << pending << ")" << std::endl;
                auto ok = nghttp2_session_consume_stream(session, id, len);
                pending -= len;
            }
        });
    }

    ~Stream()
    {
        // std::cout << "[" << id << "] dtor"  << std::endl;
    }

    size_t repeat = 1000;

    nghttp2::asio_http2::generator_cb::result_type
    call_read(uint8_t *data, std::size_t len, uint32_t *data_flags)
    {
        // std::cout << "call_read (buffer size=" << len << " bytes)" << std::endl;
        assert(len >= 13);
        memcpy(data, "Hello World!\n", 13);
        // if (--repeat == 0)
             *data_flags |= NGHTTP2_DATA_FLAG_EOF;
        return 13;
    }

#if 0
    //
    // simple custom coroutine for wrapping an async API with a callback
    //
    // https://stackoverflow.com/questions/64270626/turning-a-function-call-which-takes-a-callback-into-a-coroutine
    auto api_call_async(const std::string& some_parameter)
    {
          struct awaiter : public std::suspend_always
          {
               awaiter(const std::string &parameter) : parameter_(parameter) {}
               bool await_ready() { return true; }
               void await_suspend(std::coroutine_handle<> handle)
               {

                   // use your third party lib call directly here.
                   api_call(parameter_, [handle](std::error_code ec)
                   {
                       // call the handle to resume the coroutine
                       handle();
                   });
               }

               std::string parameter_;
          };
          return awaiter(some_parameter);
    }
#endif

    // https://think-async.com/Asio/asio-1.22.1/doc/asio/overview/composition/coro.html
    // This seems to be an async transformer... not sure if we need that here
    coro<std::string_view(uint8_t* buf)> coro_test(net::any_io_executor)
    {
        co_yield "test";
    }

    awaitable<void> do_request(nghttp2_session* session)
    {
        // std::cout << "[" << id << "] call_on_request: " << id << std::endl;

        auto nva = std::vector<nghttp2_nv>();
        nva.reserve(2);
        std::string status = "200";
        std::string date = "Sat, 01 Apr 2023 09:33:09 GMT";
        nva.push_back(make_nv_ls(":status", status));
        nva.push_back(make_nv_ls("date", date));

        nghttp2_data_provider prd;
        prd.source.ptr = this;
        //
        // https://nghttp2.org/documentation/types.html#c.nghttp2_data_source_read_callback
        //
        prd.read_callback = [](nghttp2_session *session, int32_t stream_id,
                               uint8_t *buf, size_t length, uint32_t *data_flags,
                               nghttp2_data_source *source,
                              void *user_data) -> ssize_t {
            auto stream = static_cast<Stream *>(source->ptr);
            // std::cout << "[" << stream_id << "] >>> write callback (buffer size=" << length << " bytes)" << std::endl;

            if (stream->sendBuffers.empty())
            {
                stream->writer.try_receive([&](boost::system::error_code ec, std::vector<uint8_t> buffer){
                    // std::cout << "[" << stream_id << "] >>>  try_receive: got buffer, len=" << buffer.size() << std::endl;
                    stream->sendBuffers.emplace_back(std::move(buffer));
                    if (stream->sendBuffers.size() == 1)
                        stream->sendBufferView = boost::asio::buffer(stream->sendBuffers.back());
                });
            }

            if (stream->sendBuffers.size() == 1 && stream->sendBuffers[0].empty())
            {
                // std::cout << "[" << stream_id << "] >>>   EOF" << std::endl;
                *data_flags |= NGHTTP2_DATA_FLAG_EOF;
                return 0;
            }

            if (stream->sendBuffers.empty())
            {
                // std::cout << "[" << stream_id << "] >>>   no buffer, DEFERRING" << std::endl;
                stream->writer.async_receive([stream, session](boost::system::error_code ec, std::vector<uint8_t> buffer){
                    // std::cout << "[" << stream->id << "] >>> writer.async_receive: got buffer, len=" << buffer.size() << ", resuming..." << std::endl;
                    stream->sendBuffers.emplace_back(std::move(buffer));
                    if (stream->sendBuffers.size() == 1)
                        stream->sendBufferView = boost::asio::buffer(stream->sendBuffers.back());
                    nghttp2_session_resume_data(session, stream->id);
                    stream->parent.start_write();
                });
                return NGHTTP2_ERR_DEFERRED;
            }

            assert(stream->sendBuffer.front().size() > 0);
            assert(stream->sendBufferView.data() >= stream->sendBuffer.front().data());
            assert(stream->sendBufferView.data() < stream->sendBuffer.front().data() + stream->sendBuffer[0].size());

            auto copied = boost::asio::buffer_copy(boost::asio::buffer(buf, length), stream->sendBufferView);
            // std::cout << "[" << stream_id << "] >>>   copied " << copied << " bytes" << std::endl;
            if (copied == stream->sendBufferView.size())
                stream->sendBuffers.pop_front();
            else
                stream->sendBufferView = boost::asio::buffer(stream->sendBuffers.front().data() + copied, stream->sendBuffers.front().size() - copied);
            return copied;
        };

        auto rv = nghttp2_submit_response(session, id, nva.data(), nva.size(), &prd);

        for (;;)
        {
            auto buffer = co_await reader.async_receive(use_awaitable);

            auto len = buffer.size();
            unhandled -= len;

            co_await writer.async_send(boost::system::error_code{}, std::move(buffer), use_awaitable);

            if (len == 0)
                break;
        }

        // std::cout << "[" << id << "] do_request() finished" << std::endl;

        co_return;
    }

    void call_on_request(nghttp2_session* session)
    {
        co_spawn(executor(), do_request(session), [](std::exception_ptr pex) {
            try
            {
                if (pex)
                    std::rethrow_exception(pex);
            }
            catch (const std::exception &ex)
            {
                std::cerr << "REQUEST ERROR: " << ex.what() << std::endl;
            }
        });
    }

    inline const net::any_io_executor executor() const { return parent.executor(); }

public:
    Session& parent;
    data_channel reader;
    data_channel writer;
    std::deque<std::vector<uint8_t>> sendBuffers;
    boost::asio::const_buffer sendBufferView;
    int id;
};

// ========================================================================================================

int on_begin_headers_callback(nghttp2_session *session, const nghttp2_frame *frame, void *user_data)
{
  auto handler = static_cast<Session *>(user_data);

  if (frame->hd.type != NGHTTP2_HEADERS ||
      frame->headers.cat != NGHTTP2_HCAT_REQUEST) {
    return 0;
  }

  handler->create_stream(frame->hd.stream_id);
  return 0;
}


int on_frame_recv_callback(nghttp2_session *session, const nghttp2_frame *frame, void *user_data)
{
  auto handler = static_cast<Session *>(user_data);
  auto strm = handler->find_stream(frame->hd.stream_id);

  // std::cout << "[" << frame->hd.stream_id << "] on_frame_recv_callback " << std::endl;

  switch (frame->hd.type) {
  case NGHTTP2_DATA:
    if (!strm) {
      break;
    }

    if (frame->hd.flags & NGHTTP2_FLAG_END_STREAM) {
      strm->call_on_data(session, frame->hd.stream_id, nullptr, 0);
    }

    break;

  case NGHTTP2_HEADERS: {
    if (!strm || frame->headers.cat != NGHTTP2_HCAT_REQUEST) {
      break;
    }

    strm->call_on_request(session);
    if (frame->hd.flags & NGHTTP2_FLAG_END_STREAM) {
      strm->call_on_data(session, frame->hd.stream_id, nullptr, 0);
    }

    handler->start_write();

    break;
  }
  }

  return 0;
}

int on_data_chunk_recv_callback(nghttp2_session *session, uint8_t flags,
                                int32_t stream_id, const uint8_t *data,
                                size_t len, void *user_data) {
  auto handler = static_cast<Session *>(user_data);
  auto strm = handler->find_stream(stream_id);

  if (!strm) {
    return 0;
  }

  strm->call_on_data(session, stream_id, data, len);
  handler->start_write(); // might re-open windows

  return 0;
}


int on_frame_send_callback(nghttp2_session *session, const nghttp2_frame *frame, void *user_data)
{
    // std::cout << "on_frame_send_callback:" << std::endl;
    return 0;
}

int on_stream_close_callback(nghttp2_session *session, int32_t stream_id,
                             uint32_t error_code, void *user_data) {
    auto handler = static_cast<Session *>(user_data);
    auto strm = handler->find_stream(stream_id);

    // std::cout << "[" << stream_id << "] on_stream_close_callback" << std::endl;

    handler->close_stream(stream_id);

    return 0;
}

// ========================================================================================================

awaitable<void>
Session::do_session()
{
    m_stream.socket().set_option(tcp::no_delay(true));

    beast::flat_buffer buffer;

    // integration with asio's awaitable doesn't work, yet
    // co_await simplest_coro();

    //
    // detect HTTP2 client preface, abort connection if not found
    //
    if (!co_await async_detect_http2_client_preface_coro(m_stream, buffer, use_awaitable))
    {
        std::cout << "no HTTP2 client preface detected (" << buffer.size() << " bytes in buffer), closing connection" << std::endl;
        auto socket = m_stream.release_socket();
        socket.shutdown(net::ip::tcp::socket::shutdown_send);
        socket.close();
        co_return;
    }

    //
    //  read request headers
    //
    nghttp2_session_callbacks *callbacks;
    auto rv = nghttp2_session_callbacks_new(&callbacks);
    if (rv != 0)
        throw std::runtime_error("nghttp2_session_callbacks_new");

    auto cb_del = defer(nghttp2_session_callbacks_del, callbacks);

    //
    // https://nghttp2.org/documentation/nghttp2_session_server_new.html
    //
    // At a minimum, send and receive callbacks need to be specified.
    //
    nghttp2_session_callbacks_set_on_begin_headers_callback(callbacks, on_begin_headers_callback);
    // nghttp2_session_callbacks_set_on_header_callback(callbacks, on_header_callback);
    nghttp2_session_callbacks_set_on_frame_recv_callback(callbacks, on_frame_recv_callback);
    nghttp2_session_callbacks_set_on_data_chunk_recv_callback(callbacks, on_data_chunk_recv_callback);
    nghttp2_session_callbacks_set_on_stream_close_callback(callbacks, on_stream_close_callback);
    nghttp2_session_callbacks_set_on_frame_send_callback(callbacks, on_frame_send_callback);
    // nghttp2_session_callbacks_set_on_frame_not_send_callback(callbacks, on_frame_not_send_callback);

    // disable automatic WINDOW update as we are sending window updates ourselves
    // https://github.com/nghttp2/nghttp2/issues/446
    nghttp2_option* options;
    nghttp2_option_new(&options);
    nghttp2_option_set_no_auto_window_update(options, 1);

    nghttp2_session *session_;
    rv = nghttp2_session_server_new2(&session_, callbacks, this, options);
    if (rv != 0)
        throw std::runtime_error("nghttp2_session_server_new");

    nghttp2_settings_entry ent{NGHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS, 100};
    nghttp2_submit_settings(session_, NGHTTP2_FLAG_NONE, &ent, 1);

    //
    // Let NGHTTP2 parse what we have received so far.
    // This must happen after submitting the server settings.
    //
    rv = nghttp2_session_mem_recv(session_, static_cast<uint8_t*>(buffer.data().data()), buffer.size());
    if (rv < 0 || rv != buffer.size())
        throw std::runtime_error("nghttp2_session_mem_recv");
    buffer.clear();

    //
    // read/send loop
    //
    co_await (send_loop(m_stream, session_) && read_loop(m_stream, session_));
    logi("do_session() finished");
}

awaitable<void>
Session::send_loop(mystream& stream, nghttp2_session* session)
{
    // auto state = co_await boost::asio::this_coro::cancellation_state;
    // while (state.cancelled() == boost::asio::cancellation_type::none)
    for (;;)
    {
        const uint8_t *data;
        // data is valid until next call, so we don't need to copy it

        logd("nghttp2_session_mem_send...");
        auto nread = nghttp2_session_mem_send(session, &data);
        logd("nghttp2_session_mem_send... {} bytes", nread);
        if (nread < 0)
        {
            std::cout << "closing stream" << std::endl;
            stream.close();  // will also cancel the read loop
            throw std::runtime_error("nghttp2_session_mem_send");
        }
        else if (!nghttp2_session_want_write(session) && !nghttp2_session_want_read(session))
        {
            logi("write:loop: noting to read or write, leaving");
            break;
        }
        else if (nread == 0)
        {
            logd("write loop: waiting...");
            co_await m_send_channel.async_receive(use_awaitable);
            logd("write loop: waiting... done");
        }
        else
        {
            logd("writing {} bytes...", nread);
            auto [ec, nwrite] = co_await net::async_write(stream, net::buffer(data, nread));
            if (ec)
                throw std::runtime_error("net::async_write");
            logd("writing {} bytes... done, wrote {}", nread, nwrite);
            assert(nwrite == nread);
        }
    }
}

awaitable<void>
Session::read_loop(mystream& stream, nghttp2_session* session)
{
    for (;;) // while (nghttp2_session_want_read(session))
    {
        std::array<uint8_t, 64*1024> buffer;
        // std::cout << "async_read_some... " << std::endl;
        auto [ec, n] = co_await stream.async_read_some(net::buffer(buffer));
        if (ec)
        {
            logi("async_read_some... {} ({})", ec.message(), stream.socket().remote_endpoint());
            // nghttp2_session_terminate_session(session, NGHTTP2_STREAM_CLOSED);
            start_write();
            break;
        }

        logd("nghttp2_session_mem_recv...");
        auto rv = nghttp2_session_mem_recv(session, buffer.data(), n);
        logd("nghttp2_session_mem_recv... done ({})", n);
        if (rv < 0)
            throw std::runtime_error("nghttp2_session_mem_recv");
    }
}

//
// The simples possible coroutine -- to test if it ingegrates with asio's awaitable and coro
//
// https://www.youtube.com/watch?v=J7fYddslH0Q
//
struct ReturnType
{
    struct promise_type {
        ReturnType get_return_object() { return {
            std::coroutine_handle<promise_type>::from_promise(*this)
        }; }
        std::suspend_always initial_suspend() { return {}; }
        void return_void() {}
        void unhandled_exception() {}
        std::suspend_always final_suspend() noexcept { return {}; }
    };

    std::coroutine_handle<promise_type> handle;
    ReturnType(std::coroutine_handle<promise_type> h) : handle(h) {}
    void resume() { handle.resume(); }

};

ReturnType simplest_coro()
{
    std::cout << "### SIMPLEST CORO ###" << std::endl;
    co_return;
}



awaitable<void> listener()
{
    auto executor = co_await boost::asio::this_coro::executor;

    tcp::endpoint endpoint{tcp::v4(), 8080};

    auto coro = simplest_coro();
    coro.resume();

    // Open the acceptor
    tcp::acceptor acceptor(executor);
    acceptor.open(endpoint.protocol());
    acceptor.set_option(net::socket_base::reuse_address(true));
    acceptor.bind(endpoint);
    acceptor.listen();

    for (;;)
    {
        auto socket = co_await acceptor.async_accept(use_awaitable);
        logi("new connection from {}", socket.remote_endpoint());
        auto session = std::make_shared<Session>(std::move(socket));
        co_spawn(executor, session->do_session(), [session](std::exception_ptr pex) {
            try
            {
                if (pex)
                    std::rethrow_exception(pex);
            }
            catch (const std::exception &ex)
            {
                std::cerr << "ERROR: " << ex.what() << std::endl;
            }
        });
    }
}

int main()
{
    fmtlog::setLogLevel(fmtlog::LogLevel::INF);
    boost::asio::io_context io_context;

    boost::asio::signal_set signals(io_context, SIGINT, SIGTERM);
    signals.async_wait([&](auto, auto){
        std::cout << std::endl << "INTERRUPTED" << std::endl;
        io_context.stop();
    });

    co_spawn(io_context, [timer = boost::asio::steady_timer(io_context)]() mutable -> awaitable<void> {
        for (;;)
        {
            fmtlog::poll();
            timer.expires_from_now(10ms);
            co_await timer.async_wait(use_awaitable);
        }
    }, boost::asio::detached);

    co_spawn(io_context, listener(), boost::asio::detached);

    io_context.run();
    std::cout << "done" << std::endl;
}