mqtt lib changed (experimental)
This commit is contained in:
parent
c51a45cfd0
commit
6b785fedd7
@ -4,15 +4,22 @@
|
||||
#include "globals.h"
|
||||
#include "rcommand.h"
|
||||
#include <ETH.h>
|
||||
#include <PubSubClient.h>
|
||||
#include <MQTT.h>
|
||||
|
||||
#define MQTT_INTOPIC "paxcounter/in"
|
||||
#define MQTT_OUTTOPIC "paxcounter/out"
|
||||
#define MQTT_PORT 1883
|
||||
#define MQTT_SERVER "broker.hivemq.com"
|
||||
//#define MQTT_SERVER "test.mosquitto.org"
|
||||
#define MQTT_SERVER "broker.shiftr.io"
|
||||
//#define MQTT_CLIENTNAME "arduino"
|
||||
#define MQTT_USER "try"
|
||||
#define MQTT_PASSWD "try"
|
||||
#define MQTT_RETRYSEC 20 // retry reconnect every 20 seconds
|
||||
#define MQTT_KEEPALIVE 10 // seconds to keep alive server connections
|
||||
#define MQTT_KEEPALIVE 10 // keep alive interval in seconds
|
||||
#define MQTT_TIMEOUT 1000 // timeout for all mqtt commands in milliseconds
|
||||
|
||||
#ifndef MQTT_CLIENTNAME
|
||||
#define MQTT_CLIENTNAME clientId.c_str()
|
||||
#endif
|
||||
|
||||
extern TaskHandle_t mqttTask;
|
||||
|
||||
@ -22,7 +29,8 @@ void mqtt_irq(void);
|
||||
void mqtt_loop(void);
|
||||
void mqtt_client_task(void *param);
|
||||
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(MQTTClient *client, char topic[], char payload[],
|
||||
int length);
|
||||
void NetworkEvent(WiFiEvent_t event);
|
||||
esp_err_t mqtt_init(void);
|
||||
|
||||
|
@ -7,7 +7,7 @@
|
||||
|
||||
; ---> SELECT THE TARGET PLATFORM HERE! <---
|
||||
[board]
|
||||
halfile = generic.h
|
||||
;halfile = generic.h
|
||||
;halfile = ebox.h
|
||||
;halfile = eboxtube.h
|
||||
;halfile = ecopower.h
|
||||
@ -33,15 +33,15 @@ halfile = generic.h
|
||||
;halfile = tinypicomatrix.h
|
||||
;halfile = m5core.h
|
||||
;halfile = m5fire.h
|
||||
;halfile = olimexpoeiso.h
|
||||
halfile = olimexpoeiso.h
|
||||
|
||||
[platformio]
|
||||
; upload firmware to board with usb cable
|
||||
default_envs = usb
|
||||
;default_envs = usb
|
||||
; upload firmware to a jfrog bintray repository
|
||||
;default_envs = ota
|
||||
; use latest versions of libraries
|
||||
;default_envs = dev
|
||||
default_envs = dev
|
||||
description = Paxcounter is a device for metering passenger flows in realtime. It counts how many mobile devices are around.
|
||||
|
||||
[common]
|
||||
@ -49,12 +49,12 @@ description = Paxcounter is a device for metering passenger flows in realtime. I
|
||||
release_version = 1.9.996
|
||||
; DEBUG LEVEL: For production run set to 0, otherwise device will leak RAM while running!
|
||||
; 0=None, 1=Error, 2=Warn, 3=Info, 4=Debug, 5=Verbose
|
||||
debug_level = 3
|
||||
debug_level = 4
|
||||
extra_scripts = pre:build.py
|
||||
otakeyfile = ota.conf
|
||||
lorakeyfile = loraconf.h
|
||||
lmicconfigfile = lmic_config.h
|
||||
platform_espressif32 = espressif32@1.12.2
|
||||
platform_espressif32 = espressif32@1.12.4
|
||||
monitor_speed = 115200
|
||||
upload_speed = 115200 ; set by build.py and taken from hal file
|
||||
display_library = ; set by build.py and taken from hal file
|
||||
@ -79,13 +79,13 @@ lib_deps_sensors =
|
||||
BSEC Software Library@1.5.1474
|
||||
https://github.com/ricki-z/SDS011.git
|
||||
lib_deps_basic =
|
||||
ArduinoJson@^5.13.1
|
||||
ArduinoJson@<6
|
||||
76@>=1.2.4 ; #76 Timezone by Jack Christensen
|
||||
274@>=2.3.4 ; #274 RTC by Michael Miller
|
||||
SimpleButton
|
||||
AXP202X_Library@>=1.1.1 ; AXP202 PMU lib by Lewis He
|
||||
esp32-micro-sdcard
|
||||
PubSubClient@>=2.8.0
|
||||
MQTT@>=2.4.7 ; MQTT client maintained by Joel Gaehwiler
|
||||
lib_deps_all =
|
||||
${common.lib_deps_basic}
|
||||
${common.lib_deps_lora}
|
||||
|
@ -8,7 +8,7 @@
|
||||
#include <stdint.h>
|
||||
|
||||
// enable only if you want to store a local paxcount table on the device
|
||||
//#define HAS_SDCARD 2 // this board has a SDMMC card-reader/writer
|
||||
#define HAS_SDCARD 2 // this board has a SDMMC card-reader/writer
|
||||
|
||||
// enable only if you want to send paxcount via ethernet port to mqtt server
|
||||
#define HAS_MQTT 1 // use MQTT on ethernet interface
|
||||
|
@ -8,39 +8,35 @@ QueueHandle_t MQTTSendQueue;
|
||||
TaskHandle_t mqttTask;
|
||||
|
||||
Ticker mqttTimer;
|
||||
WiFiClient netClient;
|
||||
MQTTClient mqttClient;
|
||||
|
||||
WiFiClient NetClient;
|
||||
PubSubClient mqttClient(NetClient);
|
||||
esp_err_t mqtt_init(void) {
|
||||
|
||||
void NetworkEvent(WiFiEvent_t event) {
|
||||
switch (event) {
|
||||
case SYSTEM_EVENT_ETH_START:
|
||||
ESP_LOGI(TAG, "Ethernet link layer started");
|
||||
ETH.setHostname(ETH.macAddress().c_str());
|
||||
break;
|
||||
case SYSTEM_EVENT_ETH_STOP:
|
||||
ESP_LOGI(TAG, "Ethernet link layer stopped");
|
||||
break;
|
||||
case SYSTEM_EVENT_ETH_CONNECTED:
|
||||
case SYSTEM_EVENT_STA_CONNECTED:
|
||||
ESP_LOGI(TAG, "Network link connected");
|
||||
break;
|
||||
case SYSTEM_EVENT_ETH_DISCONNECTED:
|
||||
case SYSTEM_EVENT_STA_DISCONNECTED:
|
||||
ESP_LOGI(TAG, "Network link disconnected");
|
||||
break;
|
||||
case SYSTEM_EVENT_ETH_GOT_IP:
|
||||
ESP_LOGI(TAG, "ETH MAC: %s", ETH.macAddress().c_str());
|
||||
ESP_LOGI(TAG, "IPv4: %s", ETH.localIP().toString().c_str());
|
||||
ESP_LOGI(TAG, "Link Speed: %d Mbps %s", ETH.linkSpeed(),
|
||||
ETH.fullDuplex() ? "full duplex" : "half duplex");
|
||||
case SYSTEM_EVENT_STA_GOT_IP:
|
||||
mqtt_connect(MQTT_SERVER, MQTT_PORT);
|
||||
break;
|
||||
// setup network connection
|
||||
WiFi.onEvent(NetworkEvent);
|
||||
ETH.begin();
|
||||
// WiFi.mode(WIFI_STA);
|
||||
// WiFi.begin("SSID", "PW");
|
||||
|
||||
default:
|
||||
break;
|
||||
// setup mqtt client
|
||||
mqttClient.begin(MQTT_SERVER, MQTT_PORT, netClient);
|
||||
mqttClient.onMessageAdvanced(mqtt_callback);
|
||||
|
||||
assert(SEND_QUEUE_SIZE);
|
||||
MQTTSendQueue = xQueueCreate(SEND_QUEUE_SIZE, sizeof(MessageBuffer_t));
|
||||
if (MQTTSendQueue == 0) {
|
||||
ESP_LOGE(TAG, "Could not create MQTT send queue. Aborting.");
|
||||
return ESP_FAIL;
|
||||
}
|
||||
ESP_LOGI(TAG, "MQTT send queue created, size %d Bytes",
|
||||
SEND_QUEUE_SIZE * PAYLOAD_BUFFER_SIZE);
|
||||
|
||||
ESP_LOGI(TAG, "Starting MQTTloop...");
|
||||
mqttTimer.attach(MQTT_KEEPALIVE, mqtt_irq);
|
||||
xTaskCreate(mqtt_client_task, "mqttloop", 4096, (void *)NULL, 1, &mqttTask);
|
||||
|
||||
return ESP_OK;
|
||||
}
|
||||
|
||||
int mqtt_connect(const char *my_host, const uint16_t my_port) {
|
||||
@ -49,7 +45,7 @@ int mqtt_connect(const char *my_host, const uint16_t my_port) {
|
||||
// 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", MQTT_CLIENTNAME);
|
||||
|
||||
// resolve server host name
|
||||
if (WiFi.hostByName(my_host, mqtt_server_ip)) {
|
||||
@ -60,28 +56,56 @@ int mqtt_connect(const char *my_host, const uint16_t my_port) {
|
||||
return -1;
|
||||
}
|
||||
|
||||
// attempt to connect to MQTT server
|
||||
if (NetClient.connect(mqtt_server_ip, my_port, HOMECYCLE * 2 * 1000)) {
|
||||
NetClient.setTimeout(MQTT_KEEPALIVE); // seconds
|
||||
mqttClient.setServer(mqtt_server_ip, my_port);
|
||||
mqttClient.setKeepAlive(MQTT_KEEPALIVE);
|
||||
mqttClient.setCallback(mqtt_callback);
|
||||
|
||||
if (mqttClient.connect(clientId.c_str())) {
|
||||
ESP_LOGI(TAG, "MQTT server connected, subscribing...");
|
||||
mqttClient.publish(MQTT_OUTTOPIC, clientId.c_str());
|
||||
mqttClient.subscribe(MQTT_INTOPIC);
|
||||
ESP_LOGI(TAG, "MQTT topic subscribed");
|
||||
} else {
|
||||
ESP_LOGW(TAG, "MQTT server not responding, retrying later");
|
||||
return -1;
|
||||
}
|
||||
if (mqttClient.connect(MQTT_CLIENTNAME, MQTT_USER, MQTT_PASSWD)) {
|
||||
ESP_LOGI(TAG, "MQTT server connected, subscribing...");
|
||||
mqttClient.publish(MQTT_OUTTOPIC, MQTT_CLIENTNAME);
|
||||
// Clear retained messages that may have been published earlier on topic
|
||||
mqttClient.publish(MQTT_INTOPIC, "", true, 1);
|
||||
mqttClient.subscribe(MQTT_INTOPIC);
|
||||
ESP_LOGI(TAG, "MQTT topic subscribed");
|
||||
} else {
|
||||
ESP_LOGW(TAG, "MQTT server not connected, retrying later");
|
||||
ESP_LOGD(TAG, "MQTT last_error = %d / rc = %d", mqttClient.lastError(),
|
||||
mqttClient.returnCode());
|
||||
ESP_LOGW(TAG, "MQTT server not responding, retrying later");
|
||||
return -1;
|
||||
}
|
||||
}
|
||||
|
||||
void NetworkEvent(WiFiEvent_t event) {
|
||||
switch (event) {
|
||||
case SYSTEM_EVENT_ETH_START:
|
||||
case SYSTEM_EVENT_STA_START:
|
||||
ESP_LOGI(TAG, "Network link layer started");
|
||||
// ETH.setHostname(ETH.macAddress().c_str());
|
||||
break;
|
||||
case SYSTEM_EVENT_ETH_STOP:
|
||||
case SYSTEM_EVENT_STA_STOP:
|
||||
ESP_LOGI(TAG, "Network link layer stopped");
|
||||
break;
|
||||
case SYSTEM_EVENT_ETH_CONNECTED:
|
||||
case SYSTEM_EVENT_STA_CONNECTED:
|
||||
ESP_LOGI(TAG, "Network link connected");
|
||||
break;
|
||||
case SYSTEM_EVENT_ETH_DISCONNECTED:
|
||||
case SYSTEM_EVENT_STA_DISCONNECTED:
|
||||
ESP_LOGI(TAG, "Network link disconnected");
|
||||
break;
|
||||
case SYSTEM_EVENT_ETH_GOT_IP:
|
||||
ESP_LOGI(TAG, "IP: %s", ETH.localIP().toString().c_str());
|
||||
ESP_LOGI(TAG, "Link Speed: %d Mbps %s", ETH.linkSpeed(),
|
||||
ETH.fullDuplex() ? "full duplex" : "half duplex");
|
||||
mqtt_connect(MQTT_SERVER, MQTT_PORT);
|
||||
break;
|
||||
case SYSTEM_EVENT_STA_GOT_IP:
|
||||
ESP_LOGI(TAG, "IP: %s", WiFi.localIP().toString().c_str());
|
||||
mqtt_connect(MQTT_SERVER, MQTT_PORT);
|
||||
break;
|
||||
|
||||
default:
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
void mqtt_client_task(void *param) {
|
||||
|
||||
MessageBuffer_t msg;
|
||||
@ -96,11 +120,12 @@ void mqtt_client_task(void *param) {
|
||||
|
||||
// send data to mqtt server, if we are connected
|
||||
if (mqttClient.connected()) {
|
||||
mqttClient.beginPublish(MQTT_OUTTOPIC, msg.MessageSize + 2, false);
|
||||
mqttClient.write(msg.MessagePort);
|
||||
mqttClient.write('/');
|
||||
mqttClient.write(msg.Message, msg.MessageSize);
|
||||
if (mqttClient.endPublish()) {
|
||||
|
||||
char buffer[PAYLOAD_BUFFER_SIZE + 3];
|
||||
snprintf(buffer, msg.MessageSize + 3, "%s/%s", msg.MessagePort,
|
||||
msg.Message);
|
||||
|
||||
if (mqttClient.publish(MQTT_OUTTOPIC, buffer)) {
|
||||
ESP_LOGI(TAG, "%d byte(s) sent to MQTT server", msg.MessageSize + 2);
|
||||
continue;
|
||||
} else {
|
||||
@ -112,6 +137,8 @@ void mqtt_client_task(void *param) {
|
||||
} else {
|
||||
// attempt to reconnect to MQTT server
|
||||
ESP_LOGD(TAG, "MQTT client reconnecting...");
|
||||
ESP_LOGD(TAG, "MQTT last_error = %d / rc = %d", mqttClient.lastError(),
|
||||
mqttClient.returnCode());
|
||||
mqtt_enqueuedata(&msg); // postpone the undelivered message
|
||||
delay(MQTT_RETRYSEC * 1000);
|
||||
mqtt_connect(MQTT_SERVER, MQTT_PORT);
|
||||
@ -120,27 +147,6 @@ void mqtt_client_task(void *param) {
|
||||
} // while(1)
|
||||
}
|
||||
|
||||
esp_err_t mqtt_init(void) {
|
||||
|
||||
WiFi.onEvent(NetworkEvent);
|
||||
ETH.begin();
|
||||
|
||||
assert(SEND_QUEUE_SIZE);
|
||||
MQTTSendQueue = xQueueCreate(SEND_QUEUE_SIZE, sizeof(MessageBuffer_t));
|
||||
if (MQTTSendQueue == 0) {
|
||||
ESP_LOGE(TAG, "Could not create MQTT send queue. Aborting.");
|
||||
return ESP_FAIL;
|
||||
}
|
||||
ESP_LOGI(TAG, "MQTT send queue created, size %d Bytes",
|
||||
SEND_QUEUE_SIZE * PAYLOAD_BUFFER_SIZE);
|
||||
|
||||
ESP_LOGI(TAG, "Starting MQTTloop...");
|
||||
mqttTimer.attach(MQTT_KEEPALIVE, mqtt_irq);
|
||||
xTaskCreate(mqtt_client_task, "mqttloop", 4096, (void *)NULL, 1, &mqttTask);
|
||||
|
||||
return ESP_OK;
|
||||
}
|
||||
|
||||
void mqtt_enqueuedata(MessageBuffer_t *message) {
|
||||
// enqueue message in MQTT send queue
|
||||
BaseType_t ret;
|
||||
@ -164,18 +170,19 @@ void mqtt_enqueuedata(MessageBuffer_t *message) {
|
||||
ESP_LOGW(TAG, "MQTT sendqueue is full");
|
||||
}
|
||||
|
||||
void mqtt_callback(char *topic, byte *payload, unsigned int length) {
|
||||
void mqtt_callback(MQTTClient *client, char topic[], char payload[],
|
||||
int length) {
|
||||
if (topic == MQTT_INTOPIC)
|
||||
rcommand((const uint8_t *)payload, (const uint8_t)length);
|
||||
}
|
||||
|
||||
char buffer[11];
|
||||
snprintf(buffer, 10, "%X02 %X02 %X02 %X02 %X02 %X02 %X02 %X02 %X02 %X02",
|
||||
payload);
|
||||
ESP_LOGI(TAG, "MQTT: Received %u byte(s) of payload [%s]", length, buffer);
|
||||
|
||||
rcommand(payload, length);
|
||||
void mqtt_loop(void) {
|
||||
if (!mqttClient.loop())
|
||||
ESP_LOGD(TAG, "MQTT last_error = %d / rc = %d", mqttClient.lastError(),
|
||||
mqttClient.returnCode());
|
||||
}
|
||||
|
||||
void mqtt_queuereset(void) { xQueueReset(MQTTSendQueue); }
|
||||
void mqtt_irq(void) { xTaskNotify(irqHandlerTask, MQTT_IRQ, eSetBits); }
|
||||
void mqtt_loop(void) { mqttClient.loop(); }
|
||||
|
||||
#endif // HAS_MQTT
|
@ -1,6 +1,5 @@
|
||||
#include "timekeeper.h"
|
||||
|
||||
/*
|
||||
#if !(HAS_LORA)
|
||||
#if (TIME_SYNC_LORASERVER)
|
||||
#error TIME_SYNC_LORASERVER defined, but device has no LORA configured
|
||||
@ -8,7 +7,6 @@
|
||||
#error TIME_SYNC_LORAWAN defined, but device has no LORA configured
|
||||
#endif
|
||||
#endif
|
||||
*/
|
||||
|
||||
// Local logging tag
|
||||
static const char TAG[] = __FILE__;
|
||||
|
Loading…
Reference in New Issue
Block a user