Add support for lazy subscriptions

This allows defining bridges with broad subscription patterns that will
forward incoming matching subscriptions, instead of the whole pattern.
This allows for significant reduction in traffic by only receiving
traffic clients are interested in.

Author: halfgaar

CMakeLists.shared

@@ -92,6 +92,9 @@ set(FLASHMQ_HEADERS
     ${RELPATH}sharedmutexowned.h
     ${RELPATH}reentrantmap.h
     ${RELPATH}checkeduniqueptr.h
+    ${RELPATH}lazysubscriptions.h
+    ${RELPATH}trackedsubscriptions.h
+    ${RELPATH}trackedsubscriptionstate.h
 )
 
 set(FLASHMQ_IMPLS
@@ -159,4 +162,7 @@ set(FLASHMQ_IMPLS
     ${RELPATH}fmqssl.cpp
     ${RELPATH}persistencefunctions.cpp
     ${RELPATH}clientacceptqueue.cpp
+    ${RELPATH}lazysubscriptions.cpp
+    ${RELPATH}trackedsubscriptions.cpp
+    ${RELPATH}trackedsubscriptionstate.cpp
     )

bridgeconfig.cpp

@@ -87,7 +87,6 @@ bool BridgeTopicPath::operator==(const BridgeTopicPath &other) const
     return this->topic == other.topic && this->qos == other.qos;
 }
 
-
 BridgeState::BridgeState(const BridgeConfig &config) :
     c(config)
 {
@@ -210,6 +209,39 @@ void BridgeState::resetReconnectCounter()
     intervalLogged = 0;
 }
 
+void BridgeState::constructTrackedSubscriptions()
+{
+    if (mTrackedSubscriptions)
+        return;
+    mTrackedSubscriptions = std::make_unique<TrackedSubscriptionState>();
+}
+
+CheckedUniquePtr<TrackedSubscriptionState> &BridgeState::getTrackedSubscriptions()
+{
+    return mTrackedSubscriptions;
+}
+
+/**
+ * @brief BridgeState::stealTrackedSubscriptions
+ * @param source
+ *
+ * We only move the tracked subscriptions, not the mutations. The idea is that when a new BridgeState is being created (because
+ * of a config reload), that will already have registered itself as lazy subscriber and begun to receive subscription mutations.
+ * Those mutations will be processed as soon as the MQTT connection is approved. Before that, this stealing should take
+ * place. Processing those mutations on the tracked subscriptions we take, should be consistent enough.
+ */
+void BridgeState::stealTrackedSubscriptions(BridgeState &source)
+{
+    if (!mTrackedSubscriptions || !source.mTrackedSubscriptions)
+        return;
+
+    Logger::getInstance()->log(LOG_WARNING)
+            << "Migrating 'tracked subscriptions' from the old connection of " << source.c.clientidPrefix << "' is somewhat experimental. "
+            << "There is a chance (un)subscriptions received during the connection transition aren't correctly processed.";
+
+    this->mTrackedSubscriptions->replaceTrackedSubscriptions(source.mTrackedSubscriptions->stealTrackedSubscriptions());
+}
+
 /**
  * @brief BridgeConfig::setClientId is for setting the client ID on start to the one from a saved state. That's why it only works when the prefix matches.
  * @param prefix
@@ -254,6 +286,11 @@ void BridgeConfig::setSharedSubscriptionName(const std::string &share_name)
 {
     setSharedSubscriptionName(publishes, share_name);
     setSharedSubscriptionName(subscribes, share_name);
+
+    for (auto &l : lazySubscriptions)
+    {
+        l.share_name = share_name;
+    }
 }
 
 void BridgeConfig::setSharedSubscriptionName(std::vector<BridgeTopicPath> &topics, const std::string &share_name)
@@ -317,8 +354,8 @@ void BridgeConfig::isValid()
     if (address.empty())
         throw ConfigFileException("No address specified in bridge");
 
-    if (publishes.empty() && subscribes.empty())
-        throw ConfigFileException("No subscribe or publish paths defined in bridge.");
+    if (publishes.empty() && subscribes.empty() && lazySubscriptions.empty())
+        throw ConfigFileException("No (lazy) subscribe or publish paths defined in bridge.");
 
     if (!caDir.empty() && !caFile.empty())
         throw ConfigFileException("Specify only one 'ca_file' or 'ca_dir'");
@@ -351,6 +388,11 @@ void BridgeConfig::isValid()
         std::for_each(publishes.begin(), publishes.end(), check);
         std::for_each(subscribes.begin(), subscribes.end(), check);
     }
+
+    if (!lazySubscriptions.empty() && protocolVersion < ProtocolVersion::Mqtt5)
+    {
+        throw ConfigFileException("Using lazy subscriptions needs at least MQTT5");
+    }
 }
 
 std::vector<BridgeConfig> BridgeConfig::multiply() const
@@ -365,7 +407,11 @@ std::vector<BridgeConfig> BridgeConfig::multiply() const
     {
         result.push_back(*this);
 
-        if (this->connection_count > 1)
+        /*
+         * Always give lazy subscriptions a share name as to avoid problems when you change connection_count
+         * from 1 to > 1 and reload FlashMQ.
+         */
+        if (this->connection_count > 1 || !this->lazySubscriptions.empty())
         {
             result.back().setSharedSubscriptionName(share_name);
 
@@ -399,7 +445,7 @@ bool BridgeConfig::operator ==(const BridgeConfig &other) const
            && this->useSavedClientId == other.useSavedClientId && this->maxOutgoingTopicAliases == other.maxOutgoingTopicAliases
            && this->maxIncomingTopicAliases == other.maxIncomingTopicAliases && this->tcpNoDelay == other.tcpNoDelay
            && this->local_prefix == other.local_prefix && this->remote_prefix == other.remote_prefix && this->connection_count == other.connection_count
-           && this->maxBufferSize == other.maxBufferSize;
+           && this->maxBufferSize == other.maxBufferSize && this->lazySubscriptions == other.lazySubscriptions;
 }
 
 bool BridgeConfig::operator !=(const BridgeConfig &other) const
