src/asio/factory.cpp - main - kelbon/hidi

src/asio/factory.cpp
242 строки8 KB
kelbon
sync with v0.9.6
14 июн 2026, 17:57
14 июн 2026, 17:57de0bf2c
#include "http2/asio/factory.hpp"
#include "http2/asio/awaiters.hpp"
#include "http2/logger.hpp"
#include "kelcoro/job.hpp"

#include <boost/asio/read.hpp>
#include <boost/asio/ssl/host_name_verification.hpp>

namespace http2 {

[[nodiscard]] static bool try_read_impl(auto& con, std::span<byte_t> buf) noexcept {
  size_t avail = con.readen_end - con.readen_start;
  bool b = avail >= buf.size();
  if (b) {
    memcpy(buf.data(), con.readen_start, buf.size());
    con.readen_start += buf.size();
  }
  return b;
}

bool asio_tls_connection::try_read(std::span<byte_t> buf) noexcept {
  return try_read_impl(*this, buf);
}

static dd::job do_read_some(auto& c, std::span<byte_t> userbuf, io_error_code& ec,
                            std::coroutine_handle<> callback) {
  size_t readen = 0;
  const size_t userbufsz = userbuf.size();
  byte_t* const userbufend = userbuf.data() + userbufsz;
  while (readen < userbufsz) {
    readen += co_await net.read_some_many(c.sock, ec, std::span(userbuf.data() + readen, userbufend),
                                          std::span(c.readen));
    if (ec) [[unlikely]]
      break;
  }
  c.readen_end += readen - userbufsz;
  co_await dd::this_coro::destroy_and_transfer_control_to(callback);
}

void asio_tls_connection::start_read(std::coroutine_handle<> h, std::span<byte_t> buf, io_error_code& ec) {
  // assumes only one reader at one time
  size_t avail = readen_end - readen_start;
  assert(avail < buf.size());  // start_read must be invoked only if try_read failed
  memcpy(buf.data(), readen_start, avail);
  readen_start = readen_end = readen;
  (void)do_read_some(*this, suffix(buf, buf.size() - avail), ec, h);
}

size_t asio_tls_connection::try_write(std::span<const byte_t> buf, io_error_code& ec) noexcept {
  size_t written = sock.write_some(asio::buffer(buf.data(), buf.size()), ec);
  if (ec) {
    if (ec == asio::error::would_block)
      ec.clear();  // not a error
  }
  return written;
}

void asio_tls_connection::start_write(std::coroutine_handle<> h, std::span<byte_t const> buf,
                                      io_error_code& ec) {
  asio::async_write(sock, asio::buffer(buf.data(), buf.size()), [&, h](const io_error_code& e, size_t) {
    if (e) [[unlikely]]
      ec = e;
    h.resume();
  });
}

static void close_tcp_sock(auto& tcp_sock) {
  if (!tcp_sock.is_open())
    return;
  io_error_code ec;
  ec = tcp_sock.cancel(ec);
  // dont stop on errors, i need to stop connection somehow
  (void)ec;
  // Do not do SSL shutdown, useless errors and wasting time
  ec = tcp_sock.shutdown(asio::socket_base::shutdown_both, ec);
  (void)ec;
  ec = tcp_sock.close(ec);
  (void)ec;
}

void asio_tls_connection::shutdown() noexcept {
  auto& tcp_sock = sock.lowest_layer();
  close_tcp_sock(tcp_sock);
}

bool asio_connection::try_read(std::span<byte_t> buf) noexcept {
  return try_read_impl(*this, buf);
}

void asio_connection::start_read(std::coroutine_handle<> h, std::span<byte_t> buf, io_error_code& ec) {
  // assumes only one reader at one time
  size_t avail = readen_end - readen_start;
  assert(avail < buf.size());  // start_read must be invoked only if try_read failed
  memcpy(buf.data(), readen_start, avail);
  readen_start = readen_end = readen;
  (void)do_read_some(*this, suffix(buf, buf.size() - avail), ec, h);
}

size_t asio_connection::try_write(std::span<const byte_t> buf, io_error_code& ec) noexcept {
  size_t written = sock.write_some(asio::buffer(buf.data(), buf.size()), ec);
  if (ec) {
    if (ec == asio::error::would_block)
      ec.clear();  // not a error
  }
  return written;
}

void asio_connection::start_write(std::coroutine_handle<> h, std::span<const byte_t> buf, io_error_code& ec) {
  asio::async_write(sock, asio::buffer(buf.data(), buf.size()), [&, h](const io_error_code& e, size_t) {
    if (e) [[unlikely]]
      ec = e;
    h.resume();
  });
}

void asio_connection::shutdown() noexcept {
  close_tcp_sock(sock);
}

any_transport_factory default_transport_factory(boost::asio::io_context& ctx) {
  return any_transport_factory(new asio_factory(ctx, {}));
}

any_transport_factory default_tls_transport_factory(boost::asio::io_context& ctx,
                                                    std::vector<std::filesystem::path> certs) {
  tcp_connection_options options;
  options.additional_ssl_certificates = std::move(certs);
  return any_transport_factory(new asio_tls_factory(ctx, std::move(options)));
}

asio_factory::asio_factory(boost::asio::io_context& ctx, tcp_connection_options opts)
    : ioctx(ctx), options(std::move(opts)) {
}

dd::task<any_connection_t> asio_factory::createConnection(endpoint endpoint, deadline_t deadline) {
  using tcp = asio::ip::tcp;

  tcp::resolver resolver(ioctx);

  asio::steady_timer timer(ioctx);
  bool timeoutflag = false;

  timer.expires_at(deadline.tp);
  timer.async_wait([&resolver, &timeoutflag](const io_error_code& ec) {
    if (ec != asio::error::operation_aborted) {
      timeoutflag = true;
      resolver.cancel();
    }
  });

  io_error_code ec;
  auto results = co_await net.resolve(resolver, endpoint, ec);
  if (timeoutflag)
    throw timeout_exception();

  if (results.empty() || ec)
    throw network_exception("[TCP] cannot resolve host: {}, err: {}", endpoint.to_string(), ec.message());
  tcp::socket tcp_sock(ioctx);

  timer.cancel();
  timer.expires_at(deadline.tp);
  timer.async_wait([&tcp_sock, &timeoutflag](const io_error_code& ec) {
    if (ec != asio::error::operation_aborted) {
      timeoutflag = true;
      close_tcp_sock(tcp_sock);
    }
  });

  co_await net.connect(tcp_sock, results, ec);

  if (ec)
    throw network_exception("[TCP] cannot connect to {}, err: {}", endpoint.to_string(), ec.message());
  options.apply(tcp_sock);
  co_return any_connection_t(new asio_connection(std::move(tcp_sock)));
}

asio_tls_factory::asio_tls_factory(asio::io_context& ioctx, tcp_connection_options opts)
    : ioctx(ioctx),
      options(std::move(opts)),
      sslctx(make_ssl_context_for_http2(options.additional_ssl_certificates)) {
}

dd::task<any_connection_t> asio_tls_factory::createConnection(endpoint endpoint, deadline_t deadline) {
  namespace ssl = asio::ssl;
  using tcp = asio::ip::tcp;

  tcp::resolver resolver(ioctx);

  asio::steady_timer timer(ioctx);
  bool timeoutflag = false;

  timer.expires_at(deadline.tp);
  timer.async_wait([&resolver, &timeoutflag](const io_error_code& ec) {
    if (ec != asio::error::operation_aborted) {
      timeoutflag = true;
      resolver.cancel();
    }
  });

  io_error_code ec;
  auto results = co_await net.resolve(resolver, endpoint, ec);
  if (timeoutflag)
    throw timeout_exception();
  if (results.empty() || ec)
    throw network_exception("[TCP] cannot resolve host: {}, err: {}", endpoint.to_string(), ec.what());
  asio::ip::tcp::socket tcp_sock(ioctx);

  timer.cancel();
  timer.expires_at(deadline.tp);
  timer.async_wait([&tcp_sock, &timeoutflag](const io_error_code& ec) {
    if (ec != asio::error::operation_aborted) {
      timeoutflag = true;
      close_tcp_sock(tcp_sock);
    }
  });

  co_await net.connect(tcp_sock, std::move(results), ec);

  if (timeoutflag)
    throw timeout_exception();
  if (ec)
    throw network_exception("[TCP] cannot connect to {}, err: {}", endpoint.to_string(), ec.message());
  options.apply(tcp_sock);
  assert(sslctx);
  std::unique_ptr<asio_tls_connection> res(new asio_tls_connection(std::move(tcp_sock), sslctx));
  if (options.host_for_name_verification) {
    res->sock.set_verify_mode(ssl::verify_peer);
    res->sock.set_verify_callback(asio::ssl::host_name_verification(*options.host_for_name_verification));
  } else {
    res->sock.set_verify_mode(ssl::verify_none);
  }
  if (!options.is_primal_connection)
    SSL_set_mode(res->sock.native_handle(), SSL_MODE_RELEASE_BUFFERS);
  co_await net.handshake(res->sock, ssl::stream_base::handshake_type::client, ec);
  if (timeoutflag)
    throw timeout_exception();
  if (ec)
    throw network_exception("[TCP/SSL] cannot ssl handshake: {}", ec.message());
  co_return any_connection_t(std::move(res));
}

}  // namespace http2