Retrieve store from app instance

Instead of the thread data, which didn't make sense.

Retrieve store from app instance
Instead of the thread data, which didn't make sense.
Wiebe Cazemier
1 parent b62854b1
Showing 7 changed files with 27 additions and 23 deletions
FlashMQTests/tst_maintests.cpp
client.cpp
mainapp.cpp
mainapp.h
mqttpacket.cpp
threaddata.cpp
threaddata.h
@@ -997,7 +997,7 @@ void MainTests::testSavingSessions()
     {
         std::shared_ptr<Settings> settings(new Settings());
         std::shared_ptr<SubscriptionStore> store(new SubscriptionStore());
-        std::shared_ptr<ThreadData> t(new ThreadData(0, store, settings));
+        std::shared_ptr<ThreadData> t(new ThreadData(0, settings));
         // Kind of a hack...
         Authentication auth(*settings.get());
@@ -1110,7 +1110,7 @@ void MainTests::testParsePacketHelper(const std::string &amp;topic, char from_qos, b
     std::shared_ptr<Settings> settings(new Settings());
     settings->logDebug = false;
     std::shared_ptr<SubscriptionStore> store(new SubscriptionStore());
-    std::shared_ptr<ThreadData> t(new ThreadData(0, store, settings));
+    std::shared_ptr<ThreadData> t(new ThreadData(0, settings));
     // Kind of a hack...
     Authentication auth(*settings.get());
@@ -69,7 +69,7 @@ Client::~Client()
     logger->logf(LOG_NOTICE, "Removing client '%s'. Reason(s): %s", repr().c_str(), disconnectReason.c_str());
-    std::shared_ptr<SubscriptionStore> &store = this->threadData->getSubscriptionStore();
+    std::shared_ptr<SubscriptionStore> store = MainApp::getMainApp()->getSubscriptionStore();
     if (willPublish)
     {
@@ -420,7 +420,7 @@ void Client::sendOrQueueWill()
     if (!this->willPublish)
         return;
-    std::shared_ptr<SubscriptionStore> &store = this->threadData->getSubscriptionStore();
+    std::shared_ptr<SubscriptionStore> store = MainApp::getMainApp()->getSubscriptionStore();
     store->queueWillMessage(willPublish, session);
     this->willPublish.reset();
 }
@@ -486,7 +486,7 @@ void MainApp::start()
             Authentication auth(settingsLocalCopy);
             ThreadGlobals::assign(&auth);
-            std::shared_ptr<ThreadData> threaddata = std::make_shared<ThreadData>(0, subscriptionStore, settings);
+            std::shared_ptr<ThreadData> threaddata = std::make_shared<ThreadData>(0, settings);
             std::shared_ptr<Client> client = std::make_shared<Client>(fd, threaddata, nullptr, fuzzWebsockets, nullptr, settings.get(), true);
             std::shared_ptr<Client> subscriber = std::make_shared<Client>(fdnull, threaddata, nullptr, fuzzWebsockets, nullptr, settings.get(), true);
@@ -534,7 +534,7 @@ void MainApp::start()
     for (int i = 0; i < num_threads; i++)
     {
-        std::shared_ptr<ThreadData> t = std::make_shared<ThreadData>(i, subscriptionStore, settings);
+        std::shared_ptr<ThreadData> t = std::make_shared<ThreadData>(i, settings);
         t->start(&do_thread_work);
         threads.push_back(t);
     }
@@ -785,3 +785,8 @@ void MainApp::queueCleanup()
     wakeUpThread();
 }
+std::shared_ptr<SubscriptionStore> MainApp::getSubscriptionStore()
+{
+    return this->subscriptionStore;
+}
+
@@ -110,6 +110,8 @@ public:
     void queueConfigReload();
     void queueCleanup();
+
+    std::shared_ptr<SubscriptionStore> getSubscriptionStore();
 };
 #endif // MAINAPP_H
@@ -326,7 +326,7 @@ void MqttPacket::handleConnect()
     if (sender->hasConnectPacketSeen())
         throw ProtocolError("Client already sent a CONNECT.", ReasonCodes::ProtocolError);
-    std::shared_ptr<SubscriptionStore> subscriptionStore = sender->getThreadData()->getSubscriptionStore();
+    std::shared_ptr<SubscriptionStore> subscriptionStore = MainApp::getMainApp()->getSubscriptionStore();
     sender->getThreadData()->mqttConnectCounter.inc();
@@ -775,7 +775,7 @@ void MqttPacket::handleExtendedAuth()
         if (finalResult == ReasonCodes::Success)
         {
             sender->sendConnackSuccess();
-            std::shared_ptr<SubscriptionStore> subscriptionStore = sender->getThreadData()->getSubscriptionStore();
+            std::shared_ptr<SubscriptionStore> subscriptionStore = MainApp::getMainApp()->getSubscriptionStore();
             subscriptionStore->registerClientAndKickExistingOne(sender);
         }
         else
@@ -926,7 +926,7 @@ void MqttPacket::handleSubscribe()
         if (authentication.aclCheck(sender->getClientId(), sender->getUsername(), topic, subtopics, AclAccess::subscribe, qos, false, getUserProperties()) == AuthResult::success)
         {
             logger->logf(LOG_SUBSCRIBE, "Client '%s' subscribed to '%s' QoS %d", sender->repr().c_str(), topic.c_str(), qos);
-            sender->getThreadData()->getSubscriptionStore()->addSubscription(sender, topic, subtopics, qos);
+            MainApp::getMainApp()->getSubscriptionStore()->addSubscription(sender, topic, subtopics, qos);
             subs_reponse_codes.push_back(static_cast<ReasonCodes>(qos));
         }
         else
