Merge pull request #2133 from meshtastic/master

Master downstream merge

arch/esp32/esp32.ini

@@ -26,7 +26,7 @@ build_flags =
   -DCONFIG_NIMBLE_CPP_LOG_LEVEL=2
   -DCONFIG_BT_NIMBLE_MAX_CCCDS=20
   -DESP_OPENSSL_SUPPRESS_LEGACY_WARNING
-  -DDDEBUG_HEAP
+  -DDEBUG_HEAP
 
 lib_deps =
   ${arduino_base.lib_deps}

arch/esp32/esp32s2.ini

@@ -27,7 +27,7 @@ build_flags =
   -DCONFIG_BT_NIMBLE_MAX_CCCDS=20
   -DESP_OPENSSL_SUPPRESS_LEGACY_WARNING
   -DHAS_BLUETOOTH=0
-  -DDDEBUG_HEAP
+  -DDEBUG_HEAP
   
 lib_deps =
   ${arduino_base.lib_deps}

arch/esp32/esp32s3.ini

@@ -26,7 +26,7 @@ build_flags =
   -DCONFIG_NIMBLE_CPP_LOG_LEVEL=2
   -DCONFIG_BT_NIMBLE_MAX_CCCDS=20
   -DESP_OPENSSL_SUPPRESS_LEGACY_WARNING
-  -DDDEBUG_HEAP
+  -DDEBUG_HEAP
   
 lib_deps =
   ${arduino_base.lib_deps}

src/mqtt/MQTT.cpp

@@ -21,6 +21,10 @@ String statusTopic = "msh/2/stat/";
 String cryptTopic = "msh/2/c/";   // msh/2/c/CHANNELID/NODEID
 String jsonTopic = "msh/2/json/"; // msh/2/json/CHANNELID/NODEID
 
+static MemoryDynamic<ServiceEnvelope> staticMqttPool;
+
+Allocator<ServiceEnvelope> &mqttPool = staticMqttPool;
+
 void MQTT::mqttCallback(char *topic, byte *payload, unsigned int length)
 {
     mqtt->onPublish(topic, payload, length);
@@ -142,7 +146,7 @@ void mqttInit()
     new MQTT();
 }
 
-MQTT::MQTT() : concurrency::OSThread("mqtt"), pubSub(mqttClient)
+MQTT::MQTT() : concurrency::OSThread("mqtt"), pubSub(mqttClient), mqttQueue(MAX_MQTT_QUEUE)
 {
     if(moduleConfig.mqtt.enabled) {
 
@@ -297,7 +301,7 @@ int32_t MQTT::runOnce()
                     }
                     mqttPool.release(env);
                 }
-                return 20;
+                return 200;
             } else {
                 return 30000;
             }
@@ -322,17 +326,17 @@ void MQTT::onSend(const MeshPacket &mp, ChannelIndex chIndex)
     if (ch.settings.uplink_enabled) {
         const char *channelId = channels.getGlobalId(chIndex); // FIXME, for now we just use the human name for the channel
 
-        ServiceEnvelope env = ServiceEnvelope_init_default;
+        ServiceEnvelope *env = mqttPool.allocZeroed();
-        env.channel_id = (char *)channelId;
+        env->channel_id = (char *)channelId;
-        env.gateway_id = owner.id;
+        env->gateway_id = owner.id;
-        env.packet = (MeshPacket *)&mp;
+        env->packet = (MeshPacket *)&mp;
 
         // don't bother sending if not connected...
         if (pubSub.connected()) {
 
             // FIXME - this size calculation is super sloppy, but it will go away once we dynamically alloc meshpackets
             static uint8_t bytes[MeshPacket_size + 64];
-            size_t numBytes = pb_encode_to_bytes(bytes, sizeof(bytes), &ServiceEnvelope_msg, &env);
+            size_t numBytes = pb_encode_to_bytes(bytes, sizeof(bytes), &ServiceEnvelope_msg, env);
 
             String topic = cryptTopic + channelId + "/" + owner.id;
             LOG_DEBUG("publish %s, %u bytes\n", topic.c_str(), numBytes);
@@ -348,7 +352,19 @@ void MQTT::onSend(const MeshPacket &mp, ChannelIndex chIndex)
                     pubSub.publish(topicJson.c_str(), jsonString.c_str(), false);
                 }
             }
+        } else {
+            LOG_INFO("MQTT not connected, queueing packet\n");
+            if (mqttQueue.numFree() == 0) {
+                LOG_WARN("NOTE: MQTT queue is full, discarding oldest\n");
+                ServiceEnvelope *d = mqttQueue.dequeuePtr(0);
+                if (d)
+                    mqttPool.release(d);
             }
+            // make a copy of serviceEnvelope and queue it
+            ServiceEnvelope *copied = mqttPool.allocCopy(*env);
+            assert(mqttQueue.enqueue(copied, 0));
+        }
+        mqttPool.release(env);
     }
 }
 

src/mqtt/MQTT.h

@@ -55,6 +55,7 @@ class MQTT : private concurrency::OSThread
     bool connected();
     
   protected:
+    PointerQueue<ServiceEnvelope> mqttQueue;
 
     int reconnectCount = 0;