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.

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