diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index cd2a7a81..e5fdc1e4 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -52,6 +52,7 @@ make_test(test_run) make_test(test_conn_check_health) make_test(test_conn_exec) make_test(test_conn_push) +make_test(test_conn_monitor) make_test(test_conn_reconnect) make_test(test_conn_exec_cancel) make_test(test_conn_echo_stress) diff --git a/test/test_conn_check_health.cpp b/test/test_conn_check_health.cpp index 78efa82c..8be5ecf2 100644 --- a/test/test_conn_check_health.cpp +++ b/test/test_conn_check_health.cpp @@ -5,16 +5,15 @@ */ #include +#include #include -#include -#define BOOST_TEST_MODULE check_health -#include +#include +#include #include "common.hpp" -#include -#include +#include namespace net = boost::asio; namespace redis = boost::redis; @@ -22,116 +21,65 @@ using error_code = boost::system::error_code; using connection = 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; -// TODO: Test cancel(health_check) - namespace { -struct push_callback { - connection* conn1; - connection* conn2; - generic_response* resp2; - request* req1; - int i = 0; - boost::asio::coroutine coro{}; - - void operator()(error_code ec = {}, std::size_t = 0) - { - BOOST_ASIO_CORO_REENTER(coro) for (;;) - { - BOOST_ASIO_CORO_YIELD - conn2->async_receive(*this); - if (ec) { - std::clog << "Exiting." << std::endl; - return; - } - - BOOST_TEST(resp2->has_value()); - BOOST_TEST(!resp2->value().empty()); - std::clog << "Event> " << resp2->value().front().value << std::endl; - consume_one(*resp2); - - ++i; - - if (i == 5) { - std::clog << "Pausing the server" << std::endl; - // Pause the redis server to test if the health-check exits. - BOOST_ASIO_CORO_YIELD - conn1->async_exec(*req1, ignore, *this); - std::clog << "After pausing> " << ec.message() << std::endl; - // Don't know in CI we are getting: Got RESP3 simple-error. - //BOOST_TEST(!ec); - conn2->cancel(operation::run); - conn2->cancel(operation::receive); - conn2->cancel(operation::reconnection); - return; - } - } - }; -}; - -BOOST_AUTO_TEST_CASE(check_health) +void test_check_health() { + // Setup net::io_context ioc; - connection conn1{ioc}; + connection conn{ioc}; + // This request will block forever, causing the connection to become unresponsive request req1; - req1.push("CLIENT", "PAUSE", "10000", "ALL"); - - auto cfg1 = make_test_config(); - cfg1.health_check_id = "conn1"; - cfg1.reconnect_wait_interval = std::chrono::seconds::zero(); - - bool run1_finished = false, run2_finished = false, exec_finished = false; - - conn1.async_run(cfg1, {}, [&](error_code ec) { - run1_finished = true; - std::cout << "async_run 1 completed: " << ec.message() << std::endl; - BOOST_TEST(ec != error_code()); - }); - - //-------------------------------- - - // It looks like client pause does not work for clients that are - // sending MONITOR. I will therefore open a second connection. - connection conn2{ioc}; - - auto cfg2 = make_test_config(); - cfg2.health_check_id = "conn2"; - conn2.async_run(cfg2, {}, [&](error_code ec) { - run2_finished = true; - std::cout << "async_run 2 completed: " << ec.message() << std::endl; - BOOST_TEST(ec != error_code()); - }); + req1.push("BLPOP", "any", 0); + // This request should be executed after reconnection request req2; - req2.push("MONITOR"); - generic_response resp2; - conn2.set_receive_response(resp2); + req2.push("PING", "after_reconnection"); + req2.get_config().cancel_if_unresponded = false; + req2.get_config().cancel_on_connection_lost = false; - conn2.async_exec(req2, ignore, [&exec_finished](error_code ec, std::size_t) { - exec_finished = true; - std::cout << "async_exec: " << std::endl; - BOOST_TEST(ec == error_code()); + // Make the test run faster + auto cfg = make_test_config(); + cfg.health_check_interval = 500ms; + cfg.reconnect_wait_interval = 100ms; + + bool run_finished = false, exec1_finished = false, exec2_finished = false; + + conn.async_run(cfg, [&](error_code ec) { + run_finished = true; + BOOST_TEST_EQ(ec, net::error::operation_aborted); }); - //-------------------------------- + // This request will complete after the health checker deems the connection + // as unresponsive and triggers a reconnection (it's configured to be cancelled + // on connection lost). + conn.async_exec(req1, ignore, [&](error_code ec, std::size_t) { + exec1_finished = true; + BOOST_TEST_EQ(ec, net::error::operation_aborted); - push_callback{&conn1, &conn2, &resp2, &req1}(); // Starts reading pushes. + // Execute the second request. This one will succeed after reconnection + conn.async_exec(req2, ignore, [&](error_code ec2, std::size_t) { + exec2_finished = true; + BOOST_TEST_EQ(ec2, error_code()); + conn.cancel(); + }); + }); - ioc.run_for(2 * test_timeout); + ioc.run_for(test_timeout); - BOOST_TEST(run1_finished); - BOOST_TEST(run2_finished); - BOOST_TEST(exec_finished); - - // Waits before exiting otherwise it might cause subsequent tests - // to fail. - std::this_thread::sleep_for(std::chrono::seconds{10}); + BOOST_TEST(run_finished); + BOOST_TEST(exec1_finished); + BOOST_TEST(exec2_finished); } -} // namespace \ No newline at end of file +} // namespace + +int main() +{ + test_check_health(); + + return boost::report_errors(); +} \ No newline at end of file diff --git a/test/test_conn_monitor.cpp b/test/test_conn_monitor.cpp new file mode 100644 index 00000000..24d22ee1 --- /dev/null +++ b/test/test_conn_monitor.cpp @@ -0,0 +1,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 +#include +#include +#include + +#include +#include + +#include "common.hpp" + +#include + +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(); +} \ No newline at end of file