Adding MQTT for local APRS Packet storage.

data/is-cfg.json

@@ -65,5 +65,13 @@
 			}
 		]
 	},
+	"mqtt": {
+		"active": false,
+		"server": "",
+		"port": 1883,
+		"name": "",
+		"password": "",
+		"topic": "LoraAPRS/Data"
+	},
 	"ntp_server": "pool.ntp.org"
 }

platformio.ini

@@ -13,6 +13,7 @@ lib_deps =
 	peterus/APRS-Decoder-Lib @ 0.0.6
 	peterus/esp-logger @ 0.0.1
 	peterus/ESP-FTP-Server-Lib @ 0.9.5
+	knolleary/PubSubClient@^2.8
 check_tool = cppcheck
 check_flags =
 	cppcheck: --suppress=*:*.pio\* --inline-suppr -DCPPCHECK --force lib -ilib/TimeLib -ilib/LoRa -ilib/NTPClient

src/LoRa_APRS_iGate.cpp

@@ -16,9 +16,10 @@
 #include "TaskOTA.h"
 #include "TaskRouter.h"
 #include "TaskWifi.h"
+#include "TaskMQTT.h"
 #include "project_configuration.h"
 
-#define VERSION "22.7.0"
+#define VERSION "22.7.0_zj"
 
 String create_lat_aprs(double lat);
 String create_long_aprs(double lng);
@@ -26,6 +27,7 @@ String create_long_aprs(double lng);
 TaskQueue<std::shared_ptr<APRSMessage>> toAprsIs;
 TaskQueue<std::shared_ptr<APRSMessage>> fromModem;
 TaskQueue<std::shared_ptr<APRSMessage>> toModem;
+TaskQueue<std::shared_ptr<APRSMessage>> toMQTT;
 
 System        LoRaSystem;
 Configuration userConfig;
@@ -37,8 +39,9 @@ WifiTask    wifiTask;
 OTATask     otaTask;
 NTPTask     ntpTask;
 FTPTask     ftpTask;
+MQTTTask    mqttTask(toMQTT);
 AprsIsTask  aprsIsTask(toAprsIs);
-RouterTask  routerTask(fromModem, toModem, toAprsIs);
+RouterTask  routerTask(fromModem, toModem, toAprsIs, toMQTT);
 
 void setup() {
   Serial.begin(115200);
@@ -114,6 +117,10 @@ void setup() {
     LoRaSystem.getTaskManager().addTask(&aprsIsTask);
   }
 
+  if (userConfig.mqtt.active) {
+    LoRaSystem.getTaskManager().addTask(&mqttTask);
+  }
+
   LoRaSystem.getTaskManager().setup(LoRaSystem);
 
   LoRaSystem.getDisplay().showSpashScreen("LoRa APRS iGate", VERSION);

src/Task.h

@@ -12,6 +12,7 @@ enum TaskNames
   TaskWifi,
   TaskRouter,
   TaskSize,
+  TaskMQTT,
 };
 
 #define TASK_APRS_IS "AprsIsTask"
@@ -22,5 +23,6 @@ enum TaskNames
 #define TASK_OTA     "OTATask"
 #define TASK_WIFI    "WifiTask"
 #define TASK_ROUTER  "RouterTask"
+#define TASK_MQTT    "MQTTTask"
 
 #endif

src/TaskMQTT.cpp

