Add support for reloading bridges without restart

Author: halfgaar

bridgeconfig.cpp

@@ -13,6 +13,11 @@ bool BridgeTopicPath::isValidQos() const
     return qos < 3;
 }
 
+bool BridgeTopicPath::operator==(const BridgeTopicPath &other) const
+{
+    return this->topic == other.topic && this->qos == other.qos;
+}
+
 
 BridgeState::BridgeState(const BridgeConfig &config) :
     c(config)
@@ -177,4 +182,23 @@ void BridgeConfig::isValid()
     setClientId();
 }
 
+bool BridgeConfig::operator ==(const BridgeConfig &other) const
+{
+    return this->address == other.address && this->port == other.port && this->inet_protocol == other.inet_protocol && this->tlsMode == other.tlsMode
+           && this->caFile == other.caFile && this->caDir == other.caDir && this->protocolVersion == other.protocolVersion
+           && this->bridgeProtocolBit == other.bridgeProtocolBit && this->keepalive == other.keepalive && this->clientidPrefix == other.clientidPrefix
+           && this->publishes == other.publishes && this->subscribes == other.subscribes && this->local_username == other.local_username
+           && this->remote_username == other.remote_username && this->remote_password == other.remote_password && this->remoteCleanStart == other.remoteCleanStart
+           && this->localCleanStart == other.localCleanStart && this->remoteSessionExpiryInterval == other.remoteSessionExpiryInterval
+           && this->localSessionExpiryInterval == other.localSessionExpiryInterval && this->remoteRetainAvailable == other.remoteRetainAvailable
+           && this->useSavedClientId == other.useSavedClientId && this->maxOutgoingTopicAliases == other.maxOutgoingTopicAliases
+           && this->maxIncomingTopicAliases == other.maxIncomingTopicAliases;
+}
+
+bool BridgeConfig::operator !=(const BridgeConfig &other) const
+{
+    bool r = *this == other;
+    return !r;
+}
+
 

bridgeconfig.h

@@ -23,6 +23,7 @@ struct BridgeTopicPath
     uint8_t qos = 0;
 
     bool isValidQos() const;
+    bool operator==(const BridgeTopicPath &other) const;
 };
 
 class BridgeConfig
@@ -55,10 +56,15 @@ class BridgeConfig
     bool remoteRetainAvailable = true;
     std::vector<BridgeTopicPath> subscribes;
     std::vector<BridgeTopicPath> publishes;
+    std::weak_ptr<ThreadData> owner;
+    bool queueForDelete = false;
 
     void setClientId(const std::string &prefix, const std::string &id);
     const std::string &getClientid() const;
     void isValid();
+
+    bool operator ==(const BridgeConfig &other) const;
+    bool operator !=(const BridgeConfig &other) const;
 };
 
 class BridgeState

bridgeinfodb.cpp

@@ -7,12 +7,14 @@ BridgeInfoForSerializing::BridgeInfoForSerializing(const std::shared_ptr<BridgeC
 
 }
 
-std::list<BridgeInfoForSerializing> BridgeInfoForSerializing::getBridgeInfosForSerializing(const std::list<std::shared_ptr<BridgeConfig>> input)
+std::list<BridgeInfoForSerializing> BridgeInfoForSerializing::getBridgeInfosForSerializing(const std::unordered_map<std::string, std::shared_ptr<BridgeConfig>> &input)
 {
     std::list<BridgeInfoForSerializing> result;
 
-    for (const std::shared_ptr<BridgeConfig> &bridge : input)
+    for (auto &pair : input)
     {
+        const std::shared_ptr<BridgeConfig> &bridge = pair.second;
+
         if (!bridge)
             continue;
 

bridgeinfodb.h

@@ -17,7 +17,7 @@ struct BridgeInfoForSerializing
     BridgeInfoForSerializing() = default;
     BridgeInfoForSerializing(const std::shared_ptr<BridgeConfig> bridge);
 
-    static std::list<BridgeInfoForSerializing> getBridgeInfosForSerializing(const std::list<std::shared_ptr<BridgeConfig>> input);
+    static std::list<BridgeInfoForSerializing> getBridgeInfosForSerializing(const std::unordered_map<std::string, std::shared_ptr<BridgeConfig>> &input);
 };
 
 class BridgeInfoDb : private PersistenceFile

dnsresolver.h

@@ -45,6 +45,9 @@ class DnsResolver
     void freeStuff();
 public:
     DnsResolver();
+    DnsResolver(const DnsResolver &other) = delete;
+    DnsResolver(DnsResolver &&other) = delete;
+    DnsResolver &operator=(const DnsResolver &other) = delete;
     ~DnsResolver();
 
     void query(const std::string &text, ListenerProtocol protocol, std::chrono::milliseconds timeout);

mainapp.cpp

@@ -276,7 +276,7 @@ void MainApp::queuePublishStatsOnDollarTopic()
  */
 void MainApp::saveStateInThread()
 {
-    std::list<BridgeInfoForSerializing> bridgeInfos = BridgeInfoForSerializing::getBridgeInfosForSerializing(this->bridges);
+    std::list<BridgeInfoForSerializing> bridgeInfos = BridgeInfoForSerializing::getBridgeInfosForSerializing(this->bridgeConfigs);
 
     auto f = std::bind(&MainApp::saveState, this->settings, bridgeInfos, true);
     this->bgWorker.addTask(f);
@@ -334,12 +334,54 @@ void MainApp::queueRetainedMessageExpiration()
     }
 }
 
