1
0
mirror of https://github.com/adafruit/Adafruit_MQTT_Library.git synced 2025-07-21 18:22:06 +03:00

add new processSubscriptionPacket func

This commit is contained in:
brentru
2022-10-28 13:01:06 -04:00
parent 0ba089d1e8
commit 5c00e84d08
2 changed files with 31 additions and 49 deletions

View File

@ -222,6 +222,29 @@ int8_t Adafruit_MQTT::connect(const char *user, const char *pass) {
return connect();
}
void Adafruit_MQTT::processSubscriptionPacket(Adafruit_MQTT_Subscribe *sub) {
if (sub->callback_uint32t != NULL) {
// execute callback in integer mode
uint32_t data = 0;
data = atoi((char *)sub->lastread);
sub->callback_uint32t(data);
} else if (sub->callback_double != NULL) {
// execute callback in doublefloat mode
double data = 0;
data = atof((char *)sub->lastread);
sub->callback_double(data);
} else if (sub->callback_buffer != NULL) {
// execute callback in buffer mode
DEBUG_PRINTLN("processPacketsUntil called the callback_buffer!");
sub->callback_buffer((char *)sub->lastread, sub->datalen);
} else if (sub->callback_io != NULL) {
// execute callback in io mode
((sub->io_mqtt)->*(sub->callback_io))((char *)sub->lastread, sub->datalen);
}
// mark subscription message as "read""
sub->new_message = false;
}
uint16_t Adafruit_MQTT::processPacketsUntil(uint8_t *buffer,
uint8_t waitforpackettype,
uint16_t timeout) {
@ -242,31 +265,9 @@ uint16_t Adafruit_MQTT::processPacketsUntil(uint8_t *buffer,
return len;
} else {
if (packetType == MQTT_CTRL_PUBLISH) {
Adafruit_MQTT_Subscribe *sub = handleSubscriptionPacket(len);
if (sub) {
DEBUG_PRINTLN("processPacketsUntil got subscription!");
if (sub->callback_uint32t != NULL) {
// huh lets do the callback in integer mode
uint32_t data = 0;
data = atoi((char *)sub->lastread);
sub->callback_uint32t(data);
} else if (sub->callback_double != NULL) {
// huh lets do the callback in doublefloat mode
double data = 0;
data = atof((char *)sub->lastread);
sub->callback_double(data);
} else if (sub->callback_buffer != NULL) {
// huh lets do the callback in buffer mode
DEBUG_PRINTLN("processPacketsUntil called the callback_buffer!");
sub->callback_buffer((char *)sub->lastread, sub->datalen);
} else if (sub->callback_io != NULL) {
// huh lets do the callback in io mode
((sub->io_mqtt)->*(sub->callback_io))((char *)sub->lastread,
sub->datalen);
}
}
if (sub)
processSubscriptionPacket(sub);
} else {
ERROR_PRINTLN(F("Dropped a packet"));
}
@ -506,29 +507,8 @@ void Adafruit_MQTT::processPackets(int16_t timeout) {
while (elapsed < (uint32_t)timeout) {
DEBUG_PRINTLN("L480: readSubscription() called by processPackets()");
Adafruit_MQTT_Subscribe *sub = readSubscription(timeout - elapsed);
if (sub) {
DEBUG_PRINTLN("processPackets got subscription!");
if (sub->callback_uint32t != NULL) {
// huh lets do the callback in integer mode
uint32_t data = 0;
data = atoi((char *)sub->lastread);
sub->callback_uint32t(data);
} else if (sub->callback_double != NULL) {
// huh lets do the callback in doublefloat mode
double data = 0;
data = atof((char *)sub->lastread);
sub->callback_double(data);
} else if (sub->callback_buffer != NULL) {
// huh lets do the callback in buffer mode
DEBUG_PRINTLN("processPackets callback_buffer!");
sub->callback_buffer((char *)sub->lastread, sub->datalen);
} else if (sub->callback_io != NULL) {
// huh lets do the callback in io mode
((sub->io_mqtt)->*(sub->callback_io))((char *)sub->lastread,
sub->datalen);
}
}
if (sub)
processSubscriptionPacket(sub);
// keep track over elapsed time
endtime = millis();
if (endtime < starttime) {