connection.hpp

/* Copyright (c) 2018-2023 Marcelo Zimbres Silva (mzimbres@gmail.com)
 *
 * Distributed under the Boost Software License, Version 1.0. (See
 * accompanying file LICENSE.txt)
 */
 
#ifndef BOOST_REDIS_CONNECTION_HPP
#define BOOST_REDIS_CONNECTION_HPP
 
#include <boost/redis/detail/connection_base.hpp>
#include <boost/redis/logger.hpp>
#include <boost/redis/config.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/coroutine.hpp>
#include <boost/asio/steady_timer.hpp>
#include <boost/asio/any_io_executor.hpp>
#include <boost/asio/any_completion_handler.hpp>
 
#include <chrono>
#include <memory>
#include <limits>
 
namespace boost::redis {
namespace detail
{
template <class Connection, class Logger>
struct reconnection_op {
   Connection* conn_ = nullptr;
   Logger logger_;
   asio::coroutine coro_{};
 
   template <class Self>
   void operator()(Self& self, system::error_code ec = {})
   {
      BOOST_ASIO_CORO_REENTER (coro_) for (;;)
      {
         BOOST_ASIO_CORO_YIELD
         conn_->impl_.async_run(conn_->cfg_, logger_, std::move(self));
         conn_->cancel(operation::receive);
         logger_.on_connection_lost(ec);
         if (!conn_->will_reconnect() || is_cancelled(self)) {
            conn_->cancel(operation::reconnection);
            self.complete(!!ec ? ec : asio::error::operation_aborted);
            return;
         }
 
         conn_->timer_.expires_after(conn_->cfg_.reconnect_wait_interval);
         BOOST_ASIO_CORO_YIELD
         conn_->timer_.async_wait(std::move(self));
         BOOST_REDIS_CHECK_OP0(;)
         if (!conn_->will_reconnect()) {
            self.complete(asio::error::operation_aborted);
            return;
         }
         conn_->reset_stream();
      }
   }
};
} // detail
 
template <class Executor>
class basic_connection {
public:
   using executor_type = Executor;
 
   executor_type get_executor() noexcept
      { return impl_.get_executor(); }
 
   template <class Executor1>
   struct rebind_executor
   {
      using other = basic_connection<Executor1>;
   };
 
   explicit
   basic_connection(
      executor_type ex,
      asio::ssl::context ctx = asio::ssl::context{asio::ssl::context::tlsv12_client},
      std::size_t max_read_size = (std::numeric_limits<std::size_t>::max)())
   : impl_{ex, std::move(ctx), max_read_size}
   , timer_{ex}
   { }
 
   explicit
   basic_connection(
      asio::io_context& ioc,
      asio::ssl::context ctx = asio::ssl::context{asio::ssl::context::tlsv12_client},
      std::size_t max_read_size = (std::numeric_limits<std::size_t>::max)())
   : basic_connection(ioc.get_executor(), std::move(ctx), max_read_size)
   { }
 
   template <
      class Logger = logger,
      class CompletionToken = asio::default_completion_token_t<executor_type>>
   auto
   async_run(
      config const& cfg = {},
      Logger l = Logger{},
      CompletionToken token = CompletionToken{})
   {
      using this_type = basic_connection<executor_type>;
 
      cfg_ = cfg;
      l.set_prefix(cfg_.log_prefix);
      return asio::async_compose
         < CompletionToken
         , void(system::error_code)
         >(detail::reconnection_op<this_type, Logger>{this, l}, token, timer_);
   }
 
   template <class CompletionToken = asio::default_completion_token_t<executor_type>>
   auto async_receive(CompletionToken token = CompletionToken{})
      { return impl_.async_receive(std::move(token)); }
 
   
   std::size_t receive(system::error_code& ec)
   {
      return impl_.receive(ec);
   }
 
   template <
      class Response = ignore_t,
      class CompletionToken = asio::default_completion_token_t<executor_type>
   >
   [[deprecated("Set the response with set_receive_response and use the other overload.")]]
   auto
   async_receive(
      Response& response,
      CompletionToken token = CompletionToken{})
   {
      return impl_.async_receive(response, token);
   }
 
