Adds subscription tracking support for (p)unsubscribe without arguments

include/boost/redis/impl/request.ipp

@@ -68,6 +68,17 @@ void request::add_pubsub_arg(detail::pubsub_change_type type, std::string_view v
    pubsub_changes_.push_back({type, offset, value.size()});
 }
 
+void request::push_pubsub_all(std::string_view cmd, detail::pubsub_change_type type)
+{
+   resp3::add_header(payload_, resp3::type::array, 1);
+   resp3::add_bulk(payload_, cmd);
+
+   // Track the change. These change types have no associated channel or pattern
+   pubsub_changes_.push_back({type, 0u, 0u});
+
+   ++commands_;  // these commands don't have a response
+}
+
 void request::hello() { push("HELLO", "3"); }
 
 void request::hello(std::string_view username, std::string_view password)

include/boost/redis/impl/subscription_tracker.ipp

@@ -15,6 +15,13 @@
 
 namespace boost::redis::detail {
 
+// Given a request and a change, returns an owning string
+// with the channel or pattern name affected by the change
+inline std::string get_channel_owning(const request& req, const pubsub_change& ch)
+{
+   return std::string(req.payload().substr(ch.channel_offset, ch.channel_size));
+}
+
 void subscription_tracker::clear()
 {
    channels_.clear();
@@ -24,13 +31,16 @@ void subscription_tracker::clear()
 void subscription_tracker::commit_changes(const request& req)
 {
    for (const auto& ch : request_access::pubsub_changes(req)) {
-      std::string channel{req.payload().substr(ch.channel_offset, ch.channel_size)};
       switch (ch.type) {
-         case pubsub_change_type::subscribe:    channels_.insert(std::move(channel)); break;
-         case pubsub_change_type::unsubscribe:  channels_.erase(std::move(channel)); break;
-         case pubsub_change_type::psubscribe:   pchannels_.insert(std::move(channel)); break;
-         case pubsub_change_type::punsubscribe: pchannels_.erase(std::move(channel)); break;
-         default:                               BOOST_ASSERT(false);
+         case pubsub_change_type::subscribe:   channels_.insert(get_channel_owning(req, ch)); break;
+         case pubsub_change_type::unsubscribe: channels_.erase(get_channel_owning(req, ch)); break;
+         case pubsub_change_type::psubscribe:  pchannels_.insert(get_channel_owning(req, ch)); break;
+         case pubsub_change_type::punsubscribe:
+            pchannels_.erase(get_channel_owning(req, ch));
+            break;
+         case pubsub_change_type::unsubscribe_all:  channels_.clear(); break;
+         case pubsub_change_type::punsubscribe_all: pchannels_.clear(); break;
+         default:                                   BOOST_ASSERT(false);
       }
    }
 }

include/boost/redis/request.hpp

@@ -32,10 +32,13 @@ enum class pubsub_change_type
    unsubscribe,
    psubscribe,
    punsubscribe,
+   unsubscribe_all,
+   punsubscribe_all,
 };
 
 struct pubsub_change {
    pubsub_change_type type;
+   // Unused for unsubscribe_all and punsubscribe_all
    std::size_t channel_offset;
    std::size_t channel_size;
 };
@@ -576,6 +579,29 @@ class request {
          channels_end);
    }
 