@@ -995,7 +995,7 @@ void MqttPacket::handleUnsubscribe()
         if (topic.empty())
             throw ProtocolError("Subscribe topic is empty.", ReasonCodes::MalformedPacket);
-        sender->getThreadData()->getSubscriptionStore()->removeSubscription(sender, topic);
+        MainApp::getMainApp()->getSubscriptionStore()->removeSubscription(sender, topic);
         logger->logf(LOG_UNSUBSCRIBE, "Client '%s' unsubscribed from '%s'", sender->repr().c_str(), topic.c_str());
     }
@@ -1155,7 +1155,7 @@ void MqttPacket::handlePublish()
             if (publishData.retain)
             {
                 publishData.payload = getPayloadCopy();
-                sender->getThreadData()->getSubscriptionStore()->setRetainedMessage(publishData, publishData.subtopics);
+                MainApp::getMainApp()->getSubscriptionStore()->setRetainedMessage(publishData, publishData.subtopics);
             }
             // Set dup flag to 0, because that must not be propagated [MQTT-3.3.1-3].
@@ -1164,7 +1164,7 @@ void MqttPacket::handlePublish()
             first_byte = bites[0];
             PublishCopyFactory factory(this);
-            sender->getThreadData()->getSubscriptionStore()->queuePacketAtSubscribers(factory);
+            MainApp::getMainApp()->getSubscriptionStore()->queuePacketAtSubscribers(factory);
         }
         else
         {
@@ -28,8 +28,7 @@ KeepAliveCheck::KeepAliveCheck(const std::shared_ptr&lt;Client&gt; client) :
 }
-ThreadData::ThreadData(int threadnr, std::shared_ptr<SubscriptionStore> &subscriptionStore, std::shared_ptr<Settings> settings) :
-    subscriptionStore(subscriptionStore),
+ThreadData::ThreadData(int threadnr, std::shared_ptr<Settings> settings) :
     settingsLocalCopy(*settings.get()),
     authentication(settingsLocalCopy),
     threadnr(threadnr)
@@ -178,6 +177,8 @@ void ThreadData::publishStatsOnDollarTopic(std::vector&lt;std::shared_ptr&lt;ThreadDat
     publishStat("$SYS/broker/load/messages/sent/total", sentMessageCount);
     publishStat("$SYS/broker/load/messages/sent/persecond", sentMessageCountPerSecond);
+    std::shared_ptr<SubscriptionStore> subscriptionStore = MainApp::getMainApp()->getSubscriptionStore();
+
     publishStat("$SYS/broker/retained messages/count", subscriptionStore->getRetainedMessageCount());
     publishStat("$SYS/broker/sessions/total", subscriptionStore->getSessionCount());
@@ -190,17 +191,20 @@ void ThreadData::publishStat(const std::string &amp;topic, uint64_t n)
     const std::string payload = std::to_string(n);
     Publish p(topic, payload, 0);
     PublishCopyFactory factory(&p);
+    std::shared_ptr<SubscriptionStore> subscriptionStore = MainApp::getMainApp()->getSubscriptionStore();
     subscriptionStore->queuePacketAtSubscribers(factory, true);
     subscriptionStore->setRetainedMessage(p, factory.getSubtopics());
 }
 void ThreadData::sendQueuedWills()
 {
+    std::shared_ptr<SubscriptionStore> subscriptionStore = MainApp::getMainApp()->getSubscriptionStore();
     subscriptionStore->sendQueuedWillMessages();
 }
 void ThreadData::removeExpiredSessions()
 {
+    std::shared_ptr<SubscriptionStore> subscriptionStore = MainApp::getMainApp()->getSubscriptionStore();
     subscriptionStore->removeExpiredSessionsClients();
 }
@@ -357,11 +361,6 @@ void ThreadData::removeClient(std::shared_ptr&lt;Client&gt; client)
     clients_by_fd.erase(client->getFd());
 }
-std::shared_ptr<SubscriptionStore> &ThreadData::getSubscriptionStore()
-{
-    return subscriptionStore;
-}
-
 void ThreadData::queueDoKeepAliveCheck()
 {
     std::lock_guard<std::mutex> locker(taskQueueMutex);
@@ -28,11 +28,11 @@ License along with FlashMQ. If not, see &lt;https://www.gnu.org/licenses/&gt;.
 #include <mutex>
 #include <functional>
 #include <chrono>
+#include <forward_list>
 #include "forward_declarations.h"
 #include "client.h"
-#include "subscriptionstore.h"
 #include "utils.h"
 #include "configfileparser.h"
 #include "authplugin.h"
@@ -53,7 +53,6 @@ class ThreadData
 {
     std::unordered_map<int, std::shared_ptr<Client>> clients_by_fd;
     std::mutex clients_by_fd_mutex;
-    std::shared_ptr<SubscriptionStore> subscriptionStore;
     Logger *logger;
     std::mutex clientsToRemoveMutex;
@@ -94,7 +93,7 @@ public:
     DerivableCounter sentMessageCounter;
     DerivableCounter mqttConnectCounter;
-    ThreadData(int threadnr, std::shared_ptr<SubscriptionStore> &subscriptionStore, std::shared_ptr<Settings> settings);
+    ThreadData(int threadnr, std::shared_ptr<Settings> settings);
     ThreadData(const ThreadData &other) = delete;
     ThreadData(ThreadData &&other) = delete;
@@ -105,7 +104,6 @@ public:
     void removeClientQueued(const std::shared_ptr<Client> &client);
     void removeClientQueued(int fd);
     void removeClient(std::shared_ptr<Client> client);
-    std::shared_ptr<SubscriptionStore> &getSubscriptionStore();
     void initAuthPlugin();
     void cleanupAuthPlugin();