Add configurable session expiration

Wiebe Cazemier
1 parent 436a5ffa
Showing 8 changed files with 39 additions and 20 deletions
configfileparser.cpp
mainapp.cpp
session.cpp
session.h
settings.h
subscriptionstore.cpp
subscriptionstore.h
timer.cpp
@@ -105,6 +105,7 @@ ConfigFileParser::ConfigFileParser(const std::string &amp;path) :
     validKeys.insert("mosquitto_acl_file");
     validKeys.insert("allow_anonymous");
     validKeys.insert("rlimit_nofile");
+    validKeys.insert("expire_sessions_after_seconds");
     validListenKeys.insert("port");
     validListenKeys.insert("protocol");
@@ -390,6 +391,16 @@ void ConfigFileParser::loadFile(bool test)
                     }
                     tmpSettings->rlimitNoFile = newVal;
                 }
+
+                if (key == "expire_sessions_after_seconds")
+                {
+                    int64_t newVal = std::stoi(value);
+                    if (newVal < 0 || (newVal > 0 && newVal <= 300)) // 0 means disable
+                    {
+                        throw ConfigFileException(formatString("expire_sessions_after_seconds value '%d' is invalid. Valid values are 0, or 300 or higher.", newVal));
+                    }
+                    tmpSettings->expireSessionsAfterSeconds = newVal;
+                }
             }
         }
         catch (std::invalid_argument &ex) // catch for the stoi()
@@ -179,8 +179,13 @@ MainApp::MainApp(const std::string &amp;configFilePath) :
     // TODO: override in conf possibility.
     logger->logf(LOG_NOTICE, "%d CPUs are detected, making as many threads.", num_threads);
-    auto f = std::bind(&MainApp::queueCleanup, this);
-    timer.addCallback(f, 86400000, "session expiration");
+    if (settings->expireSessionsAfterSeconds > 0)
+    {
+        auto f = std::bind(&MainApp::queueCleanup, this);
+        const uint64_t derrivedSessionCheckInterval = std::max<uint64_t>((settings->expireSessionsAfterSeconds)*1000*2, 600000);
+        const uint64_t sessionCheckInterval = std::min<uint64_t>(derrivedSessionCheckInterval, 86400000);
+        timer.addCallback(f, sessionCheckInterval, "session expiration");
+    }
     auto fKeepAlive = std::bind(&MainApp::queueKeepAliveCheckAtAllThreads, this);
     timer.addCallback(fKeepAlive, 30000, "keep-alive check");
@@ -726,7 +731,7 @@ void MainApp::queueCleanup()
 {
     std::lock_guard<std::mutex> locker(eventMutex);
-    auto f = std::bind(&SubscriptionStore::removeExpiredSessionsClients, subscriptionStore.get());
+    auto f = std::bind(&SubscriptionStore::removeExpiredSessionsClients, subscriptionStore.get(), settings->expireSessionsAfterSeconds);
     taskQueue.push_front(f);
     wakeUpThread();
@@ -164,17 +164,23 @@ uint64_t Session::sendPendingQosMessages()
     return count;
 }
-std::chrono::time_point<std::chrono::steady_clock> zeroTime;
-std::chrono::seconds expireAfter(EXPIRE_SESSION_AFTER);
-
-void Session::touch(std::chrono::time_point<std::chrono::steady_clock> val)
+/**
+ * @brief Session::touch with a time value allowed touching without causing another sys/lib call to get the time.
+ * @param newval
+ */
+void Session::touch(std::chrono::time_point<std::chrono::steady_clock> newval)
 {
-    std::chrono::time_point<std::chrono::steady_clock> newval = val > zeroTime ? val : std::chrono::steady_clock::now();
     lastTouched = newval;
 }
-bool Session::hasExpired()
+void Session::touch()
+{
+    lastTouched = std::chrono::steady_clock::now();
+}
+
+bool Session::hasExpired(int expireAfterSeconds)
 {
+    std::chrono::seconds expireAfter(expireAfterSeconds);
     std::chrono::time_point<std::chrono::steady_clock> now = std::chrono::steady_clock::now();
     return clientDisconnected() && (lastTouched + expireAfter) < now;
 }
@@ -30,9 +30,6 @@ License along with FlashMQ. If not, see &lt;https://www.gnu.org/licenses/&gt;.
 #define MAX_QOS_MSG_PENDING_PER_CLIENT 32
 #define MAX_QOS_BYTES_PENDING_PER_CLIENT 4096
-// TODO make setting
-#define EXPIRE_SESSION_AFTER 1209600
-
 struct QueuedQosPacket
 {
     uint16_t id;
@@ -67,8 +64,9 @@ public:
     void writePacket(const MqttPacket &packet, char max_qos, uint64_t &count);
     void clearQosMessage(uint16_t packet_id);
     uint64_t sendPendingQosMessages();
-    void touch(std::chrono::time_point<std::chrono::steady_clock> val = std::chrono::time_point<std::chrono::steady_clock>());
-    bool hasExpired();
+    void touch(std::chrono::time_point<std::chrono::steady_clock> val);
+    void touch();
+    bool hasExpired(int expireAfterSeconds);
     void addIncomingQoS2MessageId(uint16_t packet_id);
     bool incomingQoS2MessageIdInTransit(uint16_t packet_id) const;
@@ -46,6 +46,7 @@ public:
     std::string mosquittoAclFile;
     bool allowAnonymous = false;
     int rlimitNoFile = 1000000;
+    uint64_t expireSessionsAfterSeconds = 1209600;
     std::list<std::shared_ptr<Listener>> listeners; // Default one is created later, when none are defined.
     AuthOptCompatWrap &getAuthOptsCompat();
@@ -417,7 +417,7 @@ int SubscriptionNode::cleanSubscriptions()
 }
 // This is not MQTT compliant, but the standard doesn't keep real world constraints into account.
-void SubscriptionStore::removeExpiredSessionsClients()
+void SubscriptionStore::removeExpiredSessionsClients(int expireSessionsAfterSeconds)
 {
     RWLockGuard lock_guard(&subscriptionsRwlock);
     lock_guard.wrlock();
@@ -429,11 +429,9 @@ void SubscriptionStore::removeExpiredSessionsClients()
     {
         std::shared_ptr<Session> &session = session_it->second;
-        if (session->hasExpired())
+        if (session->hasExpired(expireSessionsAfterSeconds))
         {
-#ifndef NDEBUG
             logger->logf(LOG_DEBUG, "Removing expired session from store %s", session->getClientId().c_str());
-#endif
             session_it = sessionsById.erase(session_it);
         }
         else
@@ -99,7 +99,7 @@ public:
     void setRetainedMessage(const std::string &topic, const std::string &payload, char qos);
-    void removeExpiredSessionsClients();
+    void removeExpiredSessionsClients(int expireSessionsAfterSeconds);
 };
 #endif // SUBSCRIPTIONSTORE_H
@@ -83,7 +83,7 @@ void Timer::stop()
 void Timer::addCallback(std::function<void ()> f, uint64_t interval_ms, const std::string &name)
 {
-    logger->logf(LOG_DEBUG, "Adding event '%s' to the timer.", name.c_str());
+    logger->logf(LOG_DEBUG, "Adding event '%s' to the timer with an interval of %ld ms.", name.c_str(), interval_ms);
     CallbackEntry c;
     c.f = f;