Reduces the time elapsed by test_conn_health_check and adds test_conn_monitor (#316)

Splits test_conn_health_check into a test to verify that the health checker works and another to verify that MONITOR is handled correctly.
Tests no longer wait or use CLIENT PAUSE

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)

test/test_conn_check_health.cpp

@@ -5,16 +5,15 @@
  */
 
 #include <boost/redis/connection.hpp>
+#include <boost/redis/ignore.hpp>
 #include <boost/redis/response.hpp>
 
-#include <cstddef>
-#define BOOST_TEST_MODULE check_health
-#include <boost/test/included/unit_test.hpp>
+#include <boost/asio/error.hpp>
+#include <boost/core/lightweight_test.hpp>
 
 #include "common.hpp"
 
-#include <iostream>
-#include <thread>
+#include <cstddef>
 
 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
+}  // namespace
+
+int main()
+{
+   test_check_health();
+
+   return boost::report_errors();
+}

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 <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();
+}