boost_redis 1.4.2
A redis client library
runner.hpp
1/* Copyright (c) 2018-2022 Marcelo Zimbres Silva (mzimbres@gmail.com)
2 *
3 * Distributed under the Boost Software License, Version 1.0. (See
4 * accompanying file LICENSE.txt)
5 */
6
7#ifndef BOOST_REDIS_RUNNER_HPP
8#define BOOST_REDIS_RUNNER_HPP
9
10#include <boost/redis/detail/health_checker.hpp>
11#include <boost/redis/config.hpp>
12#include <boost/redis/response.hpp>
13#include <boost/redis/detail/helper.hpp>
14#include <boost/redis/error.hpp>
15#include <boost/redis/logger.hpp>
16#include <boost/redis/operation.hpp>
17#include <boost/redis/detail/connector.hpp>
18#include <boost/redis/detail/resolver.hpp>
19#include <boost/redis/detail/handshaker.hpp>
20#include <boost/asio/compose.hpp>
21#include <boost/asio/connect.hpp>
22#include <boost/asio/coroutine.hpp>
23#include <boost/asio/experimental/parallel_group.hpp>
24#include <boost/asio/ip/tcp.hpp>
25#include <boost/asio/steady_timer.hpp>
26#include <string>
27#include <memory>
28#include <chrono>
29
30namespace boost::redis::detail
31{
32
33template <class Runner, class Connection, class Logger>
34struct hello_op {
35 Runner* runner_ = nullptr;
36 Connection* conn_ = nullptr;
37 Logger logger_;
38 asio::coroutine coro_{};
39
40 template <class Self>
41 void operator()(Self& self, system::error_code ec = {}, std::size_t = 0)
42 {
43 BOOST_ASIO_CORO_REENTER (coro_)
44 {
45 runner_->hello_req_.clear();
46 if (runner_->hello_resp_.has_value())
47 runner_->hello_resp_.value().clear();
48 runner_->add_hello();
49
50 BOOST_ASIO_CORO_YIELD
51 conn_->async_exec(runner_->hello_req_, runner_->hello_resp_, std::move(self));
52 logger_.on_hello(ec, runner_->hello_resp_);
53 BOOST_REDIS_CHECK_OP0(conn_->cancel(operation::run);)
54 self.complete(ec);
55 }
56 }
57};
58
59template <class Runner, class Connection, class Logger>
60class runner_op {
61private:
62 Runner* runner_ = nullptr;
63 Connection* conn_ = nullptr;
64 Logger logger_;
65 asio::coroutine coro_{};
66
67public:
68 runner_op(Runner* runner, Connection* conn, Logger l)
69 : runner_{runner}
70 , conn_{conn}
71 , logger_{l}
72 {}
73
74 template <class Self>
75 void operator()( Self& self
76 , std::array<std::size_t, 3> order = {}
77 , system::error_code ec0 = {}
78 , system::error_code ec1 = {}
79 , system::error_code ec2 = {}
80 , std::size_t = 0)
81 {
82 BOOST_ASIO_CORO_REENTER (coro_)
83 {
84 BOOST_ASIO_CORO_YIELD
85 asio::experimental::make_parallel_group(
86 [this](auto token) { return runner_->async_run_all(*conn_, logger_, token); },
87 [this](auto token) { return runner_->health_checker_.async_check_health(*conn_, token); },
88 [this](auto token) { return runner_->async_hello(*conn_, logger_, token); }
89 ).async_wait(
90 asio::experimental::wait_for_all(),
91 std::move(self));
92
93 if (is_cancelled(self)) {
94 self.complete(asio::error::operation_aborted);
95 return;
96 }
97
99 self.complete(ec0);
100 return;
101 }
102
103 if (order[0] == 2 && !!ec2) {
104 self.complete(ec2);
105 return;
106 }
107
108 if (order[0] == 1 && ec1 == error::pong_timeout) {
109 self.complete(ec1);
110 return;
111 }
112
113 self.complete(ec0);
114 }
115 }
116};
117
118template <class Runner, class Connection, class Logger>
119struct run_all_op {
120 Runner* runner_ = nullptr;
121 Connection* conn_ = nullptr;
122 Logger logger_;
123 asio::coroutine coro_{};
124
125 template <class Self>
126 void operator()(Self& self, system::error_code ec = {}, std::size_t = 0)
127 {
128 BOOST_ASIO_CORO_REENTER (coro_)
129 {
130 BOOST_ASIO_CORO_YIELD
131 runner_->resv_.async_resolve(std::move(self));
132 logger_.on_resolve(ec, runner_->resv_.results());
133 BOOST_REDIS_CHECK_OP0(conn_->cancel(operation::run);)
134
135 BOOST_ASIO_CORO_YIELD
136 runner_->ctor_.async_connect(conn_->next_layer().next_layer(), runner_->resv_.results(), std::move(self));
137 logger_.on_connect(ec, runner_->ctor_.endpoint());
138 BOOST_REDIS_CHECK_OP0(conn_->cancel(operation::run);)
139
140 if (conn_->use_ssl()) {
141 BOOST_ASIO_CORO_YIELD
142 runner_->hsher_.async_handshake(conn_->next_layer(), std::move(self));
143 logger_.on_ssl_handshake(ec);
144 BOOST_REDIS_CHECK_OP0(conn_->cancel(operation::run);)
145 }
146
147 BOOST_ASIO_CORO_YIELD
148 conn_->async_run_lean(runner_->cfg_, logger_, std::move(self));
149 BOOST_REDIS_CHECK_OP0(;)
150 self.complete(ec);
151 }
152 }
153};
154
155template <class Executor>
156class runner {
157public:
158 runner(Executor ex, config cfg)
159 : resv_{ex}
160 , ctor_{ex}
161 , hsher_{ex}
162 , health_checker_{ex}
163 , cfg_{cfg}
164 { }
165
166 std::size_t cancel(operation op)
167 {
168 resv_.cancel(op);
169 ctor_.cancel(op);
170 hsher_.cancel(op);
171 health_checker_.cancel(op);
172 return 0U;
173 }
174
175 void set_config(config const& cfg)
176 {
177 cfg_ = cfg;
178 resv_.set_config(cfg);
179 ctor_.set_config(cfg);
180 hsher_.set_config(cfg);
181 health_checker_.set_config(cfg);
182 }
183
184 template <class Connection, class Logger, class CompletionToken>
185 auto async_run(Connection& conn, Logger l, CompletionToken token)
186 {
187 return asio::async_compose
188 < CompletionToken
189 , void(system::error_code)
190 >(runner_op<runner, Connection, Logger>{this, &conn, l}, token, conn);
191 }
192
193 config const& get_config() const noexcept {return cfg_;}
194
195private:
196 using resolver_type = resolver<Executor>;
197 using connector_type = connector<Executor>;
198 using handshaker_type = detail::handshaker<Executor>;
199 using health_checker_type = health_checker<Executor>;
200 using timer_type = typename connector_type::timer_type;
201
202 template <class, class, class> friend struct run_all_op;
203 template <class, class, class> friend class runner_op;
204 template <class, class, class> friend struct hello_op;
205
206 template <class Connection, class Logger, class CompletionToken>
207 auto async_run_all(Connection& conn, Logger l, CompletionToken token)
208 {
209 return asio::async_compose
210 < CompletionToken
211 , void(system::error_code)
212 >(run_all_op<runner, Connection, Logger>{this, &conn, l}, token, conn);
213 }
214
215 template <class Connection, class Logger, class CompletionToken>
216 auto async_hello(Connection& conn, Logger l, CompletionToken token)
217 {
218 return asio::async_compose
219 < CompletionToken
220 , void(system::error_code)
221 >(hello_op<runner, Connection, Logger>{this, &conn, l}, token, conn);
222 }
223
224 void add_hello()
225 {
226 if (!cfg_.username.empty() && !cfg_.password.empty() && !cfg_.clientname.empty())
227 hello_req_.push("HELLO", "3", "AUTH", cfg_.username, cfg_.password, "SETNAME", cfg_.clientname);
228 else if (cfg_.username.empty() && cfg_.password.empty() && cfg_.clientname.empty())
229 hello_req_.push("HELLO", "3");
230 else if (cfg_.clientname.empty())
231 hello_req_.push("HELLO", "3", "AUTH", cfg_.username, cfg_.password);
232 else
233 hello_req_.push("HELLO", "3", "SETNAME", cfg_.clientname);
234
235 if (cfg_.database_index)
236 hello_req_.push("SELECT", cfg_.database_index.value());
237 }
238
239 resolver_type resv_;
240 connector_type ctor_;
241 handshaker_type hsher_;
242 health_checker_type health_checker_;
243 request hello_req_;
244 generic_response hello_resp_;
245 config cfg_;
246};
247
248} // boost::redis::detail
249
250#endif // BOOST_REDIS_RUNNER_HPP
adapter::result< std::vector< resp3::node > > generic_response
A generic response to a request.
Definition: response.hpp:33
operation
Connection operations that can be cancelled.
Definition: operation.hpp:18
@ resolve_timeout
Resolve timeout.
@ pong_timeout
Connect timeout.
@ connect_timeout
Connect timeout.
@ run
Refers to connection::async_run operations.