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.

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