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.

853 lines
24 KiB

8 years ago
8 years ago
6 years ago
7 years ago
6 years ago
7 years ago
6 years ago
6 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-2018 by Xose Pérez <xose dot perez at gmail dot com>
  4. */
  5. #if MQTT_SUPPORT
  6. #include <EEPROM_Rotate.h>
  7. #include <ESP8266WiFi.h>
  8. #include <ESP8266mDNS.h>
  9. #include <ArduinoJson.h>
  10. #include <vector>
  11. #include <Ticker.h>
  12. #if MQTT_USE_ASYNC // Using AsyncMqttClient
  13. #include <AsyncMqttClient.h>
  14. AsyncMqttClient _mqtt;
  15. #else // Using PubSubClient
  16. #include <PubSubClient.h>
  17. PubSubClient _mqtt;
  18. bool _mqtt_connected = false;
  19. WiFiClient _mqtt_client;
  20. #if ASYNC_TCP_SSL_ENABLED
  21. WiFiClientSecure _mqtt_client_secure;
  22. #endif // ASYNC_TCP_SSL_ENABLED
  23. #endif // MQTT_USE_ASYNC
  24. bool _mqtt_enabled = MQTT_ENABLED;
  25. bool _mqtt_use_json = false;
  26. unsigned long _mqtt_reconnect_delay = MQTT_RECONNECT_DELAY_MIN;
  27. unsigned char _mqtt_qos = MQTT_QOS;
  28. bool _mqtt_retain = MQTT_RETAIN;
  29. unsigned long _mqtt_keepalive = MQTT_KEEPALIVE;
  30. String _mqtt_topic;
  31. String _mqtt_topic_json;
  32. String _mqtt_setter;
  33. String _mqtt_getter;
  34. bool _mqtt_forward;
  35. char *_mqtt_user = 0;
  36. char *_mqtt_pass = 0;
  37. char *_mqtt_will;
  38. char *_mqtt_clientid;
  39. #if MQTT_SKIP_RETAINED
  40. unsigned long _mqtt_connected_at = 0;
  41. #endif
  42. std::vector<mqtt_callback_f> _mqtt_callbacks;
  43. typedef struct {
  44. unsigned char parent = 255;
  45. char * topic;
  46. char * message = NULL;
  47. } mqtt_message_t;
  48. std::vector<mqtt_message_t> _mqtt_queue;
  49. Ticker _mqtt_flush_ticker;
  50. // -----------------------------------------------------------------------------
  51. // Private
  52. // -----------------------------------------------------------------------------
  53. void _mqttConnect() {
  54. // Do not connect if disabled
  55. if (!_mqtt_enabled) return;
  56. // Do not connect if already connected
  57. if (_mqtt.connected()) return;
  58. // Check reconnect interval
  59. static unsigned long last = 0;
  60. if (millis() - last < _mqtt_reconnect_delay) return;
  61. last = millis();
  62. // Increase the reconnect delay
  63. _mqtt_reconnect_delay += MQTT_RECONNECT_DELAY_STEP;
  64. if (_mqtt_reconnect_delay > MQTT_RECONNECT_DELAY_MAX) {
  65. _mqtt_reconnect_delay = MQTT_RECONNECT_DELAY_MAX;
  66. }
  67. String h = getSetting("mqttServer", MQTT_SERVER);
  68. #if MDNS_CLIENT_SUPPORT
  69. h = mdnsResolve(h);
  70. #endif
  71. char * host = strdup(h.c_str());
  72. unsigned int port = getSetting("mqttPort", MQTT_PORT).toInt();
  73. if (_mqtt_user) free(_mqtt_user);
  74. if (_mqtt_pass) free(_mqtt_pass);
  75. if (_mqtt_will) free(_mqtt_will);
  76. if (_mqtt_clientid) free(_mqtt_clientid);
  77. _mqtt_user = strdup(getSetting("mqttUser", MQTT_USER).c_str());
  78. _mqtt_pass = strdup(getSetting("mqttPassword", MQTT_PASS).c_str());
  79. _mqtt_will = strdup(mqttTopic(MQTT_TOPIC_STATUS, false).c_str());
  80. _mqtt_clientid = strdup(getSetting("mqttClientID", getIdentifier()).c_str());
  81. DEBUG_MSG_P(PSTR("[MQTT] Connecting to broker at %s:%d\n"), host, port);
  82. #if MQTT_USE_ASYNC
  83. _mqtt.setServer(host, port);
  84. _mqtt.setClientId(_mqtt_clientid);
  85. _mqtt.setKeepAlive(_mqtt_keepalive);
  86. _mqtt.setCleanSession(false);
  87. _mqtt.setWill(_mqtt_will, _mqtt_qos, _mqtt_retain, "0");
  88. if ((strlen(_mqtt_user) > 0) && (strlen(_mqtt_pass) > 0)) {
  89. DEBUG_MSG_P(PSTR("[MQTT] Connecting as user %s\n"), _mqtt_user);
  90. _mqtt.setCredentials(_mqtt_user, _mqtt_pass);
  91. }
  92. #if ASYNC_TCP_SSL_ENABLED
  93. bool secure = getSetting("mqttUseSSL", MQTT_SSL_ENABLED).toInt() == 1;
  94. _mqtt.setSecure(secure);
  95. if (secure) {
  96. DEBUG_MSG_P(PSTR("[MQTT] Using SSL\n"));
  97. unsigned char fp[20] = {0};
  98. if (sslFingerPrintArray(getSetting("mqttFP", MQTT_SSL_FINGERPRINT).c_str(), fp)) {
  99. _mqtt.addServerFingerprint(fp);
  100. } else {
  101. DEBUG_MSG_P(PSTR("[MQTT] Wrong fingerprint\n"));
  102. }
  103. }
  104. #endif // ASYNC_TCP_SSL_ENABLED
  105. DEBUG_MSG_P(PSTR("[MQTT] Client ID: %s\n"), _mqtt_clientid);
  106. DEBUG_MSG_P(PSTR("[MQTT] QoS: %d\n"), _mqtt_qos);
  107. DEBUG_MSG_P(PSTR("[MQTT] Retain flag: %d\n"), _mqtt_retain ? 1 : 0);
  108. DEBUG_MSG_P(PSTR("[MQTT] Keepalive time: %ds\n"), _mqtt_keepalive);
  109. DEBUG_MSG_P(PSTR("[MQTT] Will topic: %s\n"), _mqtt_will);
  110. _mqtt.connect();
  111. #else // not MQTT_USE_ASYNC
  112. bool response = true;
  113. #if ASYNC_TCP_SSL_ENABLED
  114. bool secure = getSetting("mqttUseSSL", MQTT_SSL_ENABLED).toInt() == 1;
  115. if (secure) {
  116. DEBUG_MSG_P(PSTR("[MQTT] Using SSL\n"));
  117. if (_mqtt_client_secure.connect(host, port)) {
  118. char fp[60] = {0};
  119. if (sslFingerPrintChar(getSetting("mqttFP", MQTT_SSL_FINGERPRINT).c_str(), fp)) {
  120. if (_mqtt_client_secure.verify(fp, host)) {
  121. _mqtt.setClient(_mqtt_client_secure);
  122. } else {
  123. DEBUG_MSG_P(PSTR("[MQTT] Invalid fingerprint\n"));
  124. response = false;
  125. }
  126. _mqtt_client_secure.stop();
  127. yield();
  128. } else {
  129. DEBUG_MSG_P(PSTR("[MQTT] Wrong fingerprint\n"));
  130. response = false;
  131. }
  132. } else {
  133. DEBUG_MSG_P(PSTR("[MQTT] Client connection failed\n"));
  134. response = false;
  135. }
  136. } else {
  137. _mqtt.setClient(_mqtt_client);
  138. }
  139. #else // not ASYNC_TCP_SSL_ENABLED
  140. _mqtt.setClient(_mqtt_client);
  141. #endif // ASYNC_TCP_SSL_ENABLED
  142. if (response) {
  143. _mqtt.setServer(host, port);
  144. if ((strlen(_mqtt_user) > 0) && (strlen(_mqtt_pass) > 0)) {
  145. DEBUG_MSG_P(PSTR("[MQTT] Connecting as user %s\n"), _mqtt_user);
  146. response = _mqtt.connect(_mqtt_clientid, _mqtt_user, _mqtt_pass, _mqtt_will, _mqtt_qos, _mqtt_retain, "0");
  147. } else {
  148. response = _mqtt.connect(_mqtt_clientid, _mqtt_will, _mqtt_qos, _mqtt_retain, "0");
  149. }
  150. DEBUG_MSG_P(PSTR("[MQTT] Client ID: %s\n"), _mqtt_clientid);
  151. DEBUG_MSG_P(PSTR("[MQTT] QoS: %d\n"), _mqtt_qos);
  152. DEBUG_MSG_P(PSTR("[MQTT] Retain flag: %d\n"), _mqtt_retain ? 1 : 0);
  153. DEBUG_MSG_P(PSTR("[MQTT] Keepalive time: %ds\n"), _mqtt_keepalive);
  154. DEBUG_MSG_P(PSTR("[MQTT] Will topic: %s\n"), _mqtt_will);
  155. }
  156. if (response) {
  157. _mqttOnConnect();
  158. } else {
  159. DEBUG_MSG_P(PSTR("[MQTT] Connection failed\n"));
  160. }
  161. #endif // MQTT_USE_ASYNC
  162. free(host);
  163. }
  164. void _mqttConfigure() {
  165. // Get base topic
  166. _mqtt_topic = getSetting("mqttTopic", MQTT_TOPIC);
  167. if (_mqtt_topic.endsWith("/")) _mqtt_topic.remove(_mqtt_topic.length()-1);
  168. // Placeholders
  169. _mqtt_topic.replace("{hostname}", getSetting("hostname"));
  170. _mqtt_topic.replace("{magnitude}", "#");
  171. if (_mqtt_topic.indexOf("#") == -1) _mqtt_topic = _mqtt_topic + "/#";
  172. String mac = WiFi.macAddress();
  173. mac.replace(":", "");
  174. _mqtt_topic.replace("{mac}", mac);
  175. // Getters and setters
  176. _mqtt_setter = getSetting("mqttSetter", MQTT_SETTER);
  177. _mqtt_getter = getSetting("mqttGetter", MQTT_GETTER);
  178. _mqtt_forward = !_mqtt_getter.equals(_mqtt_setter);
  179. // MQTT options
  180. _mqtt_qos = getSetting("mqttQoS", MQTT_QOS).toInt();
  181. _mqtt_retain = getSetting("mqttRetain", MQTT_RETAIN).toInt() == 1;
  182. _mqtt_keepalive = getSetting("mqttKeep", MQTT_KEEPALIVE).toInt();
  183. if (getSetting("mqttClientID").length() == 0) delSetting("mqttClientID");
  184. // Enable
  185. if (getSetting("mqttServer", MQTT_SERVER).length() == 0) {
  186. mqttEnabled(false);
  187. } else {
  188. _mqtt_enabled = getSetting("mqttEnabled", MQTT_ENABLED).toInt() == 1;
  189. }
  190. _mqtt_use_json = (getSetting("mqttUseJson", MQTT_USE_JSON).toInt() == 1);
  191. mqttQueueTopic(MQTT_TOPIC_JSON);
  192. _mqtt_reconnect_delay = MQTT_RECONNECT_DELAY_MIN;
  193. }
  194. void _mqttBackwards() {
  195. String mqttTopic = getSetting("mqttTopic", MQTT_TOPIC);
  196. if (mqttTopic.indexOf("{identifier}") > 0) {
  197. mqttTopic.replace("{identifier}", "{hostname}");
  198. setSetting("mqttTopic", mqttTopic);
  199. }
  200. }
  201. unsigned long _mqttNextMessageId() {
  202. static unsigned long id = 0;
  203. // just reboot, get last count from EEPROM
  204. if (id == 0) {
  205. // read id from EEPROM and shift it
  206. id = EEPROMr.read(EEPROM_MESSAGE_ID);
  207. if (id == 0xFF) {
  208. // There was nothing in EEPROM,
  209. // next message is first message
  210. id = 0;
  211. } else {
  212. id = (id << 8) + EEPROMr.read(EEPROM_MESSAGE_ID + 1);
  213. id = (id << 8) + EEPROMr.read(EEPROM_MESSAGE_ID + 2);
  214. id = (id << 8) + EEPROMr.read(EEPROM_MESSAGE_ID + 3);
  215. // Calculate next block and start from there
  216. id = MQTT_MESSAGE_ID_SHIFT * (1 + (id / MQTT_MESSAGE_ID_SHIFT));
  217. }
  218. }
  219. // Save to EEPROM every MQTT_MESSAGE_ID_SHIFT
  220. if (id % MQTT_MESSAGE_ID_SHIFT == 0) {
  221. EEPROMr.write(EEPROM_MESSAGE_ID + 0, (id >> 24) & 0xFF);
  222. EEPROMr.write(EEPROM_MESSAGE_ID + 1, (id >> 16) & 0xFF);
  223. EEPROMr.write(EEPROM_MESSAGE_ID + 2, (id >> 8) & 0xFF);
  224. EEPROMr.write(EEPROM_MESSAGE_ID + 3, (id >> 0) & 0xFF);
  225. EEPROMr.commit();
  226. }
  227. id++;
  228. return id;
  229. }
  230. // -----------------------------------------------------------------------------
  231. // WEB
  232. // -----------------------------------------------------------------------------
  233. #if WEB_SUPPORT
  234. bool _mqttWebSocketOnReceive(const char * key, JsonVariant& value) {
  235. return (strncmp(key, "mqtt", 3) == 0);
  236. }
  237. void _mqttWebSocketOnSend(JsonObject& root) {
  238. root["mqttVisible"] = 1;
  239. root["mqttStatus"] = mqttConnected();
  240. root["mqttEnabled"] = mqttEnabled();
  241. root["mqttServer"] = getSetting("mqttServer", MQTT_SERVER);
  242. root["mqttPort"] = getSetting("mqttPort", MQTT_PORT);
  243. root["mqttUser"] = getSetting("mqttUser", MQTT_USER);
  244. root["mqttClientID"] = getSetting("mqttClientID");
  245. root["mqttPassword"] = getSetting("mqttPassword", MQTT_PASS);
  246. root["mqttKeep"] = _mqtt_keepalive;
  247. root["mqttRetain"] = _mqtt_retain;
  248. root["mqttQoS"] = _mqtt_qos;
  249. #if ASYNC_TCP_SSL_ENABLED
  250. root["mqttsslVisible"] = 1;
  251. root["mqttUseSSL"] = getSetting("mqttUseSSL", MQTT_SSL_ENABLED).toInt() == 1;
  252. root["mqttFP"] = getSetting("mqttFP", MQTT_SSL_FINGERPRINT);
  253. #endif
  254. root["mqttTopic"] = getSetting("mqttTopic", MQTT_TOPIC);
  255. root["mqttUseJson"] = getSetting("mqttUseJson", MQTT_USE_JSON).toInt() == 1;
  256. }
  257. #endif
  258. // -----------------------------------------------------------------------------
  259. // SETTINGS
  260. // -----------------------------------------------------------------------------
  261. #if TERMINAL_SUPPORT
  262. void _mqttInitCommands() {
  263. settingsRegisterCommand(F("MQTT.RESET"), [](Embedis* e) {
  264. _mqttConfigure();
  265. mqttDisconnect();
  266. DEBUG_MSG_P(PSTR("+OK\n"));
  267. });
  268. }
  269. #endif // TERMINAL_SUPPORT
  270. // -----------------------------------------------------------------------------
  271. // MQTT Callbacks
  272. // -----------------------------------------------------------------------------
  273. void _mqttCallback(unsigned int type, const char * topic, const char * payload) {
  274. if (type == MQTT_CONNECT_EVENT) {
  275. // Subscribe to internal action topics
  276. mqttSubscribe(MQTT_TOPIC_ACTION);
  277. // Flag system to send heartbeat
  278. systemSendHeartbeat();
  279. }
  280. if (type == MQTT_MESSAGE_EVENT) {
  281. // Match topic
  282. String t = mqttMagnitude((char *) topic);
  283. // Actions
  284. if (t.equals(MQTT_TOPIC_ACTION)) {
  285. if (strcmp(payload, MQTT_ACTION_RESET) == 0) {
  286. deferredReset(100, CUSTOM_RESET_MQTT);
  287. }
  288. }
  289. }
  290. }
  291. void _mqttOnConnect() {
  292. DEBUG_MSG_P(PSTR("[MQTT] Connected!\n"));
  293. _mqtt_reconnect_delay = MQTT_RECONNECT_DELAY_MIN;
  294. #if MQTT_SKIP_RETAINED
  295. _mqtt_connected_at = millis();
  296. #endif
  297. // Clean subscriptions
  298. mqttUnsubscribeRaw("#");
  299. // Send connect event to subscribers
  300. for (unsigned char i = 0; i < _mqtt_callbacks.size(); i++) {
  301. (_mqtt_callbacks[i])(MQTT_CONNECT_EVENT, NULL, NULL);
  302. }
  303. }
  304. void _mqttOnDisconnect() {
  305. DEBUG_MSG_P(PSTR("[MQTT] Disconnected!\n"));
  306. // Send disconnect event to subscribers
  307. for (unsigned char i = 0; i < _mqtt_callbacks.size(); i++) {
  308. (_mqtt_callbacks[i])(MQTT_DISCONNECT_EVENT, NULL, NULL);
  309. }
  310. }
  311. void _mqttOnMessage(char* topic, char* payload, unsigned int len) {
  312. if (len == 0) return;
  313. char message[len + 1];
  314. strlcpy(message, (char *) payload, len + 1);
  315. #if MQTT_SKIP_RETAINED
  316. if (millis() - _mqtt_connected_at < MQTT_SKIP_TIME) {
  317. DEBUG_MSG_P(PSTR("[MQTT] Received %s => %s - SKIPPED\n"), topic, message);
  318. return;
  319. }
  320. #endif
  321. DEBUG_MSG_P(PSTR("[MQTT] Received %s => %s\n"), topic, message);
  322. // Send message event to subscribers
  323. for (unsigned char i = 0; i < _mqtt_callbacks.size(); i++) {
  324. (_mqtt_callbacks[i])(MQTT_MESSAGE_EVENT, topic, message);
  325. }
  326. }
  327. // -----------------------------------------------------------------------------
  328. // Public API
  329. // -----------------------------------------------------------------------------
  330. /**
  331. Returns the magnitude part of a topic
  332. @param topic the full MQTT topic
  333. @return String object with the magnitude part.
  334. */
  335. String mqttMagnitude(char * topic) {
  336. String pattern = _mqtt_topic + _mqtt_setter;
  337. int position = pattern.indexOf("#");
  338. if (position == -1) return String();
  339. String start = pattern.substring(0, position);
  340. String end = pattern.substring(position + 1);
  341. String magnitude = String(topic);
  342. if (magnitude.startsWith(start) && magnitude.endsWith(end)) {
  343. magnitude.replace(start, "");
  344. magnitude.replace(end, "");
  345. } else {
  346. magnitude = String();
  347. }
  348. return magnitude;
  349. }
  350. /**
  351. Returns a full MQTT topic from the magnitude
  352. @param magnitude the magnitude part of the topic.
  353. @param is_set whether to build a command topic (true)
  354. or a state topic (false).
  355. @return String full MQTT topic.
  356. */
  357. String mqttTopic(const char * magnitude, bool is_set) {
  358. String output = _mqtt_topic;
  359. output.replace("#", magnitude);
  360. output += is_set ? _mqtt_setter : _mqtt_getter;
  361. return output;
  362. }
  363. /**
  364. Returns a full MQTT topic from the magnitude
  365. @param magnitude the magnitude part of the topic.
  366. @param index index of the magnitude when more than one such magnitudes.
  367. @param is_set whether to build a command topic (true)
  368. or a state topic (false).
  369. @return String full MQTT topic.
  370. */
  371. String mqttTopic(const char * magnitude, unsigned int index, bool is_set) {
  372. char buffer[strlen(magnitude)+5];
  373. snprintf_P(buffer, sizeof(buffer), PSTR("%s/%d"), magnitude, index);
  374. return mqttTopic(buffer, is_set);
  375. }
  376. // -----------------------------------------------------------------------------
  377. void mqttSendRaw(const char * topic, const char * message, bool retain) {
  378. if (_mqtt.connected()) {
  379. #if MQTT_USE_ASYNC
  380. unsigned int packetId = _mqtt.publish(topic, _mqtt_qos, retain, message);
  381. DEBUG_MSG_P(PSTR("[MQTT] Sending %s => %s (PID %d)\n"), topic, message, packetId);
  382. #else
  383. _mqtt.publish(topic, message, retain);
  384. DEBUG_MSG_P(PSTR("[MQTT] Sending %s => %s\n"), topic, message);
  385. #endif
  386. }
  387. }
  388. void mqttSendRaw(const char * topic, const char * message) {
  389. mqttSendRaw (topic, message, _mqtt_retain);
  390. }
  391. void mqttSend(const char * topic, const char * message, bool force, bool retain) {
  392. bool useJson = force ? false : _mqtt_use_json;
  393. // Equeue message
  394. if (useJson) {
  395. // Set default queue topic
  396. mqttQueueTopic(MQTT_TOPIC_JSON);
  397. // Enqueue new message
  398. mqttEnqueue(topic, message);
  399. // Reset flush timer
  400. _mqtt_flush_ticker.once_ms(MQTT_USE_JSON_DELAY, mqttFlush);
  401. // Send it right away
  402. } else {
  403. mqttSendRaw(mqttTopic(topic, false).c_str(), message, retain);
  404. }
  405. }
  406. void mqttSend(const char * topic, const char * message, bool force) {
  407. mqttSend(topic, message, force, _mqtt_retain);
  408. }
  409. void mqttSend(const char * topic, const char * message) {
  410. mqttSend(topic, message, false);
  411. }
  412. void mqttSend(const char * topic, unsigned int index, const char * message, bool force, bool retain) {
  413. char buffer[strlen(topic)+5];
  414. snprintf_P(buffer, sizeof(buffer), PSTR("%s/%d"), topic, index);
  415. mqttSend(buffer, message, force, retain);
  416. }
  417. void mqttSend(const char * topic, unsigned int index, const char * message, bool force) {
  418. mqttSend(topic, index, message, force, _mqtt_retain);
  419. }
  420. void mqttSend(const char * topic, unsigned int index, const char * message) {
  421. mqttSend(topic, index, message, false);
  422. }
  423. // -----------------------------------------------------------------------------
  424. unsigned char _mqttBuildTree(JsonObject& root, char parent) {
  425. unsigned char count = 0;
  426. // Add enqueued messages
  427. for (unsigned char i=0; i<_mqtt_queue.size(); i++) {
  428. mqtt_message_t element = _mqtt_queue[i];
  429. if (element.parent == parent) {
  430. ++count;
  431. JsonObject& elements = root.createNestedObject(element.topic);
  432. unsigned char num = _mqttBuildTree(elements, i);
  433. if (0 == num) {
  434. root.set(element.topic, element.message);
  435. }
  436. }
  437. }
  438. return count;
  439. }
  440. void mqttFlush() {
  441. if (!_mqtt.connected()) return;
  442. if (_mqtt_queue.size() == 0) return;
  443. // Build tree recursively
  444. DynamicJsonBuffer jsonBuffer;
  445. JsonObject& root = jsonBuffer.createObject();
  446. _mqttBuildTree(root, 255);
  447. // Add extra propeties
  448. #if NTP_SUPPORT && MQTT_ENQUEUE_DATETIME
  449. if (ntpSynced()) root[MQTT_TOPIC_TIME] = ntpDateTime();
  450. #endif
  451. #if MQTT_ENQUEUE_MAC
  452. root[MQTT_TOPIC_MAC] = WiFi.macAddress();
  453. #endif
  454. #if MQTT_ENQUEUE_HOSTNAME
  455. root[MQTT_TOPIC_HOSTNAME] = getSetting("hostname");
  456. #endif
  457. #if MQTT_ENQUEUE_IP
  458. root[MQTT_TOPIC_IP] = getIP();
  459. #endif
  460. #if MQTT_ENQUEUE_MESSAGE_ID
  461. root[MQTT_TOPIC_MESSAGE_ID] = _mqttNextMessageId();
  462. #endif
  463. // Send
  464. String output;
  465. root.printTo(output);
  466. jsonBuffer.clear();
  467. mqttSendRaw(_mqtt_topic_json.c_str(), output.c_str(), false);
  468. // Clear queue
  469. for (unsigned char i = 0; i < _mqtt_queue.size(); i++) {
  470. mqtt_message_t element = _mqtt_queue[i];
  471. free(element.topic);
  472. if (element.message) {
  473. free(element.message);
  474. }
  475. }
  476. _mqtt_queue.clear();
  477. }
  478. void mqttQueueTopic(const char * topic) {
  479. String t = mqttTopic(topic, false);
  480. if (!t.equals(_mqtt_topic_json)) {
  481. mqttFlush();
  482. _mqtt_topic_json = t;
  483. }
  484. }
  485. int8_t mqttEnqueue(const char * topic, const char * message, unsigned char parent) {
  486. // Queue is not meant to send message "offline"
  487. // We must prevent the queue does not get full while offline
  488. if (!_mqtt.connected()) return -1;
  489. // Force flusing the queue if the MQTT_QUEUE_MAX_SIZE has been reached
  490. if (_mqtt_queue.size() >= MQTT_QUEUE_MAX_SIZE) mqttFlush();
  491. int8_t index = _mqtt_queue.size();
  492. // Enqueue new message
  493. mqtt_message_t element;
  494. element.parent = parent;
  495. element.topic = strdup(topic);
  496. if (NULL != message) {
  497. element.message = strdup(message);
  498. }
  499. _mqtt_queue.push_back(element);
  500. return index;
  501. }
  502. int8_t mqttEnqueue(const char * topic, const char * message) {
  503. return mqttEnqueue(topic, message, 255);
  504. }
  505. // -----------------------------------------------------------------------------
  506. void mqttSubscribeRaw(const char * topic) {
  507. if (_mqtt.connected() && (strlen(topic) > 0)) {
  508. #if MQTT_USE_ASYNC
  509. unsigned int packetId = _mqtt.subscribe(topic, _mqtt_qos);
  510. DEBUG_MSG_P(PSTR("[MQTT] Subscribing to %s (PID %d)\n"), topic, packetId);
  511. #else
  512. _mqtt.subscribe(topic, _mqtt_qos);
  513. DEBUG_MSG_P(PSTR("[MQTT] Subscribing to %s\n"), topic);
  514. #endif
  515. }
  516. }
  517. void mqttSubscribe(const char * topic) {
  518. mqttSubscribeRaw(mqttTopic(topic, true).c_str());
  519. }
  520. void mqttUnsubscribeRaw(const char * topic) {
  521. if (_mqtt.connected() && (strlen(topic) > 0)) {
  522. #if MQTT_USE_ASYNC
  523. unsigned int packetId = _mqtt.unsubscribe(topic);
  524. DEBUG_MSG_P(PSTR("[MQTT] Unsubscribing to %s (PID %d)\n"), topic, packetId);
  525. #else
  526. _mqtt.unsubscribe(topic);
  527. DEBUG_MSG_P(PSTR("[MQTT] Unsubscribing to %s\n"), topic);
  528. #endif
  529. }
  530. }
  531. void mqttUnsubscribe(const char * topic) {
  532. mqttUnsubscribeRaw(mqttTopic(topic, true).c_str());
  533. }
  534. // -----------------------------------------------------------------------------
  535. void mqttEnabled(bool status) {
  536. _mqtt_enabled = status;
  537. }
  538. bool mqttEnabled() {
  539. return _mqtt_enabled;
  540. }
  541. bool mqttConnected() {
  542. return _mqtt.connected();
  543. }
  544. void mqttDisconnect() {
  545. if (_mqtt.connected()) {
  546. DEBUG_MSG_P(PSTR("[MQTT] Disconnecting\n"));
  547. _mqtt.disconnect();
  548. }
  549. }
  550. bool mqttForward() {
  551. return _mqtt_forward;
  552. }
  553. void mqttRegister(mqtt_callback_f callback) {
  554. _mqtt_callbacks.push_back(callback);
  555. }
  556. void mqttSetBroker(IPAddress ip, unsigned int port) {
  557. setSetting("mqttServer", ip.toString());
  558. setSetting("mqttPort", port);
  559. mqttEnabled(MQTT_AUTOCONNECT);
  560. }
  561. void mqttSetBrokerIfNone(IPAddress ip, unsigned int port) {
  562. if (getSetting("mqttServer", MQTT_SERVER).length() == 0) mqttSetBroker(ip, port);
  563. }
  564. void mqttReset() {
  565. _mqttConfigure();
  566. mqttDisconnect();
  567. }
  568. // -----------------------------------------------------------------------------
  569. // Initialization
  570. // -----------------------------------------------------------------------------
  571. void mqttSetup() {
  572. _mqttBackwards();
  573. DEBUG_MSG_P(PSTR("[MQTT] Async %s, SSL %s, Autoconnect %s\n"),
  574. MQTT_USE_ASYNC ? "ENABLED" : "DISABLED",
  575. ASYNC_TCP_SSL_ENABLED ? "ENABLED" : "DISABLED",
  576. MQTT_AUTOCONNECT ? "ENABLED" : "DISABLED"
  577. );
  578. #if MQTT_USE_ASYNC
  579. _mqtt.onConnect([](bool sessionPresent) {
  580. _mqttOnConnect();
  581. });
  582. _mqtt.onDisconnect([](AsyncMqttClientDisconnectReason reason) {
  583. if (reason == AsyncMqttClientDisconnectReason::TCP_DISCONNECTED) {
  584. DEBUG_MSG_P(PSTR("[MQTT] TCP Disconnected\n"));
  585. }
  586. if (reason == AsyncMqttClientDisconnectReason::MQTT_IDENTIFIER_REJECTED) {
  587. DEBUG_MSG_P(PSTR("[MQTT] Identifier Rejected\n"));
  588. }
  589. if (reason == AsyncMqttClientDisconnectReason::MQTT_SERVER_UNAVAILABLE) {
  590. DEBUG_MSG_P(PSTR("[MQTT] Server unavailable\n"));
  591. }
  592. if (reason == AsyncMqttClientDisconnectReason::MQTT_MALFORMED_CREDENTIALS) {
  593. DEBUG_MSG_P(PSTR("[MQTT] Malformed credentials\n"));
  594. }
  595. if (reason == AsyncMqttClientDisconnectReason::MQTT_NOT_AUTHORIZED) {
  596. DEBUG_MSG_P(PSTR("[MQTT] Not authorized\n"));
  597. }
  598. #if ASYNC_TCP_SSL_ENABLED
  599. if (reason == AsyncMqttClientDisconnectReason::TLS_BAD_FINGERPRINT) {
  600. DEBUG_MSG_P(PSTR("[MQTT] Bad fingerprint\n"));
  601. }
  602. #endif
  603. _mqttOnDisconnect();
  604. });
  605. _mqtt.onMessage([](char* topic, char* payload, AsyncMqttClientMessageProperties properties, size_t len, size_t index, size_t total) {
  606. _mqttOnMessage(topic, payload, len);
  607. });
  608. _mqtt.onSubscribe([](uint16_t packetId, uint8_t qos) {
  609. DEBUG_MSG_P(PSTR("[MQTT] Subscribe ACK for PID %d\n"), packetId);
  610. });
  611. _mqtt.onPublish([](uint16_t packetId) {
  612. DEBUG_MSG_P(PSTR("[MQTT] Publish ACK for PID %d\n"), packetId);
  613. });
  614. #else // not MQTT_USE_ASYNC
  615. _mqtt.setCallback([](char* topic, byte* payload, unsigned int length) {
  616. _mqttOnMessage(topic, (char *) payload, length);
  617. });
  618. #endif // MQTT_USE_ASYNC
  619. _mqttConfigure();
  620. mqttRegister(_mqttCallback);
  621. #if WEB_SUPPORT
  622. wsOnSendRegister(_mqttWebSocketOnSend);
  623. wsOnReceiveRegister(_mqttWebSocketOnReceive);
  624. #endif
  625. #if TERMINAL_SUPPORT
  626. _mqttInitCommands();
  627. #endif
  628. // Main callbacks
  629. espurnaRegisterLoop(mqttLoop);
  630. espurnaRegisterReload(_mqttConfigure);
  631. }
  632. void mqttLoop() {
  633. if (WiFi.status() != WL_CONNECTED) return;
  634. #if MQTT_USE_ASYNC
  635. _mqttConnect();
  636. #else // not MQTT_USE_ASYNC
  637. if (_mqtt.connected()) {
  638. _mqtt.loop();
  639. } else {
  640. if (_mqtt_connected) {
  641. _mqttOnDisconnect();
  642. _mqtt_connected = false;
  643. }
  644. _mqttConnect();
  645. }
  646. #endif
  647. }
  648. #endif // MQTT_SUPPORT