From ebfdcd487b305b4774139c6a8c8eb90c5f8ad289 Mon Sep 17 00:00:00 2001 From: Ruben Perez Date: Wed, 29 Jul 2026 16:50:53 +0200 Subject: [PATCH 1/5] Initial impl --- include/boost/redis/impl/request.ipp | 11 ++++ .../boost/redis/impl/subscription_tracker.ipp | 21 +++++--- include/boost/redis/request.hpp | 52 +++++++++++++++++++ 3 files changed, 78 insertions(+), 6 deletions(-) diff --git a/include/boost/redis/impl/request.ipp b/include/boost/redis/impl/request.ipp index e00d0f76..1706f8f1 100644 --- a/include/boost/redis/impl/request.ipp +++ b/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) diff --git a/include/boost/redis/impl/subscription_tracker.ipp b/include/boost/redis/impl/subscription_tracker.ipp index ec23675f..1924159a 100644 --- a/include/boost/redis/impl/subscription_tracker.ipp +++ b/include/boost/redis/impl/subscription_tracker.ipp @@ -15,6 +15,12 @@ namespace boost::redis::detail { +// Given a +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 +30,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); } } } diff --git a/include/boost/redis/request.hpp b/include/boost/redis/request.hpp index b1e02a56..5c8d74ba 100644 --- a/include/boost/redis/request.hpp +++ b/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 void push_pubsub( std::string_view cmd, From c21a6d7978e9f26c8a6ecb8fd44b1c24fba7c242 Mon Sep 17 00:00:00 2001 From: Ruben Perez Date: Wed, 29 Jul 2026 17:47:56 +0200 Subject: [PATCH 2/5] test_request --- test/test_request.cpp | 107 ++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 102 insertions(+), 5 deletions(-) diff --git a/test/test_request.cpp b/test/test_request.cpp index a43469fb..d158d584 100644 --- a/test/test_request.cpp +++ b/test/test_request.cpp @@ -30,11 +30,13 @@ namespace { const char* to_string(pubsub_change_type type) { switch (type) { - case pubsub_change_type::subscribe: return "subscribe"; - case pubsub_change_type::unsubscribe: return "unsubscribe"; - case pubsub_change_type::psubscribe: return "psubscribe"; - case pubsub_change_type::punsubscribe: return "punsubscribe"; - default: return ""; + case pubsub_change_type::subscribe: return "subscribe"; + case pubsub_change_type::unsubscribe: return "unsubscribe"; + case pubsub_change_type::psubscribe: return "psubscribe"; + case pubsub_change_type::punsubscribe: return "punsubscribe"; + case pubsub_change_type::unsubscribe_all: return "unsubscribe_all"; + case pubsub_change_type::punsubscribe_all: return "punsubscribe_all"; + default: return ""; } } @@ -328,6 +330,22 @@ void test_unsubscribe_initializer_list() fix.check_unsubscribe(); } +// The version with no arguments (unsubscribe from all channels) works +void test_unsubscribe_no_args() +{ + request req; + + req.unsubscribe(); + + BOOST_TEST_EQ(req.payload(), "*1\r\n$11\r\nUNSUBSCRIBE\r\n"); + BOOST_TEST_EQ(req.get_commands(), 1u); + BOOST_TEST_EQ(req.get_expected_responses(), 0u); + const pubsub_change_str expected_changes[] = { + {pubsub_change_type::unsubscribe_all, ""}, + }; + check_pubsub_changes(req, expected_changes); +} + // --- psubscribe --- void test_psubscribe_iterators() { @@ -442,6 +460,50 @@ void test_punsubscribe_initializer_list() fix.check_punsubscribe(); } +// The version with no arguments (unsubscribe from all patterns) works +void test_punsubscribe_no_args() +{ + request req; + + req.punsubscribe(); + + BOOST_TEST_EQ(req.payload(), "*1\r\n$12\r\nPUNSUBSCRIBE\r\n"); + BOOST_TEST_EQ(req.get_commands(), 1u); + BOOST_TEST_EQ(req.get_expected_responses(), 0u); + const pubsub_change_str expected_changes[] = { + {pubsub_change_type::punsubscribe_all, ""}, + }; + check_pubsub_changes(req, expected_changes); +} + +// Empty channel and pattern names are legal, and are not confused +// with the argument-less overloads +void test_pubsub_empty_channel_name() +{ + request req; + + req.subscribe({""}); + req.unsubscribe({""}); + req.psubscribe({""}); + req.punsubscribe({""}); + + constexpr std::string_view expected = + "*2\r\n$9\r\nSUBSCRIBE\r\n$0\r\n\r\n" + "*2\r\n$11\r\nUNSUBSCRIBE\r\n$0\r\n\r\n" + "*2\r\n$10\r\nPSUBSCRIBE\r\n$0\r\n\r\n" + "*2\r\n$12\r\nPUNSUBSCRIBE\r\n$0\r\n\r\n"; + BOOST_TEST_EQ(req.payload(), expected); + BOOST_TEST_EQ(req.get_commands(), 4u); + BOOST_TEST_EQ(req.get_expected_responses(), 0u); + const pubsub_change_str expected_changes[] = { + {pubsub_change_type::subscribe, ""}, + {pubsub_change_type::unsubscribe, ""}, + {pubsub_change_type::psubscribe, ""}, + {pubsub_change_type::punsubscribe, ""}, + }; + check_pubsub_changes(req, expected_changes); +} + // Mixing regular commands and pubsub commands is OK void test_mix_pubsub_regular() { @@ -645,6 +707,37 @@ void test_append_pubsub() check_pubsub_changes(req1, expected_changes); } +// Append correctly handles the argument-less unsubscribe overloads, +// which have no associated channel or pattern +void test_append_pubsub_all() +{ + request req1; + req1.subscribe({"ch1"}); + req1.unsubscribe(); + + request req2; + req2.psubscribe({"ch2*"}); + req2.punsubscribe(); + + req1.append(req2); + + constexpr std::string_view expected = + "*2\r\n$9\r\nSUBSCRIBE\r\n$3\r\nch1\r\n" + "*1\r\n$11\r\nUNSUBSCRIBE\r\n" + "*2\r\n$10\r\nPSUBSCRIBE\r\n$4\r\nch2*\r\n" + "*1\r\n$12\r\nPUNSUBSCRIBE\r\n"; + BOOST_TEST_EQ(req1.payload(), expected); + BOOST_TEST_EQ(req1.get_commands(), 4u); + BOOST_TEST_EQ(req1.get_expected_responses(), 0u); + const pubsub_change_str expected_changes[] = { + {pubsub_change_type::subscribe, "ch1" }, + {pubsub_change_type::unsubscribe_all, "" }, + {pubsub_change_type::psubscribe, "ch2*"}, + {pubsub_change_type::punsubscribe_all, "" }, + }; + check_pubsub_changes(req1, expected_changes); +} + // If the target is empty and the source has pubsub changes, that's OK void test_append_pubsub_target_empty() { @@ -724,6 +817,7 @@ int main() test_unsubscribe_iterators_convertible_string_view(); test_unsubscribe_range(); test_unsubscribe_initializer_list(); + test_unsubscribe_no_args(); test_psubscribe_iterators(); test_psubscribe_iterators_empty(); @@ -736,7 +830,9 @@ int main() test_punsubscribe_iterators_convertible_string_view(); test_punsubscribe_range(); test_punsubscribe_initializer_list(); + test_punsubscribe_no_args(); + test_pubsub_empty_channel_name(); test_mix_pubsub_regular(); test_hello(); @@ -751,6 +847,7 @@ int main() test_append_source_empty(); test_append_both_empty(); test_append_pubsub(); + test_append_pubsub_all(); test_append_pubsub_target_empty(); test_clear(); From 09acd5a94d330e8f5b4beefe334e3e131297d61f Mon Sep 17 00:00:00 2001 From: Ruben Perez Date: Wed, 29 Jul 2026 17:52:15 +0200 Subject: [PATCH 3/5] test_subscription_tracker --- test/test_subscription_tracker.cpp | 116 +++++++++++++++++++++++++++++ 1 file changed, 116 insertions(+) diff --git a/test/test_subscription_tracker.cpp b/test/test_subscription_tracker.cpp index 7f5d5aef..6b8dacd4 100644 --- a/test/test_subscription_tracker.cpp +++ b/test/test_subscription_tracker.cpp @@ -115,6 +115,91 @@ void test_unsubscribe() BOOST_TEST_EQ(req_output.payload(), req_expected.payload()); } +// An argument-less unsubscribe removes all the channels, but no patterns +void test_unsubscribe_all() +{ + subscription_tracker tracker; + request req1, req2, req_output, req_expected; + + // Add some changes to the tracker + req1.subscribe({"ch1", "ch2"}); + req1.psubscribe({"ch1*", "ch2*"}); + tracker.commit_changes(req1); + + // Unsubscribe from all channels + req2.unsubscribe(); + tracker.commit_changes(req2); + + // Only the patterns survive + tracker.compose_subscribe_request(req_output); + req_expected.push("PSUBSCRIBE", "ch1*", "ch2*"); + BOOST_TEST_EQ(req_output.payload(), req_expected.payload()); +} + +// An argument-less punsubscribe removes all the patterns, but no channels +void test_punsubscribe_all() +{ + subscription_tracker tracker; + request req1, req2, req_output, req_expected; + + // Add some changes to the tracker + req1.subscribe({"ch1", "ch2"}); + req1.psubscribe({"ch1*", "ch2*"}); + tracker.commit_changes(req1); + + // Unsubscribe from all patterns + req2.punsubscribe(); + tracker.commit_changes(req2); + + // Only the channels survive + tracker.compose_subscribe_request(req_output); + req_expected.push("SUBSCRIBE", "ch1", "ch2"); + BOOST_TEST_EQ(req_output.payload(), req_expected.payload()); +} + +// An argument-less unsubscribe on an empty state is not a problem +void test_unsubscribe_all_empty_state() +{ + subscription_tracker tracker; + request req, req_output; + + req.unsubscribe(); + req.punsubscribe(); + tracker.commit_changes(req); + + tracker.compose_subscribe_request(req_output); + BOOST_TEST_EQ(req_output.payload(), ""); +} + +// Empty channel/pattern names are tracked like any other name, +// and are not affected by the argument-less overloads being present +void test_empty_channel_name() +{ + subscription_tracker tracker; + request req1, req2, req_output, req_expected; + + // Subscribe to a channel and a pattern with an empty name + req1.subscribe({""}); + req1.psubscribe({""}); + tracker.commit_changes(req1); + + // They are restored + tracker.compose_subscribe_request(req_output); + req_expected.push("SUBSCRIBE", ""); + req_expected.push("PSUBSCRIBE", ""); + BOOST_TEST_EQ(req_output.payload(), req_expected.payload()); + + // Unsubscribing from the empty channel only affects the channel + req2.unsubscribe({""}); + tracker.commit_changes(req2); + + req_output.clear(); + req_expected.clear(); + tracker.compose_subscribe_request(req_output); + req_expected.push("PSUBSCRIBE", ""); + BOOST_TEST_EQ(req_output.payload(), req_expected.payload()); +} + // After an unsubscribe, we can subscribe again void test_resubscribe() { @@ -145,6 +230,32 @@ void test_resubscribe() BOOST_TEST_EQ(req_output.payload(), req_expected.payload()); } +// Changes are applied in order, so subscriptions added after an +// argument-less unsubscribe in the same request are preserved +void test_unsubscribe_all_then_subscribe() +{ + subscription_tracker tracker; + request req1, req2, req_output, req_expected; + + // Add some changes to the tracker + req1.subscribe({"ch1", "ch2"}); + req1.psubscribe({"ch1*", "ch2*"}); + tracker.commit_changes(req1); + + // Clear everything, then subscribe again, in a single request + req2.unsubscribe(); + req2.punsubscribe(); + req2.subscribe({"ch3"}); + req2.psubscribe({"ch3*"}); + tracker.commit_changes(req2); + + // Only the new subscriptions are present + tracker.compose_subscribe_request(req_output); + req_expected.push("SUBSCRIBE", "ch3"); + req_expected.push("PSUBSCRIBE", "ch3*"); + BOOST_TEST_EQ(req_output.payload(), req_expected.payload()); +} + // Subscribing twice is not a problem void test_subscribe_twice() { @@ -265,7 +376,12 @@ int main() test_subscribe_psubscribe(); test_subscribe_psubscribe_same_arg(); test_unsubscribe(); + test_unsubscribe_all(); + test_punsubscribe_all(); + test_unsubscribe_all_empty_state(); + test_empty_channel_name(); test_resubscribe(); + test_unsubscribe_all_then_subscribe(); test_subscribe_twice(); test_lone_unsubscribe(); test_empty(); From c13314c2cb4890cc2ecb5d7d5ebf92c026935ddb Mon Sep 17 00:00:00 2001 From: Ruben Perez Date: Wed, 29 Jul 2026 18:02:43 +0200 Subject: [PATCH 4/5] Integration test --- test/test_conn_push2.cpp | 101 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 101 insertions(+) diff --git a/test/test_conn_push2.cpp b/test/test_conn_push2.cpp index d089b208..a8ef21dd 100644 --- a/test/test_conn_push2.cpp +++ b/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 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(); } From 8afebebe5c082311d54557cc46af7120c8150403 Mon Sep 17 00:00:00 2001 From: Ruben Perez Date: Wed, 29 Jul 2026 18:06:28 +0200 Subject: [PATCH 5/5] Fill comment --- include/boost/redis/impl/subscription_tracker.ipp | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/include/boost/redis/impl/subscription_tracker.ipp b/include/boost/redis/impl/subscription_tracker.ipp index 1924159a..4165f529 100644 --- a/include/boost/redis/impl/subscription_tracker.ipp +++ b/include/boost/redis/impl/subscription_tracker.ipp @@ -15,7 +15,8 @@ namespace boost::redis::detail { -// Given a +// 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));