@@ -0,0 +1,88 @@
+#include <logger.h>
+
+#include "Task.h"
+#include "TaskMQTT.h"
+#include "project_configuration.h"
+
+#include <WiFi.h>
+#include <WiFiMulti.h>
+
+#include <ArduinoJson.h>
+
+WiFiClient wiFiClient;
+PubSubClient _MQTT(wiFiClient);
+
+
+MQTTTask::MQTTTask(TaskQueue<std::shared_ptr<APRSMessage>> &toMQTT) : Task(TASK_MQTT, TaskMQTT), _beginCalled(false), _toMQTT(toMQTT) {
+}
+
+MQTTTask::~MQTTTask() {
+}
+
+bool MQTTTask::setup(System &system) { 
+  _MQTT.setServer(system.getUserConfig()->mqtt.server.c_str(), system.getUserConfig()->mqtt.port);
+
+  return true;
+}
+
+bool MQTTTask::loop(System &system) {
+  if (!_beginCalled) {
+    _beginCalled = true;
+  }
+
+  if (!system.isWifiEthConnected()) {
+    return false;
+  }
+  
+  if (!_MQTT.connected()) {
+    connect(system);
+  }
+
+  if (!_toMQTT.empty()) {
+    std::shared_ptr<APRSMessage> msg = _toMQTT.getElement();
+
+    DynamicJsonDocument _Data(1024);
+    String _r;
+
+    _Data["Source"] = msg->getSource();
+    _Data["Destination"] = msg->getDestination();
+    _Data["Path"] = msg->getPath();
+    _Data["Type"] = msg->getType().toString();
+    String _body = msg->getBody()->encode();
+    _body.replace("\n", "");
+    _Data["Data"] = _body;
+
+    serializeJson(_Data, _r);
+
+    logPrintD("Send MQTT: ");
+    logPrintlnD(_r);
+    
+    String _topic = String(system.getUserConfig()->mqtt.topic);
+
+    if (!_topic.endsWith("/")) {
+      _topic = _topic + "/";
+    }
+    _topic = _topic + system.getUserConfig()->callsign;
+
+    _MQTT.publish(_topic.c_str(), _r.c_str());
+  }
+  _MQTT.loop();
+  return true;
+}
+
+bool MQTTTask::connect(const System &system) {
+  logPrintI("Connecting to MQTT broker: ");  
+  logPrintI(system.getUserConfig()->mqtt.server);
+  logPrintI(" on port ");  
+  logPrintlnI(String(system.getUserConfig()->mqtt.port));
+    
+  if (_MQTT.connect(system.getUserConfig()->callsign.c_str(),system.getUserConfig()->mqtt.name.c_str() ,system.getUserConfig()->mqtt.password.c_str() )) {
+    logPrintI("Connected to MQTT broker as: ");  
+    logPrintlnI(system.getUserConfig()->callsign);
+
+    return true;
+  } else {
+    logPrintlnI("Connecting to MQTT broker faild. Try again later.");  
+  } 
+  return false;
+}

src/TaskMQTT.h

@@ -0,0 +1,22 @@
+#ifndef TASK_MQTT_H_
+#define TASK_MQTT_H_
+
+#include <APRSMessage.h>
+#include <TaskManager.h>
+#include <PubSubClient.h>
+
+class MQTTTask : public Task {
+public:
+  MQTTTask(TaskQueue<std::shared_ptr<APRSMessage>> &toMQTT);
+  virtual ~MQTTTask();
+
+  virtual bool setup(System &system) override;
+  virtual bool loop(System &system) override;
+  
+private:
+  bool      _beginCalled;
+  TaskQueue<std::shared_ptr<APRSMessage>> &_toMQTT;  
+  bool connect(const System &system);  
+};
+
+#endif

src/TaskRouter.cpp

@@ -9,7 +9,7 @@
 String create_lat_aprs(double lat);
 String create_long_aprs(double lng);
 
