TcpClient.cpp

#include "TcpClient.hpp"

#include <boost/asio/experimental/awaitable_operators.hpp>
#include <print>
#include <string>

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

using boost::asio::awaitable;
using boost::asio::deferred;
using boost::asio::detached;
using boost::asio::use_awaitable;
using boost::asio::ip::tcp;

namespace {
auto connect(std::shared_ptr<tcp::socket> socket, const tcp::endpoint &endpoint) -> awaitable<void>;
auto read(std::shared_ptr<tcp::socket> socket, std::shared_ptr<Channel> read_chan,
          std::shared_ptr<std::chrono::steady_clock::time_point> deadline) -> awaitable<void>;
auto write(std::shared_ptr<tcp::socket> socket, std::shared_ptr<Channel> write_chan,
           std::shared_ptr<std::chrono::steady_clock::time_point> deadline) -> awaitable<void>;
auto watchdog(std::shared_ptr<std::chrono::steady_clock::time_point> deadline) -> awaitable<void>;
auto wait_to_reconnect() -> awaitable<void>;
}  // namespace

auto start_tcp_client(const std::string &host, uint16_t port, std::shared_ptr<Channel> to_socket,
                      std::shared_ptr<Channel> from_socket) -> awaitable<void> {
   const auto ex = co_await boost::asio::this_coro::executor;

   const auto endpoint = tcp::endpoint{boost::asio::ip::address::from_string(host), port};

   for (;;) {
      const auto socket = std::make_shared<tcp::socket>(ex);
      co_await connect(socket, endpoint);

      std::println(stdout, "Connected to {}:{}", endpoint.address().to_string(), endpoint.port());

      const auto deadline = std::make_shared<std::chrono::steady_clock::time_point>();

      co_spawn(ex, write(socket, to_socket, deadline), detached);
      co_spawn(ex, read(socket, from_socket, deadline), detached);

      if (const auto result =
              co_await (socket->async_wait(tcp::socket::wait_error, use_awaitable) || watchdog(deadline));
          result.index() == 0) {
         std::println(stderr, "Connection failed");
      } else {
         std::println(stderr, "Inactivity timeout");
         socket->shutdown(boost::asio::socket_base::shutdown_both);
         socket->close();
      }

      co_await wait_to_reconnect();
   }
}

namespace {
auto connect(std::shared_ptr<tcp::socket> socket, const tcp::endpoint &endpoint) -> awaitable<void> {
   const auto ex = co_await boost::asio::this_coro::executor;

   std::println(stdout, "Attempting to connect to {}:{}", endpoint.address().to_string(), endpoint.port());
   for (;;) {
      if (const auto result = co_await (socket->async_connect(endpoint, use_awaitable) ||
                                        boost::asio::steady_timer{ex, 1s}.async_wait(use_awaitable));
          result.index() == 0) {
         // First branch: connect finished
         break;
      }
      std::println(stdout, "Failed to connect to {}:{}. Trying again.", endpoint.address().to_string(),
                   endpoint.port());
   }
}

auto read(std::shared_ptr<tcp::socket> socket, std::shared_ptr<Channel> read_chan,
          std::shared_ptr<std::chrono::steady_clock::time_point> deadline) -> awaitable<void> {
   auto ex = co_await boost::asio::this_coro::executor;

   auto data = std::array<char, 4096>{};
   for (;;) {
      *deadline = std::chrono::steady_clock::now() + 10s;
      const auto n = co_await socket->async_read_some(boost::asio::buffer(data), deferred);
      co_await read_chan->async_send(boost::system::error_code{}, std::string(data.data(), data.data() + n), deferred);
   }
}

auto write(std::shared_ptr<tcp::socket> socket, std::shared_ptr<Channel> write_chan,
           std::shared_ptr<std::chrono::steady_clock::time_point> deadline) -> awaitable<void> {
   for (;;) {
      *deadline = std::chrono::steady_clock::now() + 10s;
      auto buf = co_await write_chan->async_receive(deferred);
      co_await boost::asio::async_write(*socket, boost::asio::buffer(buf), deferred);
   }
}

auto watchdog(std::shared_ptr<std::chrono::steady_clock::time_point> deadline) -> awaitable<void> {
   auto timer = boost::asio::steady_timer{co_await boost::asio::this_coro::executor};
   auto now = std::chrono::steady_clock::now();
   while (*deadline > now) {
      timer.expires_at(*deadline);
      co_await timer.async_wait(deferred);
      now = std::chrono::steady_clock::now();
   }
}

auto wait_to_reconnect() -> awaitable<void> {
   const auto ex = co_await boost::asio::this_coro::executor;

   for (auto i = 5s; i > 0s; --i) {
      std::println(stdout, "Attempting to reconnection in {} seconds...", i.count());
      co_await boost::asio::steady_timer{ex, 1s}.async_wait(deferred);
   }
}
}  // namespace