Fork of the espurna firmware for `mhsw` switches
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

397 lines
12 KiB

8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
  1. /*
  2. MQTT MODULE
  3. Copyright (C) 2016-2017 by Xose Pérez <xose dot perez at gmail dot com>
  4. */
  5. #include <ESP8266WiFi.h>
  6. #include <ArduinoJson.h>
  7. #include <vector>
  8. #include <Ticker.h>
  9. const char *mqtt_user = 0;
  10. const char *mqtt_pass = 0;
  11. #if MQTT_USE_ASYNC
  12. #include <AsyncMqttClient.h>
  13. AsyncMqttClient mqtt;
  14. #else
  15. #include <PubSubClient.h>
  16. WiFiClient mqttWiFiClient;
  17. PubSubClient mqtt(mqttWiFiClient);
  18. bool _mqttConnected = false;
  19. #endif
  20. String mqttTopic;
  21. bool _mqttForward;
  22. char *_mqttUser = 0;
  23. char *_mqttPass = 0;
  24. char *_mqttWill;
  25. std::vector<void (*)(unsigned int, const char *, const char *)> _mqtt_callbacks;
  26. #if MQTT_SKIP_RETAINED
  27. unsigned long mqttConnectedAt = 0;
  28. #endif
  29. typedef struct {
  30. char * topic;
  31. char * message;
  32. } mqtt_message_t;
  33. std::vector<mqtt_message_t> _mqtt_queue;
  34. Ticker mqttFlushTicker;
  35. // -----------------------------------------------------------------------------
  36. // Public API
  37. // -----------------------------------------------------------------------------
  38. bool mqttConnected() {
  39. return mqtt.connected();
  40. }
  41. void mqttDisconnect() {
  42. mqtt.disconnect();
  43. }
  44. void buildTopics() {
  45. // Replace identifier
  46. mqttTopic = getSetting("mqttTopic", MQTT_TOPIC);
  47. mqttTopic.replace("{identifier}", getSetting("hostname"));
  48. if (!mqttTopic.endsWith("/")) mqttTopic = mqttTopic + "/";
  49. }
  50. bool mqttForward() {
  51. return _mqttForward;
  52. }
  53. String mqttSubtopic(char * topic) {
  54. String response;
  55. String t = String(topic);
  56. String mqttSetter = getSetting("mqttSetter", MQTT_USE_SETTER);
  57. if (t.startsWith(mqttTopic) && t.endsWith(mqttSetter)) {
  58. response = t.substring(mqttTopic.length(), t.length() - mqttSetter.length());
  59. }
  60. return response;
  61. }
  62. void mqttSendRaw(const char * topic, const char * message) {
  63. if (mqtt.connected()) {
  64. #if MQTT_USE_ASYNC
  65. unsigned int packetId = mqtt.publish(topic, MQTT_QOS, MQTT_RETAIN, message);
  66. DEBUG_MSG_P(PSTR("[MQTT] Sending %s => %s (PID %d)\n"), topic, message, packetId);
  67. #else
  68. mqtt.publish(topic, message, MQTT_RETAIN);
  69. DEBUG_MSG_P(PSTR("[MQTT] Sending %s => %s\n"), topic, message);
  70. #endif
  71. }
  72. }
  73. void _mqttFlush() {
  74. if (_mqtt_queue.size() == 0) return;
  75. DynamicJsonBuffer jsonBuffer;
  76. JsonObject& root = jsonBuffer.createObject();
  77. for (unsigned char i=0; i<_mqtt_queue.size(); i++) {
  78. mqtt_message_t element = _mqtt_queue[i];
  79. root[element.topic] = element.message;
  80. }
  81. if (ntpConnected()) root[MQTT_TOPIC_TIME] = ntpDateTime();
  82. root[MQTT_TOPIC_HOSTNAME] = getSetting("hostname", HOSTNAME);
  83. root[MQTT_TOPIC_IP] = getIP();
  84. String output;
  85. root.printTo(output);
  86. String path = mqttTopic + String(MQTT_TOPIC_JSON);
  87. mqttSendRaw(path.c_str(), output.c_str());
  88. for (unsigned char i = 0; i < _mqtt_queue.size(); i++) {
  89. mqtt_message_t element = _mqtt_queue[i];
  90. free(element.topic);
  91. free(element.message);
  92. }
  93. _mqtt_queue.clear();
  94. }
  95. void mqttSend(const char * topic, const char * message, bool force) {
  96. bool useJson = force ? false : getSetting("mqttUseJson", MQTT_USE_JSON).toInt() == 1;
  97. if (useJson) {
  98. mqtt_message_t element;
  99. element.topic = strdup(topic);
  100. element.message = strdup(message);
  101. _mqtt_queue.push_back(element);
  102. mqttFlushTicker.once_ms(MQTT_USE_JSON_DELAY, _mqttFlush);
  103. } else {
  104. String mqttGetter = getSetting("mqttGetter", MQTT_USE_GETTER);
  105. String path = mqttTopic + String(topic) + mqttGetter;
  106. mqttSendRaw(path.c_str(), message);
  107. }
  108. }
  109. void mqttSend(const char * topic, const char * message) {
  110. mqttSend(topic, message, false);
  111. }
  112. void mqttSend(const char * topic, unsigned int index, const char * message, bool force) {
  113. char buffer[strlen(topic)+5];
  114. sprintf(buffer, "%s/%d", topic, index);
  115. mqttSend(buffer, message, force);
  116. }
  117. void mqttSend(const char * topic, unsigned int index, const char * message) {
  118. mqttSend(topic, index, message, false);
  119. }
  120. void mqttSubscribeRaw(const char * topic) {
  121. if (mqtt.connected() && (strlen(topic) > 0)) {
  122. #if MQTT_USE_ASYNC
  123. unsigned int packetId = mqtt.subscribe(topic, MQTT_QOS);
  124. DEBUG_MSG_P(PSTR("[MQTT] Subscribing to %s (PID %d)\n"), topic, packetId);
  125. #else
  126. mqtt.subscribe(topic, MQTT_QOS);
  127. DEBUG_MSG_P(PSTR("[MQTT] Subscribing to %s\n"), topic);
  128. #endif
  129. }
  130. }
  131. void mqttSubscribe(const char * topic) {
  132. String mqttSetter = getSetting("mqttSetter", MQTT_USE_SETTER);
  133. String path = mqttTopic + String(topic) + mqttSetter;
  134. mqttSubscribeRaw(path.c_str());
  135. }
  136. // -----------------------------------------------------------------------------
  137. // Callbacks
  138. // -----------------------------------------------------------------------------
  139. void mqttRegister(void (*callback)(unsigned int, const char *, const char *)) {
  140. _mqtt_callbacks.push_back(callback);
  141. }
  142. void _mqttOnConnect() {
  143. DEBUG_MSG_P(PSTR("[MQTT] Connected!\n"));
  144. #if MQTT_SKIP_RETAINED
  145. mqttConnectedAt = millis();
  146. #endif
  147. // Build MQTT topics
  148. buildTopics();
  149. // Send first Heartbeat
  150. heartbeat();
  151. // Subscribe to system topics
  152. mqttSubscribe(MQTT_TOPIC_ACTION);
  153. // Send connect event to subscribers
  154. for (unsigned char i = 0; i < _mqtt_callbacks.size(); i++) {
  155. (*_mqtt_callbacks[i])(MQTT_CONNECT_EVENT, NULL, NULL);
  156. }
  157. }
  158. void _mqttOnDisconnect() {
  159. DEBUG_MSG_P(PSTR("[MQTT] Disconnected!\n"));
  160. // Send disconnect event to subscribers
  161. for (unsigned char i = 0; i < _mqtt_callbacks.size(); i++) {
  162. (*_mqtt_callbacks[i])(MQTT_DISCONNECT_EVENT, NULL, NULL);
  163. }
  164. }
  165. void _mqttOnMessage(char* topic, char* payload, unsigned int len) {
  166. if (len == 0) return;
  167. char message[len + 1];
  168. strlcpy(message, (char *) payload, len + 1);
  169. DEBUG_MSG_P(PSTR("[MQTT] Received %s => %s"), topic, message);
  170. #if MQTT_SKIP_RETAINED
  171. if (millis() - mqttConnectedAt < MQTT_SKIP_TIME) {
  172. DEBUG_MSG_P(PSTR(" - SKIPPED\n"));
  173. return;
  174. }
  175. #endif
  176. DEBUG_MSG_P(PSTR("\n"));
  177. // Check system topics
  178. String t = mqttSubtopic((char *) topic);
  179. if (t.equals(MQTT_TOPIC_ACTION)) {
  180. if (strcmp(message, MQTT_ACTION_RESET) == 0) {
  181. customReset(CUSTOM_RESET_MQTT);
  182. ESP.restart();
  183. }
  184. }
  185. // Send message event to subscribers
  186. for (unsigned char i = 0; i < _mqtt_callbacks.size(); i++) {
  187. (*_mqtt_callbacks[i])(MQTT_MESSAGE_EVENT, topic, message);
  188. }
  189. }
  190. void mqttConnect() {
  191. if (!mqtt.connected()) {
  192. if (getSetting("mqttServer", MQTT_SERVER).length() == 0) return;
  193. // Last option: reconnect to wifi after MQTT_MAX_TRIES attemps in a row
  194. #if MQTT_MAX_TRIES > 0
  195. static unsigned int tries = 0;
  196. static unsigned long last_try = millis();
  197. if (millis() - last_try < MQTT_TRY_INTERVAL) {
  198. if (++tries > MQTT_MAX_TRIES) {
  199. DEBUG_MSG_P(PSTR("[MQTT] MQTT_MAX_TRIES met, disconnecting from WiFi\n"));
  200. wifiDisconnect();
  201. tries = 0;
  202. return;
  203. }
  204. } else {
  205. tries = 0;
  206. }
  207. last_try = millis();
  208. #endif
  209. mqtt.disconnect();
  210. if (_mqttUser) free(_mqttUser);
  211. if (_mqttPass) free(_mqttPass);
  212. char * host = strdup(getSetting("mqttServer", MQTT_SERVER).c_str());
  213. unsigned int port = getSetting("mqttPort", MQTT_PORT).toInt();
  214. _mqttUser = strdup(getSetting("mqttUser").c_str());
  215. _mqttPass = strdup(getSetting("mqttPassword").c_str());
  216. if (_mqttWill) free(_mqttWill);
  217. _mqttWill = strdup((mqttTopic + MQTT_TOPIC_STATUS).c_str());
  218. DEBUG_MSG_P(PSTR("[MQTT] Connecting to broker at %s:%d"), host, port);
  219. mqtt.setServer(host, port);
  220. #if MQTT_USE_ASYNC
  221. mqtt.setKeepAlive(MQTT_KEEPALIVE).setCleanSession(false);
  222. mqtt.setWill(_mqttWill, MQTT_QOS, MQTT_RETAIN, "0");
  223. if ((strlen(_mqttUser) > 0) && (strlen(_mqttPass) > 0)) {
  224. DEBUG_MSG_P(PSTR(" as user '%s'."), _mqttUser);
  225. mqtt.setCredentials(_mqttUser, _mqttPass);
  226. }
  227. DEBUG_MSG_P(PSTR("\n"));
  228. mqtt.connect();
  229. #else
  230. bool response;
  231. if ((strlen(_mqttUser) > 0) && (strlen(_mqttPass) > 0)) {
  232. DEBUG_MSG_P(PSTR(" as user '%s'\n"), _mqttUser);
  233. response = mqtt.connect(getIdentifier().c_str(), _mqttUser, _mqttPass, _mqttWill, MQTT_QOS, MQTT_RETAIN, "0");
  234. } else {
  235. DEBUG_MSG_P(PSTR("\n"));
  236. response = mqtt.connect(getIdentifier().c_str(), _mqttWill, MQTT_QOS, MQTT_RETAIN, "0");
  237. }
  238. if (response) {
  239. _mqttOnConnect();
  240. _mqttConnected = true;
  241. } else {
  242. DEBUG_MSG_P(PSTR("[MQTT] Connection failed\n"));
  243. }
  244. #endif
  245. free(host);
  246. String mqttSetter = getSetting("mqttSetter", MQTT_USE_SETTER);
  247. String mqttGetter = getSetting("mqttGetter", MQTT_USE_GETTER);
  248. _mqttForward = !mqttGetter.equals(mqttSetter);
  249. }
  250. }
  251. void mqttSetup() {
  252. #if MQTT_USE_ASYNC
  253. mqtt.onConnect([](bool sessionPresent) {
  254. _mqttOnConnect();
  255. });
  256. mqtt.onDisconnect([](AsyncMqttClientDisconnectReason reason) {
  257. if (reason == AsyncMqttClientDisconnectReason::TCP_DISCONNECTED) {
  258. DEBUG_MSG_P(PSTR("[MQTT] TCP Disconnected\n"));
  259. }
  260. if (reason == AsyncMqttClientDisconnectReason::MQTT_IDENTIFIER_REJECTED) {
  261. DEBUG_MSG_P(PSTR("[MQTT] Identifier Rejected\n"));
  262. }
  263. if (reason == AsyncMqttClientDisconnectReason::MQTT_SERVER_UNAVAILABLE) {
  264. DEBUG_MSG_P(PSTR("[MQTT] Server unavailable\n"));
  265. }
  266. if (reason == AsyncMqttClientDisconnectReason::MQTT_MALFORMED_CREDENTIALS) {
  267. DEBUG_MSG_P(PSTR("[MQTT] Malformed credentials\n"));
  268. }
  269. if (reason == AsyncMqttClientDisconnectReason::MQTT_NOT_AUTHORIZED) {
  270. DEBUG_MSG_P(PSTR("[MQTT] Not authorized\n"));
  271. }
  272. if (reason == AsyncMqttClientDisconnectReason::TLS_BAD_FINGERPRINT) {
  273. DEBUG_MSG_P(PSTR("[MQTT] Bad fingerprint\n"));
  274. }
  275. _mqttOnDisconnect();
  276. });
  277. mqtt.onMessage([](char* topic, char* payload, AsyncMqttClientMessageProperties properties, size_t len, size_t index, size_t total) {
  278. _mqttOnMessage(topic, payload, len);
  279. });
  280. mqtt.onSubscribe([](uint16_t packetId, uint8_t qos) {
  281. DEBUG_MSG_P(PSTR("[MQTT] Subscribe ACK for PID %d\n"), packetId);
  282. });
  283. mqtt.onPublish([](uint16_t packetId) {
  284. DEBUG_MSG_P(PSTR("[MQTT] Publish ACK for PID %d\n"), packetId);
  285. });
  286. #else
  287. mqtt.setCallback([](char* topic, byte* payload, unsigned int length) {
  288. _mqttOnMessage(topic, (char *) payload, length);
  289. });
  290. #endif
  291. buildTopics();
  292. }
  293. void mqttLoop() {
  294. static unsigned long lastPeriod = 0;
  295. if (WiFi.status() == WL_CONNECTED) {
  296. if (!mqtt.connected()) {
  297. #if not MQTT_USE_ASYNC
  298. if (_mqttConnected) {
  299. _mqttOnDisconnect();
  300. _mqttConnected = false;
  301. }
  302. #endif
  303. unsigned long currPeriod = millis() / MQTT_RECONNECT_DELAY;
  304. if (currPeriod != lastPeriod) {
  305. lastPeriod = currPeriod;
  306. mqttConnect();
  307. }
  308. #if not MQTT_USE_ASYNC
  309. } else {
  310. mqtt.loop();
  311. #endif
  312. }
  313. }
  314. }