Rudimentary lock based cross-thread publish

Wiebe Cazemier
1 parent e29e0f6d
Showing 2 changed files with 10 additions and 1 deletions
client.cpp
subscriptionstore.cpp
@@ -77,6 +77,8 @@ bool Client::readFdIntoBuffer()
 void Client::writeMqttPacket(const MqttPacket &packet)
 {
+    std::lock_guard<std::mutex> locker(writeBufMutex);
+
     if (packet.packetType == PacketType::PUBLISH && wwi > CLIENT_MAX_BUFFER_SIZE)
         return;
@@ -99,6 +101,8 @@ void Client::writeMqttPacketLocked(const MqttPacket &amp;packet)
 // Ping responses are always the same, so hardcoding it for optimization.
 void Client::writePingResp()
 {
+    std::lock_guard<std::mutex> locker(writeBufMutex);
+
     std::cout << "Sending ping response to " << repr() << std::endl;
     if (2 > getWriteBufMaxWriteSize())
@@ -112,6 +116,8 @@ void Client::writePingResp()
 bool Client::writeBufIntoFd()
 {
+    std::lock_guard<std::mutex> locker(writeBufMutex);
+
     int n;
     while ((n = write(fd, &writebuf[wri], getWriteBufBytesUsed())) != 0)
     {
@@ -28,6 +28,9 @@ void SubscriptionStore::queueAtClientsTemp(std::string &amp;topic, const MqttPacket 
     for(const Client_p &client : subscriptions[topic])
     {
+        client->writeMqttPacket(packet);
+
+        /*
         if (client->getThreadData()->threadnr == sender->getThreadData()->threadnr)
         {
             client->writeMqttPacket(packet); // TODO: with my current hack way, this is wrong. Not using a lock only works with my previous idea of queueing.
@@ -39,6 +42,6 @@ void SubscriptionStore::queueAtClientsTemp(std::string &amp;topic, const MqttPacket 
             //client->writeMqttPacketLocked(packet);
             //client->getThreadData()->addToReadyForDequeuing(client);
             //client->getThreadData()->wakeUpThread();
-        }
+        }*/
     }
 }