+   /**
+    * @brief Appends an argument-less UNSUBSCRIBE command to the end of the request.
+    *
+    * The resulting command is `UNSUBSCRIBE`, which unsubscribes
+    * from all the channels the connection is subscribed to.
+    * Pattern subscriptions (created by `PSUBSCRIBE`) are not affected.
+    *
+    * Subscriptions removed using this function are tracked
+    * to enable PubSub state restoration. After successfully executing
+    * the request, the connection will store any newly subscribed channels and patterns.
+    * Every time a reconnection happens,
+    * a suitable `SUBSCRIBE`/`PSUBSCRIBE` command is issued automatically,
+    * to restore the subscriptions that were active before the reconnection.
+    *
+    * PubSub store restoration only happens when using @ref subscribe,
+    * @ref unsubscribe, @ref psubscribe or @ref punsubscribe.
+    * Subscription commands added by @ref push or @ref push_range are not tracked.
+    */
+   void unsubscribe()
+   {
+      push_pubsub_all("UNSUBSCRIBE", detail::pubsub_change_type::unsubscribe_all);
+   }
+
    /**
     * @brief Appends a PSUBSCRIBE command to the end of the request.
     *
@@ -724,6 +750,29 @@ class request {
          patterns_end);
    }
 
+   /**
+    * @brief Appends an argument-less PUNSUBSCRIBE command to the end of the request.
+    *
+    * The resulting command is `PUNSUBSCRIBE`, which unsubscribes
+    * from all the patterns the connection is subscribed to.
+    * Channel subscriptions (created by `SUBSCRIBE`) are not affected.
+    *
+    * Subscriptions removed using this function are tracked
+    * to enable PubSub state restoration. After successfully executing
+    * the request, the connection will store any newly subscribed channels and patterns.
+    * Every time a reconnection happens,
+    * a suitable `SUBSCRIBE`/`PSUBSCRIBE` command is issued automatically,
+    * to restore the subscriptions that were active before the reconnection.
+    *
+    * PubSub store restoration only happens when using @ref subscribe,
+    * @ref unsubscribe, @ref psubscribe or @ref punsubscribe.
+    * Subscription commands added by @ref push or @ref push_range are not tracked.
+    */
+   void punsubscribe()
+   {
+      push_pubsub_all("PUNSUBSCRIBE", detail::pubsub_change_type::punsubscribe_all);
+   }
+
    /** @brief Appends a HELLO 3 command to the end of the request.
     *
     * Equivalent to adding the Redis command `HELLO 3`.
@@ -790,6 +839,9 @@ class request {
 
    void add_pubsub_arg(detail::pubsub_change_type type, std::string_view value);
 
+   // Adds an argument-less UNSUBSCRIBE/PUNSUBSCRIBE command, with tracking
+   void push_pubsub_all(std::string_view cmd, detail::pubsub_change_type type);
+
    template <class ForwardIt>
    void push_pubsub(
       std::string_view cmd,

test/test_conn_push2.cpp

@@ -834,6 +834,106 @@ struct test_pubsub_state_restoration_impl {
 };
 void test_pubsub_state_restoration() { test_pubsub_state_restoration_impl{}.run(); }
 
+// (P)UNSUBSCRIBE (without arguments) can be used with pubsub state restoration
+struct test_pubsub_state_restoration_unsubscribe_all_impl {
+   net::io_context ioc;
+   connection conn{ioc};
+   request req{};
+   response<std::string> resp_str{};
+   bool exec_finished = false;
+
+   void sub()
+   {
+      // Subscribe to some channels and patterns
+      req.clear();
+      req.subscribe({"ch1", "ch2", "ch3"});              // active: 1, 2, 3
+      req.psubscribe({"ch1*", "ch2*", "ch3*", "ch4*"});  // active: 1, 2, 3, 4
+      conn.async_exec(req, ignore, [this](error_code ec, std::size_t) {
+         BOOST_TEST_EQ(ec, error_code());
+         unsub();
+      });
+   }
+
+   void unsub()
+   {
+      // Unsubscribe from all channels and patterns.
+      req.clear();
+      req.unsubscribe();
+      req.punsubscribe();
+
+      // Leave one subscribed channel to make checks stronger.
+      req.subscribe({"ch9"});
+
+      // Validate that we're subscribed to what we expect
+      req.push("CLIENT", "INFO");
+
+      conn.async_exec(req, resp_str, [this](error_code ec, std::size_t) {
+         BOOST_TEST_EQ(ec, error_code());
+
+         // We are subscribed to 1 channel and 0 patterns
+         BOOST_TEST(std::get<0>(resp_str).has_value());
+         BOOST_TEST_EQ(find_client_info(std::get<0>(resp_str).value(), "sub"), "1");
+         BOOST_TEST_EQ(find_client_info(std::get<0>(resp_str).value(), "psub"), "0");
+
+         quit();
+      });
+   }
+
+   void quit()
+   {
+      // Trigger a reconnection
+      req.clear();
+      req.push("QUIT");
+
+      conn.async_exec(req, ignore, [this](error_code, std::size_t) {
+         // we don't know if this request will complete successfully or not
+         client_info();
+      });
+   }
+
+   void client_info()
+   {
+      req.clear();
+      req.push("CLIENT", "INFO");
+      req.get_config().cancel_if_unresponded = false;
+
+      conn.async_exec(req, resp_str, [this](error_code ec, std::size_t) {
+         BOOST_TEST_EQ(ec, error_code());
+
+         // We are subscribed to 1 channel and 0 patterns
+         BOOST_TEST(std::get<0>(resp_str).has_value());
+         BOOST_TEST_EQ(find_client_info(std::get<0>(resp_str).value(), "sub"), "1");
+         BOOST_TEST_EQ(find_client_info(std::get<0>(resp_str).value(), "psub"), "0");
+
+         exec_finished = true;
+         conn.cancel();
+      });
+   }
+
+   void run()
+   {
+      // Start the request chain
+      sub();
+
+      // Start running
+      bool run_finished = false;
+      conn.async_run(make_test_config(), [&run_finished](error_code ec) {
+         BOOST_TEST_EQ(ec, net::error::operation_aborted);
+         run_finished = true;
+      });
+
+      ioc.run_for(test_timeout);
+
+      // Done
+      BOOST_TEST(exec_finished);
+      BOOST_TEST(run_finished);
+   }
+};
+void test_pubsub_state_restoration_unsubscribe_all()
+{
+   test_pubsub_state_restoration_unsubscribe_all_impl{}.run();
+}
+
 }  // namespace
 
 int main()
@@ -853,6 +953,7 @@ int main()
    test_push_consumer();
    test_unsubscribe();
    test_pubsub_state_restoration();
+   test_pubsub_state_restoration_unsubscribe_all();
 
    return boost::report_errors();
 }