#include "services/mqtt_service.h" #include "services/config_service.h" #include "services/wifi_service.h" #include "services/ble_service.h" #include "device_state.h" #include "config.h" #include #include #include // MQTT client (needs WiFiClient) static WiFiClient wifiClient; static PubSubClient mqtt(wifiClient); MqttService& MqttService::getInstance() { static MqttService instance; return instance; } void MqttService::messageCallback(char* topic, byte* payload, unsigned int length) { getInstance().handleMessage(topic, payload, length); } void MqttService::setup() { mqtt.setCallback(messageCallback); mqtt.setBufferSize(512); } void MqttService::loop() { connectToMQTT(); mqtt.loop(); // Periodic status report if (mqttConnected && millis() - lastStatusReport >= STATUS_REPORT_INTERVAL) { lastStatusReport = millis(); publishStatus(); } } void MqttService::connectToMQTT() { if (!configService.isConfigured() || !wifiService.isConnected()) { return; } if (mqtt.connected()) { if (!mqttConnected) { mqttConnected = true; deviceState.setMqttConnected(true); } return; } // Don't spam connection attempts if (millis() - lastMqttAttempt < MQTT_RECONNECT_INTERVAL) { return; } lastMqttAttempt = millis(); mqttConnected = false; deviceState.setMqttConnected(false); mqtt.setServer( configService.getMqttHost().c_str(), configService.getMqttPort() ); Serial.print("Connecting to MQTT at "); Serial.print(configService.getMqttHost()); Serial.print(":"); Serial.println(configService.getMqttPort()); String clientId = String("esp32-") + String(random(0xffff), HEX); // LWT for disconnect detection String statusTopic = String("ble-sim/") + configService.getDeviceId() + "/status"; String willMessage = "{\"online\":false}"; if (mqtt.connect(clientId.c_str(), statusTopic.c_str(), 1, true, willMessage.c_str())) { mqttConnected = true; deviceState.setMqttConnected(true); Serial.println("MQTT connected with LWT!"); // Subscribe to control topics String configTopic = String("ble-sim/") + configService.getDeviceId() + "/config"; String setTopic = String("ble-sim/") + configService.getDeviceId() + "/set"; mqtt.subscribe(configTopic.c_str()); mqtt.subscribe(setTopic.c_str()); Serial.print("Subscribed to: "); Serial.println(configTopic); publishStatus(); } else { Serial.print("MQTT connection failed, rc="); Serial.println(mqtt.state()); } } void MqttService::handleMessage(char* topic, byte* payload, unsigned int length) { JsonDocument doc; DeserializationError error = deserializeJson(doc, payload, length); if (error) { Serial.print("JSON parse error: "); Serial.println(error.c_str()); return; } String topicStr = String(topic); String baseTopic = String("ble-sim/") + configService.getDeviceId(); // Handle device type configuration if (topicStr == baseTopic + "/config") { String type = doc["type"] | ""; Serial.print("Configuring as: "); Serial.println(type); if (type == "heart_rate") { deviceState.setDeviceType(DeviceType::HEART_RATE); bleService.setupHeartRate(); } else if (type == "treadmill") { deviceState.setDeviceType(DeviceType::TREADMILL); bleService.setupTreadmill(); } else { deviceState.setDeviceType(DeviceType::NONE); bleService.stop(); } publishStatus(); } // Handle value updates else if (topicStr == baseTopic + "/set") { if (doc["heart_rate"].is()) { deviceState.setHeartRate(doc["heart_rate"]); Serial.print("Heart rate set to: "); Serial.println(deviceState.getValues().heartRate); } if (doc["battery"].is()) { deviceState.setBatteryLevel(doc["battery"]); bleService.updateBattery(deviceState.getValues().batteryLevel); Serial.print("Battery level set to: "); Serial.println(deviceState.getValues().batteryLevel); } if (doc["speed"].is()) { deviceState.setTreadmillSpeed(doc["speed"]); Serial.print("Speed set to: "); Serial.println(doc["speed"].as()); } if (doc["incline"].is()) { deviceState.setTreadmillIncline(doc["incline"]); Serial.print("Incline set to: "); Serial.println(doc["incline"].as()); } if (doc["distance"].is()) { deviceState.setTreadmillDistance(doc["distance"]); Serial.print("Distance set to: "); Serial.println(deviceState.getValues().treadmillDistance); } publishValues(); } } void MqttService::publishStatus() { if (!mqtt.connected()) return; JsonDocument doc; doc["online"] = true; doc["firmware_version"] = FIRMWARE_VERSION; doc["type"] = deviceState.getDeviceTypeString(); doc["ble_started"] = deviceState.isBleStarted(); doc["ip"] = wifiService.getIP(); String payload; serializeJson(doc, payload); String topic = String("ble-sim/") + configService.getDeviceId() + "/status"; mqtt.publish(topic.c_str(), payload.c_str(), true); } void MqttService::publishValues() { if (!mqtt.connected()) return; JsonDocument doc; const auto& values = deviceState.getValues(); if (deviceState.getDeviceType() == DeviceType::HEART_RATE) { doc["heart_rate"] = values.heartRate; doc["battery"] = values.batteryLevel; } else if (deviceState.getDeviceType() == DeviceType::TREADMILL) { doc["speed"] = values.treadmillSpeed / 100.0; doc["incline"] = values.treadmillIncline / 10.0; doc["distance"] = values.treadmillDistance; } String payload; serializeJson(doc, payload); String topic = String("ble-sim/") + configService.getDeviceId() + "/values"; mqtt.publish(topic.c_str(), payload.c_str()); }