-void MainApp::createBridge(std::shared_ptr<ThreadData> &thread, const std::shared_ptr<BridgeConfig> &bridgeConfig)
+void MainApp::sendBridgesToThreads()
 {
-    std::shared_ptr<BridgeState> bridgeState = std::make_shared<BridgeState>(*bridgeConfig);
+    if (threads.empty())
+        return;
+
+    int i = 0;
+    auto bridge_pos = this->bridgeConfigs.begin();
+    while (bridge_pos != this->bridgeConfigs.end())
+    {
+        auto cur = bridge_pos;
+        bridge_pos++;
+
+        std::shared_ptr<BridgeConfig> bridge = cur->second;
+
+        if (!bridge)
+            continue;
+
+        std::shared_ptr<ThreadData> owner = bridge->owner.lock();
 
-    bridgeState->threadData = thread;
-    thread->giveBridge(bridgeState);
+        if (!owner)
+        {
+            owner = threads.at(i++ % threads.size());
+            bridge->owner = owner;
+        }
+
+        if (bridge->queueForDelete)
+        {
+            owner->removeBridgeQueued(bridge, "Bridge disappeared from config");
+            this->bridgeConfigs.erase(cur);
+        }
+        else
+        {
+            std::shared_ptr<BridgeState> bridgeState = std::make_shared<BridgeState>(*bridge);
+            bridgeState->threadData = owner;
+            owner->giveBridge(bridgeState);
+        }
+    }
+}
+
+void MainApp::queueSendBridgesToThreads()
+{
+    {
+        std::lock_guard<std::mutex> locker(eventMutex);
+        auto f = std::bind(&MainApp::sendBridgesToThreads, this);
+        taskQueue.push_back(f);
+    }
+
+    wakeUpThread();
 }
 
 void MainApp::queueBridgeReconnectAllThreads(bool alsoQueueNexts)
@@ -409,12 +451,13 @@ void MainApp::saveBridgeInfo(const std::string &filePath, const std::list<Bridge
     bridgeInfoDb.saveInfo(bridgeInfos);
 }
 
-void MainApp::loadBridgeInfo()
+std::list<std::shared_ptr<BridgeConfig>> MainApp::loadBridgeInfo(Settings &settings)
 {
-    this->bridges = settings.stealBridges();
+    Logger *logger = Logger::getInstance();
+    std::list<std::shared_ptr<BridgeConfig>> bridges = settings.stealBridges();
 
     if (settings.storageDir.empty())
-        return;
+        return bridges;
 
     const std::string filePath = settings.getBridgeNamesDBFile();
 
@@ -428,7 +471,7 @@ void MainApp::loadBridgeInfo()
 
         for(const BridgeInfoForSerializing &info : bridgeInfos)
         {
-            for(std::shared_ptr<BridgeConfig> &bridgeConfig : this->bridges)
+            for(std::shared_ptr<BridgeConfig> &bridgeConfig : bridges)
             {
                 if (!bridgeConfig->useSavedClientId)
                     continue;
@@ -447,7 +490,7 @@ void MainApp::loadBridgeInfo()
         logger->logf(LOG_WARNING, "File '%s' is not there (yet)", filePath.c_str());
     }
 
-
+    return bridges;
 }
 
 void MainApp::initMainApp(int argc, char *argv[])
@@ -671,17 +714,10 @@ void MainApp::start()
 
     timer.start();
 
-    uint next_thread_index = 0;
+    sendBridgesToThreads();
+    queueBridgeReconnectAllThreads(true);
 
-    {
-        for(std::shared_ptr<BridgeConfig> &bridge : this->bridges)
-        {
-            std::shared_ptr<ThreadData> &thread = threads[next_thread_index++ % threads.size()];
-            createBridge(thread, bridge);
-        }
-
-        queueBridgeReconnectAllThreads(true);
-    }
+    uint next_thread_index = 0;
 
     this->bgWorker.start();
 
@@ -858,7 +894,7 @@ void MainApp::start()
 
     this->bgWorker.waitForStop();
 
-    std::list<BridgeInfoForSerializing> bridgeInfos = BridgeInfoForSerializing::getBridgeInfosForSerializing(this->bridges);
+    std::list<BridgeInfoForSerializing> bridgeInfos = BridgeInfoForSerializing::getBridgeInfosForSerializing(this->bridgeConfigs);
     saveState(this->settings, bridgeInfos, false);
 }
 
