Progresses with the subscriber.

Author: mzimbres

examples/cpp17_intro.cpp

@@ -7,6 +7,7 @@
 #include <iostream>
 #include <boost/redis/run.hpp>
 #include <boost/redis/address.hpp>
+#include <boost/redis/logger.hpp>
 #include <boost/redis/src.hpp>
 
 namespace net = boost::asio;
@@ -15,6 +16,7 @@ using boost::redis::request;
 using boost::redis::response;
 using boost::redis::ignore_t;
 using boost::redis::async_run;
+using boost::redis::logger;
 using boost::redis::address;
 using namespace std::chrono_literals;
 
@@ -39,7 +41,7 @@ auto main(int argc, char * argv[]) -> int
       net::io_context ioc;
       connection conn{ioc};
 
-      async_run(conn, addr, 10s, 10s, [&](auto){
+      async_run(conn, addr, 10s, 10s, logger{}, [&](auto){
          conn.cancel();
       });
 

examples/cpp17_intro_sync.cpp

@@ -8,6 +8,7 @@
 #include <boost/redis/request.hpp>
 #include <boost/redis/check_health.hpp>
 #include <boost/redis/run.hpp>
+#include <boost/redis/logger.hpp>
 #include <boost/asio/deferred.hpp>
 #include <boost/asio/use_future.hpp>
 #include <tuple>
@@ -27,6 +28,7 @@ using boost::redis::response;
 using boost::redis::ignore_t;
 using boost::redis::async_run;
 using boost::redis::address;
+using boost::redis::logger;
 using boost::redis::async_check_health;
 using namespace std::chrono_literals;
 
@@ -56,7 +58,7 @@ auto main(int argc, char * argv[]) -> int
       // Starts a thread that will can io_context::run on which the
       // connection will run.
       std::thread t{[&ioc, conn, addr]() {
-         async_run(*conn, addr, 10s, 10s, [conn](auto){
+         async_run(*conn, addr, 10s, 10s, logger{}, [conn](auto){
             conn->cancel();
          });
 

examples/cpp20_subscriber.cpp

@@ -4,24 +4,25 @@
  * accompanying file LICENSE.txt)
  */
 
-#include <boost/redis/run.hpp>
-#include <boost/redis/check_health.hpp>
+#include <boost/redis/experimental/connector.hpp>
+#include <boost/redis/logger.hpp>
 #include <boost/asio/awaitable.hpp>
+#include <boost/asio/detached.hpp>
 #include <boost/asio/use_awaitable.hpp>
+#include <boost/asio/co_spawn.hpp>
+#include <boost/asio/consign.hpp>
 #include <iostream>
 
 #if defined(BOOST_ASIO_HAS_CO_AWAIT)
-#include <boost/asio/experimental/awaitable_operators.hpp>
 
 namespace net = boost::asio;
-using namespace net::experimental::awaitable_operators;
-using steady_timer = net::use_awaitable_t<>::as_default_on_t<net::steady_timer>;
-using boost::redis::request;
-using boost::redis::async_run;
-using boost::redis::generic_response;
-using boost::redis::async_check_health;
-using boost::redis::address;
-using connection = net::use_awaitable_t<>::as_default_on_t<boost::redis::connection>;
+namespace redis = boost::redis;
+using redis::generic_response;
+using redis::address;
+using redis::logger;
+using redis::experimental::async_connect;
+using redis::experimental::connect_config;
+using connection = net::use_awaitable_t<>::as_default_on_t<redis::connection>;
 
 /* This example will subscribe and read pushes indefinitely.
  *
@@ -39,12 +40,16 @@ using connection = net::use_awaitable_t<>::as_default_on_t<boost::redis::connect
  * > CLIENT kill TYPE pubsub
  */
 
-// Receives pushes.
+// Receives server pushes.
 auto receiver(std::shared_ptr<connection> conn) -> net::awaitable<void>
 {
    for (generic_response resp;;) {
       co_await conn->async_receive(resp);
-      std::cout << resp.value().at(1).value << " " << resp.value().at(2).value << " " << resp.value().at(3).value << std::endl;
+      std::cout
+         << resp.value().at(1).value
+         << " " << resp.value().at(2).value
+         << " " << resp.value().at(3).value
+         << std::endl;
       resp.value().clear();
    }
 }
@@ -53,19 +58,10 @@ auto co_main(address const& addr) -> net::awaitable<void>
 {
    auto ex = co_await net::this_coro::executor;
    auto conn = std::make_shared<connection>(ex);
-   steady_timer timer{ex};
-
-   request req;
-   req.push("HELLO", 3);
-   req.push("SUBSCRIBE", "channel");
-
-   // The loop will reconnect on connection lost. To exit type Ctrl-C twice.
-   for (;;) {
-      co_await ((async_run(*conn, addr) || async_check_health(*conn) || receiver(conn)) && conn->async_exec(req));
-      conn->reset_stream();
-      timer.expires_after(std::chrono::seconds{1});
-      co_await timer.async_wait();
-   }
+   connect_config cfg;
+   cfg.addr = addr;
+   net::co_spawn(ex, receiver(conn), net::detached);
+   redis::experimental::async_connect(*conn, cfg, logger{}, net::consign(net::detached, conn));
 }
 
 #endif // defined(BOOST_ASIO_HAS_CO_AWAIT)

include/boost/redis/check_health.hpp

@@ -38,7 +38,7 @@ class check_health_op {
       {
          checker->prom_.emplace(conn->async_exec(checker->req_, checker->resp_, asio::experimental::use_promise));
 
-         checker->timer_.expires_after(checker->interval_);
+         checker->timer_.expires_after(checker->timeout_);
          BOOST_ASIO_CORO_YIELD
          checker->timer_.async_wait(std::move(self));
          if (ec || is_cancelled(self) || checker->resp_.value().empty()) {
@@ -70,7 +70,7 @@ class health_checker {
       std::string const& msg,
       std::chrono::steady_clock::duration interval)
    : timer_{ex}
-   , interval_{interval}
+   , timeout_{interval}
    {
       req_.push("PING", msg);
    }
@@ -93,13 +93,20 @@ class health_checker {
       prom_.reset();
    }
 
+   void cancel()
+   {
+      timer_.cancel();
+      if (prom_)
+         prom_.cancel();
+   }
+
 private:
    template <class, class> friend class check_health_op;
    timer_type timer_;
    std::optional<promise_type> prom_;
    redis::request req_;
    redis::generic_response resp_;
-   std::chrono::steady_clock::duration interval_;
+   std::chrono::steady_clock::duration timeout_;
 };
 
 } // detail

include/boost/redis/detail/runner.hpp

@@ -48,6 +48,8 @@ struct resolve_op {
             asio::experimental::wait_for_one(),
             std::move(self));
 
+         runner->logger_.on_resolve(ec1, res);
+
          if (is_cancelled(self)) {
             self.complete(asio::error::operation_aborted);
             return;
@@ -86,10 +88,10 @@ struct connect_op {
 
    template <class Self>
    void operator()( Self& self
-                  , std::array<std::size_t, 2> order = {}
-                  , system::error_code ec1 = {}
-                  , asio::ip::tcp::endpoint const& = {}
-                  , system::error_code ec2 = {})
+                  , std::array<std::size_t, 2> const& order = {}
+                  , system::error_code const& ec1 = {}
+                  , asio::ip::tcp::endpoint const& ep= {}
+                  , system::error_code const& ec2 = {})
    {
       BOOST_ASIO_CORO_REENTER (coro)
       {
@@ -105,6 +107,8 @@ struct connect_op {
             asio::experimental::wait_for_one(),
             std::move(self));
 
+         runner->logger_.on_connect(ec1, ep);
+
          if (is_cancelled(self)) {
             self.complete(asio::error::operation_aborted);
             return;
@@ -160,10 +164,21 @@ struct runner_op {
    }
 };
 
-template <class Executor>
+template <class Executor, class Logger>
 class runner {
 public:
-   runner(Executor ex, address addr): resv_{ex}, timer_{ex}, addr_{addr} {}
+   using timer_type =
+      asio::basic_waitable_timer<
+         std::chrono::steady_clock,
+         asio::wait_traits<std::chrono::steady_clock>,
+         Executor>;
+
+   runner(Executor ex, address addr, Logger l = Logger{})
+   : resv_{ex}
+   , timer_{ex}
+   , addr_{addr}
+   , logger_{l}
+   {}
 
    template <class CompletionToken>
    auto async_resolve(CompletionToken&& token)
@@ -197,14 +212,14 @@ class runner {
          >(runner_op<runner, Connection>{this, &conn, resolve_timeout, connect_timeout}, token, resv_);
    }
 
+   void cancel()
+   {
+      resv_.cancel();
+      timer_.cancel();
+   }
+
 private:
    using resolver_type = asio::ip::basic_resolver<asio::ip::tcp, Executor>;
-   using timer_type =
-      asio::basic_waitable_timer<
-         std::chrono::steady_clock,
-         asio::wait_traits<std::chrono::steady_clock>,
-         Executor>;
-
 
    template <class, class> friend struct runner_op;
    template <class, class> friend struct connect_op;
@@ -214,6 +229,7 @@ class runner {
    timer_type timer_;
    address addr_;
    asio::ip::tcp::resolver::results_type endpoints_;
+   Logger logger_;
 };
 
 } // boost::redis::detail

include/boost/redis/logger.hpp

@@ -0,0 +1,45 @@
+/* Copyright (c) 2018-2022 Marcelo Zimbres Silva ([email protected])
+ *
+ * Distributed under the Boost Software License, Version 1.0. (See
+ * accompanying file LICENSE.txt)
+ */
+
+#ifndef BOOST_REDIS_LOGGER_HPP
+#define BOOST_REDIS_LOGGER_HPP
+
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/system/error_code.hpp>
+#include <iostream>
+
+namespace boost::redis {
+
+// TODO: Move to ipp file.
+// TODO: Implement filter.
+class logger {
+public:
+   void on_resolve(system::error_code const& ec, asio::ip::tcp::resolver::results_type const& res)
+   {
+      // TODO: Print the endpoints
+      std::clog << "on_resolve: " << ec.message() << std::endl;
+   }
+
+   void on_connect(system::error_code const& ec, asio::ip::tcp::endpoint const& ep)
+   {
+      // TODO: Print the endpoint
+      std::clog << "on_connect: " << ec.message() << std::endl;
+   }
+
+   void on_connection_lost()
+   {
+      std::clog << "on_connection_lost: " << std::endl;
+   }
+
+   void on_hello(system::error_code const& ec)
+   {
+      std::clog << "on_hello: " << ec.message() << std::endl;
+   }
+};
+
+} // boost::redis
+
+#endif // BOOST_REDIS_LOGGER_HPP

include/boost/redis/run.hpp

@@ -7,16 +7,17 @@
 #ifndef BOOST_REDIS_RUN_HPP
 #define BOOST_REDIS_RUN_HPP
 
-// Has to included before promise.hpp to build on msvc.
 #include <boost/redis/detail/runner.hpp>
 #include <boost/redis/connection.hpp>
 #include <boost/redis/address.hpp>
+#include <boost/redis/logger.hpp>
 #include <boost/asio/compose.hpp>
 #include <boost/asio/consign.hpp>
 #include <memory>
 #include <chrono>
 
-namespace boost::redis {
+namespace boost::redis
+{
 
 /** @brief Call async_run on the connection.
  *  @ingroup high-level-api
@@ -35,6 +36,7 @@ namespace boost::redis {
  */
 template <
    class Socket,
+   class Logger = logger,
    class CompletionToken = asio::default_completion_token_t<typename Socket::executor_type>
 >
 auto
@@ -43,11 +45,12 @@ async_run(
    address addr = address{"127.0.0.1", "6379"},
    std::chrono::steady_clock::duration resolve_timeout = std::chrono::seconds{10},
    std::chrono::steady_clock::duration connect_timeout = std::chrono::seconds{10},
+   Logger l = Logger{},
    CompletionToken token = CompletionToken{})
 {
    using executor_type = typename Socket::executor_type;
-   using runner_type = detail::runner<executor_type>;
-   auto runner = std::make_shared<runner_type>(conn.get_executor(), addr);
+   using runner_type = detail::runner<executor_type, Logger>;
+   auto runner = std::make_shared<runner_type>(conn.get_executor(), addr, l);
 
    return
       runner->async_run(

tests/conn_check_health.cpp

@@ -5,6 +5,7 @@
  */
 
 #include <boost/redis/run.hpp>
+#include <boost/redis/logger.hpp>
 #include <boost/redis/check_health.hpp>
 #include <boost/system/errc.hpp>
 #define BOOST_TEST_MODULE check-health
@@ -22,6 +23,7 @@ using boost::redis::operation;
 using boost::redis::generic_response;
 using boost::redis::async_check_health;
 using boost::redis::async_run;
+using boost::redis::logger;
 using boost::redis::address;
 using namespace std::chrono_literals;
 
@@ -104,12 +106,12 @@ BOOST_AUTO_TEST_CASE(check_health)
    generic_response resp;
    push_callback{&conn, &conn2, &resp, &req2}(); // Starts reading pushes.
 
-   async_run(conn, address{}, 10s, 10s, [](auto ec){
+   async_run(conn, address{}, 10s, 10s, logger{}, [](auto ec){
       std::cout << "B" << std::endl;
       BOOST_TEST(!!ec);
    });
 
-   async_run(conn2, address{}, 10s, 10s, [](auto ec){
+   async_run(conn2, address{}, 10s, 10s, logger{}, [](auto ec){
       std::cout << "C" << std::endl;
       BOOST_TEST(!!ec);
    });