7#include <boost/redis/connection.hpp>
9#include <boost/asio/awaitable.hpp>
10#include <boost/asio/co_spawn.hpp>
11#include <boost/asio/consign.hpp>
12#include <boost/asio/detached.hpp>
13#include <boost/asio/signal_set.hpp>
17#if defined(BOOST_ASIO_HAS_CO_AWAIT)
24namespace net = boost::asio;
32auto stream_reader(std::shared_ptr<connection> conn) -> net::awaitable<void>
34 std::string redisStreamKey_;
38 std::string stream_id{
"$"};
39 std::string
const field =
"myfield";
42 req.push(
"XREAD",
"BLOCK",
"0",
"STREAMS",
"test-topic", stream_id);
43 co_await conn->async_exec(req, resp);
59 std::size_t item_index = 0;
60 while (item_index < std::size(resp.value())) {
61 auto const& val = resp.value().at(item_index).value;
63 if (field.compare(val) == 0) {
67 stream_id = resp.value().at(item_index - 2).value;
68 std::cout <<
"StreamId: " << stream_id <<
", "
69 <<
"MyField: " << resp.value().at(item_index + 1).value << std::endl;
83auto co_main(config cfg) -> net::awaitable<void>
85 auto ex =
co_await net::this_coro::executor;
86 auto conn = std::make_shared<connection>(ex);
87 net::co_spawn(ex, stream_reader(conn), net::detached);
90 cfg.health_check_interval = std::chrono::seconds::zero();
91 conn->async_run(cfg, net::consign(net::detached, conn));
93 signal_set sig_set(ex, SIGINT, SIGTERM);
94 co_await sig_set.async_wait();
A basic_connection that type erases the executor.
operation
Connection operations that can be cancelled.
adapter::result< std::vector< resp3::node > > generic_response
A generic response to a request.
Configure parameters used by the connection classes.