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.

621 lines
18 KiB

8 years ago
8 years ago
7 years ago
7 years ago
7 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. #if MQTT_SUPPORT
  6. #include <ESP8266WiFi.h>
  7. #include <ESP8266mDNS.h>
  8. #include <ArduinoJson.h>
  9. #include <vector>
  10. #include <Ticker.h>
  11. #if MQTT_USE_ASYNC // Using AsyncMqttClient
  12. #include <AsyncMqttClient.h>
  13. AsyncMqttClient _mqtt;
  14. #else // Using PubSubClient
  15. #include <PubSubClient.h>
  16. PubSubClient _mqtt;
  17. bool _mqtt_connected = false;
  18. WiFiClient _mqtt_client;
  19. #if ASYNC_TCP_SSL_ENABLED
  20. WiFiClientSecure _mqtt_client_secure;
  21. #endif // ASYNC_TCP_SSL_ENABLED
  22. #endif // MQTT_USE_ASYNC
  23. bool _mqtt_enabled = MQTT_ENABLED;
  24. bool _mqtt_use_json = false;
  25. unsigned long _mqtt_reconnect_delay = MQTT_RECONNECT_DELAY_MIN;
  26. unsigned char _mqtt_qos = MQTT_QOS;
  27. bool _mqtt_retain = MQTT_RETAIN;
  28. unsigned char _mqtt_keepalive = MQTT_KEEPALIVE;
  29. String _mqtt_topic;
  30. String _mqtt_setter;
  31. String _mqtt_getter;
  32. bool _mqtt_forward;
  33. char *_mqtt_user = 0;
  34. char *_mqtt_pass = 0;
  35. char *_mqtt_will;
  36. #if MQTT_SKIP_RETAINED
  37. unsigned long _mqtt_connected_at = 0;
  38. #endif
  39. std::vector<mqtt_callback_f> _mqtt_callbacks;
  40. typedef struct {
  41. char * topic;
  42. char * message;
  43. } mqtt_message_t;
  44. std::vector<mqtt_message_t> _mqtt_queue;
  45. Ticker _mqtt_flush_ticker;
  46. // -----------------------------------------------------------------------------
  47. // Public API
  48. // -----------------------------------------------------------------------------
  49. bool mqttConnected() {
  50. return _mqtt.connected();
  51. }
  52. void mqttDisconnect() {
  53. if (_mqtt.connected()) {
  54. DEBUG_MSG_P(PSTR("[MQTT] Disconnecting\n"));
  55. _mqtt.disconnect();
  56. }
  57. }
  58. bool mqttForward() {
  59. return _mqtt_forward;
  60. }
  61. String mqttSubtopic(char * topic) {
  62. String response;
  63. String t = String(topic);
  64. if (t.startsWith(_mqtt_topic) && t.endsWith(_mqtt_setter)) {
  65. response = t.substring(_mqtt_topic.length(), t.length() - _mqtt_setter.length());
  66. }
  67. return response;
  68. }
  69. void mqttSendRaw(const char * topic, const char * message) {
  70. if (_mqtt.connected()) {
  71. #if MQTT_USE_ASYNC
  72. unsigned int packetId = _mqtt.publish(topic, _mqtt_qos, _mqtt_retain, message);
  73. DEBUG_MSG_P(PSTR("[MQTT] Sending %s => %s (PID %d)\n"), topic, message, packetId);
  74. #else
  75. _mqtt.publish(topic, message, _mqtt_retain);
  76. DEBUG_MSG_P(PSTR("[MQTT] Sending %s => %s\n"), topic, message);
  77. #endif
  78. }
  79. }
  80. String getTopic(const char * topic, bool set) {
  81. String output = _mqtt_topic + String(topic);
  82. if (set) output += _mqtt_setter;
  83. return output;
  84. }
  85. String getTopic(const char * topic, unsigned int index, bool set) {
  86. char buffer[strlen(topic)+5];
  87. snprintf_P(buffer, sizeof(buffer), PSTR("%s/%d"), topic, index);
  88. return getTopic(buffer, set);
  89. }
  90. void _mqttFlush() {
  91. if (_mqtt_queue.size() == 0) return;
  92. DynamicJsonBuffer jsonBuffer;
  93. JsonObject& root = jsonBuffer.createObject();
  94. for (unsigned char i=0; i<_mqtt_queue.size(); i++) {
  95. mqtt_message_t element = _mqtt_queue[i];
  96. root[element.topic] = element.message;
  97. }
  98. #if NTP_SUPPORT
  99. if (ntpConnected()) root[MQTT_TOPIC_TIME] = ntpDateTime();
  100. #endif
  101. root[MQTT_TOPIC_HOSTNAME] = getSetting("hostname");
  102. root[MQTT_TOPIC_IP] = getIP();
  103. String output;
  104. root.printTo(output);
  105. String path = _mqtt_topic + String(MQTT_TOPIC_JSON);
  106. mqttSendRaw(path.c_str(), output.c_str());
  107. for (unsigned char i = 0; i < _mqtt_queue.size(); i++) {
  108. mqtt_message_t element = _mqtt_queue[i];
  109. free(element.topic);
  110. free(element.message);
  111. }
  112. _mqtt_queue.clear();
  113. }
  114. void mqttSend(const char * topic, const char * message, bool force) {
  115. bool useJson = force ? false : _mqtt_use_json;
  116. if (useJson) {
  117. if (_mqtt_queue.size() >= MQTT_QUEUE_MAX_SIZE) _mqttFlush();
  118. mqtt_message_t element;
  119. element.topic = strdup(topic);
  120. element.message = strdup(message);
  121. _mqtt_queue.push_back(element);
  122. _mqtt_flush_ticker.once_ms(MQTT_USE_JSON_DELAY, _mqttFlush);
  123. } else {
  124. String path = _mqtt_topic + String(topic) + _mqtt_getter;
  125. mqttSendRaw(path.c_str(), message);
  126. }
  127. }
  128. void mqttSend(const char * topic, const char * message) {
  129. mqttSend(topic, message, false);
  130. }
  131. void mqttSend(const char * topic, unsigned int index, const char * message, bool force) {
  132. char buffer[strlen(topic)+5];
  133. snprintf_P(buffer, sizeof(buffer), PSTR("%s/%d"), topic, index);
  134. mqttSend(buffer, message, force);
  135. }
  136. void mqttSend(const char * topic, unsigned int index, const char * message) {
  137. mqttSend(topic, index, message, false);
  138. }
  139. void mqttSubscribeRaw(const char * topic) {
  140. if (_mqtt.connected() && (strlen(topic) > 0)) {
  141. #if MQTT_USE_ASYNC
  142. unsigned int packetId = _mqtt.subscribe(topic, _mqtt_qos);
  143. DEBUG_MSG_P(PSTR("[MQTT] Subscribing to %s (PID %d)\n"), topic, packetId);
  144. #else
  145. _mqtt.subscribe(topic, _mqtt_qos);
  146. DEBUG_MSG_P(PSTR("[MQTT] Subscribing to %s\n"), topic);
  147. #endif
  148. }
  149. }
  150. void mqttSubscribe(const char * topic) {
  151. String path = _mqtt_topic + String(topic) + _mqtt_setter;
  152. mqttSubscribeRaw(path.c_str());
  153. }
  154. void mqttUnsubscribeRaw(const char * topic) {
  155. if (_mqtt.connected() && (strlen(topic) > 0)) {
  156. #if MQTT_USE_ASYNC
  157. unsigned int packetId = _mqtt.unsubscribe(topic);
  158. DEBUG_MSG_P(PSTR("[MQTT] Unsubscribing to %s (PID %d)\n"), topic, packetId);
  159. #else
  160. _mqtt.unsubscribe(topic);
  161. DEBUG_MSG_P(PSTR("[MQTT] Unsubscribing to %s\n"), topic);
  162. #endif
  163. }
  164. }
  165. void mqttRegister(mqtt_callback_f callback) {
  166. _mqtt_callbacks.push_back(callback);
  167. }
  168. // -----------------------------------------------------------------------------
  169. // Callbacks
  170. // -----------------------------------------------------------------------------
  171. void _mqttWebSocketOnSend(JsonObject& root) {
  172. root["mqttVisible"] = 1;
  173. root["mqttStatus"] = mqttConnected();
  174. root["mqttEnabled"] = mqttEnabled();
  175. root["mqttServer"] = getSetting("mqttServer", MQTT_SERVER);
  176. root["mqttPort"] = getSetting("mqttPort", MQTT_PORT);
  177. root["mqttUser"] = getSetting("mqttUser");
  178. root["mqttPassword"] = getSetting("mqttPassword");
  179. root["mqttKeep"] = _mqtt_keepalive;
  180. root["mqttRetain"] = _mqtt_retain;
  181. root["mqttQoS"] = _mqtt_qos;
  182. #if ASYNC_TCP_SSL_ENABLED
  183. root["mqttsslVisible"] = 1;
  184. root["mqttUseSSL"] = getSetting("mqttUseSSL", 0).toInt() == 1;
  185. root["mqttFP"] = getSetting("mqttFP");
  186. #endif
  187. root["mqttTopic"] = getSetting("mqttTopic", MQTT_TOPIC);
  188. root["mqttUseJson"] = getSetting("mqttUseJson", MQTT_USE_JSON).toInt() == 1;
  189. }
  190. void _mqttCallback(unsigned int type, const char * topic, const char * payload) {
  191. if (type == MQTT_CONNECT_EVENT) {
  192. mqttSubscribe(MQTT_TOPIC_ACTION);
  193. }
  194. if (type == MQTT_MESSAGE_EVENT) {
  195. // Match topic
  196. String t = mqttSubtopic((char *) topic);
  197. // Actions
  198. if (t.equals(MQTT_TOPIC_ACTION)) {
  199. if (strcmp(payload, MQTT_ACTION_RESET) == 0) {
  200. deferredReset(100, CUSTOM_RESET_MQTT);
  201. }
  202. }
  203. }
  204. }
  205. void _mqttOnConnect() {
  206. DEBUG_MSG_P(PSTR("[MQTT] Connected!\n"));
  207. _mqtt_reconnect_delay = MQTT_RECONNECT_DELAY_MIN;
  208. #if MQTT_SKIP_RETAINED
  209. _mqtt_connected_at = millis();
  210. #endif
  211. // Clean subscriptions
  212. mqttUnsubscribeRaw("#");
  213. // Send connect event to subscribers
  214. for (unsigned char i = 0; i < _mqtt_callbacks.size(); i++) {
  215. (_mqtt_callbacks[i])(MQTT_CONNECT_EVENT, NULL, NULL);
  216. }
  217. }
  218. void _mqttOnDisconnect() {
  219. DEBUG_MSG_P(PSTR("[MQTT] Disconnected!\n"));
  220. // Send disconnect event to subscribers
  221. for (unsigned char i = 0; i < _mqtt_callbacks.size(); i++) {
  222. (_mqtt_callbacks[i])(MQTT_DISCONNECT_EVENT, NULL, NULL);
  223. }
  224. }
  225. void _mqttOnMessage(char* topic, char* payload, unsigned int len) {
  226. if (len == 0) return;
  227. char message[len + 1];
  228. strlcpy(message, (char *) payload, len + 1);
  229. #if MQTT_SKIP_RETAINED
  230. if (millis() - _mqtt_connected_at < MQTT_SKIP_TIME) {
  231. DEBUG_MSG_P(PSTR("[MQTT] Received %s => %s - SKIPPED\n"), topic, message);
  232. return;
  233. }
  234. #endif
  235. DEBUG_MSG_P(PSTR("[MQTT] Received %s => %s\n"), topic, message);
  236. // Send message event to subscribers
  237. for (unsigned char i = 0; i < _mqtt_callbacks.size(); i++) {
  238. (_mqtt_callbacks[i])(MQTT_MESSAGE_EVENT, topic, message);
  239. }
  240. }
  241. #if MQTT_USE_ASYNC
  242. bool mqttFormatFP(const char * fingerprint, unsigned char * bytearray) {
  243. // check length (20 2-character digits ':' or ' ' separated => 20*2+19 = 59)
  244. if (strlen(fingerprint) != 59) return false;
  245. DEBUG_MSG_P(PSTR("[MQTT] Fingerprint %s\n"), fingerprint);
  246. // walk the fingerprint
  247. for (unsigned int i=0; i<20; i++) {
  248. bytearray[i] = strtol(fingerprint + 3*i, NULL, 16);
  249. }
  250. return true;
  251. }
  252. #else
  253. bool mqttFormatFP(const char * fingerprint, char * destination) {
  254. // check length (20 2-character digits ':' or ' ' separated => 20*2+19 = 59)
  255. if (strlen(fingerprint) != 59) return false;
  256. DEBUG_MSG_P(PSTR("[MQTT] Fingerprint %s\n"), fingerprint);
  257. // copy it
  258. strncpy(destination, fingerprint, 59);
  259. // walk the fingerprint replacing ':' for ' '
  260. for (unsigned char i = 0; i<59; i++) {
  261. if (destination[i] == ':') destination[i] = ' ';
  262. }
  263. return true;
  264. }
  265. #endif
  266. void mqttEnabled(bool status) {
  267. _mqtt_enabled = status;
  268. setSetting("mqttEnabled", status ? 1 : 0);
  269. }
  270. bool mqttEnabled() {
  271. return _mqtt_enabled;
  272. }
  273. void mqttConnect() {
  274. // Do not connect if disabled
  275. if (!_mqtt_enabled) return;
  276. // Do not connect if already connected
  277. if (_mqtt.connected()) return;
  278. // Check reconnect interval
  279. static unsigned long last = 0;
  280. if (millis() - last < _mqtt_reconnect_delay) return;
  281. last = millis();
  282. // Increase the reconnect delay
  283. _mqtt_reconnect_delay += MQTT_RECONNECT_DELAY_STEP;
  284. if (_mqtt_reconnect_delay > MQTT_RECONNECT_DELAY_MAX) {
  285. _mqtt_reconnect_delay = MQTT_RECONNECT_DELAY_MAX;
  286. }
  287. char * host = strdup(getSetting("mqttServer", MQTT_SERVER).c_str());
  288. if (strlen(host) == 0) return;
  289. unsigned int port = getSetting("mqttPort", MQTT_PORT).toInt();
  290. if (_mqtt_user) free(_mqtt_user);
  291. if (_mqtt_pass) free(_mqtt_pass);
  292. if (_mqtt_will) free(_mqtt_will);
  293. _mqtt_user = strdup(getSetting("mqttUser", MQTT_USER).c_str());
  294. _mqtt_pass = strdup(getSetting("mqttPassword", MQTT_PASS).c_str());
  295. _mqtt_will = strdup((_mqtt_topic + MQTT_TOPIC_STATUS).c_str());
  296. DEBUG_MSG_P(PSTR("[MQTT] Connecting to broker at %s:%d\n"), host, port);
  297. #if MQTT_USE_ASYNC
  298. _mqtt.setServer(host, port);
  299. _mqtt.setKeepAlive(_mqtt_keepalive).setCleanSession(false);
  300. _mqtt.setWill(_mqtt_will, _mqtt_qos, _mqtt_retain, "0");
  301. if ((strlen(_mqtt_user) > 0) && (strlen(_mqtt_pass) > 0)) {
  302. DEBUG_MSG_P(PSTR("[MQTT] Connecting as user %s\n"), _mqtt_user);
  303. _mqtt.setCredentials(_mqtt_user, _mqtt_pass);
  304. }
  305. #if ASYNC_TCP_SSL_ENABLED
  306. bool secure = getSetting("mqttUseSSL", MQTT_SSL_ENABLED).toInt() == 1;
  307. _mqtt.setSecure(secure);
  308. if (secure) {
  309. DEBUG_MSG_P(PSTR("[MQTT] Using SSL\n"));
  310. unsigned char fp[20] = {0};
  311. if (mqttFormatFP(getSetting("mqttFP", MQTT_SSL_FINGERPRINT).c_str(), fp)) {
  312. _mqtt.addServerFingerprint(fp);
  313. } else {
  314. DEBUG_MSG_P(PSTR("[MQTT] Wrong fingerprint\n"));
  315. }
  316. }
  317. #endif // ASYNC_TCP_SSL_ENABLED
  318. DEBUG_MSG_P(PSTR("[MQTT] Will topic: %s\n"), _mqtt_will);
  319. DEBUG_MSG_P(PSTR("[MQTT] QoS: %d\n"), _mqtt_qos);
  320. DEBUG_MSG_P(PSTR("[MQTT] Retain flag: %d\n"), _mqtt_retain ? 1 : 0);
  321. _mqtt.connect();
  322. #else // not MQTT_USE_ASYNC
  323. bool response = true;
  324. #if ASYNC_TCP_SSL_ENABLED
  325. bool secure = getSetting("mqttUseSSL", MQTT_SSL_ENABLED).toInt() == 1;
  326. if (secure) {
  327. DEBUG_MSG_P(PSTR("[MQTT] Using SSL\n"));
  328. if (_mqtt_client_secure.connect(host, port)) {
  329. char fp[60] = {0};
  330. if (mqttFormatFP(getSetting("mqttFP", MQTT_SSL_FINGERPRINT).c_str(), fp)) {
  331. if (_mqtt_client_secure.verify(fp, host)) {
  332. _mqtt.setClient(_mqtt_client_secure);
  333. } else {
  334. DEBUG_MSG_P(PSTR("[MQTT] Invalid fingerprint\n"));
  335. response = false;
  336. }
  337. _mqtt_client_secure.stop();
  338. yield();
  339. } else {
  340. DEBUG_MSG_P(PSTR("[MQTT] Wrong fingerprint\n"));
  341. response = false;
  342. }
  343. } else {
  344. DEBUG_MSG_P(PSTR("[MQTT] Client connection failed\n"));
  345. response = false;
  346. }
  347. } else {
  348. _mqtt.setClient(_mqtt_client);
  349. }
  350. #else // not ASYNC_TCP_SSL_ENABLED
  351. _mqtt.setClient(_mqtt_client);
  352. #endif // ASYNC_TCP_SSL_ENABLED
  353. if (response) {
  354. _mqtt.setServer(host, port);
  355. if ((strlen(_mqtt_user) > 0) && (strlen(_mqtt_pass) > 0)) {
  356. DEBUG_MSG_P(PSTR("[MQTT] Connecting as user %s\n"), _mqtt_user);
  357. response = _mqtt.connect(getIdentifier().c_str(), _mqtt_user, _mqtt_pass, _mqtt_will, _mqtt_qos, _mqtt_retain, "0");
  358. } else {
  359. response = _mqtt.connect(getIdentifier().c_str(), _mqtt_will, _mqtt_qos, _mqtt_retain, "0");
  360. }
  361. DEBUG_MSG_P(PSTR("[MQTT] Will topic: %s\n"), _mqtt_will);
  362. DEBUG_MSG_P(PSTR("[MQTT] QoS: %d\n"), _mqtt_qos);
  363. DEBUG_MSG_P(PSTR("[MQTT] Retain flag: %d\n"), _mqtt_retain ? 1 : 0);
  364. }
  365. if (response) {
  366. _mqttOnConnect();
  367. } else {
  368. DEBUG_MSG_P(PSTR("[MQTT] Connection failed\n"));
  369. }
  370. #endif // MQTT_USE_ASYNC
  371. free(host);
  372. }
  373. void mqttConfigure() {
  374. // Replace identifier
  375. _mqtt_topic = getSetting("mqttTopic", MQTT_TOPIC);
  376. _mqtt_topic.replace("{identifier}", getSetting("hostname"));
  377. if (!_mqtt_topic.endsWith("/")) _mqtt_topic = _mqtt_topic + "/";
  378. // Getters and setters
  379. _mqtt_setter = getSetting("mqttSetter", MQTT_USE_SETTER);
  380. _mqtt_getter = getSetting("mqttGetter", MQTT_USE_GETTER);
  381. _mqtt_forward = !_mqtt_getter.equals(_mqtt_setter);
  382. // MQTT options
  383. _mqtt_qos = getSetting("mqttQoS", MQTT_QOS).toInt();
  384. _mqtt_retain = getSetting("mqttRetain", MQTT_RETAIN).toInt() == 1;
  385. _mqtt_keepalive = getSetting("mqttKeep", MQTT_KEEPALIVE).toInt();
  386. // Enable
  387. if (getSetting("mqttServer", MQTT_SERVER).length() == 0) {
  388. mqttEnabled(false);
  389. } else {
  390. _mqtt_enabled = getSetting("mqttEnabled", MQTT_ENABLED).toInt() == 1;
  391. }
  392. _mqtt_use_json = (getSetting("mqttUseJson", MQTT_USE_JSON).toInt() == 1);
  393. _mqtt_reconnect_delay = MQTT_RECONNECT_DELAY_MIN;
  394. }
  395. void mqttSetBroker(IPAddress ip, unsigned int port) {
  396. setSetting("mqttServer", ip.toString());
  397. setSetting("mqttPort", port);
  398. mqttEnabled(MQTT_AUTOCONNECT);
  399. }
  400. void mqttSetBrokerIfNone(IPAddress ip, unsigned int port) {
  401. if (!hasSetting("mqttServer")) mqttSetBroker(ip, port);
  402. }
  403. void mqttSetup() {
  404. DEBUG_MSG_P(PSTR("[MQTT] MQTT_USE_ASYNC = %d\n"), MQTT_USE_ASYNC);
  405. DEBUG_MSG_P(PSTR("[MQTT] MQTT_AUTOCONNECT = %d\n"), MQTT_AUTOCONNECT);
  406. #if MQTT_USE_ASYNC
  407. _mqtt.onConnect([](bool sessionPresent) {
  408. _mqttOnConnect();
  409. });
  410. _mqtt.onDisconnect([](AsyncMqttClientDisconnectReason reason) {
  411. if (reason == AsyncMqttClientDisconnectReason::TCP_DISCONNECTED) {
  412. DEBUG_MSG_P(PSTR("[MQTT] TCP Disconnected\n"));
  413. }
  414. if (reason == AsyncMqttClientDisconnectReason::MQTT_IDENTIFIER_REJECTED) {
  415. DEBUG_MSG_P(PSTR("[MQTT] Identifier Rejected\n"));
  416. }
  417. if (reason == AsyncMqttClientDisconnectReason::MQTT_SERVER_UNAVAILABLE) {
  418. DEBUG_MSG_P(PSTR("[MQTT] Server unavailable\n"));
  419. }
  420. if (reason == AsyncMqttClientDisconnectReason::MQTT_MALFORMED_CREDENTIALS) {
  421. DEBUG_MSG_P(PSTR("[MQTT] Malformed credentials\n"));
  422. }
  423. if (reason == AsyncMqttClientDisconnectReason::MQTT_NOT_AUTHORIZED) {
  424. DEBUG_MSG_P(PSTR("[MQTT] Not authorized\n"));
  425. }
  426. #if ASYNC_TCP_SSL_ENABLED
  427. if (reason == AsyncMqttClientDisconnectReason::TLS_BAD_FINGERPRINT) {
  428. DEBUG_MSG_P(PSTR("[MQTT] Bad fingerprint\n"));
  429. }
  430. #endif
  431. _mqttOnDisconnect();
  432. });
  433. _mqtt.onMessage([](char* topic, char* payload, AsyncMqttClientMessageProperties properties, size_t len, size_t index, size_t total) {
  434. _mqttOnMessage(topic, payload, len);
  435. });
  436. _mqtt.onSubscribe([](uint16_t packetId, uint8_t qos) {
  437. DEBUG_MSG_P(PSTR("[MQTT] Subscribe ACK for PID %d\n"), packetId);
  438. });
  439. _mqtt.onPublish([](uint16_t packetId) {
  440. DEBUG_MSG_P(PSTR("[MQTT] Publish ACK for PID %d\n"), packetId);
  441. });
  442. #else // not MQTT_USE_ASYNC
  443. DEBUG_MSG_P(PSTR("[MQTT] Using SYNC MQTT library\n"));
  444. _mqtt.setCallback([](char* topic, byte* payload, unsigned int length) {
  445. _mqttOnMessage(topic, (char *) payload, length);
  446. });
  447. #endif // MQTT_USE_ASYNC
  448. mqttConfigure();
  449. mqttRegister(_mqttCallback);
  450. #if WEB_SUPPORT
  451. wsOnSendRegister(_mqttWebSocketOnSend);
  452. #endif
  453. }
  454. void mqttLoop() {
  455. if (WiFi.status() != WL_CONNECTED) return;
  456. #if MQTT_USE_ASYNC
  457. mqttConnect();
  458. #else // not MQTT_USE_ASYNC
  459. if (_mqtt.connected()) {
  460. _mqtt.loop();
  461. } else {
  462. if (_mqtt_connected) {
  463. _mqttOnDisconnect();
  464. _mqtt_connected = false;
  465. }
  466. mqttConnect();
  467. }
  468. #endif
  469. }
  470. #endif // MQTT_SUPPORT