Untested new format of saving sessions

But, this is a safe point before I will refactor it. I will remove the appStartTime and session last touched. With the new queued removals, this is no longer necessary.

Untested new format of saving sessions
But, this is a safe point before I will refactor it. I will remove the appStartTime and session last touched. With the new queued removals, this is no longer necessary.
Wiebe Cazemier
1 parent 41ee727a
Showing 6 changed files with 40 additions and 5 deletions
mqttpacket.cpp
session.cpp
sessionsandsubscriptionsdb.cpp
settings.cpp
settings.h
subscriptionstore.cpp
@@ -343,7 +343,7 @@ void MqttPacket::handleConnect()
         uint16_t keep_alive = readTwoBytesToUInt16();
  
         uint16_t max_qos_packets = settings.maxQosMsgPendingPerClient;
-        uint32_t session_expire = settings.expireSessionsAfterSeconds > 0 ? settings.expireSessionsAfterSeconds : std::numeric_limits<uint32_t>::max();
+        uint32_t session_expire = settings.getExpireSessionAfterSeconds();
         uint32_t max_outgoing_packet_size = settings.maxPacketSize;
         uint16_t max_outgoing_topic_aliases = 0; // Default MUST BE 0, meaning server won't initiate aliases
         bool request_response_information = false;
@@ -49,7 +49,10 @@ void Session::setProgramStartedAtUnixTimestamp(const int64_t unix_timestamp)
     appStartTime = std::chrono::steady_clock::now() - age_in_s;
 }
  
-
+/**
+ * @brief Session::getSessionRelativeAgeInMs is used to get the value to store on disk when saving sessions.
+ * @return
+ */
 int64_t Session::getSessionRelativeAgeInMs() const
 {
     const std::chrono::milliseconds sessionAge = std::chrono::duration_cast<std::chrono::milliseconds>(lastTouched - appStartTime);
@@ -57,6 +60,10 @@ int64_t Session::getSessionRelativeAgeInMs() const
     return sInMs;
 }
  
+/**
+ * @brief Session::setSessionTouch is the set 'lastTouched' value relative to the app start time when a session is loaded from disk.
+ * @param ageInMs
+ */
 void Session::setSessionTouch(int64_t ageInMs)
 {
     std::chrono::milliseconds ms(ageInMs);
@@ -17,6 +17,7 @@ License along with FlashMQ. If not, see &lt;https://www.gnu.org/licenses/&gt;.
  
 #include "sessionsandsubscriptionsdb.h"
 #include "mqttpacket.h"
+#include "threadglobals.h"
  
 #include "cassert"
  
@@ -58,18 +59,20 @@ void SessionsAndSubscriptionsDB::openRead()
  
 SessionsAndSubscriptionsResult SessionsAndSubscriptionsDB::readDataV2()
 {
+    const Settings *settings = ThreadGlobals::getSettings();
+
     SessionsAndSubscriptionsResult result;
  
     while (!feof(f))
     {
         bool eofFound = false;
  
-        const int64_t programStartAge = readInt64(eofFound);
+        const int64_t programStartStamp = readInt64(eofFound);
         if (eofFound)
             continue;
  
-        logger->logf(LOG_DEBUG, "Setting first app start time to timestamp %ld", programStartAge);
-        Session::setProgramStartedAtUnixTimestamp(programStartAge);
+        logger->logf(LOG_DEBUG, "Setting first app start time to timestamp %ld", programStartStamp);
+        Session::setProgramStartedAtUnixTimestamp(programStartStamp);
  
         const uint32_t nrOfSessions = readUint32(eofFound);
  
@@ -146,6 +149,16 @@ SessionsAndSubscriptionsResult SessionsAndSubscriptionsDB::readDataV2()
             int64_t sessionAge = readInt64(eofFound);
             logger->logf(LOG_DEBUG, "Loaded session age: %ld ms.", sessionAge);
             ses->setSessionTouch(sessionAge);
+
+            const uint32_t sessionExpiryInterval = std::min<uint32_t>(readUint32(eofFound), settings->getExpireSessionAfterSeconds());
+            const uint16_t maxQosPending = std::min<uint16_t>(readUint16(eofFound), settings->maxQosMsgPendingPerClient);
+
+            // TODO: perhaps I should calculate a new sessionExpiryInterval, minus the time it was off?
+
+            // Setting the sessionExpiryInterval back to what it was is somewhat naive, in that when you have the
+            // server off for a week, you basically suspended time and will delay all session destructions. But,
+            // I'm chosing that option versus kicking out all sessions if the server was off for a longer period.
+            ses->setSessionProperties(maxQosPending, sessionExpiryInterval, 0, ProtocolVersion::Mqtt5); // The protocol version is just dummy, to get the behavior I want.
         }
  
         const uint32_t nrOfSubscriptions = readUint32(eofFound);
@@ -259,6 +272,9 @@ void SessionsAndSubscriptionsDB::saveData(const std::vector&lt;std::unique_ptr&lt;Sess
         const int64_t sInMs = ses->getSessionRelativeAgeInMs();
         logger->logf(LOG_DEBUG, "Writing session age: %ld ms.", sInMs);
         writeInt64(sInMs);
+
+        writeUint32(ses->sessionExpiryInterval);
+        writeUint16(ses->maxQosMsgPending);
     }
  
     writeUint32(subscriptions.size());
@@ -45,3 +45,12 @@ std::string Settings::getSessionsDBFile() const
     std::string path = formatString("%s/%s", storageDir.c_str(), "sessions.db");
     return path;
 }
+
+/**
+ * @brief because 0 means 'forever', we have to translate this.
+ * @return
+ */
+uint32_t Settings::getExpireSessionAfterSeconds() const
+{
+    return expireSessionsAfterSeconds > 0 ? expireSessionsAfterSeconds : std::numeric_limits<uint32_t>::max();
+}
@@ -69,6 +69,8 @@ public:
  
     std::string getRetainedMessagesDBFile() const;
     std::string getSessionsDBFile() const;
+
+    uint32_t getExpireSessionAfterSeconds() const;
 };
  
 #endif // SETTINGS_H
@@ -855,6 +855,7 @@ void SubscriptionStore::loadSessionsAndSubscriptions(const std::string &amp;filePath
         for (std::shared_ptr<Session> &session : loadedData.sessions)
         {
             sessionsById[session->getClientId()] = session;
+            queueSessionRemoval(session);
         }
  
         std::vector<std::string> subtopics;