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.

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