redis/detail/resp3_handshaker.hpp

0001 /* Copyright (c) 2018-2024 Marcelo Zimbres Silva (mzimbres@gmail.com)
0002  *
0003  * Distributed under the Boost Software License, Version 1.0. (See
0004  * accompanying file LICENSE.txt)
0005  */
0006
0007 #ifndef BOOST_REDIS_RUNNER_HPP
0008 #define BOOST_REDIS_RUNNER_HPP
0009
0010 #include <boost/redis/config.hpp>
0011 #include <boost/redis/request.hpp>
0012 #include <boost/redis/response.hpp>
0013 #include <boost/redis/error.hpp>
0014 #include <boost/redis/logger.hpp>
0015 #include <boost/redis/operation.hpp>
0016 #include <boost/asio/compose.hpp>
0017 #include <boost/asio/coroutine.hpp>
0018 //#include <boost/asio/ip/tcp.hpp>
0019 #include <string>
0020 #include <memory>
0021 #include <chrono>
0022
0023 namespace boost::redis::detail
0024 {
0025
0026 void push_hello(config const& cfg, request& req);
0027
0028 // TODO: Can we avoid this whole function whose only purpose is to
0029 // check for an error in the hello response and complete with an error
0030 // so that the parallel group that starts it can exit?
0031 template <class Handshaker, class Connection, class Logger>
0032 struct hello_op {
0033    Handshaker* handshaker_ = nullptr;
0034    Connection* conn_ = nullptr;
0035    Logger logger_;
0036    asio::coroutine coro_{};
0037
0038    template <class Self>
0039    void operator()(Self& self, system::error_code ec = {}, std::size_t = 0)
0040    {
0041       BOOST_ASIO_CORO_REENTER (coro_)
0042       {
0043          handshaker_->add_hello();
0044
0045          BOOST_ASIO_CORO_YIELD
0046          conn_->async_exec(handshaker_->hello_req_, any_adapter(handshaker_->hello_resp_), std::move(self));
0047          logger_.on_hello(ec, handshaker_->hello_resp_);
0048
0049          if (ec) {
0050             conn_->cancel(operation::run);
0051             self.complete(ec);
0052             return;
0053          }
0054
0055          if (handshaker_->has_error_in_response()) {
0056             conn_->cancel(operation::run);
0057             self.complete(error::resp3_hello);
0058             return;
0059          }
0060
0061          self.complete({});
0062       }
0063    }
0064 };
0065
0066 template <class Executor>
0067 class resp3_handshaker {
0068 public:
0069    void set_config(config const& cfg)
0070       { cfg_ = cfg; }
0071
0072    template <class Connection, class Logger, class CompletionToken>
0073    auto async_hello(Connection& conn, Logger l, CompletionToken token)
0074    {
0075       return asio::async_compose
0076          < CompletionToken
0077          , void(system::error_code)
0078          >(hello_op<resp3_handshaker, Connection, Logger>{this, &conn, l}, token, conn);
0079    }
0080
0081 private:
0082    template <class, class, class> friend struct hello_op;
0083
0084    void add_hello()
0085    {
0086       hello_req_.clear();
0087       if (hello_resp_.has_value())
0088          hello_resp_.value().clear();
0089       push_hello(cfg_, hello_req_);
0090    }
0091
0092    bool has_error_in_response() const noexcept
0093    {
0094       if (!hello_resp_.has_value())
0095          return true;
0096
0097       auto f = [](auto const& e)
0098       {
0099          switch (e.data_type) {
0100             case resp3::type::simple_error:
0101             case resp3::type::blob_error: return true;
0102             default: return false;
0103          }
0104       };
0105
0106       return std::any_of(std::cbegin(hello_resp_.value()), std::cend(hello_resp_.value()), f);
0107    }
0108
0109    request hello_req_;
0110    generic_response hello_resp_;
0111    config cfg_;
0112 };
0113
0114 } // boost::redis::detail
0115
0116 #endif // BOOST_REDIS_RUNNER_HPP