Boost.Redis: example/cpp20_chat_room.cpp Source File

Boost.Redis 1.84.0 A redis client library

Loading...
Searching...
No Matches
cpp20_chat_room.cpp
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#include <boost/redis/connection.hpp>
8#include <boost/asio/signal_set.hpp>
9#include <boost/asio/co_spawn.hpp>
10#include <boost/asio/detached.hpp>
11#include <boost/asio/redirect_error.hpp>
12#include <boost/asio/posix/stream_descriptor.hpp>
13#include <unistd.h>
14#include <iostream>
15
16#if defined(BOOST_ASIO_HAS_CO_AWAIT)
17#if defined(BOOST_ASIO_HAS_POSIX_STREAM_DESCRIPTOR)
18
19namespace asio = boost::asio;
20using asio::posix::stream_descriptor;
21using asio::signal_set;
22using boost::asio::async_read_until;
23using boost::asio::awaitable;
24using boost::asio::co_spawn;
25using boost::asio::consign;
26using boost::asio::detached;
27using boost::asio::dynamic_buffer;
28using boost::asio::redirect_error;
29using boost::asio::use_awaitable;
30using boost::redis::config;
31using boost::redis::connection;
32using boost::redis::generic_response;
33using boost::redis::ignore;
34using boost::redis::request;
35using boost::system::error_code;
36using namespace std::chrono_literals;
37
38// Chat over Redis pubsub. To test, run this program from multiple
39// terminals and type messages to stdin.
40
41auto
42receiver(std::shared_ptr<connection> conn) -> awaitable<void>
43{
44 request req;
45 req.push("SUBSCRIBE", "channel");
46
47 generic_response resp;
48 conn->set_receive_response(resp);
49
50 while (conn->will_reconnect()) {
51
52 // Subscribe to channels.
53 co_await conn->async_exec(req, ignore);
54
55 // Loop reading Redis push messages.
56 for (error_code ec;;) {
57 co_await conn->async_receive(redirect_error(use_awaitable, ec));
58 if (ec)
59 break; // Connection lost, break so we can reconnect to channels.
60 std::cout
61 << resp.value().at(1).value
62 << " " << resp.value().at(2).value
63 << " " << resp.value().at(3).value
64 << std::endl;
65 resp.value().clear();
66 }
67 }
68}
69
70// Publishes stdin messages to a Redis channel.
71auto publisher(std::shared_ptr<stream_descriptor> in, std::shared_ptr<connection> conn) -> awaitable<void>
72{
73 for (std::string msg;;) {
74 auto n = co_await async_read_until(*in, dynamic_buffer(msg, 1024), "\n");
75 request req;
76 req.push("PUBLISH", "channel", msg);
77 co_await conn->async_exec(req, ignore);
78 msg.erase(0, n);
79 }
80}
81
82// Called from the main function (see main.cpp)
83auto co_main(config cfg) -> awaitable<void>
84{
85 auto ex = co_await asio::this_coro::executor;
86 auto conn = std::make_shared<connection>(ex);
87 auto stream = std::make_shared<stream_descriptor>(ex, ::dup(STDIN_FILENO));
88
89 co_spawn(ex, receiver(conn), detached);
90 co_spawn(ex, publisher(stream, conn), detached);
91 conn->async_run(cfg, {}, consign(detached, conn));
92
93 signal_set sig_set{ex, SIGINT, SIGTERM};
94 co_await sig_set.async_wait();
95 conn->cancel();
96 stream->cancel();
97}
98
99#else // defined(BOOST_ASIO_HAS_POSIX_STREAM_DESCRIPTOR)
100auto co_main(config const&) -> awaitable<void>
101{
102 std::cout << "Requires support for posix streams." << std::endl;
103 co_return;
104}
105#endif // defined(BOOST_ASIO_HAS_POSIX_STREAM_DESCRIPTOR)
106#endif // defined(BOOST_ASIO_HAS_CO_AWAIT)
boost::redis::connection
A basic_connection that type erases the executor.
Definition connection.hpp:1231
boost::redis::request
Creates Redis requests.
Definition request.hpp:46
boost::redis::ignore
ignore_t ignore
Global ignore object.
boost::redis::generic_response
adapter::result< std::vector< resp3::node > > generic_response
A generic response to a request.
Definition response.hpp:35
boost::redis::config
Configure parameters used by the connection classes.
Definition config.hpp:30