http_echo_awaitable_chunked.cpp

//
// Copyright (c) 2016-2019 Vinnie Falco (vinnie dot falco at gmail dot com)
//
// Distributed under the Boost Software License, Version 1.0. (See accompanying
// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
//
// Official repository: https://github.com/boostorg/beast
//

//------------------------------------------------------------------------------
//
// Example: HTTP server, coroutine
//
//------------------------------------------------------------------------------

#include <boost/beast/core.hpp>
#include <boost/beast/core/error.hpp>
#include <boost/beast/http.hpp>
#include <boost/beast/http/basic_parser.hpp>
#include <boost/beast/http/dynamic_body.hpp>
#include <boost/beast/http/impl/write.hpp>
#include <boost/beast/http/parser.hpp>
#include <boost/beast/http/serializer.hpp>
#include <boost/beast/http/write.hpp>
#include <boost/beast/version.hpp>

#include <algorithm>
#include <cstdlib>
#include <iostream>
#include <memory>
#include <string>
#include <system_error>
#include <thread>
#include <vector>

namespace beast = boost::beast;         // from <boost/beast.hpp>
namespace http = beast::http;           // from <boost/beast/http.hpp>
namespace net = boost::asio;            // from <boost/asio.hpp>
using tcp = boost::asio::ip::tcp;       // from <boost/asio/ip/tcp.hpp>

#include <boost/asio/co_spawn.hpp>
#include <boost/asio/detached.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/signal_set.hpp>
#include <boost/asio/write.hpp>
#include <boost/asio/experimental/as_tuple.hpp>
#include <boost/asio/redirect_error.hpp>

using boost::asio::awaitable;
using boost::asio::co_spawn;
using boost::asio::detached;
using boost::asio::use_awaitable_t;

using tcp_acceptor = use_awaitable_t<>::as_default_on_t<tcp::acceptor>;
using tcp_socket = use_awaitable_t<>::as_default_on_t<tcp::socket>;
namespace this_coro = boost::asio::this_coro;

using boost::asio::use_awaitable;
using boost::asio::redirect_error;

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


// Handles an HTTP server connection
awaitable<void>
do_session(boost::asio::ip::tcp::socket&& socket)
{
    beast::tcp_stream stream(std::move(socket));

    bool close = false;
    beast::error_code ec;

    // This buffer is required to persist across reads
    beast::flat_buffer buffer;

    for (;;)
    {
        // Set the timeout.
        stream.expires_after(std::chrono::seconds(30));

        http::request_parser<http::buffer_body> parser;
        co_await http::async_read_header(stream, buffer, parser, use_awaitable);
        auto& req = parser.get();

        http::response<http::buffer_body> res{http::status::ok, parser.get().version()};
        res.set(http::field::server, "Beast");
        res.set(http::field::content_type, "text/plain");
        if (req.has_content_length())
            res.content_length(parser.content_length());
        else
            res.chunked(req.chunked());

        // https://www.boost.org/doc/libs/1_78_0/libs/beast/doc/html/beast/using_http/chunked_encoding.html
        auto on_chunk_header = [](std::uint64_t size, boost::string_view extensions, boost::system::error_code& ec) {
            std::cout << "on_chunk_header: size=" << size << " extensions=" << extensions << " ec=" << ec << std::endl;
        };
        parser.on_chunk_header(on_chunk_header);

        auto on_chunk_body = [](std::uint64_t size, boost::string_view body, boost::system::error_code& ec) -> size_t{
            std::cout << "on_chunk_body: size=" << size << " ec=" << ec << std::endl;
            return size;
        };
        parser.on_chunk_body(on_chunk_body);

        res.body().data = nullptr;
        res.body().more = true;

        // We need the serializer here because the serializer requires
        // a non-const file_body, and the message oriented version of
        // http::write only works with const messages.
        http::response_serializer<http::buffer_body> sr{res};
        co_await http::async_write_header(stream, sr, use_awaitable);

        // https://www.boost.org/doc/libs/1_78_0/libs/beast/doc/html/beast/using_http/parser_stream_operations/incremental_read.html
        // https://www.boost.org/doc/libs/1_78_0/libs/beast/doc/html/beast/using_http/serializer_stream_operations.html
        do
        {
            char buf[1024];
            parser.get().body().data = buf;
            parser.get().body().size = sizeof(buf);

            auto n = co_await http::async_read_some(stream, buffer, parser, redirect_error(use_awaitable, ec));
            // auto n = co_await http::async_read(stream, buffer, parser, redirect_error(use_awaitable, ec));
            std::cout << "read : ec=" << ec << " n=" << n << " / " << parser.get().body().size << std::endl;
            if (ec == http::error::need_buffer)
                ec = {};
            if (ec)
                throw std::system_error(ec, "read");

            res.body().data = buf;
            res.body().size = sizeof(buf) - parser.get().body().size;
            res.body().more = !parser.is_done();

            n = co_await http::async_write(stream, sr, redirect_error(use_awaitable, ec));
            std::cout << "write: ec=" << ec << " n=" << n << " / " << parser.get().body().size << std::endl;
            if (ec == http::error::need_buffer)
                ec = {};
            if (ec)
                throw std::system_error(ec, "read");
        }
        while (!sr.is_done());
        stream.close();
    }

    // Send a TCP shutdown
    stream.socket().shutdown(tcp::socket::shutdown_send, ec);

    // At this point the connection is closed gracefully
}

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

// Accepts incoming connections and launches the sessions
awaitable<void> listener()
{
  auto executor = co_await this_coro::executor;
  tcp_acceptor acceptor(executor);

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

  beast::error_code ec;

    // Open the acceptor
    acceptor.open(endpoint.protocol(), ec);
    if (ec)
        throw std::system_error(ec, "open");

    // Allow address reuse
    acceptor.set_option(net::socket_base::reuse_address(true), ec);
    if (ec)
        throw std::system_error(ec, "set_option");

    // Bind to the server address
    acceptor.bind(endpoint, ec);
    if (ec)
        throw std::system_error(ec, "bind");

    // Start listening for connections
    acceptor.listen(net::socket_base::max_listen_connections, ec);
    if (ec)
        throw std::system_error(ec, "listen");


  for (;;)
  {
    auto socket = co_await acceptor.async_accept();
    co_spawn(executor, do_session(std::move(socket)), detached);
  }
}

int main()
{
  try
  {
    boost::asio::io_context io_context(1);

    boost::asio::signal_set signals(io_context, SIGINT, SIGTERM);
    signals.async_wait([&](auto, auto){ io_context.stop(); });

    co_spawn(io_context, listener(), [&](std::exception_ptr pex) {
        try
        {
            if (pex)
                std::rethrow_exception(pex);
        }
        catch (const std::exception &ex)
        {
            std::cerr << ex.what() << std::endl;
            io_context.stop();
        }
    });

    io_context.run();
  }
  catch (std::exception& e)
  {
    std::printf("Exception: %s\n", e.what());
  }
}