Use encoded ServiceEnvelope in mqttQueue (#5619)

src/mqtt/MQTT.cpp

@@ -24,6 +24,7 @@
 #include <Throttle.h>
 #include <assert.h>
 #include <pb_decode.h>
+#include <utility>
 
 MQTT *mqtt;
 
@@ -31,10 +32,6 @@ namespace
 {
 constexpr int reconnectMax = 5;
 
-static MemoryDynamic<meshtastic_ServiceEnvelope> staticMqttPool;
-
-Allocator<meshtastic_ServiceEnvelope> &mqttPool = staticMqttPool;
-
 // FIXME - this size calculation is super sloppy, but it will go away once we dynamically alloc meshpackets
 static uint8_t bytes[meshtastic_MqttClientProxyMessage_size + 30]; // 12 for channel name and 16 for nodeid
 
@@ -528,39 +525,37 @@ void MQTT::publishNodeInfo()
 }
 void MQTT::publishQueuedMessages()
 {
-    if (!mqttQueue.isEmpty()) {
+    if (mqttQueue.isEmpty())
-        LOG_DEBUG("Publish enqueued MQTT message");
+        return;
-        meshtastic_ServiceEnvelope *env = mqttQueue.dequeuePtr(0);
-        size_t numBytes = pb_encode_to_bytes(bytes, sizeof(bytes), &meshtastic_ServiceEnvelope_msg, env);
-        std::string topic;
-        if (env->packet->pki_encrypted) {
-            topic = cryptTopic + "PKI/" + owner.id;
-        } else {
-            topic = cryptTopic + env->channel_id + "/" + owner.id;
-        }
-        LOG_INFO("publish %s, %u bytes from queue", topic.c_str(), numBytes);
 
-        publish(topic.c_str(), bytes, numBytes, false);
+    LOG_DEBUG("Publish enqueued MQTT message");
+    const std::unique_ptr<QueueEntry> entry(mqttQueue.dequeuePtr(0));
+    LOG_INFO("publish %s, %u bytes from queue", entry->topic.c_str(), entry->envBytes.size());
+    publish(entry->topic.c_str(), entry->envBytes.data(), entry->envBytes.size(), false);
 
 #if !defined(ARCH_NRF52) ||                                                                                                      \
     defined(NRF52_USE_JSON) // JSON is not supported on nRF52, see issue #2804 ### Fixed by using ArduinoJson ###
-        if (moduleConfig.mqtt.json_enabled) {
+    if (!moduleConfig.mqtt.json_enabled)
+        return;
+
     // handle json topic
-            auto jsonString = MeshPacketSerializer::JsonSerialize(env->packet);
+    const DecodedServiceEnvelope env(entry->envBytes.data(), entry->envBytes.size());
-            if (jsonString.length() != 0) {
+    if (!env.validDecode || env.packet == NULL || env.channel_id == NULL)
+        return;
+
+    auto jsonString = MeshPacketSerializer::JsonSerialize(env.packet);
+    if (jsonString.length() == 0)
+        return;
+
     std::string topicJson;
-                if (env->packet->pki_encrypted) {
+    if (env.packet->pki_encrypted) {
         topicJson = jsonTopic + "PKI/" + owner.id;
     } else {
-                    topicJson = jsonTopic + env->channel_id + "/" + owner.id;
+        topicJson = jsonTopic + env.channel_id + "/" + owner.id;
     }
     LOG_INFO("JSON publish message to %s, %u bytes: %s", topicJson.c_str(), jsonString.length(), jsonString.c_str());
     publish(topicJson.c_str(), jsonString.c_str(), false);
-            }
-        }
 #endif // ARCH_NRF52 NRF52_USE_JSON
-        mqttPool.release(env);
-    }
 }
 
 void MQTT::onSend(const meshtastic_MeshPacket &mp_encrypted, const meshtastic_MeshPacket &mp_decoded, ChannelIndex chIndex)
@@ -599,59 +594,56 @@ void MQTT::onSend(const meshtastic_MeshPacket &mp_encrypted, const meshtastic_Me
     // Either encrypted packet (we couldn't decrypt) is marked as pki_encrypted, or we could decode the PKI encrypted packet
     bool isPKIEncrypted = mp_encrypted.pki_encrypted || mp_decoded.pki_encrypted;
     // If it was to a channel, check uplink enabled, else must be pki_encrypted
-    if ((ch.settings.uplink_enabled && !isPKIEncrypted) || isPKIEncrypted) {
+    if (!(ch.settings.uplink_enabled || isPKIEncrypted))
+        return;
     const char *channelId = isPKIEncrypted ? "PKI" : channels.getGlobalId(chIndex);
 
-        meshtastic_ServiceEnvelope *env = mqttPool.allocZeroed();
-        env->channel_id = (char *)channelId;
-        env->gateway_id = owner.id;
-
     LOG_DEBUG("MQTT onSend - Publish ");
+    const meshtastic_MeshPacket *p;
     if (moduleConfig.mqtt.encryption_enabled) {
-            env->packet = (meshtastic_MeshPacket *)&mp_encrypted;
+        p = &mp_encrypted;
         LOG_DEBUG("encrypted message");
     } else if (mp_decoded.which_payload_variant == meshtastic_MeshPacket_decoded_tag) {
-            env->packet = (meshtastic_MeshPacket *)&mp_decoded;
+        p = &mp_decoded;
-            LOG_DEBUG("portnum %i message", env->packet->decoded.portnum);
+        LOG_DEBUG("portnum %i message", mp_decoded.decoded.portnum);
     } else {
         LOG_DEBUG("nothing, pkt not decrypted");
-            mqttPool.release(env);
         return; // Don't upload a still-encrypted PKI packet if not encryption_enabled
     }
 
-        if (moduleConfig.mqtt.proxy_to_client_enabled || this->isConnectedDirectly()) {
+    const meshtastic_ServiceEnvelope env = {
-            size_t numBytes = pb_encode_to_bytes(bytes, sizeof(bytes), &meshtastic_ServiceEnvelope_msg, env);
+        .packet = const_cast<meshtastic_MeshPacket *>(p), .channel_id = const_cast<char *>(channelId), .gateway_id = owner.id};
+    size_t numBytes = pb_encode_to_bytes(bytes, sizeof(bytes), &meshtastic_ServiceEnvelope_msg, &env);
     std::string topic = cryptTopic + channelId + "/" + owner.id;
-            LOG_DEBUG("MQTT Publish %s, %u bytes", topic.c_str(), numBytes);
 
+    if (moduleConfig.mqtt.proxy_to_client_enabled || this->isConnectedDirectly()) {
+        LOG_DEBUG("MQTT Publish %s, %u bytes", topic.c_str(), numBytes);
         publish(topic.c_str(), bytes, numBytes, false);
 
 #if !defined(ARCH_NRF52) ||                                                                                                      \
     defined(NRF52_USE_JSON) // JSON is not supported on nRF52, see issue #2804 ### Fixed by using ArduinoJson ###
-            if (moduleConfig.mqtt.json_enabled) {
+        if (!moduleConfig.mqtt.json_enabled)
+            return;
         // handle json topic
-                auto jsonString = MeshPacketSerializer::JsonSerialize((meshtastic_MeshPacket *)&mp_decoded);
+        auto jsonString = MeshPacketSerializer::JsonSerialize(&mp_decoded);
-                if (jsonString.length() != 0) {
+        if (jsonString.length() == 0)
+            return;
         std::string topicJson = jsonTopic + channelId + "/" + owner.id;
-                    LOG_INFO("JSON publish message to %s, %u bytes: %s", topicJson.c_str(), jsonString.length(),
+        LOG_INFO("JSON publish message to %s, %u bytes: %s", topicJson.c_str(), jsonString.length(), jsonString.c_str());
-                             jsonString.c_str());
         publish(topicJson.c_str(), jsonString.c_str(), false);
-                }
-            }
 #endif // ARCH_NRF52 NRF52_USE_JSON
     } else {
         LOG_INFO("MQTT not connected, queue packet");
+        QueueEntry *entry;
         if (mqttQueue.numFree() == 0) {
             LOG_WARN("MQTT queue is full, discard oldest");
-                meshtastic_ServiceEnvelope *d = mqttQueue.dequeuePtr(0);
+            entry = mqttQueue.dequeuePtr(0);
-                if (d)
+        } else {
-                    mqttPool.release(d);
+            entry = new QueueEntry;
         }
-            // make a copy of serviceEnvelope and queue it
+        entry->topic = std::move(topic);
-            meshtastic_ServiceEnvelope *copied = mqttPool.allocCopy(*env);
+        entry->envBytes.assign(bytes, numBytes);
-            assert(mqttQueue.enqueue(copied, 0));
+        assert(mqttQueue.enqueue(entry, 0));
-        }
-        mqttPool.release(env);
     }
 }
 
@@ -660,9 +652,9 @@ void MQTT::perhapsReportToMap()
     if (!moduleConfig.mqtt.map_reporting_enabled || !(moduleConfig.mqtt.proxy_to_client_enabled || isConnectedDirectly()))
         return;
 
-    if (Throttle::isWithinTimespanMs(last_report_to_map, map_publish_interval_msecs)) {
+    if (Throttle::isWithinTimespanMs(last_report_to_map, map_publish_interval_msecs))
         return;
-    } else {
+
     if (map_position_precision == 0 || (localPosition.latitude_i == 0 && localPosition.longitude_i == 0)) {
         last_report_to_map = millis();
         if (map_position_precision == 0)
@@ -672,11 +664,6 @@ void MQTT::perhapsReportToMap()
         return;
     }
 
-        // Allocate ServiceEnvelope and fill it
-        meshtastic_ServiceEnvelope *se = mqttPool.allocZeroed();
-        se->channel_id = (char *)channels.getGlobalId(channels.getPrimaryIndex()); // Use primary channel as the channel_id
-        se->gateway_id = owner.id;
-
     // Allocate MeshPacket and fill it
     meshtastic_MeshPacket *mp = packetPool.allocZeroed();
     mp->which_payload_variant = meshtastic_MeshPacket_decoded_tag;
@@ -710,23 +697,25 @@ void MQTT::perhapsReportToMap()
 
     mapReport.num_online_local_nodes = nodeDB->getNumOnlineMeshNodes(true);
 
-        // Encode MapReport message and set it to MeshPacket in ServiceEnvelope
+    // Encode MapReport message into the MeshPacket
-        mp->decoded.payload.size = pb_encode_to_bytes(mp->decoded.payload.bytes, sizeof(mp->decoded.payload.bytes),
+    mp->decoded.payload.size =
-                                                      &meshtastic_MapReport_msg, &mapReport);
+        pb_encode_to_bytes(mp->decoded.payload.bytes, sizeof(mp->decoded.payload.bytes), &meshtastic_MapReport_msg, &mapReport);
-        se->packet = mp;
 
-        size_t numBytes = pb_encode_to_bytes(bytes, sizeof(bytes), &meshtastic_ServiceEnvelope_msg, se);
+    // Encode the MeshPacket into a binary ServiceEnvelope and publish
+    const meshtastic_ServiceEnvelope se = {
+        .packet = mp,
+        .channel_id = (char *)channels.getGlobalId(channels.getPrimaryIndex()), // Use primary channel as the channel_id
+        .gateway_id = owner.id};
+    size_t numBytes = pb_encode_to_bytes(bytes, sizeof(bytes), &meshtastic_ServiceEnvelope_msg, &se);
 
     LOG_INFO("MQTT Publish map report to %s", mapTopic.c_str());
     publish(mapTopic.c_str(), bytes, numBytes, false);
 
-        // Release the allocated memory for ServiceEnvelope and MeshPacket
+    // Release the allocated memory for MeshPacket
-        mqttPool.release(se);
     packetPool.release(mp);
 
     // Update the last report time
     last_report_to_map = millis();
-    }
 }
 
 bool MQTT::isPrivateIpAddress(const char address[])

src/mqtt/MQTT.h

@@ -78,7 +78,11 @@ class MQTT : private concurrency::OSThread
     void start() { setIntervalFromNow(0); };
 
   protected:
-    PointerQueue<meshtastic_ServiceEnvelope> mqttQueue;
+    struct QueueEntry {
+        std::string topic;
+        std::basic_string<uint8_t> envBytes; // binary/pb_encode_to_bytes ServiceEnvelope
+    };
+    PointerQueue<QueueEntry> mqttQueue;
 
     int reconnectCount = 0;