-RouterTask::RouterTask(TaskQueue<std::shared_ptr<APRSMessage>> &fromModem, TaskQueue<std::shared_ptr<APRSMessage>> &toModem, TaskQueue<std::shared_ptr<APRSMessage>> &toAprsIs) : Task(TASK_ROUTER, TaskRouter), _fromModem(fromModem), _toModem(toModem), _toAprsIs(toAprsIs) {
+RouterTask::RouterTask(TaskQueue<std::shared_ptr<APRSMessage>> &fromModem, TaskQueue<std::shared_ptr<APRSMessage>> &toModem, TaskQueue<std::shared_ptr<APRSMessage>> &toAprsIs, TaskQueue<std::shared_ptr<APRSMessage>> &toMQTT) : Task(TASK_ROUTER, TaskRouter), _fromModem(fromModem), _toModem(toModem), _toAprsIs(toAprsIs), _toMQTT(toMQTT) {
 }
 
 RouterTask::~RouterTask() {
@@ -33,6 +33,11 @@ bool RouterTask::loop(System &system) {
   // do routing
   if (!_fromModem.empty()) {
     std::shared_ptr<APRSMessage> modemMsg = _fromModem.getElement();
+    std::shared_ptr<APRSMessage> DataMQTT = modemMsg;
+
+    if (system.getUserConfig()->mqtt.active) {
+      _toMQTT.addElement(DataMQTT);
+    }
 
     if (system.getUserConfig()->aprs_is.active && modemMsg->getSource() != system.getUserConfig()->callsign) {
       std::shared_ptr<APRSMessage> aprsIsMsg = std::make_shared<APRSMessage>(*modemMsg);

src/TaskRouter.h

@@ -3,10 +3,12 @@
 
 #include <APRSMessage.h>
 #include <TaskManager.h>
+#include <TaskMQTT.h>
 
 class RouterTask : public Task {
 public:
-  RouterTask(TaskQueue<std::shared_ptr<APRSMessage>> &fromModem, TaskQueue<std::shared_ptr<APRSMessage>> &toModem, TaskQueue<std::shared_ptr<APRSMessage>> &toAprsIs);
+  RouterTask(TaskQueue<std::shared_ptr<APRSMessage>> &fromModem, TaskQueue<std::shared_ptr<APRSMessage>> &toModem, TaskQueue<std::shared_ptr<APRSMessage>> &toAprsIs, TaskQueue<std::shared_ptr<APRSMessage>> &toMQTT
+);
   virtual ~RouterTask();
 
   virtual bool setup(System &system) override;
@@ -16,6 +18,7 @@ private:
   TaskQueue<std::shared_ptr<APRSMessage>> &_fromModem;
   TaskQueue<std::shared_ptr<APRSMessage>> &_toModem;
   TaskQueue<std::shared_ptr<APRSMessage>> &_toAprsIs;
+  TaskQueue<std::shared_ptr<APRSMessage>> &_toMQTT;
 
   std::shared_ptr<APRSMessage> _beaconMsg;
   Timer                        _beacon_timer;

src/project_configuration.cpp

@@ -77,6 +77,14 @@ void ProjectConfigurationManagement::readProjectConfiguration(DynamicJsonDocumen
     us.password = "ftp";
     conf.ftp.users.push_back(us);
   }
+  if (data.containsKey("mqtt")) {
+    conf.mqtt.active = data["mqtt"]["active"] | false;
+    conf.mqtt.server = data["mqtt"]["server"].as<String>();
+    conf.mqtt.port = data["mqtt"]["port"].as<uint16_t>();
+    conf.mqtt.name = data["mqtt"]["name"].as<String>();
+    conf.mqtt.password = data["mqtt"]["password"].as<String>();
+    conf.mqtt.topic = data["mqtt"]["topic"].as<String>();
+  }
   if (data.containsKey("ntp_server"))
     conf.ntpServer = data["ntp_server"].as<String>();
 
@@ -133,6 +141,12 @@ void ProjectConfigurationManagement::writeProjectConfiguration(Configuration &co
     v["name"]     = u.name;
     v["password"] = u.password;
   }
+  data["mqtt"]["active"]                   = conf.mqtt.active;
+  data["mqtt"]["server"]                   = conf.mqtt.server;
+  data["mqtt"]["port"]                     = conf.mqtt.port;
+  data["mqtt"]["name"]                     = conf.mqtt.name;
+  data["mqtt"]["password"]                 = conf.mqtt.password;
+  data["mqtt"]["topic"]                    = conf.mqtt.topic;  
   data["ntp_server"] = conf.ntpServer;
 
   data["board"] = conf.board;

src/project_configuration.h

@@ -117,6 +117,16 @@ public:
     std::list<User> users;
   };
 
+  class MQTT {
+  public:
+    bool active;
+    String server;
+    uint16_t port;
+    String name;
+    String password;
+    String topic;
+  };
+
   Configuration() : callsign("NOCALL-10"), board(""), ntpServer("pool.ntp.org"){};
 
   String  callsign;
@@ -128,6 +138,7 @@ public:
   LoRa    lora;
   Display display;
   Ftp     ftp;
+  MQTT    mqtt;
   String  board;
   String  ntpServer;
 };