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.

409 lines
12 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 <ArduinoJson.h>
  7. #include <vector>
  8. #include <Ticker.h>
  9. const char *mqtt_user = 0;
  10. const char *mqtt_pass = 0;
  11. #if MQTT_USE_ASYNC
  12. #include <AsyncMqttClient.h>
  13. AsyncMqttClient mqtt;
  14. #else
  15. #include <PubSubClient.h>
  16. WiFiClient mqttWiFiClient;
  17. PubSubClient mqtt(mqttWiFiClient);
  18. bool _mqttConnected = false;
  19. #endif
  20. String _mqttTopic;
  21. String _mqttSetter;
  22. String _mqttGetter;
  23. bool _mqttForward;
  24. char *_mqttUser = 0;
  25. char *_mqttPass = 0;
  26. char *_mqttWill;
  27. std::vector<void (*)(unsigned int, const char *, const char *)> _mqtt_callbacks;
  28. #if MQTT_SKIP_RETAINED
  29. unsigned long mqttConnectedAt = 0;
  30. #endif
  31. typedef struct {
  32. char * topic;
  33. char * message;
  34. } mqtt_message_t;
  35. std::vector<mqtt_message_t> _mqtt_queue;
  36. Ticker mqttFlushTicker;
  37. // -----------------------------------------------------------------------------
  38. // Public API
  39. // -----------------------------------------------------------------------------
  40. bool mqttConnected() {
  41. return mqtt.connected();
  42. }
  43. void mqttDisconnect() {
  44. mqtt.disconnect();
  45. }
  46. bool mqttForward() {
  47. return _mqttForward;
  48. }
  49. String mqttSubtopic(char * topic) {
  50. String response;
  51. String t = String(topic);
  52. if (t.startsWith(_mqttTopic) && t.endsWith(_mqttSetter)) {
  53. response = t.substring(_mqttTopic.length(), t.length() - _mqttSetter.length());
  54. }
  55. return response;
  56. }
  57. void mqttSendRaw(const char * topic, const char * message) {
  58. if (mqtt.connected()) {
  59. #if MQTT_USE_ASYNC
  60. unsigned int packetId = mqtt.publish(topic, MQTT_QOS, MQTT_RETAIN, message);
  61. DEBUG_MSG_P(PSTR("[MQTT] Sending %s => %s (PID %d)\n"), topic, message, packetId);
  62. #else
  63. mqtt.publish(topic, message, MQTT_RETAIN);
  64. DEBUG_MSG_P(PSTR("[MQTT] Sending %s => %s\n"), topic, message);
  65. #endif
  66. }
  67. }
  68. void _mqttFlush() {
  69. if (_mqtt_queue.size() == 0) return;
  70. DynamicJsonBuffer jsonBuffer;
  71. JsonObject& root = jsonBuffer.createObject();
  72. for (unsigned char i=0; i<_mqtt_queue.size(); i++) {
  73. mqtt_message_t element = _mqtt_queue[i];
  74. root[element.topic] = element.message;
  75. }
  76. if (ntpConnected()) root[MQTT_TOPIC_TIME] = ntpDateTime();
  77. root[MQTT_TOPIC_HOSTNAME] = getSetting("hostname", HOSTNAME);
  78. root[MQTT_TOPIC_IP] = getIP();
  79. String output;
  80. root.printTo(output);
  81. String path = _mqttTopic + String(MQTT_TOPIC_JSON);
  82. mqttSendRaw(path.c_str(), output.c_str());
  83. for (unsigned char i = 0; i < _mqtt_queue.size(); i++) {
  84. mqtt_message_t element = _mqtt_queue[i];
  85. free(element.topic);
  86. free(element.message);
  87. }
  88. _mqtt_queue.clear();
  89. }
  90. void mqttSend(const char * topic, const char * message, bool force) {
  91. bool useJson = force ? false : getSetting("mqttUseJson", MQTT_USE_JSON).toInt() == 1;
  92. if (useJson) {
  93. mqtt_message_t element;
  94. element.topic = strdup(topic);
  95. element.message = strdup(message);
  96. _mqtt_queue.push_back(element);
  97. mqttFlushTicker.once_ms(MQTT_USE_JSON_DELAY, _mqttFlush);
  98. } else {
  99. String path = _mqttTopic + String(topic) + _mqttGetter;
  100. mqttSendRaw(path.c_str(), message);
  101. }
  102. }
  103. void mqttSend(const char * topic, const char * message) {
  104. mqttSend(topic, message, false);
  105. }
  106. void mqttSend(const char * topic, unsigned int index, const char * message, bool force) {
  107. char buffer[strlen(topic)+5];
  108. sprintf(buffer, "%s/%d", topic, index);
  109. mqttSend(buffer, message, force);
  110. }
  111. void mqttSend(const char * topic, unsigned int index, const char * message) {
  112. mqttSend(topic, index, message, false);
  113. }
  114. void mqttSubscribeRaw(const char * topic) {
  115. if (mqtt.connected() && (strlen(topic) > 0)) {
  116. #if MQTT_USE_ASYNC
  117. unsigned int packetId = mqtt.subscribe(topic, MQTT_QOS);
  118. DEBUG_MSG_P(PSTR("[MQTT] Subscribing to %s (PID %d)\n"), topic, packetId);
  119. #else
  120. mqtt.subscribe(topic, MQTT_QOS);
  121. DEBUG_MSG_P(PSTR("[MQTT] Subscribing to %s\n"), topic);
  122. #endif
  123. }
  124. }
  125. void mqttSubscribe(const char * topic) {
  126. String path = _mqttTopic + String(topic) + _mqttSetter;
  127. mqttSubscribeRaw(path.c_str());
  128. }
  129. void mqttRegister(void (*callback)(unsigned int, const char *, const char *)) {
  130. _mqtt_callbacks.push_back(callback);
  131. }
  132. // -----------------------------------------------------------------------------
  133. // Callbacks
  134. // -----------------------------------------------------------------------------
  135. void _mqttCallback(unsigned int type, const char * topic, const char * payload) {
  136. if (type == MQTT_CONNECT_EVENT) {
  137. mqttSubscribe(MQTT_TOPIC_ACTION);
  138. }
  139. if (type == MQTT_MESSAGE_EVENT) {
  140. // Match topic
  141. String t = mqttSubtopic((char *) topic);
  142. // Actions
  143. if (t.equals(MQTT_TOPIC_ACTION)) {
  144. if (strcmp(payload, MQTT_ACTION_RESET) == 0) {
  145. customReset(CUSTOM_RESET_MQTT);
  146. ESP.restart();
  147. }
  148. }
  149. }
  150. }
  151. void _mqttOnConnect() {
  152. DEBUG_MSG_P(PSTR("[MQTT] Connected!\n"));
  153. #if MQTT_SKIP_RETAINED
  154. mqttConnectedAt = millis();
  155. #endif
  156. // Build MQTT topics
  157. mqttConfigure();
  158. // Send first Heartbeat
  159. heartbeat();
  160. // Send connect event to subscribers
  161. for (unsigned char i = 0; i < _mqtt_callbacks.size(); i++) {
  162. (*_mqtt_callbacks[i])(MQTT_CONNECT_EVENT, NULL, NULL);
  163. }
  164. }
  165. void _mqttOnDisconnect() {
  166. DEBUG_MSG_P(PSTR("[MQTT] Disconnected!\n"));
  167. // Send disconnect event to subscribers
  168. for (unsigned char i = 0; i < _mqtt_callbacks.size(); i++) {
  169. (*_mqtt_callbacks[i])(MQTT_DISCONNECT_EVENT, NULL, NULL);
  170. }
  171. }
  172. void _mqttOnMessage(char* topic, char* payload, unsigned int len) {
  173. if (len == 0) return;
  174. char message[len + 1];
  175. strlcpy(message, (char *) payload, len + 1);
  176. DEBUG_MSG_P(PSTR("[MQTT] Received %s => %s"), topic, message);
  177. #if MQTT_SKIP_RETAINED
  178. if (millis() - mqttConnectedAt < MQTT_SKIP_TIME) {
  179. DEBUG_MSG_P(PSTR(" - SKIPPED\n"));
  180. return;
  181. }
  182. #endif
  183. DEBUG_MSG_P(PSTR("\n"));
  184. // Send message event to subscribers
  185. for (unsigned char i = 0; i < _mqtt_callbacks.size(); i++) {
  186. (*_mqtt_callbacks[i])(MQTT_MESSAGE_EVENT, topic, message);
  187. }
  188. }
  189. void mqttConnect() {
  190. if (!mqtt.connected()) {
  191. if (getSetting("mqttServer", MQTT_SERVER).length() == 0) return;
  192. // Last option: reconnect to wifi after MQTT_MAX_TRIES attemps in a row
  193. #if MQTT_MAX_TRIES > 0
  194. static unsigned int tries = 0;
  195. static unsigned long last_try = millis();
  196. if (millis() - last_try < MQTT_TRY_INTERVAL) {
  197. if (++tries > MQTT_MAX_TRIES) {
  198. DEBUG_MSG_P(PSTR("[MQTT] MQTT_MAX_TRIES met, disconnecting from WiFi\n"));
  199. wifiDisconnect();
  200. tries = 0;
  201. return;
  202. }
  203. } else {
  204. tries = 0;
  205. }
  206. last_try = millis();
  207. #endif
  208. mqtt.disconnect();
  209. if (_mqttUser) free(_mqttUser);
  210. if (_mqttPass) free(_mqttPass);
  211. char * host = strdup(getSetting("mqttServer", MQTT_SERVER).c_str());
  212. unsigned int port = getSetting("mqttPort", MQTT_PORT).toInt();
  213. _mqttUser = strdup(getSetting("mqttUser").c_str());
  214. _mqttPass = strdup(getSetting("mqttPassword").c_str());
  215. if (_mqttWill) free(_mqttWill);
  216. _mqttWill = strdup((_mqttTopic + MQTT_TOPIC_STATUS).c_str());
  217. DEBUG_MSG_P(PSTR("[MQTT] Connecting to broker at %s:%d"), host, port);
  218. mqtt.setServer(host, port);
  219. #if MQTT_USE_ASYNC
  220. mqtt.setKeepAlive(MQTT_KEEPALIVE).setCleanSession(false);
  221. mqtt.setWill(_mqttWill, MQTT_QOS, MQTT_RETAIN, "0");
  222. if ((strlen(_mqttUser) > 0) && (strlen(_mqttPass) > 0)) {
  223. DEBUG_MSG_P(PSTR(" as user '%s'."), _mqttUser);
  224. mqtt.setCredentials(_mqttUser, _mqttPass);
  225. }
  226. DEBUG_MSG_P(PSTR("\n"));
  227. mqtt.connect();
  228. #else
  229. bool response;
  230. if ((strlen(_mqttUser) > 0) && (strlen(_mqttPass) > 0)) {
  231. DEBUG_MSG_P(PSTR(" as user '%s'\n"), _mqttUser);
  232. response = mqtt.connect(getIdentifier().c_str(), _mqttUser, _mqttPass, _mqttWill, MQTT_QOS, MQTT_RETAIN, "0");
  233. } else {
  234. DEBUG_MSG_P(PSTR("\n"));
  235. response = mqtt.connect(getIdentifier().c_str(), _mqttWill, MQTT_QOS, MQTT_RETAIN, "0");
  236. }
  237. if (response) {
  238. _mqttOnConnect();
  239. _mqttConnected = true;
  240. } else {
  241. DEBUG_MSG_P(PSTR("[MQTT] Connection failed\n"));
  242. }
  243. #endif
  244. free(host);
  245. }
  246. }
  247. void mqttConfigure() {
  248. // Replace identifier
  249. _mqttTopic = getSetting("mqttTopic", MQTT_TOPIC);
  250. _mqttTopic.replace("{identifier}", getSetting("hostname"));
  251. if (!_mqttTopic.endsWith("/")) _mqttTopic = _mqttTopic + "/";
  252. _mqttSetter = getSetting("mqttSetter", MQTT_USE_SETTER);
  253. _mqttGetter = getSetting("mqttGetter", MQTT_USE_GETTER);
  254. _mqttForward = !_mqttGetter.equals(_mqttSetter);
  255. }
  256. void mqttSetup() {
  257. #if MQTT_USE_ASYNC
  258. mqtt.onConnect([](bool sessionPresent) {
  259. _mqttOnConnect();
  260. });
  261. mqtt.onDisconnect([](AsyncMqttClientDisconnectReason reason) {
  262. if (reason == AsyncMqttClientDisconnectReason::TCP_DISCONNECTED) {
  263. DEBUG_MSG_P(PSTR("[MQTT] TCP Disconnected\n"));
  264. }
  265. if (reason == AsyncMqttClientDisconnectReason::MQTT_IDENTIFIER_REJECTED) {
  266. DEBUG_MSG_P(PSTR("[MQTT] Identifier Rejected\n"));
  267. }
  268. if (reason == AsyncMqttClientDisconnectReason::MQTT_SERVER_UNAVAILABLE) {
  269. DEBUG_MSG_P(PSTR("[MQTT] Server unavailable\n"));
  270. }
  271. if (reason == AsyncMqttClientDisconnectReason::MQTT_MALFORMED_CREDENTIALS) {
  272. DEBUG_MSG_P(PSTR("[MQTT] Malformed credentials\n"));
  273. }
  274. if (reason == AsyncMqttClientDisconnectReason::MQTT_NOT_AUTHORIZED) {
  275. DEBUG_MSG_P(PSTR("[MQTT] Not authorized\n"));
  276. }
  277. #if ASYNC_TCP_SSL_ENABLED
  278. if (reason == AsyncMqttClientDisconnectReason::TLS_BAD_FINGERPRINT) {
  279. DEBUG_MSG_P(PSTR("[MQTT] Bad fingerprint\n"));
  280. }
  281. #endif
  282. _mqttOnDisconnect();
  283. });
  284. mqtt.onMessage([](char* topic, char* payload, AsyncMqttClientMessageProperties properties, size_t len, size_t index, size_t total) {
  285. _mqttOnMessage(topic, payload, len);
  286. });
  287. mqtt.onSubscribe([](uint16_t packetId, uint8_t qos) {
  288. DEBUG_MSG_P(PSTR("[MQTT] Subscribe ACK for PID %d\n"), packetId);
  289. });
  290. mqtt.onPublish([](uint16_t packetId) {
  291. DEBUG_MSG_P(PSTR("[MQTT] Publish ACK for PID %d\n"), packetId);
  292. });
  293. #else
  294. mqtt.setCallback([](char* topic, byte* payload, unsigned int length) {
  295. _mqttOnMessage(topic, (char *) payload, length);
  296. });
  297. #endif
  298. mqttRegister(_mqttCallback);
  299. }
  300. void mqttLoop() {
  301. static unsigned long lastPeriod = 0;
  302. if (WiFi.status() == WL_CONNECTED) {
  303. if (!mqtt.connected()) {
  304. #if not MQTT_USE_ASYNC
  305. if (_mqttConnected) {
  306. _mqttOnDisconnect();
  307. _mqttConnected = false;
  308. }
  309. #endif
  310. unsigned long currPeriod = millis() / MQTT_RECONNECT_DELAY;
  311. if (currPeriod != lastPeriod) {
  312. lastPeriod = currPeriod;
  313. mqttConnect();
  314. }
  315. #if not MQTT_USE_ASYNC
  316. } else {
  317. mqtt.loop();
  318. #endif
  319. }
  320. }
  321. }