Adds support for PubSub state restoration

README.md

@@ -93,32 +93,39 @@ The coroutine below shows how to use it
 
 
 ```cpp
-auto
-receiver(std::shared_ptr<connection> conn) -> net::awaitable<void>
+auto receiver(std::shared_ptr<connection> conn) -> asio::awaitable<void>
 {
-   request req;
-   req.push("SUBSCRIBE", "channel");
-
-   flat_tree resp;
+   generic_flat_response resp;
    conn->set_receive_response(resp);
 
-   // Loop while reconnection is enabled
-   while (conn->will_reconnect()) {
-
-      // Reconnect to channels.
-      co_await conn->async_exec(req);
+   // Subscribe to the channel 'mychannel'. You can add any number of channels here.
+   request req;
+   req.subscribe({"mychannel"});
+   co_await conn->async_exec(req);
+
+   // You're now subscribed to 'mychannel'. Pushes sent over this channel will be stored
+   // in resp. If the connection encounters a network error and reconnects to the server,
+   // it will automatically subscribe to 'mychannel' again. This is transparent to the user.
+
+   // Loop to read Redis push messages.
+   for (error_code ec;;) {
+      // Wait for pushes
+      co_await conn->async_receive2(asio::redirect_error(ec));
+
+      // Check for errors and cancellations
+      if (ec && (ec != asio::experimental::error::channel_cancelled || !conn->will_reconnect())) {
+         std::cerr << "Error during receive2: " << ec << std::endl;
+         break;
+      }
 
-      // Loop reading Redis pushes.
-      for (error_code ec;;) {
-         co_await conn->async_receive2(resp, redirect_error(ec));
-         if (ec)
-            break; // Connection lost, break so we can reconnect to channels.
+      // The response must be consumed without suspending the
+      // coroutine i.e. without the use of async operations.
+      for (auto const& elem : resp.value().get_view())
+         std::cout << elem.value << "\n";
 
-         // Use the response resp in some way and then clear it.
-         ...
+      std::cout << std::endl;
 
-         resp.clear();
-      }
+      resp.value().clear();
    }
 }
 ```

doc/modules/ROOT/pages/index.adoc

@@ -104,32 +104,39 @@ The coroutine below shows how to use it
 
 [source,cpp]
 ----
