rbxii3/firmware
1
0
Fork 0
mirror of https://github.com/meshtastic/firmware.git synced 2025-12-20 09:43:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into develop

Browse Source
This commit is contained in:
Thomas Göttgens
2023-01-12 15:01:56 +01:00
committed by GitHub
parent f905763161 09efbb75b1
commit 42122f3d0f
5 changed files with 8 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
src/mqtt/MQTT.cpp
View File

@@ -21,10 +21,6 @@ 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);
@@ -146,7 +142,7 @@ void mqttInit()
new MQTT();
}
MQTT::MQTT() : concurrency::OSThread("mqtt"), pubSub(mqttClient), mqttQueue(MAX_MQTT_QUEUE)
MQTT::MQTT() : concurrency::OSThread("mqtt"), pubSub(mqttClient)
{
if(moduleConfig.mqtt.enabled) {
@@ -301,7 +297,7 @@ int32_t MQTT::runOnce()
}
mqttPool.release(env);
}
return 200;
return 20;
} else {
return 30000;
}
@@ -326,17 +322,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 = mqttPool.allocZeroed();
env->channel_id = (char *)channelId;
env->gateway_id = owner.id;
env->packet = (MeshPacket *)&mp;
ServiceEnvelope env = ServiceEnvelope_init_default;
env.channel_id = (char *)channelId;
env.gateway_id = owner.id;
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);
@@ -352,19 +348,7 @@ 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);
}
}