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.

309 lines
8.4 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 <vector>
  7. const char *mqtt_user = 0;
  8. const char *mqtt_pass = 0;
  9. #if MQTT_USE_ASYNC
  10. #include <AsyncMqttClient.h>
  11. AsyncMqttClient mqtt;
  12. #else
  13. #include <PubSubClient.h>
  14. WiFiClient mqttWiFiClient;
  15. PubSubClient mqtt(mqttWiFiClient);
  16. bool _mqttConnected = false;
  17. #endif
  18. String mqttTopic;
  19. bool _mqttForward;
  20. char *_mqttUser = 0;
  21. char *_mqttPass = 0;
  22. std::vector<void (*)(unsigned int, const char *, const char *)> _mqtt_callbacks;
  23. #if MQTT_SKIP_RETAINED
  24. unsigned long mqttConnectedAt = 0;
  25. #endif
  26. // -----------------------------------------------------------------------------
  27. // Public API
  28. // -----------------------------------------------------------------------------
  29. bool mqttConnected() {
  30. return mqtt.connected();
  31. }
  32. void mqttDisconnect() {
  33. mqtt.disconnect();
  34. }
  35. void buildTopics() {
  36. // Replace identifier
  37. mqttTopic = getSetting("mqttTopic", MQTT_TOPIC);
  38. mqttTopic.replace("{identifier}", getSetting("hostname"));
  39. if (!mqttTopic.endsWith("/")) mqttTopic = mqttTopic + "/";
  40. }
  41. bool mqttForward() {
  42. return _mqttForward;
  43. }
  44. String mqttSubtopic(char * topic) {
  45. String response;
  46. String t = String(topic);
  47. String mqttSetter = getSetting("mqttSetter", MQTT_USE_SETTER);
  48. if (t.startsWith(mqttTopic) && t.endsWith(mqttSetter)) {
  49. response = t.substring(mqttTopic.length(), t.length() - mqttSetter.length());
  50. }
  51. return response;
  52. }
  53. void mqttSendRaw(const char * topic, const char * message) {
  54. if (mqtt.connected()) {
  55. DEBUG_MSG_P(PSTR("[MQTT] Sending %s => %s\n"), topic, message);
  56. #if MQTT_USE_ASYNC
  57. mqtt.publish(topic, MQTT_QOS, MQTT_RETAIN, message);
  58. #else
  59. mqtt.publish(topic, message, MQTT_RETAIN);
  60. #endif
  61. }
  62. }
  63. void mqttSend(const char * topic, const char * message) {
  64. String mqttGetter = getSetting("mqttGetter", MQTT_USE_GETTER);
  65. String path = mqttTopic + String(topic) + mqttGetter;
  66. mqttSendRaw(path.c_str(), message);
  67. }
  68. void mqttSend(const char * topic, unsigned int index, const char * message) {
  69. String mqttGetter = getSetting("mqttGetter", MQTT_USE_GETTER);
  70. String path = mqttTopic + String(topic) + String ("/") + String(index) + mqttGetter;;
  71. mqttSendRaw(path.c_str(), message);
  72. }
  73. void mqttSubscribeRaw(const char * topic) {
  74. if (mqtt.connected() && (strlen(topic) > 0)) {
  75. DEBUG_MSG_P(PSTR("[MQTT] Subscribing to %s\n"), topic);
  76. mqtt.subscribe(topic, MQTT_QOS);
  77. }
  78. }
  79. void mqttSubscribe(const char * topic) {
  80. String mqttSetter = getSetting("mqttSetter", MQTT_USE_SETTER);
  81. String path = mqttTopic + String(topic) + mqttSetter;
  82. mqttSubscribeRaw(path.c_str());
  83. }
  84. // -----------------------------------------------------------------------------
  85. // Callbacks
  86. // -----------------------------------------------------------------------------
  87. void mqttRegister(void (*callback)(unsigned int, const char *, const char *)) {
  88. _mqtt_callbacks.push_back(callback);
  89. }
  90. void _mqttOnConnect() {
  91. DEBUG_MSG_P(PSTR("[MQTT] Connected!\n"));
  92. #if MQTT_SKIP_RETAINED
  93. mqttConnectedAt = millis();
  94. #endif
  95. // Build MQTT topics
  96. buildTopics();
  97. // Send first Heartbeat
  98. heartbeat();
  99. // Subscribe to system topics
  100. mqttSubscribe(MQTT_TOPIC_ACTION);
  101. // Send connect event to subscribers
  102. for (unsigned char i = 0; i < _mqtt_callbacks.size(); i++) {
  103. (*_mqtt_callbacks[i])(MQTT_CONNECT_EVENT, NULL, NULL);
  104. }
  105. }
  106. void _mqttOnDisconnect() {
  107. DEBUG_MSG_P(PSTR("[MQTT] Disconnected!\n"));
  108. // Send disconnect event to subscribers
  109. for (unsigned char i = 0; i < _mqtt_callbacks.size(); i++) {
  110. (*_mqtt_callbacks[i])(MQTT_DISCONNECT_EVENT, NULL, NULL);
  111. }
  112. }
  113. void _mqttOnMessage(char* topic, char* payload, unsigned int len) {
  114. char message[len + 1];
  115. strlcpy(message, (char *) payload, len + 1);
  116. DEBUG_MSG_P(PSTR("[MQTT] Received %s => %s"), topic, message);
  117. #if MQTT_SKIP_RETAINED
  118. if (millis() - mqttConnectedAt < MQTT_SKIP_TIME) {
  119. DEBUG_MSG_P(PSTR(" - SKIPPED\n"));
  120. return;
  121. }
  122. #endif
  123. DEBUG_MSG_P(PSTR("\n"));
  124. // Check system topics
  125. String t = mqttSubtopic((char *) topic);
  126. if (t.equals(MQTT_TOPIC_ACTION)) {
  127. if (strcmp(message, MQTT_ACTION_RESET) == 0) {
  128. ESP.restart();
  129. }
  130. }
  131. // Send message event to subscribers
  132. for (unsigned char i = 0; i < _mqtt_callbacks.size(); i++) {
  133. (*_mqtt_callbacks[i])(MQTT_MESSAGE_EVENT, topic, message);
  134. }
  135. }
  136. void mqttConnect() {
  137. if (!mqtt.connected()) {
  138. if (getSetting("mqttServer", MQTT_SERVER).length() == 0) return;
  139. // Last option: reconnect to wifi after MQTT_MAX_TRIES attemps in a row
  140. #if MQTT_MAX_TRIES > 0
  141. static unsigned int tries = 0;
  142. static unsigned long last_try = millis();
  143. if (millis() - last_try < MQTT_TRY_INTERVAL) {
  144. if (++tries > MQTT_MAX_TRIES) {
  145. DEBUG_MSG_P(PSTR("[MQTT] MQTT_MAX_TRIES met, disconnecting from WiFi\n"));
  146. wifiDisconnect();
  147. tries = 0;
  148. return;
  149. }
  150. } else {
  151. tries = 0;
  152. }
  153. last_try = millis();
  154. #endif
  155. mqtt.disconnect();
  156. if (_mqttUser) free(_mqttUser);
  157. if (_mqttPass) free(_mqttPass);
  158. char * host = strdup(getSetting("mqttServer", MQTT_SERVER).c_str());
  159. unsigned int port = getSetting("mqttPort", MQTT_PORT).toInt();
  160. _mqttUser = strdup(getSetting("mqttUser").c_str());
  161. _mqttPass = strdup(getSetting("mqttPassword").c_str());
  162. char * will = strdup((mqttTopic + MQTT_TOPIC_STATUS).c_str());
  163. DEBUG_MSG_P(PSTR("[MQTT] Connecting to broker at %s:%d"), host, port);
  164. mqtt.setServer(host, port);
  165. #if MQTT_USE_ASYNC
  166. mqtt.setKeepAlive(MQTT_KEEPALIVE).setCleanSession(false);
  167. mqtt.setWill(will, MQTT_QOS, MQTT_RETAIN, "0");
  168. if ((strlen(_mqttUser) > 0) && (strlen(_mqttPass) > 0)) {
  169. DEBUG_MSG_P(PSTR(" as user '%s'."), _mqttUser);
  170. mqtt.setCredentials(_mqttUser, _mqttPass);
  171. }
  172. DEBUG_MSG_P(PSTR("\n"));
  173. mqtt.connect();
  174. #else
  175. bool response;
  176. if ((strlen(_mqttUser) > 0) && (strlen(_mqttPass) > 0)) {
  177. DEBUG_MSG_P(PSTR(" as user '%s'\n"), _mqttUser);
  178. response = mqtt.connect(getIdentifier().c_str(), _mqttUser, _mqttPass, will, MQTT_QOS, MQTT_RETAIN, "0");
  179. } else {
  180. DEBUG_MSG_P(PSTR("\n"));
  181. response = mqtt.connect(getIdentifier().c_str(), will, MQTT_QOS, MQTT_RETAIN, "0");
  182. }
  183. if (response) {
  184. _mqttOnConnect();
  185. _mqttConnected = true;
  186. } else {
  187. DEBUG_MSG_P(PSTR("[MQTT] Connection failed\n"));
  188. }
  189. #endif
  190. free(host);
  191. free(will);
  192. String mqttSetter = getSetting("mqttSetter", MQTT_USE_SETTER);
  193. String mqttGetter = getSetting("mqttGetter", MQTT_USE_GETTER);
  194. bool _mqttForward = !mqttGetter.equals(mqttSetter);
  195. }
  196. }
  197. void mqttSetup() {
  198. #if MQTT_USE_ASYNC
  199. mqtt.onConnect([](bool sessionPresent) {
  200. _mqttOnConnect();
  201. });
  202. mqtt.onDisconnect([](AsyncMqttClientDisconnectReason reason) {
  203. _mqttOnDisconnect();
  204. });
  205. mqtt.onMessage([](char* topic, char* payload, AsyncMqttClientMessageProperties properties, size_t len, size_t index, size_t total) {
  206. _mqttOnMessage(topic, payload, len);
  207. });
  208. #else
  209. mqtt.setCallback([](char* topic, byte* payload, unsigned int length) {
  210. _mqttOnMessage(topic, (char *) payload, length);
  211. });
  212. #endif
  213. buildTopics();
  214. }
  215. void mqttLoop() {
  216. static unsigned long lastPeriod = 0;
  217. if (WiFi.status() == WL_CONNECTED) {
  218. if (!mqtt.connected()) {
  219. #if not MQTT_USE_ASYNC
  220. if (_mqttConnected) {
  221. _mqttOnDisconnect();
  222. _mqttConnected = false;
  223. }
  224. #endif
  225. unsigned long currPeriod = millis() / MQTT_RECONNECT_DELAY;
  226. if (currPeriod != lastPeriod) {
  227. lastPeriod = currPeriod;
  228. mqttConnect();
  229. }
  230. #if not MQTT_USE_ASYNC
  231. } else {
  232. mqtt.loop();
  233. #endif
  234. }
  235. }
  236. }