MQTT client enhanced (experimental)

This commit is contained in:
Klaus K Wilting 2020-06-16 20:29:49 +02:00
parent 55315b7301
commit d555656852
5 changed files with 72 additions and 66 deletions

View File

@ -11,6 +11,7 @@
#define BME_IRQ 0x080 #define BME_IRQ 0x080
#define MATRIX_DISPLAY_IRQ 0x100 #define MATRIX_DISPLAY_IRQ 0x100
#define PMU_IRQ 0x200 #define PMU_IRQ 0x200
#define MQTT_IRQ 0x400
#include "globals.h" #include "globals.h"
#include "button.h" #include "button.h"
@ -20,10 +21,12 @@
#include "bmesensor.h" #include "bmesensor.h"
#include "power.h" #include "power.h"
#include "ledmatrixdisplay.h" #include "ledmatrixdisplay.h"
#include "mqttclient.h"
void irqHandler(void *pvParameters); void irqHandler(void *pvParameters);
void mask_user_IRQ(); void mask_user_IRQ();
void unmask_user_IRQ(); void unmask_user_IRQ();
void doIRQ(int irq);
#ifdef HAS_DISPLAY #ifdef HAS_DISPLAY
void IRAM_ATTR DisplayIRQ(); void IRAM_ATTR DisplayIRQ();
@ -41,5 +44,4 @@ void IRAM_ATTR ButtonIRQ();
void IRAM_ATTR PMUIRQ(); void IRAM_ATTR PMUIRQ();
#endif #endif
#endif #endif

View File

@ -6,17 +6,20 @@
#include <ETH.h> #include <ETH.h>
#include <PubSubClient.h> #include <PubSubClient.h>
#define MQTT_INTOPIC "paxcounter_in/" #define MQTT_INTOPIC "paxcounter/in"
#define MQTT_OUTTOPIC "paxcounter_out/" #define MQTT_OUTTOPIC "paxcounter/out"
#define MQTT_PORT 1883 #define MQTT_PORT 1883
#define MQTT_SERVER "broker.hivemq.com" #define MQTT_SERVER "broker.hivemq.com"
#define MQTT_RETRYSEC 10 // retry reconnect every 10 seconds //#define MQTT_SERVER "test.mosquitto.org"
#define MQTT_RETRYSEC 20 // retry reconnect every 20 seconds
#define MQTT_KEEPALIVE 10 // seconds to keep alive server connections
extern TaskHandle_t mqttTask; extern TaskHandle_t mqttTask;
extern PubSubClient mqttClient;
void mqtt_enqueuedata(MessageBuffer_t *message); void mqtt_enqueuedata(MessageBuffer_t *message);
void mqtt_queuereset(void); void mqtt_queuereset(void);
void mqtt_irq(void);
void mqtt_loop(void);
void mqtt_client_task(void *param); void mqtt_client_task(void *param);
int mqtt_connect(const char *my_host, const uint16_t my_port); int mqtt_connect(const char *my_host, const uint16_t my_port);
void mqtt_callback(char *topic, byte *payload, unsigned int length); void mqtt_callback(char *topic, byte *payload, unsigned int length);

View File