@@ -409,3 +455,26 @@ bool BridgeConfig::operator !=(const BridgeConfig &other) const
 }
 
 
+
+BridgeLazySubscription::BridgeLazySubscription(const std::string &pattern, uint8_t qos) :
+    pattern(pattern),
+    qos(qos)
+{
+
+}
+
+bool BridgeLazySubscription::operator==(const BridgeLazySubscription &other) const
+{
+    return this->pattern == other.pattern && this->qos == other.qos && this->share_name == other.share_name;
+}
+
+void BridgeLazySubscription::isValid() const
+{
+    if (pattern.empty() || !(isValidSubscribePath(pattern) && pattern.back() == '#'))
+        throw ConfigFileException("The pattern '" + pattern + "' is not a valid lazy subscription pattern. It must end with a multi-level wildcard.");
+
+    if (qos > 2)
+        throw ConfigFileException("QoS " + std::to_string(qos) + " is not a valid QoS for a lazy subscription");
+}
+
+

bridgeconfig.h

@@ -9,6 +9,8 @@
 #include "dnsresolver.h"
 #include "sslctxmanager.h"
 #include "utils.h"
+#include "trackedsubscriptionstate.h"
+#include "checkeduniqueptr.h"
 
 enum class BridgeTLSMode
 {
@@ -26,6 +28,22 @@ struct BridgeTopicPath
     bool operator==(const BridgeTopicPath &other) const;
 };
 
+struct BridgeLazySubscription
+{
+    std::string pattern;
+    uint8_t qos = 0;
+    std::string share_name;
+
+    BridgeLazySubscription() = default;
+    BridgeLazySubscription(const std::string &pattern, uint8_t qos);
+
+    bool operator==(const BridgeLazySubscription &other) const;
+
+    void isValid() const;
+
+};
+
+
 /**
  * @brief The BridgeClientGroupIds class manages the random IDs used in fmq_client_group_id and shared
  * subscription names for them.
@@ -52,6 +70,7 @@ class BridgeClientGroupIds
     void saveShareNames(const std::string &path) const;
 };
 
+
 class BridgeConfig
 {
     std::string clientid;
@@ -98,6 +117,8 @@ class BridgeConfig
     std::optional<std::string> local_prefix;
     std::optional<std::string> remote_prefix;
 
+    std::list<BridgeLazySubscription> lazySubscriptions;
+
     void setClientId(const std::string &prefix, const std::string &id);
     const std::string &getClientid() const;
     const std::optional<std::string> &getFmqClientGroupId() const;
@@ -109,13 +130,22 @@ class BridgeConfig
     bool operator !=(const BridgeConfig &other) const;
 };
 
+/**
+ * Lifetime rules:
+ *
+ * - Outlives bridge connects/disconnects/reconnects. It's meant to be one level above Client in that respect.
+ * - Gets recreated when the bridge config changes and you issue a config reload (SIGHUP). Some data may be
+ *   taken from the existin one for the same client id prefix.
+ */
 class BridgeState
 {
     bool sslInitialized = false;
     std::chrono::time_point<std::chrono::steady_clock> lastReconnectAttempt;
     int reconnectCounter = 0;
     const int baseReconnectInterval = (get_random_int<int>() % 30) + 30;
     int intervalLogged = 0;
+    CheckedUniquePtr<TrackedSubscriptionState> mTrackedSubscriptions;
+
 public:
     const BridgeConfig c;
     std::weak_ptr<Session> session;
@@ -132,6 +162,9 @@ class BridgeState
     bool timeForNewReconnectAttempt();
     void registerReconnect();
     void resetReconnectCounter();
+    void constructTrackedSubscriptions();
+    CheckedUniquePtr<TrackedSubscriptionState> &getTrackedSubscriptions();
+    void stealTrackedSubscriptions(BridgeState &other);
 };
 
 #endif // BRIDGECONFIG_H

configfileparser.cpp

@@ -302,6 +302,9 @@ ConfigFileParser::ConfigFileParser(const std::string &path) :
     validBridgeKeys.insert("minimum_tls_version");
     validBridgeKeys.insert("connection_count");
     validBridgeKeys.insert("max_buffer_size");