-auto
-receiver(std::shared_ptr<connection> conn) -> net::awaitable<void>
+auto receiver(std::shared_ptr<connection> conn) -> asio::awaitable<void>
 {
-   request req;
-   req.push("SUBSCRIBE", "channel");
-
-   flat_tree resp;
+   generic_flat_response resp;
    conn->set_receive_response(resp);
 
-   // Loop while reconnection is enabled
-   while (conn->will_reconnect()) {
-
-      // Reconnect to channels.
-      co_await conn->async_exec(req);
+   // Subscribe to the channel 'mychannel'. You can add any number of channels here.
+   request req;
+   req.subscribe({"mychannel"});
+   co_await conn->async_exec(req);
+
+   // You're now subscribed to 'mychannel'. Pushes sent over this channel will be stored
+   // in resp. If the connection encounters a network error and reconnects to the server,
+   // it will automatically subscribe to 'mychannel' again. This is transparent to the user.
+
+   // Loop to read Redis push messages.
+   for (error_code ec;;) {
+      // Wait for pushes
+      co_await conn->async_receive2(asio::redirect_error(ec));
+
+      // Check for errors and cancellations
+      if (ec && (ec != asio::experimental::error::channel_cancelled || !conn->will_reconnect())) {
+         std::cerr << "Error during receive2: " << ec << std::endl;
+         break;
+      }
 
-      // Loop reading Redis pushes.
-      for (error_code ec;;) {
-         co_await conn->async_receive2(resp, redirect_error(ec));
-         if (ec)
-            break; // Connection lost, break so we can reconnect to channels.
+      // The response must be consumed without suspending the
+      // coroutine i.e. without the use of async operations.
+      for (auto const& elem : resp.value().get_view())
+         std::cout << elem.value << "\n";
 
-         // Use the response here and then clear it.
-         ...
+      std::cout << std::endl;
 
-         resp.clear();
-      }
+      resp.value().clear();
    }
 }
 ----

doc/modules/ROOT/pages/requests_responses.adoc

@@ -184,7 +184,7 @@ must **NOT** be included in the response tuple. For example, the following reque
 ----
 request req;
 req.push("PING");
-req.push("SUBSCRIBE", "channel");
+req.subscribe({"channel"});
 req.push("QUIT");
 ----
 

example/cpp20_chat_room.cpp

@@ -14,6 +14,7 @@
 #include <boost/asio/redirect_error.hpp>
 #include <boost/asio/signal_set.hpp>
 
+#include <exception>
 #include <iostream>
 #include <unistd.h>
 
@@ -40,31 +41,44 @@ using namespace std::chrono_literals;
 // Chat over Redis pubsub. To test, run this program from multiple
 // terminals and type messages to stdin.
 
+namespace {
+
+auto rethrow_on_error = [](std::exception_ptr exc) {
+   if (exc)
+      std::rethrow_exception(exc);
+};
+
 auto receiver(std::shared_ptr<connection> conn) -> awaitable<void>
 {
-   request req;
-   req.push("SUBSCRIBE", "channel");
-
+   // Set the receive response, so pushes are stored in resp
    generic_flat_response resp;
    conn->set_receive_response(resp);
 
-   while (conn->will_reconnect()) {
-      // Subscribe to channels.
-      co_await conn->async_exec(req);
+   // Subscribe to the channel 'channel'. Using request::subscribe()
+   // (instead of request::push()) makes the connection re-subscribe
+   // to 'channel' whenever it re-connects to the server.
+   request req;
+   req.subscribe({"channel"});
+   co_await conn->async_exec(req);
 
-      // Loop reading Redis push messages.
-      for (error_code ec;;) {
-         co_await conn->async_receive2(redirect_error(ec));
-         if (ec)
-            break;  // Connection lost, break so we can reconnect to channels.
+   for (error_code ec;;) {
+      // Wait for pushes
+      co_await conn->async_receive2(asio::redirect_error(ec));
 
-         for (auto const& elem: resp.value().get_view())
-            std::cout << elem.value << "\n";
+      // Check for errors and cancellations
+      if (ec && (ec != asio::experimental::error::channel_cancelled || !conn->will_reconnect())) {
+         std::cerr << "Error during receive2: " << ec << std::endl;
+         break;
+      }
 
-         std::cout << std::endl;
+      // The response must be consumed without suspending the
+      // coroutine i.e. without the use of async operations.
+      for (auto const& elem : resp.value().get_view())
+         std::cout << elem.value << "\n";
 
-         resp.value().clear();
-      }
+      std::cout << std::endl;
+
+      resp.value().clear();
    }
 }
 
@@ -81,15 +95,17 @@ auto publisher(std::shared_ptr<stream_descriptor> in, std::shared_ptr<connection
    }
 }
 
+}  // namespace
+
 // Called from the main function (see main.cpp)
 auto co_main(config cfg) -> awaitable<void>
 {
    auto ex = co_await asio::this_coro::executor;
    auto conn = std::make_shared<connection>(ex);
    auto stream = std::make_shared<stream_descriptor>(ex, ::dup(STDIN_FILENO));
 
-   co_spawn(ex, receiver(conn), detached);
-   co_spawn(ex, publisher(stream, conn), detached);
+   co_spawn(ex, receiver(conn), rethrow_on_error);
+   co_spawn(ex, publisher(stream, conn), rethrow_on_error);
    conn->async_run(cfg, consign(detached, conn));
 
    signal_set sig_set{ex, SIGINT, SIGTERM};

example/cpp20_subscriber.cpp

@@ -10,6 +10,7 @@
 #include <boost/asio/co_spawn.hpp>
 #include <boost/asio/consign.hpp>
 #include <boost/asio/detached.hpp>
+#include <boost/asio/experimental/channel_error.hpp>
 #include <boost/asio/redirect_error.hpp>
 #include <boost/asio/signal_set.hpp>
 #include <boost/asio/use_awaitable.hpp>
@@ -32,7 +33,7 @@ using asio::signal_set;
  * To test send messages with redis-cli
  *
  *    $ redis-cli -3
- *    127.0.0.1:6379> PUBLISH channel some-message
+ *    127.0.0.1:6379> PUBLISH mychannel some-message
  *    (integer) 3
  *    127.0.0.1:6379>
  *
@@ -46,33 +47,39 @@ using asio::signal_set;
 // Receives server pushes.
 auto receiver(std::shared_ptr<connection> conn) -> asio::awaitable<void>
 {
-   request req;
-   req.push("SUBSCRIBE", "channel");
-
    generic_flat_response resp;
    conn->set_receive_response(resp);
 
-   // Loop while reconnection is enabled
-   while (conn->will_reconnect()) {
-      // Reconnect to the channels.
-      co_await conn->async_exec(req);
-
-      // Loop to read Redis push messages.
-      for (error_code ec;;) {
-         // Wait for pushes
-         co_await conn->async_receive2(asio::redirect_error(ec));
-         if (ec)
-            break;  // Connection lost, break so we can reconnect to channels.
+   // Subscribe to the channel 'mychannel'. You can add any number of channels here.
+   request req;
+   req.subscribe({"mychannel"});
+   co_await conn->async_exec(req);
+
+   // You're now subscribed to 'mychannel'. Pushes sent over this channel will be stored
+   // in resp. If the connection encounters a network error and reconnects to the server,
+   // it will automatically subscribe to 'mychannel' again. This is transparent to the user.
+   // You need to use specialized request::subscribe() function (instead of request::push)
+   // to enable this behavior.
+
+   // Loop to read Redis push messages.
+   for (error_code ec;;) {
+      // Wait for pushes
+      co_await conn->async_receive2(asio::redirect_error(ec));
+
+      // Check for errors and cancellations
+      if (ec && (ec != asio::experimental::error::channel_cancelled || !conn->will_reconnect())) {
+         std::cerr << "Error during receive2: " << ec << std::endl;
+         break;
+      }
 
-         // The response must be consumed without suspending the
-         // coroutine i.e. without the use of async operations.
-         for (auto const& elem: resp.value().get_view())
-            std::cout << elem.value << "\n";
+      // The response must be consumed without suspending the
+      // coroutine i.e. without the use of async operations.
+      for (auto const& elem : resp.value().get_view())
+         std::cout << elem.value << "\n";
 
-         std::cout << std::endl;
+      std::cout << std::endl;
 
-         resp.value().clear();
-      }
+      resp.value().clear();
    }
 }
 

include/boost/redis/connection.hpp

@@ -108,7 +108,10 @@ struct connection_impl {
       {
          while (true) {
             // Invoke the state machine
-            auto act = fsm_.resume(obj_->is_open(), self.get_cancellation_state().cancelled());
+            auto act = fsm_.resume(
+               obj_->is_open(),
+               obj_->st_,
+               self.get_cancellation_state().cancelled());
 
             // Do what the FSM said
             switch (act.type()) {
@@ -203,7 +206,7 @@ struct connection_impl {
       });
 
       return asio::async_compose<CompletionToken, void(system::error_code, std::size_t)>(
-         exec_op{this, notifier, exec_fsm(st_.mpx, std::move(info))},
+         exec_op{this, notifier, exec_fsm(std::move(info))},
          token,
          writer_cv_);
    }

include/boost/redis/detail/connection_state.hpp

@@ -11,6 +11,7 @@
 
 #include <boost/redis/config.hpp>
 #include <boost/redis/detail/multiplexer.hpp>
+#include <boost/redis/detail/subscription_tracker.hpp>
 #include <boost/redis/logger.hpp>
 #include <boost/redis/request.hpp>
 #include <boost/redis/resp3/node.hpp>
@@ -47,7 +48,9 @@ struct connection_state {
    config cfg{};
    multiplexer mpx{};
    std::string diagnostic{};  // Used by the setup request and Sentinel
+   request setup_req{};
    request ping_req{};
+   subscription_tracker tracker{};
 
    // Sentinel stuff
    lazy_random_engine eng{};

include/boost/redis/detail/exec_fsm.hpp

@@ -21,6 +21,8 @@
 
 namespace boost::redis::detail {
 
+struct connection_state;
+
 // What should we do next?
 enum class exec_action_type
 {
@@ -54,16 +56,17 @@ class exec_action {
 
 class exec_fsm {
    int resume_point_{0};
-   multiplexer* mpx_{nullptr};
    std::shared_ptr<multiplexer::elem> elem_;
 
 public:
-   exec_fsm(multiplexer& mpx, std::shared_ptr<multiplexer::elem> elem) noexcept
-   : mpx_(&mpx)
-   , elem_(std::move(elem))
+   exec_fsm(std::shared_ptr<multiplexer::elem> elem) noexcept
+   : elem_(std::move(elem))
    { }
 
-   exec_action resume(bool connection_is_open, asio::cancellation_type_t cancel_state);
+   exec_action resume(
+      bool connection_is_open,
+      connection_state& st,
+      asio::cancellation_type_t cancel_state);
 };
 
 }  // namespace boost::redis::detail