@ -55,7 +55,6 @@ void doHousekeeping() {
#ifdef HAS_MQTT #ifdef HAS_MQTT
ESP_LOGD(TAG, "MQTTloop %d bytes left | Taskstate = %d", ESP_LOGD(TAG, "MQTTloop %d bytes left | Taskstate = %d",
uxTaskGetStackHighWaterMark(mqttTask), eTaskGetState(mqttTask)); uxTaskGetStackHighWaterMark(mqttTask), eTaskGetState(mqttTask));
mqttClient.loop();
#endif #endif
#if (defined HAS_DCF77 || defined HAS_IF482) #if (defined HAS_DCF77 || defined HAS_IF482)
@ -152,7 +151,6 @@ uint32_t getFreeRAM() {
void reset_counters() { void reset_counters() {
#if ((WIFICOUNTER) || (BLECOUNTER)) #if ((WIFICOUNTER) || (BLECOUNTER))
macs.clear(); // clear all macs container macs.clear(); // clear all macs container
macs_total = 0; // reset all counters
macs_wifi = 0; macs_wifi = 0;
macs_ble = 0; macs_ble = 0;
#ifdef HAS_DISPLAY #ifdef HAS_DISPLAY

View File

@ -69,6 +69,14 @@ void irqHandler(void *pvParameters) {
} }
#endif #endif
// MQTT loop due?
#if (HAS_MQTT)
if (InterruptStatus & MQTT_IRQ) {
mqtt_loop();
InterruptStatus &= ~MQTT_IRQ;
}
#endif
// are cyclic tasks due? // are cyclic tasks due?
if (InterruptStatus & CYCLIC_IRQ) { if (InterruptStatus & CYCLIC_IRQ) {
doHousekeeping(); doHousekeeping();
@ -91,53 +99,30 @@ void irqHandler(void *pvParameters) {
} // for } // for
} // irqHandler() } // irqHandler()
// esp32 hardware timer triggered interrupt service routines // timer triggered interrupt service routines
// they notify the irq handler task // they notify the irq handler task
#ifdef HAS_DISPLAY void IRAM_ATTR doIRQ(int irq) {
void IRAM_ATTR DisplayIRQ() {
BaseType_t xHigherPriorityTaskWoken = pdFALSE; BaseType_t xHigherPriorityTaskWoken = pdFALSE;
xTaskNotifyFromISR(irqHandlerTask, irq, eSetBits, &xHigherPriorityTaskWoken);
xTaskNotifyFromISR(irqHandlerTask, DISPLAY_IRQ, eSetBits,
&xHigherPriorityTaskWoken);
if (xHigherPriorityTaskWoken) if (xHigherPriorityTaskWoken)
portYIELD_FROM_ISR(); portYIELD_FROM_ISR();
} }
#ifdef HAS_DISPLAY
void IRAM_ATTR DisplayIRQ() { doIRQ(DISPLAY_IRQ); }
#endif #endif
#ifdef HAS_MATRIX_DISPLAY #ifdef HAS_MATRIX_DISPLAY
void IRAM_ATTR MatrixDisplayIRQ() { void IRAM_ATTR MatrixDisplayIRQ() { doIRQ(MATRIX_DISPLAY_IRQ); }
BaseType_t xHigherPriorityTaskWoken = pdFALSE;
xTaskNotifyFromISR(irqHandlerTask, MATRIX_DISPLAY_IRQ, eSetBits,
&xHigherPriorityTaskWoken);
if (xHigherPriorityTaskWoken)
portYIELD_FROM_ISR();
}
#endif #endif
#ifdef HAS_BUTTON #ifdef HAS_BUTTON
void IRAM_ATTR ButtonIRQ() { void IRAM_ATTR ButtonIRQ() { doIRQ(BUTTON_IRQ); }
BaseType_t xHigherPriorityTaskWoken = pdFALSE;
xTaskNotifyFromISR(irqHandlerTask, BUTTON_IRQ, eSetBits,
&xHigherPriorityTaskWoken);
if (xHigherPriorityTaskWoken)
portYIELD_FROM_ISR();
}
#endif #endif
#ifdef HAS_PMU #ifdef HAS_PMU
void IRAM_ATTR PMUIRQ() { void IRAM_ATTR PMUIRQ() { doIRQ(PMU_IRQ); }
BaseType_t xHigherPriorityTaskWoken = pdFALSE;
xTaskNotifyFromISR(irqHandlerTask, PMU_IRQ, eSetBits,
&xHigherPriorityTaskWoken);
if (xHigherPriorityTaskWoken)
portYIELD_FROM_ISR();
}
#endif #endif
void mask_user_IRQ() { xTaskNotify(irqHandlerTask, MASK_IRQ, eSetBits); } void mask_user_IRQ() { xTaskNotify(irqHandlerTask, MASK_IRQ, eSetBits); }

View File

@ -7,8 +7,10 @@ static const char TAG[] = __FILE__;
QueueHandle_t MQTTSendQueue; QueueHandle_t MQTTSendQueue;
TaskHandle_t mqttTask; TaskHandle_t mqttTask;
WiFiClient EthClient; Ticker mqttTimer;
PubSubClient mqttClient(EthClient);
WiFiClient NetClient;
PubSubClient mqttClient(NetClient);
void NetworkEvent(WiFiEvent_t event) { void NetworkEvent(WiFiEvent_t event) {
switch (event) { switch (event) {
@ -16,22 +18,26 @@ void NetworkEvent(WiFiEvent_t event) {
ESP_LOGI(TAG, "Ethernet link layer started"); ESP_LOGI(TAG, "Ethernet link layer started");
ETH.setHostname(ETH.macAddress().c_str()); ETH.setHostname(ETH.macAddress().c_str());
break; break;
case SYSTEM_EVENT_ETH_STOP:
ESP_LOGI(TAG, "Ethernet link layer stopped");
break;
case SYSTEM_EVENT_ETH_CONNECTED: case SYSTEM_EVENT_ETH_CONNECTED:
case SYSTEM_EVENT_STA_CONNECTED:
ESP_LOGI(TAG, "Network link connected"); ESP_LOGI(TAG, "Network link connected");
break; break;
case SYSTEM_EVENT_ETH_DISCONNECTED:
case SYSTEM_EVENT_STA_DISCONNECTED:
ESP_LOGI(TAG, "Network link disconnected");
break;
case SYSTEM_EVENT_ETH_GOT_IP: case SYSTEM_EVENT_ETH_GOT_IP:
ESP_LOGI(TAG, "ETH MAC: %s", ETH.macAddress().c_str()); ESP_LOGI(TAG, "ETH MAC: %s", ETH.macAddress().c_str());
ESP_LOGI(TAG, "IPv4: %s", ETH.localIP().toString().c_str()); ESP_LOGI(TAG, "IPv4: %s", ETH.localIP().toString().c_str());
ESP_LOGI(TAG, "Link Speed: %d Mbps %s", ETH.linkSpeed(), ESP_LOGI(TAG, "Link Speed: %d Mbps %s", ETH.linkSpeed(),
ETH.fullDuplex() ? "full duplex" : "half duplex"); ETH.fullDuplex() ? "full duplex" : "half duplex");
case SYSTEM_EVENT_STA_GOT_IP:
mqtt_connect(MQTT_SERVER, MQTT_PORT); mqtt_connect(MQTT_SERVER, MQTT_PORT);
break; break;
case SYSTEM_EVENT_ETH_DISCONNECTED:
ESP_LOGI(TAG, "Network link disconnected");
break;
case SYSTEM_EVENT_ETH_STOP:
ESP_LOGI(TAG, "Ethernet link layer stopped");
break;
default: default:
break; break;
} }
@ -40,10 +46,12 @@ void NetworkEvent(WiFiEvent_t event) {
int mqtt_connect(const char *my_host, const uint16_t my_port) { int mqtt_connect(const char *my_host, const uint16_t my_port) {
IPAddress mqtt_server_ip; IPAddress mqtt_server_ip;
static String clientId = "paxcounter-" + ETH.macAddress(); // static String clientId = "paxcounter-" + ETH.macAddress();
static String clientId = "paxcounter-" + String(random(0xffff), HEX);
ESP_LOGI(TAG, "MQTT name is %s", clientId.c_str()); ESP_LOGI(TAG, "MQTT name is %s", clientId.c_str());
// resolve server // resolve server host name
if (WiFi.hostByName(my_host, mqtt_server_ip)) { if (WiFi.hostByName(my_host, mqtt_server_ip)) {
ESP_LOGI(TAG, "Attempting to connect to %s [%s]", my_host, ESP_LOGI(TAG, "Attempting to connect to %s [%s]", my_host,
mqtt_server_ip.toString().c_str()); mqtt_server_ip.toString().c_str());
@ -53,9 +61,10 @@ int mqtt_connect(const char *my_host, const uint16_t my_port) {
} }
// attempt to connect to MQTT server // attempt to connect to MQTT server
if (EthClient.connect(mqtt_server_ip, my_port, HOMECYCLE * 2 * 1000)) { if (NetClient.connect(mqtt_server_ip, my_port, HOMECYCLE * 2 * 1000)) {
NetClient.setTimeout(MQTT_KEEPALIVE); // seconds
mqttClient.setServer(mqtt_server_ip, my_port); mqttClient.setServer(mqtt_server_ip, my_port);
mqttClient.setKeepAlive(HOMECYCLE * 2); mqttClient.setKeepAlive(MQTT_KEEPALIVE);
mqttClient.setCallback(mqtt_callback); mqttClient.setCallback(mqtt_callback);
if (mqttClient.connect(clientId.c_str())) { if (mqttClient.connect(clientId.c_str())) {
@ -92,23 +101,30 @@ void mqtt_client_task(void *param) {
mqttClient.write('/'); mqttClient.write('/');
mqttClient.write(msg.Message, msg.MessageSize); mqttClient.write(msg.Message, msg.MessageSize);
if (mqttClient.endPublish()) { if (mqttClient.endPublish()) {
ESP_LOGI(TAG, "%d byte(s) sent to MQTT", msg.MessageSize + 2); ESP_LOGI(TAG, "%d byte(s) sent to MQTT server", msg.MessageSize + 2);
continue; // while(1) continue;
} else } else {
goto reconnect; mqtt_enqueuedata(&msg); // postpone the undelivered message
ESP_LOGD(TAG,
"Couldn't sent message to MQTT server, message postponed");
}
} else { } else {
// attempt to reconnect to MQTT server // attempt to reconnect to MQTT server
reconnect: ESP_LOGD(TAG, "MQTT client reconnecting...");
mqtt_enqueuedata(&msg); // postpone the undelivered message mqtt_enqueuedata(&msg); // postpone the undelivered message
delay(MQTT_RETRYSEC * 1000); delay(MQTT_RETRYSEC * 1000);
mqtt_connect(MQTT_SERVER, MQTT_PORT); mqtt_connect(MQTT_SERVER, MQTT_PORT);
} }
} // while(1) } // while(1)
} }
esp_err_t mqtt_init(void) { esp_err_t mqtt_init(void) {
WiFi.onEvent(NetworkEvent);
ETH.begin();
assert(SEND_QUEUE_SIZE); assert(SEND_QUEUE_SIZE);
MQTTSendQueue = xQueueCreate(SEND_QUEUE_SIZE, sizeof(MessageBuffer_t)); MQTTSendQueue = xQueueCreate(SEND_QUEUE_SIZE, sizeof(MessageBuffer_t));
if (MQTTSendQueue == 0) { if (MQTTSendQueue == 0) {
@ -119,10 +135,8 @@ esp_err_t mqtt_init(void) {
SEND_QUEUE_SIZE * PAYLOAD_BUFFER_SIZE); SEND_QUEUE_SIZE * PAYLOAD_BUFFER_SIZE);
ESP_LOGI(TAG, "Starting MQTTloop..."); ESP_LOGI(TAG, "Starting MQTTloop...");
xTaskCreate(mqtt_client_task, "mqttloop", 4096, (void *)NULL, 2, &mqttTask); mqttTimer.attach(MQTT_KEEPALIVE, mqtt_irq);
xTaskCreate(mqtt_client_task, "mqttloop", 4096, (void *)NULL, 1, &mqttTask);
WiFi.onEvent(NetworkEvent);
ETH.begin();
return ESP_OK; return ESP_OK;
} }
@ -151,13 +165,17 @@ void mqtt_enqueuedata(MessageBuffer_t *message) {
} }
void mqtt_callback(char *topic, byte *payload, unsigned int length) { void mqtt_callback(char *topic, byte *payload, unsigned int length) {
String s = "";
for (int i = 0; i < length; i++) char buffer[11];
s += (char)payload[i]; snprintf(buffer, 10, "%X02 %X02 %X02 %X02 %X02 %X02 %X02 %X02 %X02 %X02",
ESP_LOGD(TAG, "MQTT: Received %u byte(s) of payload [%s]", length, s); payload);
// rcommand(payload, length); ESP_LOGI(TAG, "MQTT: Received %u byte(s) of payload [%s]", length, buffer);
rcommand(payload, length);
} }
void mqtt_queuereset(void) { xQueueReset(MQTTSendQueue); } void mqtt_queuereset(void) { xQueueReset(MQTTSendQueue); }
void mqtt_irq(void) { xTaskNotify(irqHandlerTask, MQTT_IRQ, eSetBits); }
void mqtt_loop(void) { mqttClient.loop(); }
#endif // HAS_MQTT #endif // HAS_MQTT