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.

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