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
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>
14
15#include <iostream>
16#include <unistd.h>
17
18#if defined(BOOST_ASIO_HAS_CO_AWAIT)
19#if defined(BOOST_ASIO_HAS_POSIX_STREAM_DESCRIPTOR)
20
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;
39
40// Chat over Redis pubsub. To test, run this program from multiple
41// terminals and type messages to stdin.
42
43auto receiver(std::shared_ptr<connection> conn) -> awaitable<void>
44{
45 request req;
46 req.push("SUBSCRIBE", "channel");
47
49 conn->set_receive_response(resp);
50
51 while (conn->will_reconnect()) {
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 << resp.value().at(1).value << " " << resp.value().at(2).value << " "
61 << resp.value().at(3).value << std::endl;
62 resp.value().clear();
63 }
64 }
65}
66
67// Publishes stdin messages to a Redis channel.
68auto publisher(std::shared_ptr<stream_descriptor> in, std::shared_ptr<connection> conn)
69 -> awaitable<void>
70{
71 for (std::string msg;;) {
72 auto n = co_await async_read_until(*in, dynamic_buffer(msg, 1024), "\n");
73 request req;
74 req.push("PUBLISH", "channel", msg);
75 co_await conn->async_exec(req, ignore);
76 msg.erase(0, n);
77 }
78}
79
80// Called from the main function (see main.cpp)
81auto co_main(config cfg) -> awaitable<void>
82{
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));
86
87 co_spawn(ex, receiver(conn), detached);
88 co_spawn(ex, publisher(stream, conn), detached);
89 conn->async_run(cfg, consign(detached, conn));
90
91 signal_set sig_set{ex, SIGINT, SIGTERM};
92 co_await sig_set.async_wait();
93 conn->cancel();
94 stream->cancel();
95}
96
97#else // defined(BOOST_ASIO_HAS_POSIX_STREAM_DESCRIPTOR)
98auto co_main(config const&) -> awaitable<void>
99{
100 std::cout << "Requires support for posix streams." << std::endl;
101 co_return;
102}
103#endif // defined(BOOST_ASIO_HAS_POSIX_STREAM_DESCRIPTOR)
104#endif // defined(BOOST_ASIO_HAS_CO_AWAIT)
A basic_connection that type erases the executor.
Creates Redis requests.
Definition request.hpp:46
ignore_t ignore
Global ignore object.
adapter::result< std::vector< resp3::node > > generic_response
A generic response to a request.
Definition response.hpp:35
Configure parameters used by the connection classes.
Definition config.hpp:30