+
+    validBridgeLazySubscriptionKeys.insert("subscribe");
+    validBridgeLazySubscriptionKeys.insert("qos");
 }
 
 std::list<std::string> ConfigFileParser::readFileRecursively(const std::string &path) const
@@ -421,11 +424,12 @@ void ConfigFileParser::loadFile(bool test)
     std::stack<ConfigParseLevel> curParseLevel;
     std::shared_ptr<Listener> curListener;
     std::optional<BridgeConfig> curBridge;
+    std::optional<BridgeLazySubscription> curLazySubscription;
     std::list<BridgeConfig> preMultipliedBridges;
     Settings tmpSettings;
 
     curParseLevel.push(ConfigParseLevel::Root);
-    const std::set<std::string> blockNames {"listen", "bridge"};
+    const std::set<std::string> blockNames {"listen", "bridge", "lazy_subscription"};
 
     // Then once we know the config file is valid, process it.
     for (std::string &line : lines)
@@ -451,6 +455,14 @@ void ConfigFileParser::loadFile(bool test)
                 curParseLevel.push(ConfigParseLevel::Bridge);
                 curBridge = std::make_optional<BridgeConfig>();
             }
+            else if (testKeyValidity(key, "lazy_subscription", blockNames))
+            {
+                if (curParseLevel.top() != ConfigParseLevel::Bridge)
+                    throw ConfigFileException("Block '" + key + "' can only be defined in a bridge");
+
+                curParseLevel.push(ConfigParseLevel::BridgeLazySubscriptions);
+                curLazySubscription = std::make_optional<BridgeLazySubscription>();
+            }
             else
             {
                 std::ostringstream oss;
@@ -494,6 +506,12 @@ void ConfigFileParser::loadFile(bool test)
                 curBridge->isValid();
                 preMultipliedBridges.push_back(std::move(curBridge.value()));
             }
+            else if (curParseLevel.top() == ConfigParseLevel::BridgeLazySubscriptions)
+            {
+                curLazySubscription.value().isValid();
+                curBridge.value().lazySubscriptions.emplace_back(std::move(curLazySubscription.value()));
+                curLazySubscription.reset();
+            }
 
             curParseLevel.pop();
 
@@ -934,6 +952,26 @@ void ConfigFileParser::loadFile(bool test)
                 testCorrectNumberOfValues(key, number_of_expected_values, values);
                 continue;
             }
+            else if (curParseLevel.top() == ConfigParseLevel::BridgeLazySubscriptions)
+            {
+                if (testKeyValidity(key, "subscribe", validBridgeLazySubscriptionKeys))
+                {
+                    if (!isValidUtf8(valueTrimmed) || !isValidSubscribePath(valueTrimmed))
+                        throw ConfigFileException("Path '" + valueTrimmed + "' is not a valid subscribe pattern");
+
+                    if (!curLazySubscription.value().pattern.empty())
+                        throw ConfigFileException("Multiple 'subscribe' lines for lazy subscriptions is not valid. Use multiple 'lazy_subscription' blocks.");
+
+                    curLazySubscription.value().pattern = valueTrimmed;
+                }
+                else if (testKeyValidity(key, "qos", validBridgeLazySubscriptionKeys))
+                {
+                    curLazySubscription.value().qos = value_to_int_ranged<uint8_t>(key, valueTrimmed, 0, 2);
+                }
+
+                testCorrectNumberOfValues(key, number_of_expected_values, values);
+                continue;
+            }
 
             const std::string plugin_opt_ = "plugin_opt_";
             if (startsWith(key, plugin_opt_))

configfileparser.h

@@ -26,7 +26,8 @@ enum class ConfigParseLevel
 {
     Root,
     Listen,
-    Bridge
+    Bridge,
+    BridgeLazySubscriptions
 };
 
 template<typename T>
@@ -101,6 +102,7 @@ class ConfigFileParser
     std::set<std::string> validKeys;
     std::set<std::string> validListenKeys;
     std::set<std::string> validBridgeKeys;
+    std::set<std::string> validBridgeLazySubscriptionKeys;
 
     const std::regex key_value_regex = std::regex("^([\\w\\-]+)\\s+(.+)$");
     const std::regex block_regex_start = std::regex("^([a-zA-Z0-9_\\-]+) *\\{$");

globals.h

@@ -3,8 +3,10 @@
 
 #include <memory>
 #include <pthread.h>
+#include <optional>
 
 #include "subscriptionstore.h"
+#include "lazysubscriptions.h"
 #include "globalstats.h"
 #include "bridgeconfig.h"
 #include "checkedsharedptr.h"
@@ -26,6 +28,7 @@ class Globals
         GlobalStats stats;
         BridgeClientGroupIds bridgeClientGroupIds;
         MutexOwned<std::vector<std::shared_ptr<ThreadData>>> threadDatas;
+        std::optional<LazySubscriptions> lazySubscriptions;
 
         CheckedSharedPtr<ThreadData> getDeterministicThreadData();
     };