   template <
      class Response = ignore_t,
      class CompletionToken = asio::default_completion_token_t<executor_type>
   >
   auto
   async_exec(
      request const& req,
      Response& resp = ignore,
      CompletionToken&& token = CompletionToken{})
   {
      return impl_.async_exec(req, resp, std::forward<CompletionToken>(token));
   }
 
   void cancel(operation op = operation::all)
   {
      switch (op) {
         case operation::reconnection:
         case operation::all:
            cfg_.reconnect_wait_interval = std::chrono::seconds::zero();
            timer_.cancel();
            break;
         default: /* ignore */;
      }
 
      impl_.cancel(op);
   }
 
   bool will_reconnect() const noexcept
      { return cfg_.reconnect_wait_interval != std::chrono::seconds::zero();}
 
   auto const& get_ssl_context() const noexcept
      { return impl_.get_ssl_context();}
 
   void reset_stream()
      { impl_.reset_stream(); }
 
   auto& next_layer() noexcept
      { return impl_.next_layer(); }
 
   auto const& next_layer() const noexcept
      { return impl_.next_layer(); }
 
   template <class Response>
   void set_receive_response(Response& response)
      { impl_.set_receive_response(response); }
 
   usage get_usage() const noexcept
      { return impl_.get_usage(); }
 
private:
   using timer_type =
      asio::basic_waitable_timer<
         std::chrono::steady_clock,
         asio::wait_traits<std::chrono::steady_clock>,
         Executor>;
 
   template <class, class> friend struct detail::reconnection_op;
 
   config cfg_;
   detail::connection_base<executor_type> impl_;
   timer_type timer_;
};
 
class connection {
public:
   using executor_type = asio::any_io_executor;
 
   explicit
   connection(
      executor_type ex,
      asio::ssl::context ctx = asio::ssl::context{asio::ssl::context::tlsv12_client},
      std::size_t max_read_size = (std::numeric_limits<std::size_t>::max)());
 
   explicit
   connection(
      asio::io_context& ioc,
      asio::ssl::context ctx = asio::ssl::context{asio::ssl::context::tlsv12_client},
      std::size_t max_read_size = (std::numeric_limits<std::size_t>::max)());
 
   executor_type get_executor() noexcept
      { return impl_.get_executor(); }
 
   template <class CompletionToken>
   auto async_run(config const& cfg, logger l, CompletionToken token)
   {
      return asio::async_initiate<
         CompletionToken, void(boost::system::error_code)>(
            [](auto handler, connection* self, config const* cfg, logger l)
            {
               self->async_run_impl(*cfg, l, std::move(handler));
            }, token, this, &cfg, l);
   }
 
   template <class Response, class CompletionToken>
   [[deprecated("Set the response with set_receive_response and use the other overload.")]]
   auto async_receive(Response& response, CompletionToken token)
   {
      return impl_.async_receive(response, std::move(token));
   }
 
   template <class CompletionToken>
   auto async_receive(CompletionToken token)
      { return impl_.async_receive(std::move(token)); }
 
   std::size_t receive(system::error_code& ec)
   {
      return impl_.receive(ec);
   }
 
   template <class Response, class CompletionToken>
   auto async_exec(request const& req, Response& resp, CompletionToken token)
   {
      return impl_.async_exec(req, resp, std::move(token));
   }
 
   void cancel(operation op = operation::all);
 
   bool will_reconnect() const noexcept
      { return impl_.will_reconnect();}
 
   auto& next_layer() noexcept
      { return impl_.next_layer(); }
 
   auto const& next_layer() const noexcept
      { return impl_.next_layer(); }
 
   void reset_stream()
      { impl_.reset_stream();}
 
   template <class Response>
   void set_receive_response(Response& response)
      { impl_.set_receive_response(response); }
 
   usage get_usage() const noexcept
      { return impl_.get_usage(); }
 
   auto const& get_ssl_context() const noexcept
      { return impl_.get_ssl_context();}
 
private:
   void
   async_run_impl(
      config const& cfg,
      logger l,
      asio::any_completion_handler<void(boost::system::error_code)> token);
 
   basic_connection<executor_type> impl_;
};
 
} // boost::redis
 
#endif // BOOST_REDIS_CONNECTION_HPP