Publishing within the same thread works

Roughly...

Publishing within the same thread works
Roughly...
Wiebe Cazemier
1 parent a5b81d74
Showing 11 changed files with 171 additions and 8 deletions
client.cpp
client.h
main.cpp
mqttpacket.cpp
mqttpacket.h
subscriptionstore.cpp
subscriptionstore.h
threaddata.cpp
threaddata.h
types.cpp
types.h
@@ -57,7 +57,7 @@ bool Client::readFdIntoBuffer()
     return true;
 }
  
-void Client::writeMqttPacket(MqttPacket &packet)
+void Client::writeMqttPacket(const MqttPacket &packet)
 {
     if (packet.getSize() > getWriteBufMaxWriteSize())
         growWriteBuffer(packet.getSize());
@@ -115,6 +115,18 @@ std::string Client::repr()
     return a.str();
 }
  
+void Client::queueMessage(const MqttPacket &packet)
+{
+
+
+    // TODO: semaphores on stl containers?
+}
+
+void Client::queuedMessagesToBuffer()
+{
+
+}
+
 bool Client::bufferToMqttPackets(std::vector<MqttPacket> &packetQueueIn, Client_p &sender)
 {
     while (getReadBufBytesUsed() >= MQTT_HEADER_LENGH)
@@ -92,13 +92,16 @@ public:
     void setAuthenticated(bool value) { authenticated = value;}
     bool getAuthenticated() { return authenticated; }
     bool hasConnectPacketSeen() { return connectPacketSeen; }
+    ThreadData_p getThreadData() { return threadData; }
  
     void writePingResp();
-    void writeMqttPacket(MqttPacket &packet);
+    void writeMqttPacket(const MqttPacket &packet);
     bool writeBufIntoFd();
  
     std::string repr();
  
+    void queueMessage(const MqttPacket &packet);
+    void queuedMessagesToBuffer();
 };
  
 #endif // CLIENT_H
@@ -28,8 +28,22 @@ void do_thread_work(ThreadData *threadData)
  
     std::vector<MqttPacket> packetQueueIn;
  
+    uint64_t eventfd_value = 0;
+
     while (1)
     {
+        if (eventfd_value > 0)
+        {
+            for (Client_p client : threadData->getReadyForDequeueing())
+            {
+                client->queuedMessagesToBuffer();
+            }
+            threadData->clearReadyForDequeueing();
+            eventfd_value = 0;
+        }
+
+        // TODO: do all the buftofd here, not spread out over
+
         int fdcount = epoll_wait(epoll_fd, events, MAX_EVENTS, 100);
  
         if (fdcount > 0)
@@ -39,6 +53,13 @@ void do_thread_work(ThreadData *threadData)
                 struct epoll_event cur_ev = events[i];
                 int fd = cur_ev.data.fd;
  
+                // If this thread was actively woken up.
+                if (fd == threadData->event_fd)
+                {
+                    read(fd, &eventfd_value, sizeof(uint64_t));
+                    continue;
+                }
+
                 Client_p client = threadData->getClient(fd);
  
                 if (client)
@@ -3,16 +3,17 @@
 #include <iostream>
 #include <list>
  
+// constructor for parsing incoming packets
 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)
 {
-    unsigned char _packetType = (buf[0] & 0xF0) >> 4;
+    std::memcpy(&bites[0], buf, len);
+    first_byte = bites[0];
+    unsigned char _packetType = (first_byte & 0xF0) >> 4;
     packetType = (PacketType)_packetType;
     pos += fixed_header_length;
-
-    std::memcpy(&bites[0], buf, len);
 }
  
 MqttPacket::MqttPacket(const ConnAck &connAck) :
@@ -46,14 +47,42 @@ 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
 }
  
