Merge pull request #717 from kleunen/subscription_map_clean

Many cleanups and improvements to subscription and retained map

Author: redboltz

test/retained_topic_map.hpp

@@ -83,7 +83,7 @@ class retained_topic_map {
             topic,
             [this, &parent](MQTT_NS::string_view t) {
                 if (t == "+" || t == "#") {
-                    throw std::runtime_error("No wildcards allowed in retained topic name");
+                    throw_no_wildcards_allowed();
                 }
 
                 node_id_t parent_id = parent->id;
@@ -94,7 +94,7 @@ class retained_topic_map {
                 if (entry == direct_index.end()) {
                     entry = map.insert(path_entry(parent->id, t, next_node_id++)).first;
                     if (next_node_id == max_node_id) {
-                        throw std::overflow_error("Maximum number of topics reached");
+                        throw_max_stored_topics();
                     }
                 }
                 else {
@@ -136,7 +136,7 @@ class retained_topic_map {
     // Match all underlying topics when a hash entry is matched
     // perform a breadth-first iteration over all items in the tree below
     template<typename Output>
-    void match_hash_entries(node_id_t parent, Output callback, bool ignore_system) const {
+    void match_hash_entries(node_id_t parent, Output&& callback, bool ignore_system) const {
         std::deque<node_id_t> entries;
         entries.push_back(parent);
         std::deque<node_id_t> new_entries;
@@ -170,7 +170,7 @@ class retained_topic_map {
 
     // Find all topics that match the specified topic filter
     template<typename Output>
-    void find_match(MQTT_NS::string_view topic_filter, Output callback) const {
+    void find_match(MQTT_NS::string_view topic_filter, Output&& callback) const {
         std::deque<direct_const_iterator> entries;
         entries.push_back(root);
 
@@ -212,7 +212,7 @@ class retained_topic_map {
             }
         );
 
-        for (auto entry : entries) {
+        for (auto& entry : entries) {
             if (entry->value) {
                 callback(*entry->value);
             }
@@ -247,11 +247,30 @@ class retained_topic_map {
     void increase_topics(std::vector<direct_const_iterator> const &path) {
         auto& direct_index = map.template get<direct_index_tag>();
 
-        for(auto i : path) {
+        for(auto& i : path) {
             direct_index.modify(i, [](path_entry &entry){ ++entry.count; });
         }
     }
 
+    // Exceptions used
+    static void throw_max_stored_topics() { throw std::overflow_error("Retained map maximum number of topics reached"); }
+    static void throw_no_wildcards_allowed() { throw std::runtime_error("Retained map no wildcards allowed in retained topic name"); }
+
+    // Increase the map size (total number of topics stored)
+    void increase_map_size() {
+        if(map_size == std::numeric_limits<decltype(map_size)>::max()) {
+            throw_max_stored_topics();
+        }
+
+        ++map_size;
+    }
+
+    // Decrease the map size (total number of topics stored)
+    void decrease_map_size(size_t count) {
+        BOOST_ASSERT(map_size >= count);
+        map_size -= count;
+    }
+
 public:
     retained_topic_map()
     {
@@ -268,14 +287,14 @@ class retained_topic_map {
         if (path.empty()) {
             auto new_topic = this->create_topic(topic);
             direct_index.modify(new_topic, [&value](path_entry &entry) mutable { entry.value.emplace(std::forward<V>(value)); });
-            ++map_size;
+            increase_map_size();
             return 1;
         }
 
         if (!path.back()->value) {
             this->increase_topics(path);
             direct_index.modify(path.back(), [&value](path_entry &entry) mutable { entry.value.emplace(std::forward<V>(value)); });
-            ++map_size;
+            increase_map_size();
             return 1;
         }
 
@@ -285,14 +304,15 @@ class retained_topic_map {
     }
 
     // Find all stored topics that math the specified topic_filter
-    void find(MQTT_NS::string_view topic_filter, std::function< void (Value const&) > const& callback) const {
-        find_match(topic_filter, callback);
+    template<typename Output>
+    void find(MQTT_NS::string_view topic_filter, Output&& callback) const {
+        find_match(topic_filter, std::forward<Output>(callback));
     }
 
     // Remove a stored value at the specified topic
     std::size_t erase(MQTT_NS::string_view topic) {
         auto result = erase_topic(topic);
-        map_size -= result;
+        decrease_map_size(result);
         return result;
     }
 

test/subscription_map.cpp

@@ -44,13 +44,16 @@ BOOST_AUTO_TEST_CASE( test_single_subscription ) {
     std::string text = "example/test/A";
 
     single_subscription_map< std::string > map;
-    auto handle = map.insert(text, text);
+    auto handle = map.insert(text, text).first;
     BOOST_TEST(handle.second == "A");
     BOOST_TEST(map.handle_to_topic_filter(handle) == text);
-    BOOST_CHECK_THROW(map.insert(text, text), std::exception);
+    BOOST_TEST(map.insert(text, text).second == false);
     map.update(handle, "new_value");
     map.erase(handle);
 
+    BOOST_TEST(map.insert(text, text).second == true);
+    BOOST_TEST(map.erase(text) == 1);
+
     BOOST_TEST(map.size() == 0);
     BOOST_TEST(map.internal_size() == 1);
 
@@ -112,7 +115,7 @@ BOOST_AUTO_TEST_CASE( test_single_subscription ) {
 
     std::vector< single_subscription_map< std::string >::handle > handles;
     for (auto const& i : values) {
-        handles.push_back(map.insert(i, i));
+        handles.push_back(map.insert(i, i).first);
     }
 
     for (auto const& i : handles) {
@@ -164,7 +167,6 @@ BOOST_AUTO_TEST_CASE( test_multiple_subscription ) {
     BOOST_TEST(map.size() == 0);
     BOOST_TEST(map.internal_size() == 1);
 
-
     std::vector<std::string> values = {
         "example/test/A", "example/+/A", "example/#", "#"
     };
@@ -241,6 +243,26 @@ BOOST_AUTO_TEST_CASE( test_multiple_subscription ) {
 
     BOOST_TEST(map.size() == 0);
     BOOST_TEST(map.internal_size() == 1);
+
+    // Check if $ does not match # at root
+    map = multiple_subscription_map<std::string, int>();
+
+    map.insert_or_assign("#", "123", 10);
+    map.insert_or_assign("example/plus/A", "123", 10);
+
+    matches = {};
+    map.find("example/plus/A", [&matches](std::string const &a, int /*value*/) {
+        matches.push_back(a);
+    });
+    BOOST_TEST(matches.size() == 2);
+
+    matches = {};
+    map.find("$SYS/plus/A", [&matches](std::string const &a, int /*value*/) {
+        matches.push_back(a);
+    });
+    BOOST_TEST(matches.size() == 0);
+
+ //   map.dump(std::cout);
 }
 
 BOOST_AUTO_TEST_CASE( test_multiple_subscription_modify ) {
@@ -253,6 +275,7 @@ BOOST_AUTO_TEST_CASE( test_multiple_subscription_modify ) {
         }
     };
 
+
     using mi_t = multiple_subscription_map<std::string, my>;
     mi_t map;
     map.insert_or_assign("a/b/c", "123", my());