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.

1039 lines
30 KiB

8 years ago
8 years ago
7 years ago
6 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-2019 by Xose Pérez <xose dot perez at gmail dot com>
  4. Updated secure client support by Niek van der Maas < mail at niekvandermaas dot nl>
  5. */
  6. #if MQTT_SUPPORT
  7. #include <EEPROM_Rotate.h>
  8. #include <ESP8266WiFi.h>
  9. #include <ESP8266mDNS.h>
  10. #include <ArduinoJson.h>
  11. #include <vector>
  12. #include <utility>
  13. #include <Ticker.h>
  14. #include <TimeLib.h>
  15. #include "libs/SecureClientHelpers.h"
  16. #if MQTT_LIBRARY == MQTT_LIBRARY_ASYNCMQTTCLIENT
  17. AsyncMqttClient _mqtt;
  18. #else // MQTT_LIBRARY_ARDUINOMQTT / MQTT_LIBRARY_PUBSUBCLIENT
  19. WiFiClient _mqtt_client;
  20. #if SECURE_CLIENT != SECURE_CLIENT_NONE
  21. std::unique_ptr<SecureClient> _mqtt_client_secure = nullptr;
  22. #if MQTT_SECURE_CLIENT_INCLUDE_CA
  23. #include "static/mqtt_client_trusted_root_ca.h" // Assumes this header file defines a _mqtt_client_trusted_root_ca[] PROGMEM = "...PEM data..."
  24. #else
  25. #include "static/letsencrypt_isrgroot_pem.h" // Default to LetsEncrypt X3 certificate
  26. #define _mqtt_client_trusted_root_ca _ssl_letsencrypt_isrg_x3_ca
  27. #endif // MQTT_SECURE_CLIENT_INCLUDE_CA
  28. #endif // SECURE_CLIENT != SECURE_CLIENT_NONE
  29. #if MQTT_LIBRARY == MQTT_LIBRARY_ARDUINOMQTT
  30. #ifdef MQTT_MAX_PACKET_SIZE
  31. MQTTClient _mqtt(MQTT_MAX_PACKET_SIZE);
  32. #else
  33. MQTTClient _mqtt;
  34. #endif
  35. #elif MQTT_LIBRARY == MQTT_LIBRARY_PUBSUBCLIENT
  36. PubSubClient _mqtt;
  37. #endif
  38. #endif // MQTT_LIBRARY == MQTT_ASYNCMQTTCLIENT
  39. bool _mqtt_enabled = MQTT_ENABLED;
  40. bool _mqtt_use_json = false;
  41. unsigned long _mqtt_reconnect_delay = MQTT_RECONNECT_DELAY_MIN;
  42. unsigned long _mqtt_last_connection = 0;
  43. bool _mqtt_connected = false;
  44. bool _mqtt_connecting = false;
  45. unsigned char _mqtt_qos = MQTT_QOS;
  46. bool _mqtt_retain = MQTT_RETAIN;
  47. unsigned long _mqtt_keepalive = MQTT_KEEPALIVE;
  48. String _mqtt_topic;
  49. String _mqtt_topic_json;
  50. String _mqtt_setter;
  51. String _mqtt_getter;
  52. bool _mqtt_forward;
  53. String _mqtt_user;
  54. String _mqtt_pass;
  55. String _mqtt_will;
  56. String _mqtt_server;
  57. uint16_t _mqtt_port;
  58. String _mqtt_clientid;
  59. String _mqtt_payload_online;
  60. String _mqtt_payload_offline;
  61. std::vector<mqtt_callback_f> _mqtt_callbacks;
  62. struct mqtt_message_t {
  63. static const unsigned char END = 255;
  64. unsigned char parent = END;
  65. char * topic;
  66. char * message = NULL;
  67. };
  68. std::vector<mqtt_message_t> _mqtt_queue;
  69. Ticker _mqtt_flush_ticker;
  70. // -----------------------------------------------------------------------------
  71. // Private
  72. // -----------------------------------------------------------------------------
  73. #if SECURE_CLIENT == SECURE_CLIENT_AXTLS
  74. SecureClientConfig _mqtt_sc_config {
  75. "MQTT",
  76. []() -> String {
  77. return _mqtt_server;
  78. },
  79. []() -> int {
  80. return getSetting("mqttScCheck", MQTT_SECURE_CLIENT_CHECK).toInt();
  81. },
  82. []() -> String {
  83. return getSetting("mqttFP", MQTT_SSL_FINGERPRINT);
  84. },
  85. true
  86. };
  87. #endif
  88. #if SECURE_CLIENT == SECURE_CLIENT_BEARSSL
  89. SecureClientConfig _mqtt_sc_config {
  90. "MQTT",
  91. []() -> int {
  92. return getSetting("mqttScCheck", MQTT_SECURE_CLIENT_CHECK).toInt();
  93. },
  94. []() -> PGM_P {
  95. return _mqtt_client_trusted_root_ca;
  96. },
  97. []() -> String {
  98. return getSetting("mqttFP", MQTT_SSL_FINGERPRINT);
  99. },
  100. []() -> uint16_t {
  101. return getSetting("mqttScMFLN", MQTT_SECURE_CLIENT_MFLN).toInt();
  102. },
  103. true
  104. };
  105. #endif
  106. #if MQTT_LIBRARY == MQTT_LIBRARY_ASYNCMQTTCLIENT
  107. void _mqttSetupAsyncClient(bool secure = false) {
  108. _mqtt.setServer(_mqtt_server.c_str(), _mqtt_port);
  109. _mqtt.setClientId(_mqtt_clientid.c_str());
  110. _mqtt.setKeepAlive(_mqtt_keepalive);
  111. _mqtt.setCleanSession(false);
  112. _mqtt.setWill(_mqtt_will.c_str(), _mqtt_qos, _mqtt_retain, _mqtt_payload_offline.c_str());
  113. if (_mqtt_user.length() && _mqtt_pass.length()) {
  114. DEBUG_MSG_P(PSTR("[MQTT] Connecting as user %s\n"), _mqtt_user.c_str());
  115. _mqtt.setCredentials(_mqtt_user.c_str(), _mqtt_pass.c_str());
  116. }
  117. #if SECURE_CLIENT != SECURE_CLIENT_NONE
  118. if (secure) {
  119. DEBUG_MSG_P(PSTR("[MQTT] Using SSL\n"));
  120. _mqtt.setSecure(secure);
  121. }
  122. #endif // SECURE_CLIENT != SECURE_CLIENT_NONE
  123. _mqtt.connect();
  124. }
  125. #endif // MQTT_LIBRARY == MQTT_LIBRARY_ASYNCMQTTCLIENT
  126. #if (MQTT_LIBRARY == MQTT_LIBRARY_ARDUINOMQTT) || (MQTT_LIBRARY == MQTT_LIBRARY_PUBSUBCLIENT)
  127. WiFiClient& _mqttGetClient(bool secure) {
  128. #if SECURE_CLIENT != SECURE_CLIENT_NONE
  129. return (secure ? _mqtt_client_secure->get() : _mqtt_client);
  130. #else
  131. return _mqtt_client;
  132. #endif
  133. }
  134. bool _mqttSetupSyncClient(bool secure = false) {
  135. #if SECURE_CLIENT != SECURE_CLIENT_NONE
  136. if (secure) {
  137. if (!_mqtt_client_secure) _mqtt_client_secure = std::make_unique<SecureClient>(_mqtt_sc_config);
  138. return _mqtt_client_secure->beforeConnected();
  139. }
  140. #endif
  141. return true;
  142. }
  143. bool _mqttConnectSyncClient(bool secure = false) {
  144. bool result = false;
  145. #if MQTT_LIBRARY == MQTT_LIBRARY_ARDUINOMQTT
  146. _mqtt.begin(_mqtt_server.c_str(), _mqtt_port, _mqttGetClient(secure));
  147. _mqtt.setWill(_mqtt_will.c_str(), _mqtt_payload_offline.c_str(), _mqtt_retain, _mqtt_qos);
  148. result = _mqtt.connect(_mqtt_clientid.c_str(), _mqtt_user.c_str(), _mqtt_pass.c_str());
  149. #elif MQTT_LIBRARY == MQTT_LIBRARY_PUBSUBCLIENT
  150. _mqtt.setClient(_mqttGetClient(secure));
  151. _mqtt.setServer(_mqtt_server.c_str(), _mqtt_port);
  152. if (_mqtt_user.length() && _mqtt_pass.length()) {
  153. DEBUG_MSG_P(PSTR("[MQTT] Connecting as user %s\n"), _mqtt_user.c_str());
  154. result = _mqtt.connect(_mqtt_clientid.c_str(), _mqtt_user.c_str(), _mqtt_pass.c_str(), _mqtt_will.c_str(), _mqtt_qos, _mqtt_retain, _mqtt_payload_offline.c_str());
  155. } else {
  156. result = _mqtt.connect(_mqtt_clientid.c_str(), _mqtt_will.c_str(), _mqtt_qos, _mqtt_retain, _mqtt_payload_offline.c_str());
  157. }
  158. #endif
  159. #if SECURE_CLIENT != SECURE_CLIENT_NONE
  160. if (result && secure) {
  161. result = _mqtt_client_secure->afterConnected();
  162. }
  163. #endif
  164. return result;
  165. }
  166. #endif // (MQTT_LIBRARY == MQTT_LIBRARY_ARDUINOMQTT) || (MQTT_LIBRARY == MQTT_LIBRARY_PUBSUBCLIENT)
  167. void _mqttConnect() {
  168. // Do not connect if disabled
  169. if (!_mqtt_enabled) return;
  170. // Do not connect if already connected or still trying to connect
  171. if (_mqtt.connected() || _mqtt_connecting) return;
  172. // Check reconnect interval
  173. if (millis() - _mqtt_last_connection < _mqtt_reconnect_delay) return;
  174. // Increase the reconnect delay
  175. _mqtt_reconnect_delay += MQTT_RECONNECT_DELAY_STEP;
  176. if (_mqtt_reconnect_delay > MQTT_RECONNECT_DELAY_MAX) {
  177. _mqtt_reconnect_delay = MQTT_RECONNECT_DELAY_MAX;
  178. }
  179. #if MDNS_CLIENT_SUPPORT
  180. _mqtt_server = mdnsResolve(_mqtt_server);
  181. #endif
  182. DEBUG_MSG_P(PSTR("[MQTT] Connecting to broker at %s:%u\n"), _mqtt_server.c_str(), _mqtt_port);
  183. DEBUG_MSG_P(PSTR("[MQTT] Client ID: %s\n"), _mqtt_clientid.c_str());
  184. DEBUG_MSG_P(PSTR("[MQTT] QoS: %d\n"), _mqtt_qos);
  185. DEBUG_MSG_P(PSTR("[MQTT] Retain flag: %d\n"), _mqtt_retain ? 1 : 0);
  186. DEBUG_MSG_P(PSTR("[MQTT] Keepalive time: %ds\n"), _mqtt_keepalive);
  187. DEBUG_MSG_P(PSTR("[MQTT] Will topic: %s\n"), _mqtt_will.c_str());
  188. _mqtt_connecting = true;
  189. #if SECURE_CLIENT != SECURE_CLIENT_NONE
  190. const bool secure = getSetting("mqttUseSSL", MQTT_SSL_ENABLED).toInt() == 1;
  191. #else
  192. const bool secure = false;
  193. #endif
  194. #if MQTT_LIBRARY == MQTT_LIBRARY_ASYNCMQTTCLIENT
  195. _mqttSetupAsyncClient(secure);
  196. #elif (MQTT_LIBRARY == MQTT_LIBRARY_ARDUINOMQTT) || (MQTT_LIBRARY == MQTT_LIBRARY_PUBSUBCLIENT)
  197. if (_mqttSetupSyncClient(secure) && _mqttConnectSyncClient(secure)) {
  198. _mqttOnConnect();
  199. } else {
  200. DEBUG_MSG_P(PSTR("[MQTT] Connection failed\n"));
  201. _mqttOnDisconnect();
  202. }
  203. #else
  204. #error "please check that MQTT_LIBRARY is valid"
  205. #endif
  206. }
  207. void _mqttPlaceholders(String& text) {
  208. text.replace("{hostname}", getSetting("hostname"));
  209. text.replace("{magnitude}", "#");
  210. String mac = WiFi.macAddress();
  211. mac.replace(":", "");
  212. text.replace("{mac}", mac);
  213. }
  214. template<typename T>
  215. void _mqttApplySetting(T& current, T& updated) {
  216. if (current != updated) {
  217. current = std::move(updated);
  218. mqttDisconnect();
  219. }
  220. }
  221. template<typename T>
  222. void _mqttApplySetting(T& current, const T& updated) {
  223. if (current != updated) {
  224. current = updated;
  225. mqttDisconnect();
  226. }
  227. }
  228. template<typename T>
  229. void _mqttApplyTopic(T& current, const char* magnitude) {
  230. String updated = mqttTopic(magnitude, false);
  231. if (current != updated) {
  232. mqttFlush();
  233. current = std::move(updated);
  234. }
  235. }
  236. void _mqttConfigure() {
  237. // Enable only when server is set
  238. {
  239. String server = getSetting("mqttServer", MQTT_SERVER);
  240. uint16_t port = getSetting("mqttPort", MQTT_PORT).toInt();
  241. bool enabled = false;
  242. if (server.length()) {
  243. enabled = getSetting("mqttEnabled", MQTT_ENABLED).toInt() == 1;
  244. }
  245. _mqttApplySetting(_mqtt_server, server);
  246. _mqttApplySetting(_mqtt_enabled, enabled);
  247. _mqttApplySetting(_mqtt_port, port);
  248. if (!enabled) return;
  249. }
  250. // Get base topic and apply placeholders
  251. {
  252. String topic = getSetting("mqttTopic", MQTT_TOPIC);
  253. if (topic.endsWith("/")) topic.remove(topic.length()-1);
  254. // Replace things inside curly braces (like {hostname}, {mac} etc.)
  255. _mqttPlaceholders(topic);
  256. if (topic.indexOf("#") == -1) topic.concat("/#");
  257. _mqttApplySetting(_mqtt_topic, topic);
  258. _mqttApplyTopic(_mqtt_will, MQTT_TOPIC_STATUS);
  259. }
  260. // Getter and setter
  261. {
  262. String setter = getSetting("mqttSetter", MQTT_SETTER);
  263. String getter = getSetting("mqttGetter", MQTT_GETTER);
  264. bool forward = !setter.equals(getter) && RELAY_REPORT_STATUS;
  265. _mqttApplySetting(_mqtt_setter, setter);
  266. _mqttApplySetting(_mqtt_getter, getter);
  267. _mqttApplySetting(_mqtt_forward, forward);
  268. }
  269. // MQTT options
  270. {
  271. String user = getSetting("mqttUser", MQTT_USER);
  272. _mqttPlaceholders(user);
  273. String pass = getSetting("mqttPassword", MQTT_PASS);
  274. unsigned char qos = getSetting("mqttQoS", MQTT_QOS).toInt();
  275. bool retain = getSetting("mqttRetain", MQTT_RETAIN).toInt() == 1;
  276. unsigned long keepalive = getSetting("mqttKeep", MQTT_KEEPALIVE).toInt();
  277. String id = getSetting("mqttClientID", getIdentifier());
  278. _mqttPlaceholders(id);
  279. _mqttApplySetting(_mqtt_user, user);
  280. _mqttApplySetting(_mqtt_pass, pass);
  281. _mqttApplySetting(_mqtt_qos, qos);
  282. _mqttApplySetting(_mqtt_retain, retain);
  283. _mqttApplySetting(_mqtt_keepalive, keepalive);
  284. _mqttApplySetting(_mqtt_clientid, id);
  285. }
  286. // MQTT JSON
  287. {
  288. _mqttApplySetting(_mqtt_use_json, getSetting("mqttUseJson", MQTT_USE_JSON).toInt() == 1);
  289. _mqttApplyTopic(_mqtt_topic_json, MQTT_TOPIC_JSON);
  290. }
  291. // Custom payload strings
  292. settingsProcessConfig({
  293. {_mqtt_payload_online, "mqttPayloadOnline", MQTT_STATUS_ONLINE},
  294. {_mqtt_payload_offline, "mqttPayloadOffline", MQTT_STATUS_OFFLINE}
  295. });
  296. // Reset reconnect delay to reconnect sooner
  297. _mqtt_reconnect_delay = MQTT_RECONNECT_DELAY_MIN;
  298. }
  299. void _mqttBackwards() {
  300. String mqttTopic = getSetting("mqttTopic", MQTT_TOPIC);
  301. if (mqttTopic.indexOf("{identifier}") > 0) {
  302. mqttTopic.replace("{identifier}", "{hostname}");
  303. setSetting("mqttTopic", mqttTopic);
  304. }
  305. }
  306. void _mqttInfo() {
  307. DEBUG_MSG_P(PSTR(
  308. "[MQTT] "
  309. #if MQTT_LIBRARY == MQTT_LIBRARY_ASYNCMQTTCLIENT
  310. "AsyncMqttClient"
  311. #elif MQTT_LIBRARY == MQTT_LIBRARY_ARDUINOMQTT
  312. "Arduino-MQTT"
  313. #elif MQTT_LIBRARY == MQTT_LIBRARY_PUBSUBCLIENT
  314. "PubSubClient"
  315. #endif
  316. ", SSL "
  317. #if SECURE_CLIENT != SEURE_CLIENT_NONE
  318. "ENABLED"
  319. #else
  320. "DISABLED"
  321. #endif
  322. ", Autoconnect "
  323. #if MQTT_AUTOCONNECT
  324. "ENABLED"
  325. #else
  326. "DISABLED"
  327. #endif
  328. "\n"
  329. ));
  330. DEBUG_MSG_P(PSTR("[MQTT] Client %s, %s\n"),
  331. _mqtt_enabled ? "ENABLED" : "DISABLED",
  332. _mqtt.connected() ? "CONNECTED" : "DISCONNECTED"
  333. );
  334. DEBUG_MSG_P(PSTR("[MQTT] Retry %s (Now %u, Last %u, Delay %u, Step %u)\n"),
  335. _mqtt_connecting ? "CONNECTING" : "WAITING",
  336. millis(),
  337. _mqtt_last_connection,
  338. _mqtt_reconnect_delay,
  339. MQTT_RECONNECT_DELAY_STEP
  340. );
  341. }
  342. // -----------------------------------------------------------------------------
  343. // WEB
  344. // -----------------------------------------------------------------------------
  345. #if WEB_SUPPORT
  346. bool _mqttWebSocketOnKeyCheck(const char * key, JsonVariant& value) {
  347. return (strncmp(key, "mqtt", 3) == 0);
  348. }
  349. void _mqttWebSocketOnVisible(JsonObject& root) {
  350. root["mqttVisible"] = 1;
  351. #if ASYNC_TCP_SSL_ENABLED
  352. root["mqttsslVisible"] = 1;
  353. #endif
  354. }
  355. void _mqttWebSocketOnData(JsonObject& root) {
  356. root["mqttStatus"] = mqttConnected();
  357. }
  358. void _mqttWebSocketOnConnected(JsonObject& root) {
  359. root["mqttEnabled"] = mqttEnabled();
  360. root["mqttServer"] = getSetting("mqttServer", MQTT_SERVER);
  361. root["mqttPort"] = getSetting("mqttPort", MQTT_PORT);
  362. root["mqttUser"] = getSetting("mqttUser", MQTT_USER);
  363. root["mqttClientID"] = getSetting("mqttClientID");
  364. root["mqttPassword"] = getSetting("mqttPassword", MQTT_PASS);
  365. root["mqttKeep"] = _mqtt_keepalive;
  366. root["mqttRetain"] = _mqtt_retain;
  367. root["mqttQoS"] = _mqtt_qos;
  368. #if SECURE_CLIENT != SECURE_CLIENT_NONE
  369. root["mqttUseSSL"] = getSetting("mqttUseSSL", MQTT_SSL_ENABLED).toInt() == 1;
  370. root["mqttFP"] = getSetting("mqttFP", MQTT_SSL_FINGERPRINT);
  371. #endif
  372. root["mqttTopic"] = getSetting("mqttTopic", MQTT_TOPIC);
  373. root["mqttUseJson"] = getSetting("mqttUseJson", MQTT_USE_JSON).toInt() == 1;
  374. }
  375. #endif
  376. // -----------------------------------------------------------------------------
  377. // SETTINGS
  378. // -----------------------------------------------------------------------------
  379. #if TERMINAL_SUPPORT
  380. void _mqttInitCommands() {
  381. terminalRegisterCommand(F("MQTT.RESET"), [](Embedis* e) {
  382. _mqttConfigure();
  383. mqttDisconnect();
  384. terminalOK();
  385. });
  386. terminalRegisterCommand(F("MQTT.INFO"), [](Embedis* e) {
  387. _mqttInfo();
  388. terminalOK();
  389. });
  390. }
  391. #endif // TERMINAL_SUPPORT
  392. // -----------------------------------------------------------------------------
  393. // MQTT Callbacks
  394. // -----------------------------------------------------------------------------
  395. void _mqttCallback(unsigned int type, const char * topic, const char * payload) {
  396. if (type == MQTT_CONNECT_EVENT) {
  397. // Subscribe to internal action topics
  398. mqttSubscribe(MQTT_TOPIC_ACTION);
  399. // Flag system to send heartbeat
  400. systemSendHeartbeat();
  401. }
  402. if (type == MQTT_MESSAGE_EVENT) {
  403. // Match topic
  404. String t = mqttMagnitude((char *) topic);
  405. // Actions
  406. if (t.equals(MQTT_TOPIC_ACTION)) {
  407. if (strcmp(payload, MQTT_ACTION_RESET) == 0) {
  408. deferredReset(100, CUSTOM_RESET_MQTT);
  409. }
  410. }
  411. }
  412. }
  413. void _mqttOnConnect() {
  414. DEBUG_MSG_P(PSTR("[MQTT] Connected!\n"));
  415. _mqtt_reconnect_delay = MQTT_RECONNECT_DELAY_MIN;
  416. _mqtt_last_connection = millis();
  417. _mqtt_connecting = false;
  418. _mqtt_connected = true;
  419. // Clean subscriptions
  420. mqttUnsubscribeRaw("#");
  421. // Send connect event to subscribers
  422. for (unsigned char i = 0; i < _mqtt_callbacks.size(); i++) {
  423. (_mqtt_callbacks[i])(MQTT_CONNECT_EVENT, NULL, NULL);
  424. }
  425. }
  426. void _mqttOnDisconnect() {
  427. // Reset reconnection delay
  428. _mqtt_last_connection = millis();
  429. _mqtt_connecting = false;
  430. _mqtt_connected = false;
  431. DEBUG_MSG_P(PSTR("[MQTT] Disconnected!\n"));
  432. // Send disconnect event to subscribers
  433. for (unsigned char i = 0; i < _mqtt_callbacks.size(); i++) {
  434. (_mqtt_callbacks[i])(MQTT_DISCONNECT_EVENT, NULL, NULL);
  435. }
  436. }
  437. void _mqttOnMessage(char* topic, char* payload, unsigned int len) {
  438. if (len == 0) return;
  439. char message[len + 1];
  440. strlcpy(message, (char *) payload, len + 1);
  441. #if MQTT_SKIP_RETAINED
  442. if (millis() - _mqtt_last_connection < MQTT_SKIP_TIME) {
  443. DEBUG_MSG_P(PSTR("[MQTT] Received %s => %s - SKIPPED\n"), topic, message);
  444. return;
  445. }
  446. #endif
  447. DEBUG_MSG_P(PSTR("[MQTT] Received %s => %s\n"), topic, message);
  448. // Send message event to subscribers
  449. for (unsigned char i = 0; i < _mqtt_callbacks.size(); i++) {
  450. (_mqtt_callbacks[i])(MQTT_MESSAGE_EVENT, topic, message);
  451. }
  452. }
  453. // -----------------------------------------------------------------------------
  454. // Public API
  455. // -----------------------------------------------------------------------------
  456. /**
  457. Returns the magnitude part of a topic
  458. @param topic the full MQTT topic
  459. @return String object with the magnitude part.
  460. */
  461. String mqttMagnitude(char * topic) {
  462. String pattern = _mqtt_topic + _mqtt_setter;
  463. int position = pattern.indexOf("#");
  464. if (position == -1) return String();
  465. String start = pattern.substring(0, position);
  466. String end = pattern.substring(position + 1);
  467. String magnitude = String(topic);
  468. if (magnitude.startsWith(start) && magnitude.endsWith(end)) {
  469. magnitude.replace(start, "");
  470. magnitude.replace(end, "");
  471. } else {
  472. magnitude = String();
  473. }
  474. return magnitude;
  475. }
  476. /**
  477. Returns a full MQTT topic from the magnitude
  478. @param magnitude the magnitude part of the topic.
  479. @param is_set whether to build a command topic (true)
  480. or a state topic (false).
  481. @return String full MQTT topic.
  482. */
  483. String mqttTopic(const char * magnitude, bool is_set) {
  484. String output = _mqtt_topic;
  485. output.replace("#", magnitude);
  486. output += is_set ? _mqtt_setter : _mqtt_getter;
  487. return output;
  488. }
  489. /**
  490. Returns a full MQTT topic from the magnitude
  491. @param magnitude the magnitude part of the topic.
  492. @param index index of the magnitude when more than one such magnitudes.
  493. @param is_set whether to build a command topic (true)
  494. or a state topic (false).
  495. @return String full MQTT topic.
  496. */
  497. String mqttTopic(const char * magnitude, unsigned int index, bool is_set) {
  498. char buffer[strlen(magnitude)+5];
  499. snprintf_P(buffer, sizeof(buffer), PSTR("%s/%d"), magnitude, index);
  500. return mqttTopic(buffer, is_set);
  501. }
  502. // -----------------------------------------------------------------------------
  503. void mqttSendRaw(const char * topic, const char * message, bool retain) {
  504. if (_mqtt.connected()) {
  505. #if MQTT_LIBRARY == MQTT_LIBRARY_ASYNCMQTTCLIENT
  506. unsigned int packetId = _mqtt.publish(topic, _mqtt_qos, retain, message);
  507. DEBUG_MSG_P(PSTR("[MQTT] Sending %s => %s (PID %d)\n"), topic, message, packetId);
  508. #elif MQTT_LIBRARY == MQTT_LIBRARY_ARDUINOMQTT
  509. _mqtt.publish(topic, message, retain, _mqtt_qos);
  510. DEBUG_MSG_P(PSTR("[MQTT] Sending %s => %s\n"), topic, message);
  511. #elif MQTT_LIBRARY == MQTT_LIBRARY_PUBSUBCLIENT
  512. _mqtt.publish(topic, message, retain);
  513. DEBUG_MSG_P(PSTR("[MQTT] Sending %s => %s\n"), topic, message);
  514. #endif
  515. }
  516. }
  517. void mqttSendRaw(const char * topic, const char * message) {
  518. mqttSendRaw (topic, message, _mqtt_retain);
  519. }
  520. void mqttSend(const char * topic, const char * message, bool force, bool retain) {
  521. bool useJson = force ? false : _mqtt_use_json;
  522. // Equeue message
  523. if (useJson) {
  524. // Enqueue new message
  525. mqttEnqueue(topic, message);
  526. // Reset flush timer
  527. _mqtt_flush_ticker.once_ms(MQTT_USE_JSON_DELAY, mqttFlush);
  528. // Send it right away
  529. } else {
  530. mqttSendRaw(mqttTopic(topic, false).c_str(), message, retain);
  531. }
  532. }
  533. void mqttSend(const char * topic, const char * message, bool force) {
  534. mqttSend(topic, message, force, _mqtt_retain);
  535. }
  536. void mqttSend(const char * topic, const char * message) {
  537. mqttSend(topic, message, false);
  538. }
  539. void mqttSend(const char * topic, unsigned int index, const char * message, bool force, bool retain) {
  540. char buffer[strlen(topic)+5];
  541. snprintf_P(buffer, sizeof(buffer), PSTR("%s/%d"), topic, index);
  542. mqttSend(buffer, message, force, retain);
  543. }
  544. void mqttSend(const char * topic, unsigned int index, const char * message, bool force) {
  545. mqttSend(topic, index, message, force, _mqtt_retain);
  546. }
  547. void mqttSend(const char * topic, unsigned int index, const char * message) {
  548. mqttSend(topic, index, message, false);
  549. }
  550. // -----------------------------------------------------------------------------
  551. unsigned char _mqttBuildTree(JsonObject& root, char parent) {
  552. unsigned char count = 0;
  553. // Add enqueued messages
  554. for (unsigned char i=0; i<_mqtt_queue.size(); i++) {
  555. mqtt_message_t element = _mqtt_queue[i];
  556. if (element.parent == parent) {
  557. ++count;
  558. JsonObject& elements = root.createNestedObject(element.topic);
  559. unsigned char num = _mqttBuildTree(elements, i);
  560. if (0 == num) {
  561. if (isNumber(element.message)) {
  562. double value = atof(element.message);
  563. if (value == int(value)) {
  564. root.set(element.topic, int(value));
  565. } else {
  566. root.set(element.topic, value);
  567. }
  568. } else {
  569. root.set(element.topic, element.message);
  570. }
  571. }
  572. }
  573. }
  574. return count;
  575. }
  576. void mqttFlush() {
  577. if (!_mqtt.connected()) return;
  578. if (_mqtt_queue.size() == 0) return;
  579. // Build tree recursively
  580. DynamicJsonBuffer jsonBuffer(1024);
  581. JsonObject& root = jsonBuffer.createObject();
  582. _mqttBuildTree(root, mqtt_message_t::END);
  583. // Add extra propeties
  584. #if NTP_SUPPORT && MQTT_ENQUEUE_DATETIME
  585. if (ntpSynced()) root[MQTT_TOPIC_TIME] = ntpDateTime();
  586. #endif
  587. #if MQTT_ENQUEUE_MAC
  588. root[MQTT_TOPIC_MAC] = WiFi.macAddress();
  589. #endif
  590. #if MQTT_ENQUEUE_HOSTNAME
  591. root[MQTT_TOPIC_HOSTNAME] = getSetting("hostname");
  592. #endif
  593. #if MQTT_ENQUEUE_IP
  594. root[MQTT_TOPIC_IP] = getIP();
  595. #endif
  596. #if MQTT_ENQUEUE_MESSAGE_ID
  597. root[MQTT_TOPIC_MESSAGE_ID] = (Rtcmem->mqtt)++;
  598. #endif
  599. // Send
  600. String output;
  601. root.printTo(output);
  602. jsonBuffer.clear();
  603. mqttSendRaw(_mqtt_topic_json.c_str(), output.c_str(), false);
  604. // Clear queue
  605. for (unsigned char i = 0; i < _mqtt_queue.size(); i++) {
  606. mqtt_message_t element = _mqtt_queue[i];
  607. free(element.topic);
  608. if (element.message) {
  609. free(element.message);
  610. }
  611. }
  612. _mqtt_queue.clear();
  613. }
  614. int8_t mqttEnqueue(const char * topic, const char * message, unsigned char parent) {
  615. // Queue is not meant to send message "offline"
  616. // We must prevent the queue does not get full while offline
  617. if (!_mqtt.connected()) return -1;
  618. // Force flusing the queue if the MQTT_QUEUE_MAX_SIZE has been reached
  619. if (_mqtt_queue.size() >= MQTT_QUEUE_MAX_SIZE) mqttFlush();
  620. int8_t index = _mqtt_queue.size();
  621. // Enqueue new message
  622. mqtt_message_t element;
  623. element.parent = parent;
  624. element.topic = strdup(topic);
  625. if (NULL != message) {
  626. element.message = strdup(message);
  627. }
  628. _mqtt_queue.push_back(element);
  629. return index;
  630. }
  631. int8_t mqttEnqueue(const char * topic, const char * message) {
  632. return mqttEnqueue(topic, message, mqtt_message_t::END);
  633. }
  634. // -----------------------------------------------------------------------------
  635. void mqttSubscribeRaw(const char * topic) {
  636. if (_mqtt.connected() && (strlen(topic) > 0)) {
  637. #if MQTT_LIBRARY == MQTT_LIBRARY_ASYNCMQTTCLIENT
  638. unsigned int packetId = _mqtt.subscribe(topic, _mqtt_qos);
  639. DEBUG_MSG_P(PSTR("[MQTT] Subscribing to %s (PID %d)\n"), topic, packetId);
  640. #else // Arduino-MQTT or PubSubClient
  641. _mqtt.subscribe(topic, _mqtt_qos);
  642. DEBUG_MSG_P(PSTR("[MQTT] Subscribing to %s\n"), topic);
  643. #endif
  644. }
  645. }
  646. void mqttSubscribe(const char * topic) {
  647. mqttSubscribeRaw(mqttTopic(topic, true).c_str());
  648. }
  649. void mqttUnsubscribeRaw(const char * topic) {
  650. if (_mqtt.connected() && (strlen(topic) > 0)) {
  651. #if MQTT_LIBRARY == MQTT_LIBRARY_ASYNCMQTTCLIENT
  652. unsigned int packetId = _mqtt.unsubscribe(topic);
  653. DEBUG_MSG_P(PSTR("[MQTT] Unsubscribing to %s (PID %d)\n"), topic, packetId);
  654. #else // Arduino-MQTT or PubSubClient
  655. _mqtt.unsubscribe(topic);
  656. DEBUG_MSG_P(PSTR("[MQTT] Unsubscribing to %s\n"), topic);
  657. #endif
  658. }
  659. }
  660. void mqttUnsubscribe(const char * topic) {
  661. mqttUnsubscribeRaw(mqttTopic(topic, true).c_str());
  662. }
  663. // -----------------------------------------------------------------------------
  664. void mqttEnabled(bool status) {
  665. _mqtt_enabled = status;
  666. }
  667. bool mqttEnabled() {
  668. return _mqtt_enabled;
  669. }
  670. bool mqttConnected() {
  671. return _mqtt.connected();
  672. }
  673. void mqttDisconnect() {
  674. if (_mqtt.connected()) {
  675. DEBUG_MSG_P(PSTR("[MQTT] Disconnecting\n"));
  676. _mqtt.disconnect();
  677. }
  678. }
  679. bool mqttForward() {
  680. return _mqtt_forward;
  681. }
  682. void mqttRegister(mqtt_callback_f callback) {
  683. _mqtt_callbacks.push_back(callback);
  684. }
  685. void mqttSetBroker(IPAddress ip, unsigned int port) {
  686. setSetting("mqttServer", ip.toString());
  687. _mqtt_server = ip.toString();
  688. setSetting("mqttPort", port);
  689. _mqtt_port = port;
  690. mqttEnabled(MQTT_AUTOCONNECT);
  691. }
  692. void mqttSetBrokerIfNone(IPAddress ip, unsigned int port) {
  693. if (getSetting("mqttServer", MQTT_SERVER).length() == 0) mqttSetBroker(ip, port);
  694. }
  695. const String& mqttPayloadOnline() {
  696. return _mqtt_payload_online;
  697. }
  698. const String& mqttPayloadOffline() {
  699. return _mqtt_payload_offline;
  700. }
  701. const char* mqttPayloadStatus(bool status) {
  702. return status ? _mqtt_payload_online.c_str() : _mqtt_payload_offline.c_str();
  703. }
  704. void mqttSendStatus() {
  705. mqttSend(MQTT_TOPIC_STATUS, _mqtt_payload_online.c_str(), true);
  706. }
  707. // -----------------------------------------------------------------------------
  708. // Initialization
  709. // -----------------------------------------------------------------------------
  710. void mqttSetup() {
  711. _mqttBackwards();
  712. _mqttInfo();
  713. #if MQTT_LIBRARY == MQTT_LIBRARY_ASYNCMQTTCLIENT
  714. // XXX: should not place this in config, addServerFingerprint does not check for duplicates
  715. #if SECURE_CLIENT != SECURE_CLIENT_NONE
  716. {
  717. if (_mqtt_sc_config.on_fingerprint) {
  718. const String fingerprint = _mqtt_sc_config.on_fingerprint();
  719. uint8_t buffer[20] = {0};
  720. if (sslFingerPrintArray(fingerprint.c_str(), buffer)) {
  721. _mqtt.addServerFingerprint(buffer);
  722. }
  723. }
  724. }
  725. #endif // SECURE_CLIENT != SECURE_CLIENT_NONE
  726. _mqtt.onConnect([](bool sessionPresent) {
  727. _mqttOnConnect();
  728. });
  729. _mqtt.onDisconnect([](AsyncMqttClientDisconnectReason reason) {
  730. if (reason == AsyncMqttClientDisconnectReason::TCP_DISCONNECTED) {
  731. DEBUG_MSG_P(PSTR("[MQTT] TCP Disconnected\n"));
  732. }
  733. if (reason == AsyncMqttClientDisconnectReason::MQTT_IDENTIFIER_REJECTED) {
  734. DEBUG_MSG_P(PSTR("[MQTT] Identifier Rejected\n"));
  735. }
  736. if (reason == AsyncMqttClientDisconnectReason::MQTT_SERVER_UNAVAILABLE) {
  737. DEBUG_MSG_P(PSTR("[MQTT] Server unavailable\n"));
  738. }
  739. if (reason == AsyncMqttClientDisconnectReason::MQTT_MALFORMED_CREDENTIALS) {
  740. DEBUG_MSG_P(PSTR("[MQTT] Malformed credentials\n"));
  741. }
  742. if (reason == AsyncMqttClientDisconnectReason::MQTT_NOT_AUTHORIZED) {
  743. DEBUG_MSG_P(PSTR("[MQTT] Not authorized\n"));
  744. }
  745. #if SECURE_CLIENT == SECURE_CLIENT_AXTLS
  746. if (reason == AsyncMqttClientDisconnectReason::TLS_BAD_FINGERPRINT) {
  747. DEBUG_MSG_P(PSTR("[MQTT] Bad fingerprint\n"));
  748. }
  749. #endif
  750. _mqttOnDisconnect();
  751. });
  752. _mqtt.onMessage([](char* topic, char* payload, AsyncMqttClientMessageProperties properties, size_t len, size_t index, size_t total) {
  753. _mqttOnMessage(topic, payload, len);
  754. });
  755. _mqtt.onSubscribe([](uint16_t packetId, uint8_t qos) {
  756. DEBUG_MSG_P(PSTR("[MQTT] Subscribe ACK for PID %d\n"), packetId);
  757. });
  758. _mqtt.onPublish([](uint16_t packetId) {
  759. DEBUG_MSG_P(PSTR("[MQTT] Publish ACK for PID %d\n"), packetId);
  760. });
  761. #elif MQTT_LIBRARY == MQTT_LIBRARY_ARDUINOMQTT
  762. _mqtt.onMessageAdvanced([](MQTTClient *client, char topic[], char payload[], int length) {
  763. _mqttOnMessage(topic, payload, length);
  764. });
  765. #elif MQTT_LIBRARY == MQTT_LIBRARY_PUBSUBCLIENT
  766. _mqtt.setCallback([](char* topic, byte* payload, unsigned int length) {
  767. _mqttOnMessage(topic, (char *) payload, length);
  768. });
  769. #endif // MQTT_LIBRARY == MQTT_LIBRARY_ASYNCMQTTCLIENT
  770. _mqttConfigure();
  771. mqttRegister(_mqttCallback);
  772. #if WEB_SUPPORT
  773. wsRegister()
  774. .onVisible(_mqttWebSocketOnVisible)
  775. .onData(_mqttWebSocketOnData)
  776. .onConnected(_mqttWebSocketOnConnected)
  777. .onKeyCheck(_mqttWebSocketOnKeyCheck);
  778. mqttRegister([](unsigned int type, const char*, const char*) {
  779. if ((type == MQTT_CONNECT_EVENT) || (type == MQTT_DISCONNECT_EVENT)) wsPost(_mqttWebSocketOnData);
  780. });
  781. #endif
  782. #if TERMINAL_SUPPORT
  783. _mqttInitCommands();
  784. #endif
  785. // Main callbacks
  786. espurnaRegisterLoop(mqttLoop);
  787. espurnaRegisterReload(_mqttConfigure);
  788. }
  789. void mqttLoop() {
  790. if (WiFi.status() != WL_CONNECTED) return;
  791. #if MQTT_LIBRARY == MQTT_LIBRARY_ASYNCMQTTCLIENT
  792. _mqttConnect();
  793. #else // MQTT_LIBRARY != MQTT_LIBRARY_ASYNCMQTTCLIENT
  794. if (_mqtt.connected()) {
  795. _mqtt.loop();
  796. } else {
  797. if (_mqtt_connected) {
  798. _mqttOnDisconnect();
  799. }
  800. _mqttConnect();
  801. }
  802. #endif // MQTT_LIBRARY == MQTT_LIBRARY_ASYNCMQTTCLIENT
  803. }
  804. #else
  805. bool mqttForward() {
  806. return false;
  807. }
  808. #endif // MQTT_SUPPORT