Reorder stuff for storing subscription store

Wiebe Cazemier
1 parent fec1aa82
Showing 11 changed files with 105 additions and 32 deletions
CMakeLists.txt
client.cpp
client.h
forward_declarations.h
main.cpp
mqttpacket.cpp
mqttpacket.h
subscriptionstore.cpp
subscriptionstore.h
threaddata.cpp
threaddata.h
@@ -14,6 +14,8 @@ add_executable(FlashMQ
     bytestopacketparser.cpp
     mqttpacket.cpp
     exceptions.cpp
-    types.cpp)
+    types.cpp
+    subscriptionstore.cpp
+    )
 target_link_libraries(FlashMQ pthread)
@@ -115,7 +115,7 @@ std::string Client::repr()
     return a.str();
 }
-bool Client::bufferToMqttPackets(std::vector<MqttPacket> &packetQueueIn)
+bool Client::bufferToMqttPackets(std::vector<MqttPacket> &packetQueueIn, Client_p &sender)
 {
     while (getReadBufBytesUsed() >= MQTT_HEADER_LENGH)
     {
@@ -144,7 +144,7 @@ bool Client::bufferToMqttPackets(std::vector&lt;MqttPacket&gt; &amp;packetQueueIn)
         if (packet_length <= getReadBufBytesUsed())
         {
-            MqttPacket packet(&readbuf[ri], packet_length, remaining_length_i, this);
+            MqttPacket packet(&readbuf[ri], packet_length, remaining_length_i, sender);
             packetQueueIn.push_back(std::move(packet));
             ri += packet_length;
@@ -5,6 +5,8 @@
 #include <unistd.h>
 #include <vector>
+#include "forward_declarations.h"
+
 #include "threaddata.h"
 #include "mqttpacket.h"
 #include "exceptions.h"
@@ -12,11 +14,6 @@
 #define CLIENT_BUFFER_SIZE 1024
 #define MQTT_HEADER_LENGH 2
-class ThreadData;
-typedef  std::shared_ptr<ThreadData> ThreadData_p;
-
-class MqttPacket;
-
 class Client
 {
     int fd;
@@ -90,7 +87,7 @@ public:
     int getFd() { return fd;}
     bool readFdIntoBuffer();
-    bool bufferToMqttPackets(std::vector<MqttPacket> &packetQueueIn);
+    bool bufferToMqttPackets(std::vector<MqttPacket> &packetQueueIn, Client_p &sender);
     void setClientProperties(const std::string &clientId, const std::string username, bool connectPacketSeen, uint16_t keepalive);
     void setAuthenticated(bool value) { authenticated = value;}
     bool getAuthenticated() { return authenticated; }
@@ -104,6 +101,4 @@ public:
 };
-typedef std::shared_ptr<Client> Client_p;
-
 #endif // CLIENT_H
+#ifndef FORWARD_DECLARATIONS_H
+#define FORWARD_DECLARATIONS_H
+
+#include <memory>
+
+class Client;
+typedef std::shared_ptr<Client> Client_p;
+class ThreadData;
+typedef std::shared_ptr<ThreadData> ThreadData_p;
+class MqttPacket;
+class SubscriptionStore;
+
+
+#endif // FORWARD_DECLARATIONS_H
@@ -7,10 +7,12 @@
 #include <thread>
 #include <vector>
+
 #include "utils.h"
 #include "threaddata.h"
 #include "client.h"
 #include "mqttpacket.h"
+#include "subscriptionstore.h"
 #define MAX_EVENTS 1024
 #define NR_OF_THREADS 4
@@ -50,7 +52,7 @@ void do_thread_work(ThreadData *threadData)
                         }
                         else
                         {
-                            client->bufferToMqttPackets(packetQueueIn); // TODO: different, because now I need to give the packet a raw pointer.
+                            client->bufferToMqttPackets(packetQueueIn, client);
                         }
                     }
                     if (cur_ev.events & EPOLLOUT)
@@ -64,7 +66,7 @@ void do_thread_work(ThreadData *threadData)
         for (MqttPacket &packet : packetQueueIn)
         {
-            packet.handle();
+            packet.handle(threadData->getSubscriptionStore());
         }
         packetQueueIn.clear();
     }
@@ -100,11 +102,13 @@ int main()
     ev.events = EPOLLIN;
     check<std::runtime_error>(epoll_ctl(epoll_fd_accept, EPOLL_CTL_ADD, listen_fd, &ev));
+    std::shared_ptr<SubscriptionStore> subscriptionStore(new SubscriptionStore());
+
     std::vector<std::shared_ptr<ThreadData>> threads;
     for (int i = 0; i < NR_OF_THREADS; i++)
     {
-        std::shared_ptr<ThreadData> t(new ThreadData(i));
+        std::shared_ptr<ThreadData> t(new ThreadData(i, subscriptionStore));
         std::thread thread(do_thread_work, t.get());
         t->thread = std::move(thread);
         threads.push_back(t);
@@ -3,7 +3,7 @@
 #include <iostream>
 #include <list>
-MqttPacket::MqttPacket(char *buf, size_t len, size_t fixed_header_length, Client *sender) :
+MqttPacket::MqttPacket(char *buf, size_t len, size_t fixed_header_length, Client_p &sender) :
     bites(len),
     fixed_header_length(fixed_header_length),
     sender(sender)
@@ -46,14 +46,14 @@ MqttPacket::MqttPacket(const SubAck &amp;subAck) :
     bites[1] = returnList.size() + 1; // TODO: make some generic way of calculating the header and use the multi-byte length
 }
