#include "mqtt.h" namespace config { // Values should be defined in config.h uint16_t mqtt_sending_interval = MQTT_SENDING_INTERVAL; // [s] //INFO: Listen to every CO2 sensor which is connected to the server: // mosquitto_sub -h MQTT_SERVER -t 'CO2sensors/#' -p 443 --capath /etc/ssl/certs/ -u "MQTT_USER" -P "MQTT_PASSWORD" -v const char *mqtt_server = MQTT_SERVER; const uint16_t mqtt_port = MQTT_PORT; const char *mqtt_user = MQTT_USER; const char *mqtt_password = MQTT_PASSWORD; const bool allow_mqtt_commands = ALLOW_MQTT_COMMANDS; const unsigned long wait_after_fail = 900; // [s] Wait 15 minutes after an MQTT connection fail, before trying again. } #if MQTT_ENCRYPTED # if defined(ESP32) # include # endif WiFiClientSecure espClient; #else WiFiClient espClient; #endif PubSubClient mqttClient(espClient); namespace mqtt { unsigned long last_sent_at = 0; unsigned long last_failed_at = 0; bool connected = false; char publish_topic[21]; // e.g. "CO2sensors/ESPxxxxxx\0" const char *json_sensor_format; char last_successful_publish[23] = ""; void initialize(const char *sensorId) { json_sensor_format = PSTR("{\"time\":\"%s\", \"co2\":%d, \"temp\":%.1f, \"rh\":%.1f}"); snprintf(publish_topic, sizeof(publish_topic), "CO2sensors/%s", sensorId); #if MQTT_ENCRYPTED // The sensor doesn't check the fingerprint of the MQTT broker, because otherwise this fingerprint should be updated // on the sensor every 3 months. The connection can still be encrypted, though: espClient.setInsecure(); // If not available for ESP32, please update Arduino IDE / PlatformIO #endif mqttClient.setServer(config::mqtt_server, config::mqtt_port); sensor_console::defineIntCommand("mqtt", setMQTTinterval, F("60 (Sets MQTT sending interval, in s)")); sensor_console::defineCommand("send_local_ip", sendInfoAboutLocalNetwork, F("(Sends local IP and SSID via MQTT. Can be useful to find sensor)")); } void publish(const char *timestamp, int16_t co2, float temperature, float humidity) { if (WiFi.status() == WL_CONNECTED && mqttClient.connected()) { led_effects::onBoardLEDOn(); Serial.print(F("MQTT - Publishing message ... ")); char payload[75]; // Should be enough for json... snprintf(payload, sizeof(payload), json_sensor_format, timestamp, co2, temperature, humidity); // Topic is the same as clientID. e.g. 'CO2sensors/ESP3d03da' if (mqttClient.publish(publish_topic, payload)) { Serial.println(F("OK")); ntp::getLocalTime(last_successful_publish); } else { Serial.println(F("Failed.")); } led_effects::onBoardLEDOff(); } } /** * Allows sensor to be controlled by commands over MQTT * * mosquitto_pub -h MQTT_SERVER -t 'CO2sensors/SENSOR_ID/control' -p 443 --capath /etc/ssl/certs/ -u "MQTT_USER" -P "MQTT_PASSWORD" -m "reset" * mosquitto_pub -h MQTT_SERVER -t 'CO2sensors/SENSOR_ID/control' -p 443 --capath /etc/ssl/certs/ -u "MQTT_USER" -P "MQTT_PASSWORD" -m "timer 30" * mosquitto_pub -h MQTT_SERVER -t 'CO2sensors/SENSOR_ID/control' -p 443 --capath /etc/ssl/certs/ -u "MQTT_USER" -P "MQTT_PASSWORD" -m "mqtt 900" * mosquitto_pub -h MQTT_SERVER -t 'CO2sensors/SENSOR_ID/control' -p 443 --capath /etc/ssl/certs/ -u "MQTT_USER" -P "MQTT_PASSWORD" -m "calibrate 700" */ void controlSensorCallback(char *sub_topic, byte *message, unsigned int length) { if (length == 0) { return; } led_effects::onBoardLEDOn(); Serial.print(F("Message arrived on topic: ")); Serial.println(sub_topic); char command[length + 1]; for (unsigned int i = 0; i < length; i++) { command[i] = message[i]; } command[length] = 0; sensor_console::execute(command); led_effects::onBoardLEDOff(); } void reconnect() { if (last_failed_at > 0 && (seconds() - last_failed_at < config::wait_after_fail)) { // It failed less than wait_after_fail ago. Not even trying. return; } if (WiFi.status() != WL_CONNECTED) { //NOTE: Sadly, WiFi.status is sometimes WL_CONNECTED even though it's really not // No WIFI return; } Serial.print(F("MQTT - Attempting connection to ")); Serial.print(MQTT_SERVER); Serial.print(MQTT_ENCRYPTED ? F(" (Encrypted") : F(" (Unencrypted")); Serial.print(F(", port ")); Serial.print(MQTT_PORT); Serial.print(F(") ...")); led_effects::onBoardLEDOn(); // Wait for connection, at most 15s (default) mqttClient.connect(publish_topic, config::mqtt_user, config::mqtt_password); led_effects::onBoardLEDOff(); connected = mqttClient.connected(); if (connected) { if (config::allow_mqtt_commands) { char control_topic[60]; // Should be enough for "CO2sensors/ESPd03cc5/control" snprintf(control_topic, sizeof(control_topic), "%s/control", publish_topic); mqttClient.subscribe(control_topic); mqttClient.setCallback(controlSensorCallback); } Serial.println(F(" Connected.")); last_failed_at = 0; } else { last_failed_at = seconds(); Serial.print(F(" Failed! Error code=")); Serial.print(mqttClient.state()); Serial.print(F(". Will try again in ")); Serial.print(config::wait_after_fail); Serial.println("s."); } } void publishIfTimeHasCome(const char *timestamp, const int16_t &co2, const float &temp, const float &hum) { // Send message via MQTT according to sending interval unsigned long now = seconds(); if (now - last_sent_at > config::mqtt_sending_interval) { last_sent_at = now; publish(timestamp, co2, temp, hum); } } void keepConnection() { // Keep MQTT connection if (!mqttClient.connected()) { reconnect(); } mqttClient.loop(); } /***************************************************************** * Callbacks for sensor commands * *****************************************************************/ void setMQTTinterval(int32_t sending_interval) { config::mqtt_sending_interval = sending_interval; Serial.print(F("Setting MQTT sending interval to : ")); Serial.print(config::mqtt_sending_interval); Serial.println("s."); led_effects::showKITTWheel(color::green, 1); } // It can be hard to find the local IP of a sensor if it isn't connected to Serial port, and if mDNS is disabled. // If the sensor can be reach by MQTT, it can answer with info about local_ip and ssid. // The sensor will send the info to "CO2sensors/ESP123456/info". void sendInfoAboutLocalNetwork() { char info_topic[60]; // Should be enough for "CO2sensors/ESP123456/info" snprintf(info_topic, sizeof(info_topic), "%s/info", publish_topic); char payload[75]; // Should be enough for info json... const char *json_info_format = PSTR("{\"local_ip\":\"%s\", \"ssid\":\"%s\"}"); snprintf(payload, sizeof(payload), json_info_format, wifi::local_ip, WIFI_SSID); mqttClient.publish(info_topic, payload); } }