@@ -966,20 +1002,55 @@ void MainApp::loadConfig(bool reload)
         }
     }
 
-    // We are not reloading bridges, because it's hard to figure out what to do. This does mean we need to
-    // validate the config of existing ones, because they may have their certificate paths changed.
-    if (this->bridges.empty())
     {
-        if (!reload)
-            loadBridgeInfo();
-    }
-    else
-    {
-        // Note that this checks the certificate paths of how the briges came from the config file, not the actual bridges. But, at least
-        // for now, they must be the same, because bridges aren't changed after they are created.
-        for (auto &bridge : this->bridges)
+        for (auto &pair : bridgeConfigs)
+        {
+            pair.second->queueForDelete = true;
+        }
+
+        std::list<std::shared_ptr<BridgeConfig>> bridges = loadBridgeInfo(this->settings);
+
+        for (std::shared_ptr<BridgeConfig> &bridge : bridges)
+        {
+            if (!bridge)
+                continue;
+
+            auto pos = this->bridgeConfigs.find(bridge->clientidPrefix);
+            if (pos != this->bridgeConfigs.end())
+            {
+                logger->log(LOG_NOTICE) << "Assing new config to bridge '" << bridge->clientidPrefix << "' and reconnect if needed.";
+
+                std::shared_ptr<BridgeConfig> &cur = pos->second;
+
+                if (!cur)
+                    continue;
+
+                std::shared_ptr<ThreadData> owner = cur->owner.lock();
+                std::string clientid = cur->getClientid();
+                cur = bridge;
+                cur->owner = owner;
+                cur->setClientId(cur->clientidPrefix, clientid);
+            }
+            else
+            {
+                logger->log(LOG_NOTICE) << "Adding bridge '" << bridge->clientidPrefix << "'.";
+                this->bridgeConfigs[bridge->clientidPrefix] = bridge;
+            }
+        }
+
+        for (auto &pair : bridgeConfigs)
+        {
+            if (pair.second->queueForDelete)
+            {
+                logger->log(LOG_NOTICE) << "Queueing bridge '" << pair.first << "' for removal, because it disappeared from config.";
+            }
+        }
+
+        // On first load, the start() function will take care of it.
+        if (reload)
         {
-            bridge->isValid();
+            sendBridgesToThreads();
+            queueBridgeReconnectAllThreads(false);
         }
     }
 

mainapp.h

@@ -62,7 +62,7 @@ class MainApp
     std::list<std::shared_ptr<Listener>> listeners;
     std::unordered_map<int, ScopedSocket> activeListenSockets;
 
-    std::list<std::shared_ptr<BridgeConfig>> bridges;
+    std::unordered_map<std::string, std::shared_ptr<BridgeConfig>> bridgeConfigs;
     std::mutex quitMutex;
     std::string fuzzFilePath;
     OneInstanceLock oneInstanceLock;
@@ -87,14 +87,15 @@ class MainApp
     void queuePublishStatsOnDollarTopic();
     static void saveState(const Settings &settings, const std::list<BridgeInfoForSerializing> &bridgeInfos, bool sleep_after_limit);
     static void saveBridgeInfo(const std::string &filePath, const std::list<BridgeInfoForSerializing> &bridgeInfos);
-    void loadBridgeInfo();
+    static std::list<std::shared_ptr<BridgeConfig>> loadBridgeInfo(Settings &settings);
     void saveStateInThread();
     void queueSaveStateInThread();
     void queueSendQueuedWills();
     void waitForWillsQueued();
     void waitForDisconnectsInitiated();
     void queueRetainedMessageExpiration();
-    void createBridge(std::shared_ptr<ThreadData> &thread, const std::shared_ptr<BridgeConfig> &bridgeConfig);
+    void sendBridgesToThreads();
+    void queueSendBridgesToThreads();
     void queueBridgeReconnectAllThreads(bool alsoQueueNexts);
 
     MainApp(const std::string &configFilePath);