+MqttPacket::MqttPacket(const Publish &publish) :
+    bites(publish.topic.length() + publish.payload.length() + 2 + 2) // TODO: same as above
+{
+    packetType = PacketType::PUBLISH;
+    char first_byte = static_cast<char>(packetType) << 4;
+    writeByte(first_byte);
+
+    char topicLenMSB = (publish.topic.length() & 0xF0) >> 8;
+    char topicLenLSB = publish.topic.length() & 0x0F;
+    writeByte(topicLenMSB);
+    writeByte(topicLenLSB);
+    writeBytes(publish.topic.c_str(), publish.topic.length());
+
+    writeBytes(publish.payload.c_str(), publish.payload.length());
+
+    // TODO: untested. May be unnecessary, because a received packet can be resent just like that.
+}
+
 void MqttPacket::handle(std::shared_ptr<SubscriptionStore> &subscriptionStore)
 {
+    if (packetType != PacketType::CONNECT)
+    {
+        if (!sender->getAuthenticated())
+        {
+            throw ProtocolError("Non-connect packet from non-authenticated client.");
+        }
+    }
+
     if (packetType == PacketType::CONNECT)
         handleConnect();
     else if (packetType == PacketType::PINGREQ)
         sender->writePingResp();
     else if (packetType == PacketType::SUBSCRIBE)
         handleSubscribe(subscriptionStore);
+    else if (packetType == PacketType::PUBLISH)
+        handlePublish(subscriptionStore);
 }
  
 void MqttPacket::handleConnect()
@@ -123,6 +152,7 @@ void MqttPacket::handleConnect()
         // TODO: validate UTF8 encoded username/password.
  
         sender->setClientProperties(client_id, username, true, keep_alive);
+        sender->setAuthenticated(true);
  
         std::cout << "Connect: " << sender->repr() << std::endl;
  
@@ -159,6 +189,30 @@ void MqttPacket::handleSubscribe(std::shared_ptr&lt;SubscriptionStore&gt; &amp;subscriptio
     sender->writeBufIntoFd();
 }
  
+void MqttPacket::handlePublish(std::shared_ptr<SubscriptionStore> &subscriptionStore)
+{
+    uint16_t variable_header_length = readTwoBytesToUInt16();
+
+    if (variable_header_length == 0)
+        throw ProtocolError("Empty publish topic");
+
+    char qos = (first_byte & 0b00000110) >> 1;
+
+    std::string topic(readBytes(variable_header_length), variable_header_length);
+
+    if (qos)
+    {
+        throw ProtocolError("Qos not implemented.");
+        //uint16_t packet_id = readTwoBytesToUInt16();
+    }
+
+    // TODO: validate UTF8.
+    size_t payload_length = remainingAfterPos();
+    std::string payload(readBytes(payload_length), payload_length);
+
+    subscriptionStore->queueAtClientsTemp(topic, *this, sender);
+}
+
  
 Client_p MqttPacket::getSender() const
 {
@@ -197,6 +251,14 @@ void MqttPacket::writeByte(char b)
     bites[pos++] = b;
 }
  
