Fix write buffer bug: wasn't increased enough

There still is a bug: writing a very big packet. I wrote the test case already.

Fix write buffer bug: wasn't increased enough
There still is a bug: writing a very big packet. I wrote the test case already.
Wiebe Cazemier
1 parent 0262ae00
Showing 5 changed files with 60 additions and 11 deletions
FlashMQTests/tst_maintests.cpp
FlashMQTests/twoclienttestcontext.cpp
FlashMQTests/twoclienttestcontext.h
client.cpp
mqttpacket.cpp
@@ -32,6 +32,9 @@ private slots:
     void test_retained_changed();
     void test_retained_removed();
+    void test_packet_bigger_than_one_doubling();
+    void test_very_big_packet();
+
 };
 MainTests::MainTests()
@@ -267,8 +270,8 @@ void MainTests::test_retained()
     QString topic = "retaintopic";
     testContext.connectSender();
-    testContext.publishRetained(topic, payload);
-    testContext.publishRetained("dummy2", "Nobody sees this");
+    testContext.publish(topic, payload, true);
+    testContext.publish("dummy2", "Nobody sees this", true);
     testContext.connectReceiver();
     testContext.subscribeReceiver("dummy");
@@ -283,7 +286,7 @@ void MainTests::test_retained()
     testContext.receivedMessages.clear();
-    testContext.publishRetained(topic, payload);
+    testContext.publish(topic, payload, true);
     testContext.waitReceiverReceived();
     QVERIFY2(testContext.receivedMessages.count() == 1, "There must be one message in the received list");
@@ -300,11 +303,11 @@ void MainTests::test_retained_changed()
     QString topic = "retaintopic";
     testContext.connectSender();
-    testContext.publishRetained(topic, payload);
+    testContext.publish(topic, payload, true);
     payload = "Changed payload";
-    testContext.publishRetained(topic, payload);
+    testContext.publish(topic, payload, true);
     testContext.connectReceiver();
     testContext.subscribeReceiver(topic);
@@ -325,11 +328,11 @@ void MainTests::test_retained_removed()
     QString topic = "retaintopic";
     testContext.connectSender();
-    testContext.publishRetained(topic, payload);
+    testContext.publish(topic, payload, true);
     payload = "";
-    testContext.publishRetained(topic, payload);
+    testContext.publish(topic, payload, true);
     testContext.connectReceiver();
     testContext.subscribeReceiver(topic);
@@ -338,6 +341,49 @@ void MainTests::test_retained_removed()
     QVERIFY2(testContext.receivedMessages.empty(), "We erased the retained message. We shouldn't have received any.");
 }
+void MainTests::test_packet_bigger_than_one_doubling()
+{
+    TwoClientTestContext testContext;
+
+    QByteArray payload(8000, 3);
+    QString topic = "hugepacket";
+
+    testContext.connectSender();
+    testContext.connectReceiver();
+    testContext.subscribeReceiver(topic);
+
+    testContext.publish(topic, payload);
+    testContext.waitReceiverReceived();
+
+    QVERIFY2(testContext.receivedMessages.count() == 1, "There must be one message in the received list");
+
+    QMQTT::Message msg = testContext.receivedMessages.first();
+    QCOMPARE(msg.payload(), payload);
+    QVERIFY(!msg.retain());
+}
+
+// This tests our write buffer, and that it's emptied during writing already.
+void MainTests::test_very_big_packet()
+{
+    TwoClientTestContext testContext;
+
+    QByteArray payload(10*1024*1024, 3);
+    QString topic = "hugepacket";
+
+    testContext.connectSender();
+    testContext.connectReceiver();
+    testContext.subscribeReceiver(topic);
+
+    testContext.publish(topic, payload);
+    testContext.waitReceiverReceived();
+
+    QVERIFY2(testContext.receivedMessages.count() == 1, "There must be one message in the received list");
+
+    QMQTT::Message msg = testContext.receivedMessages.first();
+    QCOMPARE(msg.payload(), payload);
+    QVERIFY(!msg.retain());
+}
+
 QTEST_GUILESS_MAIN(MainTests)
 #include "tst_maintests.moc"
@@ -11,11 +11,11 @@ TwoClientTestContext::TwoClientTestContext(QObject *parent) : QObject(parent)
     receiver.reset(new QMQTT::Client(targetHost));
 }
-void TwoClientTestContext::publishRetained(const QString &topic, const QByteArray &payload)
+void TwoClientTestContext::publish(const QString &topic, const QByteArray &payload, bool retain)
 {
     QMQTT::Message msg;
     msg.setTopic(topic);
-    msg.setRetain(true);
+    msg.setRetain(retain);
     msg.setQos(0);
     msg.setPayload(payload);
     sender->publish(msg);
@@ -17,7 +17,7 @@ private slots:
 public:
     explicit TwoClientTestContext(QObject *parent = nullptr);
-    void publishRetained(const QString &topic, const QByteArray &payload);
+    void publish(const QString &topic, const QByteArray &payload, bool retain = false);
     void connectSender();
     void connectReceiver();
     void disconnectReceiver();
@@ -81,7 +81,7 @@ void Client::writeMqttPacket(const MqttPacket &amp;packet)
 {
     std::lock_guard<std::mutex> locker(writeBufMutex);
-    if (packet.getSizeIncludingNonPresentHeader() > writebuf.freeSpace())
+    while (packet.getSizeIncludingNonPresentHeader() > writebuf.freeSpace())
     {
         if (packet.packetType == PacketType::PUBLISH && writebuf.getSize() >= CLIENT_MAX_BUFFER_SIZE)
             return;
@@ -99,6 +99,7 @@ void Client::writeMqttPacket(const MqttPacket &amp;packet)
         while (len_left > 0)
         {
             const size_t len = std::min<int>(len_left, writebuf.maxWriteSize());
+            assert(len > 0);
             std::memcpy(writebuf.headPtr(), &r.bytes[src_i], len);
             writebuf.advanceHead(len);
             src_i += len;
@@ -113,6 +114,7 @@ void Client::writeMqttPacket(const MqttPacket &amp;packet)
     while (len_left > 0)
     {
         const size_t len = std::min<int>(len_left, writebuf.maxWriteSize());
+        assert(len > 0);
         std::memcpy(writebuf.headPtr(), &packet.getBites()[src_i], len);
         writebuf.advanceHead(len);
         src_i += len;
@@ -22,6 +22,7 @@ MqttPacket::MqttPacket(CirBuf &amp;buf, size_t packet_len, size_t fixed_header_lengt
     while (_packet_len > 0)
     {
         int readlen = std::min<int>(buf.maxReadSize(), _packet_len);
+        assert(readlen > 0);
         std::memcpy(&bites[i], buf.tailPtr(), readlen);
         buf.advanceTail(readlen);
         i += readlen;