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.

791 lines
23 KiB

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