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.

208 lines
5.4 KiB

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 <AsyncMqttClient.h>
  7. #include <vector>
  8. AsyncMqttClient mqtt;
  9. String mqttTopic;
  10. std::vector<void (*)(unsigned int, const char *, const char *)> _mqtt_callbacks;
  11. #if MQTT_SKIP_RETAINED
  12. unsigned long mqttConnectedAt = 0;
  13. #endif
  14. // -----------------------------------------------------------------------------
  15. // MQTT
  16. // -----------------------------------------------------------------------------
  17. bool mqttConnected() {
  18. return mqtt.connected();
  19. }
  20. void mqttDisconnect() {
  21. mqtt.disconnect();
  22. }
  23. void buildTopics() {
  24. // Replace identifier
  25. mqttTopic = getSetting("mqttTopic", MQTT_TOPIC);
  26. mqttTopic.replace("{identifier}", getSetting("hostname"));
  27. }
  28. char * mqttSubtopic(char * topic) {
  29. int pos = min(mqttTopic.length(), strlen(topic));
  30. return topic + pos;
  31. }
  32. void mqttSendRaw(const char * topic, const char * message) {
  33. if (mqtt.connected()) {
  34. DEBUG_MSG("[MQTT] Sending %s => %s\n", topic, message);
  35. mqtt.publish(topic, MQTT_QOS, MQTT_RETAIN, message);
  36. }
  37. }
  38. void mqttSend(const char * topic, const char * message) {
  39. String path = mqttTopic + String(topic);
  40. mqttSendRaw(path.c_str(), message);
  41. }
  42. void mqttSubscribeRaw(const char * topic) {
  43. if (mqtt.connected()) {
  44. DEBUG_MSG("[MQTT] Subscribing to %s\n", topic);
  45. mqtt.subscribe(topic, MQTT_QOS);
  46. }
  47. }
  48. void mqttSubscribe(const char * topic) {
  49. String path = mqttTopic + String(topic);
  50. mqttSubscribeRaw(path.c_str());
  51. }
  52. void mqttRegister(void (*callback)(unsigned int, const char *, const char *)) {
  53. _mqtt_callbacks.push_back(callback);
  54. }
  55. void _mqttOnConnect(bool sessionPresent) {
  56. DEBUG_MSG("[MQTT] Connected!\n");
  57. #if MQTT_SKIP_RETAINED
  58. mqttConnectedAt = millis();
  59. #endif
  60. // Build MQTT topics
  61. buildTopics();
  62. // Say hello and report our IP and VERSION
  63. mqttSend(MQTT_IP_TOPIC, getIP().c_str());
  64. mqttSend(MQTT_VERSION_TOPIC, APP_VERSION);
  65. // Subscribe to system topics
  66. mqttSubscribe(MQTT_ACTION_TOPIC);
  67. // Send connect event to subscribers
  68. for (unsigned char i = 0; i < _mqtt_callbacks.size(); i++) {
  69. (*_mqtt_callbacks[i])(MQTT_CONNECT_EVENT, NULL, NULL);
  70. }
  71. }
  72. void _mqttOnDisconnect(AsyncMqttClientDisconnectReason reason) {
  73. DEBUG_MSG("[MQTT] Disconnected!\n");
  74. // Send disconnect event to subscribers
  75. for (unsigned char i = 0; i < _mqtt_callbacks.size(); i++) {
  76. (*_mqtt_callbacks[i])(MQTT_DISCONNECT_EVENT, NULL, NULL);
  77. }
  78. }
  79. void _mqttOnMessage(char* topic, char* payload, AsyncMqttClientMessageProperties properties, size_t len, size_t index, size_t total) {
  80. char message[len+1];
  81. strlcpy(message, payload, len+1);
  82. DEBUG_MSG("[MQTT] Received %s => %s", topic, message);
  83. #if MQTT_SKIP_RETAINED
  84. if (millis() - mqttConnectedAt < MQTT_SKIP_TIME) {
  85. DEBUG_MSG(" - SKIPPED\n");
  86. return;
  87. }
  88. #endif
  89. DEBUG_MSG("\n");
  90. // Check system topics
  91. char * p = mqttSubtopic(topic);
  92. if (strcmp(p, MQTT_ACTION_TOPIC) == 0) {
  93. if (strcmp(payload, MQTT_ACTION_RESET) == 0) {
  94. ESP.reset();
  95. }
  96. }
  97. // Send message event to subscribers
  98. // Topic is set to the specific part each one might be checking
  99. for (unsigned char i = 0; i < _mqtt_callbacks.size(); i++) {
  100. (*_mqtt_callbacks[i])(MQTT_MESSAGE_EVENT, topic, message);
  101. }
  102. }
  103. void mqttConnect() {
  104. if (!mqtt.connected()) {
  105. char * host = strdup(getSetting("mqttServer", MQTT_SERVER).c_str());
  106. if (strlen(host) == 0) return;
  107. // Last option: reconnect to wifi after MQTT_MAX_TRIES attemps in a row
  108. #if MQTT_MAX_TRIES > 0
  109. static unsigned int tries = 0;
  110. static unsigned long last_try = millis();
  111. if (millis() - last_try < MQTT_TRY_INTERVAL) {
  112. if (++tries > MQTT_MAX_TRIES) {
  113. DEBUG_MSG("[MQTT] MQTT_MAX_TRIES met, disconnecting from WiFi\n");
  114. wifiDisconnect();
  115. tries = 0;
  116. return;
  117. }
  118. } else {
  119. tries = 0;
  120. }
  121. last_try = millis();
  122. #endif
  123. mqtt.disconnect();
  124. unsigned int port = getSetting("mqttPort", MQTT_PORT).toInt();
  125. char * user = strdup(getSetting("mqttUser").c_str());
  126. char * pass = strdup(getSetting("mqttPassword").c_str());
  127. DEBUG_MSG("[MQTT] Connecting to broker at %s", host);
  128. mqtt.setServer(host, port);
  129. mqtt.setKeepAlive(MQTT_KEEPALIVE).setCleanSession(false);
  130. mqtt.setWill((mqttTopic + MQTT_HEARTBEAT_TOPIC).c_str(), MQTT_QOS, MQTT_RETAIN, "0");
  131. if ((strlen(user) > 0) && (strlen(pass) > 0)) {
  132. DEBUG_MSG(" as user '%s'.", user);
  133. mqtt.setCredentials(user, pass);
  134. }
  135. DEBUG_MSG("\n");
  136. mqtt.connect();
  137. }
  138. }
  139. void mqttSetup() {
  140. mqtt.onConnect(_mqttOnConnect);
  141. mqtt.onDisconnect(_mqttOnDisconnect);
  142. mqtt.onMessage(_mqttOnMessage);
  143. buildTopics();
  144. }
  145. void mqttLoop() {
  146. static unsigned long lastPeriod = 0;
  147. if (WiFi.status() == WL_CONNECTED) {
  148. if (!mqtt.connected()) {
  149. unsigned long currPeriod = millis() / MQTT_RECONNECT_DELAY;
  150. if (currPeriod != lastPeriod) {
  151. lastPeriod = currPeriod;
  152. mqttConnect();
  153. }
  154. }
  155. }
  156. }