-void MqttPacket::handle()
+void MqttPacket::handle(std::shared_ptr<SubscriptionStore> &subscriptionStore)
 {
     if (packetType == PacketType::CONNECT)
         handleConnect();
     else if (packetType == PacketType::PINGREQ)
         sender->writePingResp();
     else if (packetType == PacketType::SUBSCRIBE)
-        handleSubscribe();
+        handleSubscribe(subscriptionStore);
 }
 void MqttPacket::handleConnect()
@@ -137,7 +137,7 @@ void MqttPacket::handleConnect()
     }
 }
-void MqttPacket::handleSubscribe()
+void MqttPacket::handleSubscribe(std::shared_ptr<SubscriptionStore> &subscriptionStore)
 {
     uint16_t packet_id = readTwoBytesToUInt16();
@@ -147,6 +147,7 @@ void MqttPacket::handleSubscribe()
         uint16_t topicLength = readTwoBytesToUInt16();
         std::string topic(readBytes(topicLength), topicLength);
         std::cout << sender->repr() << " Subscribed to " << topic << std::endl;
+        subscriptionStore->addSubscription(sender, topic);
         subs.push_back(std::move(topic));
     }
@@ -159,6 +160,16 @@ void MqttPacket::handleSubscribe()
 }
+Client_p MqttPacket::getSender() const
+{
+    return sender;
+}
+
+void MqttPacket::setSender(const Client_p &value)
+{
+    sender = value;
+}
+
 char *MqttPacket::readBytes(size_t length)
 {
     if (pos + length > bites.size())
@@ -6,18 +6,19 @@
 #include <vector>
 #include <exception>
+#include "forward_declarations.h"
+
 #include "client.h"
 #include "exceptions.h"
 #include "types.h"
-
-class Client;
+#include "subscriptionstore.h"
 class MqttPacket
 {
     std::vector<char> bites;
     size_t fixed_header_length = 0;
-    Client *sender;
+    Client_p sender;
     size_t pos = 0;
     ProtocolVersion protocolVersion = ProtocolVersion::None;
@@ -29,18 +30,20 @@ class MqttPacket
 public:
     PacketType packetType = PacketType::Reserved;
-    MqttPacket(char *buf, size_t len, size_t fixed_header_length, Client *sender);
+    MqttPacket(char *buf, size_t len, size_t fixed_header_length, Client_p &sender);
     MqttPacket(const ConnAck &connAck);
     MqttPacket(const SubAck &subAck);
-    void handle();
+    void handle(std::shared_ptr<SubscriptionStore> &subscriptionStore);
     void handleConnect();
-    void handleSubscribe();
+    void handleSubscribe(std::shared_ptr<SubscriptionStore> &subscriptionStore);
     void handlePing();
     size_t getSize() { return bites.size(); }
     const std::vector<char> &getBites() { return bites; }
+    Client_p getSender() const;
+    void setSender(const Client_p &value);
 };
 #endif // MQTTPACKET_H
+#include "subscriptionstore.h"
+
+SubscriptionStore::SubscriptionStore()
+{
+
+}
+
+void SubscriptionStore::addSubscription(Client_p &client, std::string &topic)
+{
+    this->subscriptions[topic].push_back(client);
+}
+#ifndef SUBSCRIPTIONSTORE_H
+#define SUBSCRIPTIONSTORE_H
+
+#include <unordered_map>
+#include <list>
+
+#include "forward_declarations.h"
+
+#include "client.h"
+
+class SubscriptionStore
+{
+    std::unordered_map<std::string, std::list<Client_p>> subscriptions;
+public:
+    SubscriptionStore();
+
+    void addSubscription(Client_p &client, std::string &topic);
+
+    // work with read copies intead of mutex/lock over the central store
+    void getReadCopy(); // TODO
+};
+
+#endif // SUBSCRIPTIONSTORE_H
 #include "threaddata.h"
-ThreadData::ThreadData(int threadnr) :
+ThreadData::ThreadData(int threadnr, std::shared_ptr<SubscriptionStore> &subscriptionStore) :
+    subscriptionStore(subscriptionStore),
     threadnr(threadnr)
 {
     epollfd = check<std::runtime_error>(epoll_create(999));
@@ -29,3 +30,8 @@ void ThreadData::removeClient(Client_p client)
 {
     clients_by_fd.erase(client->getFd());
 }
+
+std::shared_ptr<SubscriptionStore> &ThreadData::getSubscriptionStore()
+{
+    return subscriptionStore;
+}
@@ -2,18 +2,23 @@
 #define THREADDATA_H
 #include <thread>
-#include "utils.h"
+
 #include <sys/epoll.h>
 #include <sys/eventfd.h>
-#include "client.h"
 #include <map>
-class Client;
-typedef std::shared_ptr<Client> Client_p;
+#include "forward_declarations.h"
+
+#include "client.h"
+#include "subscriptionstore.h"
+#include "utils.h"
+
+
 class ThreadData
 {
     std::map<int, Client_p> clients_by_fd;
+    std::shared_ptr<SubscriptionStore> subscriptionStore;
 public:
     std::thread thread;
@@ -21,13 +26,12 @@ public:
     int epollfd = 0;
     int event_fd = 0;
-    ThreadData(int threadnr);
+    ThreadData(int threadnr, std::shared_ptr<SubscriptionStore> &subscriptionStore);
     void giveClient(Client_p client);
     Client_p getClient(int fd);
     void removeClient(Client_p client);
+    std::shared_ptr<SubscriptionStore> &getSubscriptionStore();
 };
-typedef  std::shared_ptr<ThreadData> ThreadData_p;
-
 #endif // THREADDATA_H