+void MqttPacket::writeBytes(const char *b, size_t len)
+{
+    if (pos + len > bites.size())
+        throw ProtocolError("Exceeding packet size");
+
+    memcpy(&bites[pos], b, len);
+}
+
 uint16_t MqttPacket::readTwoBytesToUInt16()
 {
     if (pos + 2 > bites.size())
@@ -19,28 +19,34 @@ class MqttPacket
     std::vector<char> bites;
     size_t fixed_header_length = 0;
     Client_p sender;
+    char first_byte;
     size_t pos = 0;
     ProtocolVersion protocolVersion = ProtocolVersion::None;
  
     char *readBytes(size_t length);
     char readByte();
     void writeByte(char b);
+    void writeBytes(const char *b, size_t len);
     uint16_t readTwoBytesToUInt16();
     size_t remainingAfterPos();
  
 public:
     PacketType packetType = PacketType::Reserved;
     MqttPacket(char *buf, size_t len, size_t fixed_header_length, Client_p &sender);
+
+    // TODO: not constructors, but static functions that return all the stuff after the fixed header, then a constructor with vector.
     MqttPacket(const ConnAck &connAck);
     MqttPacket(const SubAck &subAck);
+    MqttPacket(const Publish &publish);
  
     void handle(std::shared_ptr<SubscriptionStore> &subscriptionStore);
     void handleConnect();
     void handleSubscribe(std::shared_ptr<SubscriptionStore> &subscriptionStore);
     void handlePing();
+    void handlePublish(std::shared_ptr<SubscriptionStore> &subscriptionStore);
  
-    size_t getSize() { return bites.size(); }
-    const std::vector<char> &getBites() { return bites; }
+    size_t getSize() const { return bites.size(); }
+    const std::vector<char> &getBites() const { return bites; }
  
     Client_p getSender() const;
     void setSender(const Client_p &value);
@@ -9,3 +9,20 @@ void SubscriptionStore::addSubscription(Client_p &amp;client, std::string &amp;topic)
 {
     this->subscriptions[topic].push_back(client);
 }
+
+void SubscriptionStore::queueAtClientsTemp(std::string &topic, const MqttPacket &packet, const Client_p &sender)
+{
+    for(Client_p &client : subscriptions[topic])
+    {
+        if (client->getThreadData()->threadnr == sender->getThreadData()->threadnr)
+        {
+            client->writeMqttPacket(packet);
+            client->writeBufIntoFd();
+        }
+        else
+        {
+            client->queueMessage(packet);
+            client->getThreadData()->addToReadyForDequeuing(client);
+        }
+    }
+}
@@ -18,6 +18,8 @@ public:
  
     // work with read copies intead of mutex/lock over the central store
     void getReadCopy(); // TODO
+
+    void queueAtClientsTemp(std::string &topic, const MqttPacket &packet, const Client_p &sender);
 };
  
 #endif // SUBSCRIPTIONSTORE_H
@@ -35,3 +35,21 @@ std::shared_ptr&lt;SubscriptionStore&gt; &amp;ThreadData::getSubscriptionStore()
 {
     return subscriptionStore;
 }
+
+void ThreadData::wakeUpThread()
+{
+    uint64_t one = 1;
+    write(event_fd, &one, sizeof(uint64_t));
+}
+
+void ThreadData::addToReadyForDequeuing(Client_p &client)
+{
+    this->readyForDequeueing.insert(client);
+}
+
+void ThreadData::clearReadyForDequeueing()
+{
+    this->readyForDequeueing.clear();
+}
+
+
@@ -6,6 +6,8 @@
 #include <sys/epoll.h>
 #include <sys/eventfd.h>
 #include <map>
+#include <unordered_set>
+#include <unordered_map>
  
 #include "forward_declarations.h"
  
@@ -17,8 +19,9 @@
  
 class ThreadData
 {
-    std::map<int, Client_p> clients_by_fd;
+    std::unordered_map<int, Client_p> clients_by_fd;
     std::shared_ptr<SubscriptionStore> subscriptionStore;
+    std::unordered_set<Client_p> readyForDequeueing;
  
 public:
     std::thread thread;
@@ -32,6 +35,10 @@ public:
     Client_p getClient(int fd);
     void removeClient(Client_p client);
     std::shared_ptr<SubscriptionStore> &getSubscriptionStore();
+    void wakeUpThread();
+    void addToReadyForDequeuing(Client_p &client);
+    std::unordered_set<Client_p> &getReadyForDequeueing() { return readyForDequeueing; }
+    void clearReadyForDequeueing();
 };
  
 #endif // THREADDATA_H
@@ -15,3 +15,10 @@ SubAck::SubAck(uint16_t packet_id, const std::list&lt;std::string&gt; &amp;subs) :
         responses.push_back(SubAckReturnCodes::MaxQoS0);
     }
 }
+
+Publish::Publish(std::string &topic, std::string payload) :
+    topic(topic),
+    payload(payload)
+{
+
+}
@@ -66,4 +66,12 @@ public:
     SubAck(uint16_t packet_id, const std::list<std::string> &subs);
 };
  
+class Publish
+{
+public:
+    std::string topic;
+    std::string payload;
+    Publish(std::string &topic, std::string payload);
+};
+
 #endif // TYPES_H