Store clients as sessions

In preparation for clean session and qos.

Store clients as sessions
In preparation for clean session and qos.
Wiebe Cazemier
1 parent 43df4fc2
Showing 5 changed files with 64 additions and 18 deletions
CMakeLists.txt
session.cpp
session.h
subscriptionstore.cpp
subscriptionstore.h
@@ -13,6 +13,7 @@ add_executable(FlashMQ
     utils.cpp
     threaddata.cpp
     client.cpp
+    session.cpp
     bytestopacketparser.cpp
     mqttpacket.cpp
     exceptions.cpp
+#include "session.h"
+
+Session::Session()
+{
+
+}
+
+Session::Session(std::shared_ptr<Client> &client)
+{
+    this->client = client;
+}
+
+bool Session::clientDisconnected() const
+{
+    return client.expired();
+}
+
+std::shared_ptr<Client> Session::makeSharedClient() const
+{
+     return client.lock();
+}
+#ifndef SESSION_H
+#define SESSION_H
+
+#include <memory>
+
+class Client;
+
+class Session
+{
+    std::weak_ptr<Client> client;
+    // TODO: qos message queue, as some kind of movable pointer.
+public:
+    Session();
+    Session(std::shared_ptr<Client> &client);
+
+    bool clientDisconnected() const;
+    std::shared_ptr<Client> makeSharedClient() const;
+};
+
+#endif // SESSION_H
@@ -13,7 +13,7 @@ SubscriptionNode::SubscriptionNode(const std::string &amp;subtopic) :
 SubscriptionStore::SubscriptionStore() :
     root(new SubscriptionNode("root")),
-    clients_by_id_const(clients_by_id)
+    sessionsByIdConst(sessionsById)
 {
 }
@@ -65,19 +65,21 @@ void SubscriptionStore::registerClientAndKickExistingOne(Client_p &amp;client)
     if (client->getClientId().empty())
         throw ProtocolError("Trying to store client without an ID.");
-    std::weak_ptr<Client> existingClient = clients_by_id[client->getClientId()];
-    auto it = clients_by_id.find(client->getClientId());
-
-    if (it != clients_by_id.end() && !it->second.expired())
+    auto session_it = sessionsById.find(client->getClientId());
+    if (session_it != sessionsById.end())
     {
-        std::shared_ptr<Client> cl = it->second.lock();
-        logger->logf(LOG_NOTICE, "Disconnecting existing client with id '%s'", cl->getClientId().c_str());
-        cl->setReadyForDisconnect();
-        cl->getThreadData()->removeClient(cl);
-        cl->markAsDisconnecting();
-    }
+        Session &session = session_it->second;
-    clients_by_id[client->getClientId()] = client;
+        if (!session.clientDisconnected())
+        {
+            std::shared_ptr<Client> cl = session.makeSharedClient();
+            logger->logf(LOG_NOTICE, "Disconnecting existing client with id '%s'", cl->getClientId().c_str());
+            cl->setReadyForDisconnect();
+            cl->getThreadData()->removeClient(cl);
+            cl->markAsDisconnecting();
+        }
+    }
+    sessionsById[client->getClientId()] = client;
 }
 // TODO: should I implement cache, this needs to be changed to returning a list of clients.
@@ -85,12 +87,13 @@ void SubscriptionStore::publishNonRecursively(const MqttPacket &amp;packet, const st
 {
     for (const std::string &client_id : subscribers)
     {
-        auto client_it = clients_by_id_const.find(client_id);
-        if (client_it != clients_by_id_const.end())
+        auto session_it = sessionsByIdConst.find(client_id);
+        if (session_it != sessionsByIdConst.end())
         {
-            if (!client_it->second.expired())
+            const Session &session = session_it->second;
+            if (!session.clientDisconnected())
             {
-                Client_p c = client_it->second.lock();
+                Client_p c = session.makeSharedClient();
                 c->writeMqttPacketAndBlameThisClient(packet);
             }
         }
@@ -10,6 +10,7 @@
 #include "forward_declarations.h"
 #include "client.h"
+#include "session.h"
 #include "utils.h"
 #include "retainedmessage.h"
 #include "logger.h"
@@ -39,8 +40,8 @@ class SubscriptionStore
 {
     std::unique_ptr<SubscriptionNode> root;
     pthread_rwlock_t subscriptionsRwlock = PTHREAD_RWLOCK_INITIALIZER;
-    std::unordered_map<std::string, std::weak_ptr<Client>> clients_by_id;
-    const std::unordered_map<std::string, std::weak_ptr<Client>> &clients_by_id_const;
+    std::unordered_map<std::string, Session> sessionsById;
+    const std::unordered_map<std::string, Session> &sessionsByIdConst;
     pthread_rwlock_t retainedMessagesRwlock = PTHREAD_RWLOCK_INITIALIZER;
     std::unordered_set<RetainedMessage> retainedMessages;