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.

563 lines
16 KiB

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