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.

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