7#include <boost/redis/connection.hpp>
9#include <boost/asio/co_spawn.hpp>
10#include <boost/asio/detached.hpp>
11#include <boost/asio/posix/stream_descriptor.hpp>
12#include <boost/asio/redirect_error.hpp>
13#include <boost/asio/signal_set.hpp>
18#if defined(BOOST_ASIO_HAS_CO_AWAIT)
19#if defined(BOOST_ASIO_HAS_POSIX_STREAM_DESCRIPTOR)
21namespace asio = boost::asio;
22using asio::posix::stream_descriptor;
23using asio::signal_set;
24using boost::asio::async_read_until;
25using boost::asio::awaitable;
26using boost::asio::co_spawn;
27using boost::asio::consign;
28using boost::asio::detached;
29using boost::asio::dynamic_buffer;
30using boost::asio::redirect_error;
31using boost::asio::use_awaitable;
37using boost::system::error_code;
38using namespace std::chrono_literals;
43auto receiver(std::shared_ptr<connection> conn) -> awaitable<void>
46 req.push(
"SUBSCRIBE",
"channel");
49 conn->set_receive_response(resp);
51 while (conn->will_reconnect()) {
53 co_await conn->async_exec(req, ignore);
56 for (error_code ec;;) {
57 co_await conn->async_receive(redirect_error(use_awaitable, ec));
60 std::cout << resp.value().at(1).value <<
" " << resp.value().at(2).value <<
" "
61 << resp.value().at(3).value << std::endl;
68auto publisher(std::shared_ptr<stream_descriptor> in, std::shared_ptr<connection> conn)
71 for (std::string msg;;) {
72 auto n =
co_await async_read_until(*in, dynamic_buffer(msg, 1024),
"\n");
74 req.push(
"PUBLISH",
"channel", msg);
75 co_await conn->async_exec(req, ignore);
81auto co_main(config cfg) -> awaitable<void>
83 auto ex =
co_await asio::this_coro::executor;
84 auto conn = std::make_shared<connection>(ex);
85 auto stream = std::make_shared<stream_descriptor>(ex, ::dup(STDIN_FILENO));
87 co_spawn(ex, receiver(conn), detached);
88 co_spawn(ex, publisher(stream, conn), detached);
89 conn->async_run(cfg, consign(detached, conn));
91 signal_set sig_set{ex, SIGINT, SIGTERM};
92 co_await sig_set.async_wait();
98auto co_main(config
const&) -> awaitable<void>
100 std::cout <<
"Requires support for posix streams." << std::endl;
A basic_connection that type erases the executor.
ignore_t ignore
Global ignore object.
adapter::result< std::vector< resp3::node > > generic_response
A generic response to a request.
Configure parameters used by the connection classes.