1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
|
/* Copyright (c) 2018-2022 Marcelo Zimbres Silva (mzimbres@gmail.com)
*
* Distributed under the Boost Software License, Version 1.0. (See
* accompanying file LICENSE.txt)
*/
#include <boost/redis/connection.hpp>
#include <boost/redis/ignore.hpp>
#include <boost/redis/request.hpp>
#include <boost/redis/response.hpp>
#include <boost/asio/error.hpp>
#include <boost/core/lightweight_test.hpp>
#include "common.hpp"
#include <cstddef>
namespace net = boost::asio;
using boost::system::error_code;
using boost::redis::connection;
using boost::redis::request;
using boost::redis::ignore;
using boost::redis::operation;
using boost::redis::generic_response;
using boost::redis::consume_one;
using namespace std::chrono_literals;
namespace {
// Verifies that using the MONITOR command works properly.
// Opens a connection, issues a MONITOR, issues some commands to
// generate some traffic, and waits for several MONITOR messages to arrive.
class test_monitor {
net::io_context ioc;
connection conn{ioc};
generic_response monitor_resp;
request ping_req;
bool run_finished = false, exec_finished = false, receive_finished = false;
int num_pushes_received = 0;
void start_receive()
{
conn.async_receive([this](error_code ec, std::size_t) {
// We should expect one push entry, at least
BOOST_TEST_EQ(ec, error_code());
BOOST_TEST(monitor_resp.has_value());
BOOST_TEST_NOT(monitor_resp.value().empty());
// Log the value and consume it
std::clog << "Event> " << monitor_resp.value().front().value << std::endl;
consume_one(monitor_resp);
if (++num_pushes_received >= 5) {
receive_finished = true;
} else {
start_receive();
}
});
}
// Starts generating traffic so our receiver task can progress
void start_generating_traffic()
{
conn.async_exec(ping_req, ignore, [this](error_code ec, std::size_t) {
// PINGs should complete successfully
BOOST_TEST_EQ(ec, error_code());
// Once the receiver exits, stop sending requests and tear down the connection
if (receive_finished) {
conn.cancel();
exec_finished = true;
} else {
start_generating_traffic();
}
});
}
public:
test_monitor() = default;
void run()
{
// Setup
ping_req.push("PING", "test_monitor");
conn.set_receive_response(monitor_resp);
request monitor_req;
monitor_req.push("MONITOR");
// Run the connection
conn.async_run(make_test_config(), [&](error_code ec) {
run_finished = true;
BOOST_TEST_EQ(ec, net::error::operation_aborted);
});
// Issue the monitor, then start generating traffic
conn.async_exec(monitor_req, ignore, [&](error_code ec, std::size_t) {
BOOST_TEST_EQ(ec, error_code());
start_generating_traffic();
});
// In parallel, start a subscriber
start_receive();
ioc.run_for(test_timeout);
BOOST_TEST(run_finished);
BOOST_TEST(receive_finished);
BOOST_TEST(exec_finished);
}
};
} // namespace
int main()
{
test_monitor{}.run();
return boost::report_errors();
}
|
