mqtt lib changed (experimental)

include/mqttclient.h

@@ -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);
 

platformio.ini

@@ -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,28 +33,28 @@ 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]
 ; for release_version use max. 10 chars total, use any decimal format like "a.b.c"
 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 = [email protected].2
+platform_espressif32 = [email protected].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 [email protected]
     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}

src/hal/olimexpoeiso.h

@@ -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

src/mqttclient.cpp

@@ -8,18 +8,79 @@ QueueHandle_t MQTTSendQueue;
 TaskHandle_t mqttTask;
 
 Ticker mqttTimer;
+WiFiClient netClient;
+MQTTClient mqttClient;
 
-WiFiClient NetClient;
-PubSubClient mqttClient(NetClient);
+esp_err_t mqtt_init(void) {
+
+  // setup network connection
+  WiFi.onEvent(NetworkEvent);
+  ETH.begin();
+  // WiFi.mode(WIFI_STA);
+  // WiFi.begin("SSID", "PW");
+
+  // 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) {
+  IPAddress mqtt_server_ip;
+
+  // static String clientId = "paxcounter-" + ETH.macAddress();
+  static String clientId = "paxcounter-" + String(random(0xffff), HEX);
+
+  ESP_LOGI(TAG, "MQTT name is %s", MQTT_CLIENTNAME);
+
+  // resolve server host name
+  if (WiFi.hostByName(my_host, mqtt_server_ip)) {
+    ESP_LOGI(TAG, "Attempting to connect to %s [%s]", my_host,
+             mqtt_server_ip.toString().c_str());
+  } else {
+    ESP_LOGI(TAG, "Could not resolve %s", my_host);
+    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_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:
-    ESP_LOGI(TAG, "Ethernet link layer started");
-    ETH.setHostname(ETH.macAddress().c_str());
+  case SYSTEM_EVENT_STA_START:
+    ESP_LOGI(TAG, "Network link layer started");
+    // ETH.setHostname(ETH.macAddress().c_str());
     break;
   case SYSTEM_EVENT_ETH_STOP:
-    ESP_LOGI(TAG, "Ethernet link layer stopped");
+  case SYSTEM_EVENT_STA_STOP:
+    ESP_LOGI(TAG, "Network link layer stopped");
     break;
   case SYSTEM_EVENT_ETH_CONNECTED:
   case SYSTEM_EVENT_STA_CONNECTED:
@@ -30,11 +91,13 @@ void NetworkEvent(WiFiEvent_t event) {
     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, "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;
 
@@ -43,45 +106,6 @@ void NetworkEvent(WiFiEvent_t event) {
   }
 }
 
-int mqtt_connect(const char *my_host, const uint16_t my_port) {
-  IPAddress mqtt_server_ip;
-
-  // static String clientId = "paxcounter-" + ETH.macAddress();
-  static String clientId = "paxcounter-" + String(random(0xffff), HEX);
-
-  ESP_LOGI(TAG, "MQTT name is %s", clientId.c_str());
-
-  // resolve server host name
-  if (WiFi.hostByName(my_host, mqtt_server_ip)) {
-    ESP_LOGI(TAG, "Attempting to connect to %s [%s]", my_host,
-             mqtt_server_ip.toString().c_str());
-  } else {
-    ESP_LOGI(TAG, "Could not resolve %s", my_host);
-    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;
-    }
-  } else {
-    ESP_LOGW(TAG, "MQTT server not connected, retrying later");
-    return -1;
-  }
-}
-
 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) {
-
-  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);
+void mqtt_callback(MQTTClient *client, char topic[], char payload[],
+                   int length) {
+  if (topic == MQTT_INTOPIC)
+    rcommand((const uint8_t *)payload, (const uint8_t)length);
+}
 
-  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

src/timekeeper.cpp

@@ -1,14 +1,12 @@
 #include "timekeeper.h"
 
-/*
 #if !(HAS_LORA)
 #if (TIME_SYNC_LORASERVER)
 #error TIME_SYNC_LORASERVER defined, but device has no LORA configured
 #elif (TIME_SYNC_LORAWAN)
 #error TIME_SYNC_LORAWAN defined, but device has no LORA configured
 #endif
 #endif
-*/
 
 // Local logging tag
 static const char TAG[] = __FILE__;