bonamin/project-vesper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fixed MQTT and WS Routing - w/o SSL

Browse Source
This commit is contained in:
Vasileios Kousaridas
2025-10-13 17:34:54 +03:00
parent f696984cd1
commit 956786321a
29 changed files with 2043 additions and 1210 deletions
Download Patch File Download Diff File Expand all files Collapse all files

724
vesper/src/Communication/Communication.cpp → vesper/src/Communication/CommandHandler/CommandHandler.cpp
View File

@@ -1,421 +1,55 @@
#include "Communication.hpp"
#include "../ConfigManager/ConfigManager.hpp"
#include "../OTAManager/OTAManager.hpp"
#include "../Networking/Networking.hpp"
#include "../Logging/Logging.hpp"
#include "../Player/Player.hpp"
#include "../FileManager/FileManager.hpp"
#include "../TimeKeeper/TimeKeeper.hpp"
#include "../FirmwareValidator/FirmwareValidator.hpp"
#include "../MqttSSL/MqttSSL.hpp"
#include <WiFiClientSecure.h>
/*
* COMMANDHANDLER.CPP - Command Processing Implementation
*/
Communication* Communication::_instance = nullptr;
StaticJsonDocument<2048> Communication::_parseDocument;
#include "CommandHandler.hpp"
#include "../../ConfigManager/ConfigManager.hpp"
#include "../../OTAManager/OTAManager.hpp"
#include "../../Player/Player.hpp"
#include "../../FileManager/FileManager.hpp"
#include "../../TimeKeeper/TimeKeeper.hpp"
#include "../../FirmwareValidator/FirmwareValidator.hpp"
#include "../../ClientManager/ClientManager.hpp"
#include "../../Logging/Logging.hpp"
#include "../ResponseBuilder/ResponseBuilder.hpp"
static void connectToMqttWrapper(TimerHandle_t xTimer) {
if (Communication::_instance) {
Communication::_instance->connectToMqtt();
}
}
Communication::Communication(ConfigManager& configManager,
OTAManager& otaManager,
Networking& networking,
AsyncMqttClient& mqttClient,
AsyncWebServer& server,
AsyncWebSocket& webSocket,
AsyncUDP& udp)
CommandHandler::CommandHandler(ConfigManager& configManager, OTAManager& otaManager)
: _configManager(configManager)
, _otaManager(otaManager)
, _networking(networking)
, _mqttClient(mqttClient)
, _server(server)
, _webSocket(webSocket)
, _udp(udp)
, _player(nullptr)
, _fileManager(nullptr)
, _timeKeeper(nullptr)
, _firmwareValidator(nullptr)
, _mqttReconnectTimer(nullptr) {
, _clientManager(nullptr)
, _responseCallback(nullptr) {}
_instance = this;
CommandHandler::~CommandHandler() {}
void CommandHandler::setPlayerReference(Player* player) {
_player = player;
}
Communication::~Communication() {
if (_mqttReconnectTimer != nullptr) {
xTimerDelete(_mqttReconnectTimer, portMAX_DELAY);
_mqttReconnectTimer = nullptr;
}
_instance = nullptr;
void CommandHandler::setFileManagerReference(FileManager* fm) {
_fileManager = fm;
}
void Communication::begin() {
LOG_INFO("Initializing Communication Manager v2.1");
initMqtt();
initWebSocket();
LOG_INFO("Communication Manager initialized with multi-client support");
void CommandHandler::setTimeKeeperReference(Timekeeper* tk) {
_timeKeeper = tk;
}
void Communication::initMqtt() {
_mqttReconnectTimer = xTimerCreate("mqttTimer", pdMS_TO_TICKS(2000), pdFALSE,
(void*)0, connectToMqttWrapper);
_mqttClient.onConnect(onMqttConnect);
_mqttClient.onDisconnect(onMqttDisconnect);
_mqttClient.onSubscribe(onMqttSubscribe);
_mqttClient.onUnsubscribe(onMqttUnsubscribe);
_mqttClient.onMessage(onMqttMessage);
_mqttClient.onPublish(onMqttPublish);
auto& mqttConfig = _configManager.getMqttConfig();
// Log SSL status - AsyncMqttClient SSL is compile-time configured
MqttSSL::logSSLStatus(_mqttClient, mqttConfig.port);
// DEBUG: Log connection details
LOG_INFO("MQTT Config: host=%s, port=%d, user=%s, pass=%s",
mqttConfig.host.c_str(), mqttConfig.port,
mqttConfig.user.c_str(), mqttConfig.password.c_str());
_mqttClient.setServer(mqttConfig.host.c_str(), mqttConfig.port);
_mqttClient.setCredentials(mqttConfig.user.c_str(), mqttConfig.password.c_str());
void CommandHandler::setFirmwareValidatorReference(FirmwareValidator* fv) {
_firmwareValidator = fv;
}
void Communication::initWebSocket() {
_webSocket.onEvent(onWebSocketEvent);
_server.addHandler(&_webSocket);
void CommandHandler::setClientManagerReference(ClientManager* cm) {
_clientManager = cm;
}
void Communication::connectToMqtt() {
if (_networking.isConnected()) {
LOG_INFO("Connecting to MQTT...");
_mqttClient.connect();
} else {
LOG_WARNING("Cannot connect to MQTT: No network connection");
}
void CommandHandler::setResponseCallback(ResponseCallback callback) {
_responseCallback = callback;
}
void Communication::subscribeMqtt() {
char topic[64];
snprintf(topic, sizeof(topic), "vesper/%s/control", _configManager.getDeviceUID().c_str());
uint16_t topicId = _mqttClient.subscribe(topic, 2);
LOG_INFO("Subscribing to Command topic, QoS 2, packetId: %d", topicId);
}
void Communication::sendResponse(const String& response, const MessageContext& context) {
if (context.source == MessageSource::MQTT) {
publishToMqtt(response);
} else if (context.source == MessageSource::WEBSOCKET) {
_clientManager.sendToClient(context.clientId, response);
}
}
void Communication::sendSuccessResponse(const String& type, const String& payload, const MessageContext& context) {
String response = ResponseBuilder::success(type, payload);
sendResponse(response, context);
}
void Communication::sendErrorResponse(const String& type, const String& message, const MessageContext& context) {
String response = ResponseBuilder::error(type, message);
sendResponse(response, context);
}
void Communication::broadcastStatus(const String& statusMessage) {
publishToMqtt(statusMessage);
broadcastToAllWebSocketClients(statusMessage);
}
void Communication::broadcastStatus(const JsonDocument& statusJson) {
String statusMessage;
serializeJson(statusJson, statusMessage);
broadcastStatus(statusMessage);
}
void Communication::sendBellOverloadNotification(const std::vector<uint8_t>& bellNumbers,
const std::vector<uint16_t>& bellLoads,
const String& severity) {
StaticJsonDocument<512> overloadMsg;
overloadMsg["status"] = "INFO";
overloadMsg["type"] = "bell_overload";
JsonArray bellsArray = overloadMsg["payload"]["bells"].to<JsonArray>();
JsonArray loadsArray = overloadMsg["payload"]["loads"].to<JsonArray>();
for (size_t i = 0; i < bellNumbers.size() && i < bellLoads.size(); i++) {
bellsArray.add(bellNumbers[i] + 1);
loadsArray.add(bellLoads[i]);
}
overloadMsg["payload"]["severity"] = severity;
broadcastStatus(overloadMsg);
LOG_WARNING("Bell overload notification sent: %d bells, severity: %s",
bellNumbers.size(), severity.c_str());
}
void Communication::broadcastToMasterClients(const String& message) {
_clientManager.sendToMasterClients(message);
}
void Communication::broadcastToSecondaryClients(const String& message) {
_clientManager.sendToSecondaryClients(message);
}
void Communication::broadcastToAllWebSocketClients(const String& message) {
_clientManager.broadcastToAll(message);
}
void Communication::broadcastToAllWebSocketClients(const JsonDocument& message) {
String messageStr;
serializeJson(message, messageStr);
_clientManager.broadcastToAll(messageStr);
LOG_DEBUG("Broadcasted JSON to WebSocket clients: %s", messageStr.c_str());
}
void Communication::publishToMqtt(const String& data) {
if (_mqttClient.connected()) {
char topicData[64];
snprintf(topicData, sizeof(topicData), "vesper/%s/data", _configManager.getDeviceUID().c_str());
_mqttClient.publish(topicData, 0, true, data.c_str());
LOG_DEBUG("Published to MQTT: %s", data.c_str());
} else {
LOG_ERROR("MQTT Not Connected! Message Failed: %s", data.c_str());
}
}
void Communication::onNetworkConnected() {
LOG_DEBUG("Network connected - attempting MQTT connection");
connectToMqtt();
}
void Communication::onNetworkDisconnected() {
LOG_DEBUG("Network disconnected - stopping MQTT timer");
xTimerStop(_mqttReconnectTimer, 0);
}
void Communication::setupUdpDiscovery() {
uint16_t discoveryPort = _configManager.getNetworkConfig().discoveryPort;
if (_udp.listen(discoveryPort)) {
LOG_INFO("UDP discovery listening on port %u", discoveryPort);
_udp.onPacket([this](AsyncUDPPacket packet) {
String msg = String((const char*)packet.data(), packet.length());
LOG_DEBUG("UDP from %s:%u -> %s",
packet.remoteIP().toString().c_str(),
packet.remotePort(),
msg.c_str());
bool shouldReply = false;
if (msg.indexOf("discover") >= 0) {
shouldReply = true;
} else {
StaticJsonDocument<128> req;
DeserializationError err = deserializeJson(req, msg);
if (!err) {
shouldReply = (req["op"] == "discover" && req["svc"] == "vesper");
}
}
if (!shouldReply) return;
StaticJsonDocument<256> doc;
doc["op"] = "discover_reply";
doc["svc"] = "vesper";
doc["ver"] = 1;
doc["name"] = "Proj. Vesper v0.5";
doc["id"] = _configManager.getDeviceUID();
doc["ip"] = _networking.getLocalIP();
char wsUrl[64];
snprintf(wsUrl, sizeof(wsUrl), "ws://%s/ws", _networking.getLocalIP().c_str());
doc["ws"] = wsUrl;
doc["port"] = 80;
doc["fw"] = "1.2.3";
doc["clients"] = _clientManager.getClientCount();
String out;
serializeJson(doc, out);
_udp.writeTo((const uint8_t*)out.c_str(), out.length(),
packet.remoteIP(), packet.remotePort());
});
} else {
LOG_ERROR("Failed to start UDP discovery.");
}
}
void Communication::onMqttConnect(bool sessionPresent) {
LOG_INFO("Connected to MQTT");
if (_instance) {
_instance->subscribeMqtt();
}
}
void Communication::onMqttDisconnect(AsyncMqttClientDisconnectReason reason) {
String reasonStr;
switch (reason) {
case AsyncMqttClientDisconnectReason::TCP_DISCONNECTED:
reasonStr = "TCP_DISCONNECTED";
break;
case AsyncMqttClientDisconnectReason::MQTT_UNACCEPTABLE_PROTOCOL_VERSION:
reasonStr = "UNACCEPTABLE_PROTOCOL_VERSION";
break;
case AsyncMqttClientDisconnectReason::MQTT_IDENTIFIER_REJECTED:
reasonStr = "IDENTIFIER_REJECTED";
break;
case AsyncMqttClientDisconnectReason::MQTT_SERVER_UNAVAILABLE:
reasonStr = "SERVER_UNAVAILABLE";
break;
case AsyncMqttClientDisconnectReason::MQTT_MALFORMED_CREDENTIALS:
reasonStr = "MALFORMED_CREDENTIALS";
break;
case AsyncMqttClientDisconnectReason::MQTT_NOT_AUTHORIZED:
reasonStr = "NOT_AUTHORIZED";
break;
case AsyncMqttClientDisconnectReason::TLS_BAD_FINGERPRINT:
reasonStr = "TLS_BAD_FINGERPRINT";
break;
default:
reasonStr = "UNKNOWN(" + String((int)reason) + ")";
break;
}
LOG_WARNING("Disconnected from MQTT: %s", reasonStr.c_str());
if (_instance && _instance->_networking.isConnected()) {
xTimerStart(_instance->_mqttReconnectTimer, 0);
}
}
void Communication::onMqttMessage(char* topic, char* payload, AsyncMqttClientMessageProperties properties,
size_t len, size_t index, size_t total) {
if (!_instance) return;
char expectedTopic[64];
snprintf(expectedTopic, sizeof(expectedTopic), "vesper/%s/control", _instance->_configManager.getDeviceUID().c_str());
if (strcmp(topic, expectedTopic) == 0) {
JsonDocument command = _instance->parsePayload(payload);
MessageContext context(MessageSource::MQTT);
_instance->handleCommand(command, context);
}
}
void Communication::onMqttSubscribe(uint16_t packetId, uint8_t qos) {
LOG_INFO("Subscribe acknowledged. PacketID: %d / QoS: %d", packetId, qos);
}
void Communication::onMqttUnsubscribe(uint16_t packetId) {
LOG_INFO("Unsubscribe Acknowledged. PacketID: %d", packetId);
}
void Communication::onMqttPublish(uint16_t packetId) {
LOG_DEBUG("Publish Acknowledged. PacketID: %d", packetId);
}
void Communication::onWebSocketEvent(AsyncWebSocket* server, AsyncWebSocketClient* client,
AwsEventType type, void* arg, uint8_t* data, size_t len) {
if (!_instance) return;
switch (type) {
case WS_EVT_CONNECT:
_instance->onWebSocketConnect(client);
break;
case WS_EVT_DISCONNECT:
_instance->onWebSocketDisconnect(client);
break;
case WS_EVT_DATA:
_instance->onWebSocketReceived(client, arg, data, len);
break;
case WS_EVT_ERROR:
LOG_ERROR("WebSocket client #%u error(%u): %s", client->id(), *((uint16_t*)arg), (char*)data);
break;
default:
break;
}
}
void Communication::onWebSocketConnect(AsyncWebSocketClient* client) {
LOG_INFO("WebSocket client #%u connected from %s", client->id(), client->remoteIP().toString().c_str());
_clientManager.addClient(client, ClientManager::DeviceType::UNKNOWN);
String welcomeMsg = ResponseBuilder::success("connection", "Connected to Vesper");
_clientManager.sendToClient(client->id(), welcomeMsg);
}
void Communication::onWebSocketDisconnect(AsyncWebSocketClient* client) {
LOG_INFO("WebSocket client #%u disconnected", client->id());
_clientManager.removeClient(client->id());
_clientManager.cleanupDisconnectedClients();
}
void Communication::onWebSocketReceived(AsyncWebSocketClient* client, void* arg, uint8_t* data, size_t len) {
AwsFrameInfo* info = (AwsFrameInfo*)arg;
if (info->final && info->index == 0 && info->len == len && info->opcode == WS_TEXT) {
char* payload = (char*)malloc(len + 1);
if (!payload) {
LOG_ERROR("Failed to allocate memory for WebSocket payload");
return;
}
memcpy(payload, data, len);
payload[len] = '\0';
LOG_DEBUG("Received WebSocket message from client #%u (length %d): %s",
client->id(), len, payload);
JsonDocument command = parsePayload(payload);
MessageContext context(MessageSource::WEBSOCKET, client->id());
_clientManager.updateClientLastSeen(client->id());
handleCommand(command, context);
free(payload);
} else {
LOG_WARNING("Received fragmented or non-text WebSocket message - ignoring");
}
}
JsonDocument Communication::parsePayload(char* payload) {
_parseDocument.clear();
size_t payloadLen = strlen(payload);
LOG_DEBUG("Parsing payload (length %d): %s", payloadLen, payload);
if (payloadLen == 0) {
LOG_ERROR("Empty payload received");
return _parseDocument;
}
String cleanJson = String(payload);
cleanJson.replace("\r\n", "");
cleanJson.replace("\n", "");
cleanJson.replace("\r", "");
cleanJson.trim();
LOG_DEBUG("Cleaned JSON: %s", cleanJson.c_str());
DeserializationError error = deserializeJson(_parseDocument, cleanJson);
if (error) {
LOG_ERROR("JSON deserialization failed: %s", error.c_str());
} else {
LOG_DEBUG("JSON parsed successfully");
}
return _parseDocument;
}
void Communication::handleCommand(JsonDocument& command, const MessageContext& context) {
void CommandHandler::processCommand(JsonDocument& command, const MessageContext& context) {
String cmd = command["cmd"];
JsonVariant contents = command["contents"];
@@ -444,12 +78,28 @@ void Communication::handleCommand(JsonDocument& command, const MessageContext& c
}
}
void Communication::handlePingCommand(const MessageContext& context) {
void CommandHandler::sendResponse(const String& response, const MessageContext& context) {
if (_responseCallback) {
_responseCallback(response, context);
}
}
void CommandHandler::sendSuccessResponse(const String& type, const String& payload, const MessageContext& context) {
String response = ResponseBuilder::success(type, payload);
sendResponse(response, context);
}
void CommandHandler::sendErrorResponse(const String& type, const String& message, const MessageContext& context) {
String response = ResponseBuilder::error(type, message);
sendResponse(response, context);
}
void CommandHandler::handlePingCommand(const MessageContext& context) {
String response = ResponseBuilder::pong();
sendResponse(response, context);
}
void Communication::handleStatusCommand(const MessageContext& context) {
void CommandHandler::handleStatusCommand(const MessageContext& context) {
PlayerStatus playerStatus = _player ? _player->getStatus() : PlayerStatus::STOPPED;
uint32_t timeElapsedMs = 0;
uint64_t projectedRunTime = 0;
@@ -468,12 +118,19 @@ void Communication::handleStatusCommand(const MessageContext& context) {
sendResponse(response, context);
}
void Communication::handleIdentifyCommand(JsonVariant contents, const MessageContext& context) {
void CommandHandler::handleIdentifyCommand(JsonVariant contents, const MessageContext& context) {
if (context.source != MessageSource::WEBSOCKET) {
sendErrorResponse("identify", "Identify command only available via WebSocket", context);
return;
}
// 🛡️ SAFETY CHECK: Ensure ClientManager reference is set
if (!_clientManager) {
LOG_ERROR("ClientManager reference not set in CommandHandler!");
sendErrorResponse("identify", "Internal error: ClientManager not available", context);
return;
}
if (!contents.containsKey("device_type")) {
sendErrorResponse("identify", "Missing device_type parameter", context);
return;
@@ -489,7 +146,7 @@ void Communication::handleIdentifyCommand(JsonVariant contents, const MessageCon
}
if (deviceType != ClientManager::DeviceType::UNKNOWN) {
_clientManager.updateClientType(context.clientId, deviceType);
_clientManager->updateClientType(context.clientId, deviceType);
sendSuccessResponse("identify", "Device identified as " + deviceTypeStr, context);
LOG_INFO("Client #%u identified as %s device", context.clientId, deviceTypeStr.c_str());
} else {
@@ -497,7 +154,7 @@ void Communication::handleIdentifyCommand(JsonVariant contents, const MessageCon
}
}
void Communication::handlePlaybackCommand(JsonVariant contents, const MessageContext& context) {
void CommandHandler::handlePlaybackCommand(JsonVariant contents, const MessageContext& context) {
if (_player) {
bool success = _player->command(contents);
@@ -512,7 +169,7 @@ void Communication::handlePlaybackCommand(JsonVariant contents, const MessageCon
}
}
void Communication::handleFileManagerCommand(JsonVariant contents, const MessageContext& context) {
void CommandHandler::handleFileManagerCommand(JsonVariant contents, const MessageContext& context) {
if (!contents.containsKey("action")) {
sendErrorResponse("file_manager", "Missing action parameter", context);
return;
@@ -533,7 +190,7 @@ void Communication::handleFileManagerCommand(JsonVariant contents, const Message
}
}
void Communication::handleRelaySetupCommand(JsonVariant contents, const MessageContext& context) {
void CommandHandler::handleRelaySetupCommand(JsonVariant contents, const MessageContext& context) {
if (!contents.containsKey("action")) {
sendErrorResponse("relay_setup", "Missing action parameter", context);
return;
@@ -552,7 +209,7 @@ void Communication::handleRelaySetupCommand(JsonVariant contents, const MessageC
}
}
void Communication::handleClockSetupCommand(JsonVariant contents, const MessageContext& context) {
void CommandHandler::handleClockSetupCommand(JsonVariant contents, const MessageContext& context) {
if (!contents.containsKey("action")) {
sendErrorResponse("clock_setup", "Missing action parameter", context);
return;
@@ -585,7 +242,7 @@ void Communication::handleClockSetupCommand(JsonVariant contents, const MessageC
}
}
void Communication::handleSystemInfoCommand(JsonVariant contents, const MessageContext& context) {
void CommandHandler::handleSystemInfoCommand(JsonVariant contents, const MessageContext& context) {
if (!contents.containsKey("action")) {
sendErrorResponse("system_info", "Missing action parameter", context);
return;
@@ -606,13 +263,15 @@ void Communication::handleSystemInfoCommand(JsonVariant contents, const MessageC
handleRollbackFirmwareCommand(context);
} else if (action == "get_firmware_status") {
handleGetFirmwareStatusCommand(context);
} else if (action == "get_full_settings") {
handleGetFullSettingsCommand(context);
} else {
LOG_WARNING("Unknown system info action: %s", action.c_str());
sendErrorResponse("system_info", "Unknown action: " + action, context);
}
}
void Communication::handleListMelodiesCommand(const MessageContext& context) {
void CommandHandler::handleListMelodiesCommand(const MessageContext& context) {
if (!_fileManager) {
sendErrorResponse("list_melodies", "FileManager not available", context);
return;
@@ -639,7 +298,7 @@ void Communication::handleListMelodiesCommand(const MessageContext& context) {
sendResponse(responseStr, context);
}
void Communication::handleDownloadMelodyCommand(JsonVariant contents, const MessageContext& context) {
void CommandHandler::handleDownloadMelodyCommand(JsonVariant contents, const MessageContext& context) {
if (!_fileManager) {
sendErrorResponse("download_melody", "FileManager not available", context);
return;
@@ -653,7 +312,7 @@ void Communication::handleDownloadMelodyCommand(JsonVariant contents, const Mess
sendResponse(response, context);
}
void Communication::handleDeleteMelodyCommand(JsonVariant contents, const MessageContext& context) {
void CommandHandler::handleDeleteMelodyCommand(JsonVariant contents, const MessageContext& context) {
if (!_fileManager) {
sendErrorResponse("delete_melody", "FileManager not available", context);
return;
@@ -676,7 +335,7 @@ void Communication::handleDeleteMelodyCommand(JsonVariant contents, const Messag
}
}
void Communication::handleSetRelayTimersCommand(JsonVariant contents, const MessageContext& context) {
void CommandHandler::handleSetRelayTimersCommand(JsonVariant contents, const MessageContext& context) {
try {
_configManager.updateBellDurations(contents);
// Save configuration to ensure persistence
@@ -694,7 +353,7 @@ void Communication::handleSetRelayTimersCommand(JsonVariant contents, const Mess
}
}
void Communication::handleSetRelayOutputsCommand(JsonVariant contents, const MessageContext& context) {
void CommandHandler::handleSetRelayOutputsCommand(JsonVariant contents, const MessageContext& context) {
try {
_configManager.updateBellOutputs(contents);
// Note: Bell outputs are typically not persisted to SD card as they're more of a mapping configuration
@@ -706,7 +365,7 @@ void Communication::handleSetRelayOutputsCommand(JsonVariant contents, const Mes
}
}
void Communication::handleSetClockOutputsCommand(JsonVariant contents, const MessageContext& context) {
void CommandHandler::handleSetClockOutputsCommand(JsonVariant contents, const MessageContext& context) {
try {
_configManager.updateClockOutputs(contents);
// Save configuration to ensure persistence
@@ -724,7 +383,7 @@ void Communication::handleSetClockOutputsCommand(JsonVariant contents, const Mes
}
}
void Communication::handleSetClockTimingsCommand(JsonVariant contents, const MessageContext& context) {
void CommandHandler::handleSetClockTimingsCommand(JsonVariant contents, const MessageContext& context) {
try {
_configManager.updateClockOutputs(contents);
// Save configuration to ensure persistence
@@ -742,7 +401,7 @@ void Communication::handleSetClockTimingsCommand(JsonVariant contents, const Mes
}
}
void Communication::handleSetClockAlertsCommand(JsonVariant contents, const MessageContext& context) {
void CommandHandler::handleSetClockAlertsCommand(JsonVariant contents, const MessageContext& context) {
try {
_configManager.updateClockAlerts(contents);
// Save configuration to ensure persistence
@@ -760,7 +419,7 @@ void Communication::handleSetClockAlertsCommand(JsonVariant contents, const Mess
}
}
void Communication::handleSetClockBacklightCommand(JsonVariant contents, const MessageContext& context) {
void CommandHandler::handleSetClockBacklightCommand(JsonVariant contents, const MessageContext& context) {
try {
_configManager.updateClockBacklight(contents);
// Save configuration to ensure persistence
@@ -778,7 +437,7 @@ void Communication::handleSetClockBacklightCommand(JsonVariant contents, const M
}
}
void Communication::handleSetClockSilenceCommand(JsonVariant contents, const MessageContext& context) {
void CommandHandler::handleSetClockSilenceCommand(JsonVariant contents, const MessageContext& context) {
try {
_configManager.updateClockSilence(contents);
// Save configuration to ensure persistence
@@ -796,7 +455,7 @@ void Communication::handleSetClockSilenceCommand(JsonVariant contents, const Mes
}
}
void Communication::handleSetRtcTimeCommand(JsonVariant contents, const MessageContext& context) {
void CommandHandler::handleSetRtcTimeCommand(JsonVariant contents, const MessageContext& context) {
if (!_timeKeeper) {
sendErrorResponse("set_rtc_time", "TimeKeeper not available", context);
return;
@@ -866,7 +525,7 @@ void Communication::handleSetRtcTimeCommand(JsonVariant contents, const MessageC
}
}
void Communication::handleSetPhysicalClockTimeCommand(JsonVariant contents, const MessageContext& context) {
void CommandHandler::handleSetPhysicalClockTimeCommand(JsonVariant contents, const MessageContext& context) {
if (!contents.containsKey("hour") || !contents.containsKey("minute")) {
sendErrorResponse("set_physical_clock_time", "Missing hour or minute parameter", context);
return;
@@ -886,8 +545,12 @@ void Communication::handleSetPhysicalClockTimeCommand(JsonVariant contents, cons
return;
}
// Convert 24-hour to 12-hour format for analog clock
int clockHour = hour % 12;
if (clockHour == 0) clockHour = 12; // Midnight/Noon = 12, not 0
// Set the physical clock time using ConfigManager
_configManager.setPhysicalClockHour(hour);
_configManager.setPhysicalClockHour(clockHour);
_configManager.setPhysicalClockMinute(minute);
_configManager.setLastSyncTime(millis() / 1000);
@@ -896,15 +559,15 @@ void Communication::handleSetPhysicalClockTimeCommand(JsonVariant contents, cons
if (saved) {
sendSuccessResponse("set_physical_clock_time", "Physical clock time updated and saved successfully", context);
LOG_INFO("Physical clock time set to %02d:%02d and saved to SD", hour, minute);
LOG_INFO("Physical clock time set to %02d:%02d (12h: %02d:%02d) and saved to SD",
hour, minute, clockHour, minute);
} else {
sendErrorResponse("set_physical_clock_time", "Physical clock time updated but failed to save to SD card", context);
LOG_ERROR("Physical clock time set to %02d:%02d but failed to save to SD", hour, minute);
}
}
void Communication::handlePauseClockUpdatesCommand(JsonVariant contents, const MessageContext& context) {
void CommandHandler::handlePauseClockUpdatesCommand(JsonVariant contents, const MessageContext& context) {
if (!_timeKeeper) {
return;
}
@@ -922,7 +585,7 @@ void Communication::handlePauseClockUpdatesCommand(JsonVariant contents, const M
}
}
void Communication::handleSetClockEnabledCommand(JsonVariant contents, const MessageContext& context) {
void CommandHandler::handleSetClockEnabledCommand(JsonVariant contents, const MessageContext& context) {
if (!contents.containsKey("enabled")) {
sendErrorResponse("set_clock_enabled", "Missing enabled parameter", context);
return;
@@ -948,29 +611,7 @@ void Communication::handleSetClockEnabledCommand(JsonVariant contents, const Mes
}
}
String Communication::getPayloadContent(char* data, size_t len) {
String content = "";
for (size_t i = 0; i < len; i++) {
content.concat(data[i]);
}
return content;
}
int Communication::extractBellNumber(const String& key) {
if (key.length() >= 2) {
char firstChar = key.charAt(0);
if (firstChar == 'b' || firstChar == 'c') {
String numberPart = key.substring(1);
return numberPart.toInt();
}
}
return 0;
}
void Communication::handleGetDeviceTimeCommand(const MessageContext& context) {
void CommandHandler::handleGetDeviceTimeCommand(const MessageContext& context) {
StaticJsonDocument<256> response;
response["status"] = "SUCCESS";
response["type"] = "device_time";
@@ -1003,7 +644,7 @@ void Communication::handleGetDeviceTimeCommand(const MessageContext& context) {
LOG_DEBUG("Device time requested");
}
void Communication::handleGetClockTimeCommand(const MessageContext& context) {
void CommandHandler::handleGetClockTimeCommand(const MessageContext& context) {
StaticJsonDocument<256> response;
response["status"] = "SUCCESS";
response["type"] = "clock_time";
@@ -1025,80 +666,11 @@ void Communication::handleGetClockTimeCommand(const MessageContext& context) {
_configManager.getLastSyncTime());
}
// ════════════════════════════════════════════════════════════════════════════
// HEALTH CHECK IMPLEMENTATION
// ════════════════════════════════════════════════════════════════════════════
bool Communication::isHealthy() const {
// Check if required references are set
if (!_player) {
LOG_DEBUG("Communication: Unhealthy - Player reference not set");
return false;
}
if (!_fileManager) {
LOG_DEBUG("Communication: Unhealthy - FileManager reference not set");
return false;
}
if (!_timeKeeper) {
LOG_DEBUG("Communication: Unhealthy - TimeKeeper reference not set");
return false;
}
// Check if WebSocket server is active
if (!hasActiveWebSocketClients() && !isMqttConnected()) {
LOG_DEBUG("Communication: Unhealthy - No active connections (WebSocket or MQTT)");
return false;
}
// Check if MQTT reconnection timer exists and is functioning
if (_mqttReconnectTimer == nullptr) {
LOG_DEBUG("Communication: Unhealthy - MQTT reconnection timer not created");
return false;
}
// Check if networking dependency is healthy
if (!_networking.isConnected()) {
LOG_DEBUG("Communication: Unhealthy - No network connection");
return false;
}
return true;
}
// ════════════════════════════════════════════════════════════════════════════
// SYSTEM COMMAND IMPLEMENTATION
// ════════════════════════════════════════════════════════════════════════════
void Communication::handleSystemCommand(JsonVariant contents, const MessageContext& context) {
if (!contents.containsKey("action")) {
sendErrorResponse("system", "Missing action parameter", context);
return;
}
String action = contents["action"];
LOG_DEBUG("Processing system action: %s", action.c_str());
if (action == "commit_firmware") {
handleCommitFirmwareCommand(context);
} else if (action == "rollback_firmware") {
handleRollbackFirmwareCommand(context);
} else if (action == "get_firmware_status") {
handleGetFirmwareStatusCommand(context);
} else if (action == "set_network_config") {
handleSetNetworkConfigCommand(contents, context);
} else {
LOG_WARNING("Unknown system action: %s", action.c_str());
sendErrorResponse("system", "Unknown action: " + action, context);
}
}
// ════════════════════════════════════════════════════════════════════════════
// FIRMWARE MANAGEMENT IMPLEMENTATION
// ════════════════════════════════════════════════════════════════════════════
void Communication::handleCommitFirmwareCommand(const MessageContext& context) {
void CommandHandler::handleCommitFirmwareCommand(const MessageContext& context) {
if (!_firmwareValidator) {
sendErrorResponse("commit_firmware", "FirmwareValidator not available", context);
return;
@@ -1123,7 +695,7 @@ void Communication::handleCommitFirmwareCommand(const MessageContext& context) {
}
}
void Communication::handleRollbackFirmwareCommand(const MessageContext& context) {
void CommandHandler::handleRollbackFirmwareCommand(const MessageContext& context) {
if (!_firmwareValidator) {
sendErrorResponse("rollback_firmware", "FirmwareValidator not available", context);
return;
@@ -1144,7 +716,7 @@ void Communication::handleRollbackFirmwareCommand(const MessageContext& context)
}
}
void Communication::handleGetFirmwareStatusCommand(const MessageContext& context) {
void CommandHandler::handleGetFirmwareStatusCommand(const MessageContext& context) {
if (!_firmwareValidator) {
sendErrorResponse("get_firmware_status", "FirmwareValidator not available", context);
return;
@@ -1217,18 +789,44 @@ void Communication::handleGetFirmwareStatusCommand(const MessageContext& context
LOG_DEBUG("Firmware status requested: %s", stateStr.c_str());
}
// ════════════════════════════════════════════════════════════════════════════
// NETWORK CONFIGURATION IMPLEMENTATION
// ════════════════════════════════════════════════════════════════════════════
void CommandHandler::handleGetFullSettingsCommand(const MessageContext& context) {
LOG_DEBUG("Full settings requested");
// Get all settings as JSON string from ConfigManager
String settingsJson = _configManager.getAllSettingsAsJson();
// Parse it to embed in our response structure
StaticJsonDocument<512> response;
response["status"] = "SUCCESS";
response["type"] = "full_settings";
// Parse the settings JSON and add as payload
DynamicJsonDocument settingsDoc(4096);
DeserializationError error = deserializeJson(settingsDoc, settingsJson);
if (error) {
LOG_ERROR("Failed to parse settings JSON: %s", error.c_str());
sendErrorResponse("get_full_settings", "Failed to serialize settings", context);
return;
}
response["payload"] = settingsDoc.as<JsonObject>();
String responseStr;
serializeJson(response, responseStr);
sendResponse(responseStr, context);
LOG_DEBUG("Full settings sent (%d bytes)", responseStr.length());
}
void Communication::handleSetNetworkConfigCommand(JsonVariant contents, const MessageContext& context) {
void CommandHandler::handleSetNetworkConfigCommand(JsonVariant contents, const MessageContext& context) {
// Validate that we have at least one parameter to update
bool hasHostname = contents.containsKey("hostname");
bool hasStaticIPConfig = contents.containsKey("useStaticIP");
bool hasAPPass = contents.containsKey("apPass");
bool hasDiscoveryPort = contents.containsKey("discoveryPort");
if (!hasHostname && !hasStaticIPConfig && !hasAPPass && !hasDiscoveryPort) {
if (!hasHostname && !hasStaticIPConfig) {
sendErrorResponse("set_network_config", "No network parameters provided", context);
return;
}
@@ -1248,8 +846,6 @@ void Communication::handleSetNetworkConfigCommand(JsonVariant contents, const Me
IPAddress dns1 = currentConfig.dns1;
IPAddress dns2 = currentConfig.dns2;
String hostname = currentConfig.hostname;
String apPass = currentConfig.apPass;
uint16_t discoveryPort = currentConfig.discoveryPort;
// Update hostname if provided
if (hasHostname) {
@@ -1310,16 +906,10 @@ void Communication::handleSetNetworkConfigCommand(JsonVariant contents, const Me
// If anything changed, update and save configuration
if (configChanged) {
// Update network config using existing method
_configManager.updateNetworkConfig(useStaticIP, ip, gateway, subnet, dns1, dns2);
// Update network config (saves to SD internally)
_configManager.updateNetworkConfig(hostname, useStaticIP, ip, gateway, subnet, dns1, dns2);
// Manually update fields not handled by updateNetworkConfig
// Note: This is a workaround since NetworkConfig doesn't have setters for all fields
auto& writableConfig = const_cast<ConfigManager::NetworkConfig&>(_configManager.getNetworkConfig());
writableConfig.hostname = hostname;
// Save to SD card
bool saved = _configManager.saveNetworkConfig();
bool saved = true; // saveNetworkConfig() already called in updateNetworkConfig()
if (saved) {
String responseMsg = "Network configuration updated successfully";
@@ -1344,3 +934,63 @@ void Communication::handleSetNetworkConfigCommand(JsonVariant contents, const Me
LOG_ERROR("Unknown exception in handleSetNetworkConfigCommand");
}
}
// ════════════════════════════════════════════════════════════════════════════
// RESET DEFAULTS COMMAND
// ════════════════════════════════════════════════════════════════════════════
void CommandHandler::handleResetDefaultsCommand(const MessageContext& context) {
LOG_WARNING("⚠️ Factory reset requested. Proceeding...");
try {
// Reset all configurations to defaults
bool resetComplete = _configManager.resetAllToDefaults();
if (resetComplete) {
sendSuccessResponse("reset_defaults", "Reset to Defaults completed. Device will Restart to apply changes.", context);
LOG_WARNING("✅ Factory reset completed and all configurations saved to SD card");
} else {
sendErrorResponse("reset_defaults", "Reset to Defaults applied but failed to save some configurations to SD card", context);
LOG_ERROR("❌ Reset to Defaults applied but failed to save some configurations to SD card");
}
} catch (...) {
sendErrorResponse("reset_defaults", "Failed to perform Reset to Defaults", context);
LOG_ERROR("❌ Exception occurred during Resetting to Defaults");
}
}
// ════════════════════════════════════════════════════════════════════════════
// SYSTEM COMMAND IMPLEMENTATION
// ════════════════════════════════════════════════════════════════════════════
void CommandHandler::handleSystemCommand(JsonVariant contents, const MessageContext& context) {
if (!contents.containsKey("action")) {
sendErrorResponse("system", "Missing action parameter", context);
return;
}
String action = contents["action"];
LOG_DEBUG("Processing system action: %s", action.c_str());
if (action == "status") {
handleStatusCommand(context);
} else if (action == "reset_defaults") {
handleResetDefaultsCommand(context);
} else if (action == "commit_firmware") {
handleCommitFirmwareCommand(context);
} else if (action == "rollback_firmware") {
handleRollbackFirmwareCommand(context);
} else if (action == "get_firmware_status") {
handleGetFirmwareStatusCommand(context);
} else if (action == "set_network_config") {
handleSetNetworkConfigCommand(contents, context);
} else {
LOG_WARNING("Unknown system action: %s", action.c_str());
sendErrorResponse("system", "Unknown action: " + action, context);
}
}

141
vesper/src/Communication/CommandHandler/CommandHandler.hpp Normal file
View File

@@ -0,0 +1,141 @@
/*
* ═══════════════════════════════════════════════════════════════════════════════════
* COMMANDHANDLER.HPP - Unified Command Processing
* ═══════════════════════════════════════════════════════════════════════════════════
*
* ⚙️ COMMAND ROUTER AND PROCESSOR ⚙️
*
* Processes all incoming commands from both MQTT and WebSocket:
* • System commands (ping, status, identify)
* • Playback commands
* • File manager commands
* • Relay setup commands
* • Clock setup commands
* • Firmware management
* • Network configuration
*
* 📋 VERSION: 1.0
* 📅 DATE: 2025-10-01
* 👨‍💻 AUTHOR: Advanced Bell Systems
* ═══════════════════════════════════════════════════════════════════════════════════
*/
#pragma once
#include <Arduino.h>
#include <ArduinoJson.h>
// Forward declarations
class ConfigManager;
class OTAManager;
class Player;
class FileManager;
class Timekeeper;
class FirmwareValidator;
class ClientManager;
class CommandHandler {
public:
// Message source identification
enum class MessageSource {
MQTT,
WEBSOCKET
};
struct MessageContext {
MessageSource source;
uint32_t clientId; // Only for WebSocket
MessageContext(MessageSource src, uint32_t id = 0)
: source(src), clientId(id) {}
};
// Response callback type
using ResponseCallback = std::function<void(const String& response, const MessageContext& context)>;
explicit CommandHandler(ConfigManager& configManager,
OTAManager& otaManager);
~CommandHandler();
/**
* @brief Set component references
*/
void setPlayerReference(Player* player);
void setFileManagerReference(FileManager* fm);
void setTimeKeeperReference(Timekeeper* tk);
void setFirmwareValidatorReference(FirmwareValidator* fv);
void setClientManagerReference(ClientManager* cm);
/**
* @brief Set response callback for sending responses back
*/
void setResponseCallback(ResponseCallback callback);
/**
* @brief Process incoming command
* @param command JSON command document
* @param context Message context (source and client ID)
*/
void processCommand(JsonDocument& command, const MessageContext& context);
private:
// Dependencies
ConfigManager& _configManager;
OTAManager& _otaManager;
Player* _player;
FileManager* _fileManager;
Timekeeper* _timeKeeper;
FirmwareValidator* _firmwareValidator;
ClientManager* _clientManager;
ResponseCallback _responseCallback;
// Response helpers
void sendResponse(const String& response, const MessageContext& context);
void sendSuccessResponse(const String& type, const String& payload, const MessageContext& context);
void sendErrorResponse(const String& type, const String& message, const MessageContext& context);
// Command handlers
void handlePingCommand(const MessageContext& context);
void handleStatusCommand(const MessageContext& context);
void handleIdentifyCommand(JsonVariant contents, const MessageContext& context);
void handlePlaybackCommand(JsonVariant contents, const MessageContext& context);
void handleFileManagerCommand(JsonVariant contents, const MessageContext& context);
void handleRelaySetupCommand(JsonVariant contents, const MessageContext& context);
void handleClockSetupCommand(JsonVariant contents, const MessageContext& context);
void handleSystemInfoCommand(JsonVariant contents, const MessageContext& context);
void handleSystemCommand(JsonVariant contents, const MessageContext& context);
// File Manager sub-commands
void handleListMelodiesCommand(const MessageContext& context);
void handleDownloadMelodyCommand(JsonVariant contents, const MessageContext& context);
void handleDeleteMelodyCommand(JsonVariant contents, const MessageContext& context);
// Relay Setup sub-commands
void handleSetRelayTimersCommand(JsonVariant contents, const MessageContext& context);
void handleSetRelayOutputsCommand(JsonVariant contents, const MessageContext& context);
// Clock Setup sub-commands
void handleSetClockOutputsCommand(JsonVariant contents, const MessageContext& context);
void handleSetClockTimingsCommand(JsonVariant contents, const MessageContext& context);
void handleSetClockAlertsCommand(JsonVariant contents, const MessageContext& context);
void handleSetClockBacklightCommand(JsonVariant contents, const MessageContext& context);
void handleSetClockSilenceCommand(JsonVariant contents, const MessageContext& context);
void handleSetRtcTimeCommand(JsonVariant contents, const MessageContext& context);
void handleSetPhysicalClockTimeCommand(JsonVariant contents, const MessageContext& context);
void handlePauseClockUpdatesCommand(JsonVariant contents, const MessageContext& context);
void handleSetClockEnabledCommand(JsonVariant contents, const MessageContext& context);
// System Info sub-commands
void handleGetDeviceTimeCommand(const MessageContext& context);
void handleGetClockTimeCommand(const MessageContext& context);
void handleCommitFirmwareCommand(const MessageContext& context);
void handleRollbackFirmwareCommand(const MessageContext& context);
void handleGetFirmwareStatusCommand(const MessageContext& context);
void handleGetFullSettingsCommand(const MessageContext& context);
// Network configuration
void handleSetNetworkConfigCommand(JsonVariant contents, const MessageContext& context);
// System Config
void handleResetDefaultsCommand(const MessageContext& context);
};

232
vesper/src/Communication/Communication.hpp
View File

@@ -1,232 +0,0 @@
/*
* ═══════════════════════════════════════════════════════════════════════════════════
* COMMUNICATION.HPP - Multi-Protocol Communication Manager v3.0
* ═══════════════════════════════════════════════════════════════════════════════════
*
* 📡 THE COMMUNICATION HUB OF VESPER 📡
*
* This class manages all external communication protocols including MQTT,
* WebSocket, and UDP discovery. It provides a unified interface for
* grouped command handling and status reporting across multiple protocols.
*
* 🏗️ ARCHITECTURE:
* • Multi-protocol support with unified grouped command processing
* • Multi-client WebSocket support with device type identification
* • Automatic connection management and reconnection
* • Unified response system for consistent messaging
* • Thread-safe operation with proper resource management
* • Batch command support for efficient configuration
*
* 📡 SUPPORTED PROTOCOLS:
* • MQTT: Primary control interface with auto-reconnection
* • WebSocket: Real-time multi-client web interface communication
* • UDP Discovery: Auto-discovery service for network scanning
*
* 📱 CLIENT MANAGEMENT:
* • Support for multiple WebSocket clients (master/secondary devices)
* • Client type identification and targeted messaging
* • Automatic cleanup of disconnected clients
* • Broadcast capabilities for status updates
*
* 🔄 MESSAGE ROUTING:
* • Commands accepted from both MQTT and WebSocket
* • Responses sent only to originating protocol/client
* • Status broadcasts sent to all WebSocket clients + MQTT
* • Grouped command processing for all protocols
*
* 📋 VERSION: 3.0 (Grouped commands + batch processing)
* 📅 DATE: 2025
* 👨‍💻 AUTHOR: Advanced Bell Systems
* ═══════════════════════════════════════════════════════════════════════════════════
*/
#pragma once
#include <Arduino.h>
#include <AsyncMqttClient.h>
#include <ESPAsyncWebServer.h>
#include <AsyncUDP.h>
#include <ArduinoJson.h>
#include "ResponseBuilder.hpp"
#include "../ClientManager/ClientManager.hpp"
class ConfigManager;
class OTAManager;
class Player;
class FileManager;
class Timekeeper;
class Networking;
class FirmwareValidator;
class Communication {
public:
// Message source identification for response routing
enum class MessageSource {
MQTT,
WEBSOCKET
};
struct MessageContext {
MessageSource source;
uint32_t clientId; // Only used for WebSocket messages
MessageContext(MessageSource src, uint32_t id = 0)
: source(src), clientId(id) {}
};
explicit Communication(ConfigManager& configManager,
OTAManager& otaManager,
Networking& networking,
AsyncMqttClient& mqttClient,
AsyncWebServer& server,
AsyncWebSocket& webSocket,
AsyncUDP& udp);
~Communication();
void begin();
void setPlayerReference(Player* player) { _player = player; }
void setFileManagerReference(FileManager* fm) { _fileManager = fm; }
void setTimeKeeperReference(Timekeeper* tk) { _timeKeeper = tk; }
void setFirmwareValidatorReference(FirmwareValidator* fv) { _firmwareValidator = fv; }
void setupUdpDiscovery();
// Public methods for timer callbacks
void connectToMqtt();
void subscribeMqtt();
// Status methods
bool isMqttConnected() const { return _mqttClient.connected(); }
bool hasActiveWebSocketClients() const { return _clientManager.hasClients(); }
size_t getWebSocketClientCount() const { return _clientManager.getClientCount(); }
// Response methods - unified response system
void sendResponse(const String& response, const MessageContext& context);
void sendSuccessResponse(const String& type, const String& payload, const MessageContext& context);
void sendErrorResponse(const String& type, const String& message, const MessageContext& context);
// Broadcast methods - for status updates that go to everyone
void broadcastStatus(const String& statusMessage);
void broadcastStatus(const JsonDocument& statusJson);
void broadcastToMasterClients(const String& message);
void broadcastToSecondaryClients(const String& message);
void broadcastToAllWebSocketClients(const String& message);
void broadcastToAllWebSocketClients(const JsonDocument& message);
void publishToMqtt(const String& data);
// ═══════════════════════════════════════════════════════════════════════════════
// HEALTH CHECK METHOD
// ═══════════════════════════════════════════════════════════════════════════════
/** @brief Check if Communication is in healthy state */
bool isHealthy() const;
// Bell overload notification
void sendBellOverloadNotification(const std::vector<uint8_t>& bellNumbers,
const std::vector<uint16_t>& bellLoads,
const String& severity);
// Network connection callbacks (called by Networking)
void onNetworkConnected();
void onNetworkDisconnected();
// Static instance for callbacks
static Communication* _instance;
private:
// Dependencies
ConfigManager& _configManager;
OTAManager& _otaManager;
Networking& _networking;
AsyncMqttClient& _mqttClient;
AsyncWebServer& _server;
AsyncWebSocket& _webSocket;
AsyncUDP& _udp;
Player* _player;
FileManager* _fileManager;
Timekeeper* _timeKeeper;
FirmwareValidator* _firmwareValidator;
// Client manager
ClientManager _clientManager;
// State
TimerHandle_t _mqttReconnectTimer;
// Reusable JSON documents
static StaticJsonDocument<2048> _parseDocument;
// MQTT methods
void initMqtt();
static void onMqttConnect(bool sessionPresent);
static void onMqttDisconnect(AsyncMqttClientDisconnectReason reason);
static void onMqttMessage(char* topic, char* payload, AsyncMqttClientMessageProperties properties,
size_t len, size_t index, size_t total);
static void onMqttSubscribe(uint16_t packetId, uint8_t qos);
static void onMqttUnsubscribe(uint16_t packetId);
static void onMqttPublish(uint16_t packetId);
// WebSocket methods
void initWebSocket();
static void onWebSocketEvent(AsyncWebSocket* server, AsyncWebSocketClient* client,
AwsEventType type, void* arg, uint8_t* data, size_t len);
void onWebSocketConnect(AsyncWebSocketClient* client);
void onWebSocketDisconnect(AsyncWebSocketClient* client);
void onWebSocketReceived(AsyncWebSocketClient* client, void* arg, uint8_t* data, size_t len);
void handleClientIdentification(AsyncWebSocketClient* client, JsonDocument& command);
// Command processing - unified for both MQTT and WebSocket with grouped commands
JsonDocument parsePayload(char* payload);
void handleCommand(JsonDocument& command, const MessageContext& context);
// ═════════════════════════════════════════════════════════════════════════════════
// GROUPED COMMAND HANDLERS
// ═════════════════════════════════════════════════════════════════════════════════
// System commands
void handleSystemCommand(JsonVariant contents, const MessageContext& context);
void handleSystemInfoCommand(JsonVariant contents, const MessageContext& context);
void handlePlaybackCommand(JsonVariant contents, const MessageContext& context);
void handleFileManagerCommand(JsonVariant contents, const MessageContext& context);
void handleRelaySetupCommand(JsonVariant contents, const MessageContext& context);
void handleClockSetupCommand(JsonVariant contents, const MessageContext& context);
// System sub-commands
void handlePingCommand(const MessageContext& context);
void handleStatusCommand(const MessageContext& context);
void handleIdentifyCommand(JsonVariant contents, const MessageContext& context);
void handleGetDeviceTimeCommand(const MessageContext& context);
void handleGetClockTimeCommand(const MessageContext& context);
// Firmware management commands
void handleCommitFirmwareCommand(const MessageContext& context);
void handleRollbackFirmwareCommand(const MessageContext& context);
void handleGetFirmwareStatusCommand(const MessageContext& context);
// Network configuration command
void handleSetNetworkConfigCommand(JsonVariant contents, const MessageContext& context);
// File Manager sub-commands
void handleListMelodiesCommand(const MessageContext& context);
void handleDownloadMelodyCommand(JsonVariant contents, const MessageContext& context);
void handleDeleteMelodyCommand(JsonVariant contents, const MessageContext& context);
// Relay Setup sub-commands
void handleSetRelayTimersCommand(JsonVariant contents, const MessageContext& context);
void handleSetRelayOutputsCommand(JsonVariant contents, const MessageContext& context);
// Clock Setup sub-commands
void handleSetClockOutputsCommand(JsonVariant contents, const MessageContext& context);
void handleSetClockTimingsCommand(JsonVariant contents, const MessageContext& context);
void handleSetClockAlertsCommand(JsonVariant contents, const MessageContext& context);
void handleSetClockBacklightCommand(JsonVariant contents, const MessageContext& context);
void handleSetClockSilenceCommand(JsonVariant contents, const MessageContext& context);
void handleSetRtcTimeCommand(JsonVariant contents, const MessageContext& context);
void handleSetPhysicalClockTimeCommand(JsonVariant contents, const MessageContext& context);
void handlePauseClockUpdatesCommand(JsonVariant contents, const MessageContext& context);
void handleSetClockEnabledCommand(JsonVariant contents, const MessageContext& context);
// Utility methods
String getPayloadContent(char* data, size_t len);
int extractBellNumber(const String& key); // Extract bell number from "b1", "c1", etc.
};

300
vesper/src/Communication/CommunicationRouter/CommunicationRouter.cpp Normal file
View File

@@ -0,0 +1,300 @@
/*
* COMMUNICATIONROUTER.CPP - Communication Router Implementation
*/
#include "CommunicationRouter.hpp"
#include "../../ConfigManager/ConfigManager.hpp"
#include "../../OTAManager/OTAManager.hpp"
#include "../../Networking/Networking.hpp"
#include "../../Logging/Logging.hpp"
#include "../../Player/Player.hpp"
#include "../../FileManager/FileManager.hpp"
#include "../../TimeKeeper/TimeKeeper.hpp"
#include "../../FirmwareValidator/FirmwareValidator.hpp"
CommunicationRouter::CommunicationRouter(ConfigManager& configManager,
OTAManager& otaManager,
Networking& networking,
AsyncWebServer& server,
AsyncWebSocket& webSocket,
AsyncUDP& udp)
: _configManager(configManager)
, _otaManager(otaManager)
, _networking(networking)
, _server(server)
, _webSocket(webSocket)
, _udp(udp)
, _player(nullptr)
, _fileManager(nullptr)
, _timeKeeper(nullptr)
, _firmwareValidator(nullptr)
, _mqttClient(configManager, networking)
, _clientManager()
, _wsServer(webSocket, _clientManager)
, _commandHandler(configManager, otaManager) {}
CommunicationRouter::~CommunicationRouter() {}
void CommunicationRouter::begin() {
LOG_INFO("Initializing Communication Router v4.0 (Modular)");
// 🔥 CRITICAL: Initialize WebSocket FIRST to ensure it's always set up
// Even if MQTT fails, we want WebSocket to work!
LOG_INFO("Setting up WebSocket server...");
// Initialize WebSocket server
_wsServer.begin();
_wsServer.setCallback([this](uint32_t clientId, const JsonDocument& message) {
onWebSocketMessage(clientId, message);
});
// 🔥 CRITICAL FIX: Attach WebSocket handler to AsyncWebServer
// This MUST happen before any potential failures!
_server.addHandler(&_webSocket);
LOG_INFO("✅ WebSocket handler attached to AsyncWebServer on /ws");
//Now initialize MQTT client (can fail without breaking WebSocket)
try {
LOG_INFO("Setting up MQTT client...");
_mqttClient.begin();
_mqttClient.setCallback([this](const String& topic, const String& payload) {
onMqttMessage(topic, payload);
});
LOG_INFO("✅ MQTT client initialized");
} catch (...) {
LOG_ERROR("❌ MQTT initialization failed, but WebSocket is still available");
}
// 🔥 CRITICAL FIX: Connect ClientManager to CommandHandler
_commandHandler.setClientManagerReference(&_clientManager);
LOG_INFO("ClientManager reference set for CommandHandler");
// Setup command handler response callback
_commandHandler.setResponseCallback([this](const String& response, const CommandHandler::MessageContext& context) {
sendResponse(response, context);
});
LOG_INFO("Communication Router initialized with modular architecture");
}
void CommunicationRouter::setPlayerReference(Player* player) {
_player = player;
_commandHandler.setPlayerReference(player);
}
void CommunicationRouter::setFileManagerReference(FileManager* fm) {
_fileManager = fm;
_commandHandler.setFileManagerReference(fm);
}
void CommunicationRouter::setTimeKeeperReference(Timekeeper* tk) {
_timeKeeper = tk;
_commandHandler.setTimeKeeperReference(tk);
}
void CommunicationRouter::setFirmwareValidatorReference(FirmwareValidator* fv) {
_firmwareValidator = fv;
_commandHandler.setFirmwareValidatorReference(fv);
}
void CommunicationRouter::setupUdpDiscovery() {
uint16_t discoveryPort = _configManager.getNetworkConfig().discoveryPort;
if (_udp.listen(discoveryPort)) {
LOG_INFO("UDP discovery listening on port %u", discoveryPort);
_udp.onPacket([this](AsyncUDPPacket packet) {
String msg = String((const char*)packet.data(), packet.length());
LOG_DEBUG("UDP from %s:%u -> %s",
packet.remoteIP().toString().c_str(),
packet.remotePort(),
msg.c_str());
bool shouldReply = false;
if (msg.indexOf("discover") >= 0) {
shouldReply = true;
} else {
StaticJsonDocument<128> req;
DeserializationError err = deserializeJson(req, msg);
if (!err) {
shouldReply = (req["op"] == "discover" && req["svc"] == "vesper");
}
}
if (!shouldReply) return;
StaticJsonDocument<256> doc;
doc["op"] = "discover_reply";
doc["svc"] = "vesper";
doc["ver"] = 1;
doc["name"] = "Proj. Vesper v2.0";
doc["id"] = _configManager.getDeviceUID();
doc["ip"] = _networking.getLocalIP();
char wsUrl[64];
snprintf(wsUrl, sizeof(wsUrl), "ws://%s/ws", _networking.getLocalIP().c_str());
doc["ws"] = wsUrl;
doc["port"] = 80;
doc["fw"] = "2.0";
doc["clients"] = _clientManager.getClientCount();
String out;
serializeJson(doc, out);
_udp.writeTo((const uint8_t*)out.c_str(), out.length(),
packet.remoteIP(), packet.remotePort());
});
} else {
LOG_ERROR("Failed to start UDP discovery.");
}
}
bool CommunicationRouter::isMqttConnected() const {
return _mqttClient.isConnected();
}
bool CommunicationRouter::hasActiveWebSocketClients() const {
return _wsServer.hasClients();
}
size_t CommunicationRouter::getWebSocketClientCount() const {
return _wsServer.getClientCount();
}
bool CommunicationRouter::isHealthy() const {
// Check if required references are set
if (!_player || !_fileManager || !_timeKeeper) {
LOG_DEBUG("CommunicationRouter: Unhealthy - Missing references");
return false;
}
// Check if at least one protocol is connected
if (!isMqttConnected() && !hasActiveWebSocketClients()) {
LOG_DEBUG("CommunicationRouter: Unhealthy - No active connections");
return false;
}
// Check network connectivity
if (!_networking.isConnected()) {
LOG_DEBUG("CommunicationRouter: Unhealthy - No network connection");
return false;
}
return true;
}
void CommunicationRouter::broadcastStatus(const String& statusMessage) {
publishToMqtt(statusMessage);
_wsServer.broadcastToAll(statusMessage);
}
void CommunicationRouter::broadcastStatus(const JsonDocument& statusJson) {
String statusMessage;
serializeJson(statusJson, statusMessage);
broadcastStatus(statusMessage);
}
void CommunicationRouter::broadcastToMasterClients(const String& message) {
_wsServer.broadcastToMaster(message);
}
void CommunicationRouter::broadcastToSecondaryClients(const String& message) {
_wsServer.broadcastToSecondary(message);
}
void CommunicationRouter::broadcastToAllWebSocketClients(const String& message) {
_wsServer.broadcastToAll(message);
}
void CommunicationRouter::broadcastToAllWebSocketClients(const JsonDocument& message) {
String messageStr;
serializeJson(message, messageStr);
_wsServer.broadcastToAll(messageStr);
}
void CommunicationRouter::publishToMqtt(const String& data) {
if (_mqttClient.isConnected()) {
_mqttClient.publish("data", data, 0, false);
LOG_DEBUG("Published to MQTT: %s", data.c_str());
} else {
LOG_ERROR("MQTT Not Connected! Message Failed: %s", data.c_str());
}
}
void CommunicationRouter::sendBellOverloadNotification(const std::vector<uint8_t>& bellNumbers,
const std::vector<uint16_t>& bellLoads,
const String& severity) {
StaticJsonDocument<512> overloadMsg;
overloadMsg["status"] = "INFO";
overloadMsg["type"] = "bell_overload";
JsonArray bellsArray = overloadMsg["payload"]["bells"].to<JsonArray>();
JsonArray loadsArray = overloadMsg["payload"]["loads"].to<JsonArray>();
for (size_t i = 0; i < bellNumbers.size() && i < bellLoads.size(); i++) {
bellsArray.add(bellNumbers[i] + 1);
loadsArray.add(bellLoads[i]);
}
overloadMsg["payload"]["severity"] = severity;
broadcastStatus(overloadMsg);
LOG_WARNING("Bell overload notification sent: %d bells, severity: %s",
bellNumbers.size(), severity.c_str());
}
void CommunicationRouter::onNetworkConnected() {
LOG_DEBUG("Network connected - notifying MQTT client");
_mqttClient.onNetworkConnected();
}
void CommunicationRouter::onNetworkDisconnected() {
LOG_DEBUG("Network disconnected - notifying MQTT client");
_mqttClient.onNetworkDisconnected();
}
void CommunicationRouter::onMqttMessage(const String& topic, const String& payload) {
LOG_DEBUG("MQTT message received: %s", payload.c_str());
// Parse JSON
StaticJsonDocument<2048> doc;
DeserializationError error = deserializeJson(doc, payload);
if (error) {
LOG_ERROR("Failed to parse MQTT JSON: %s", error.c_str());
return;
}
// Create message context for MQTT
CommandHandler::MessageContext context(CommandHandler::MessageSource::MQTT);
// Forward to command handler
_commandHandler.processCommand(doc, context);
}
void CommunicationRouter::onWebSocketMessage(uint32_t clientId, const JsonDocument& message) {
// Extract command for logging
String cmd = message["cmd"] | "unknown";
LOG_INFO("📨 WebSocket message from client #%u: cmd=%s", clientId, cmd.c_str());
// Create message context for WebSocket with client ID
CommandHandler::MessageContext context(CommandHandler::MessageSource::WEBSOCKET, clientId);
// Forward to command handler (need to cast away const for now)
JsonDocument& mutableDoc = const_cast<JsonDocument&>(message);
_commandHandler.processCommand(mutableDoc, context);
LOG_DEBUG("WebSocket message from client #%u processed", clientId);
}
void CommunicationRouter::sendResponse(const String& response, const CommandHandler::MessageContext& context) {
if (context.source == CommandHandler::MessageSource::MQTT) {
LOG_DEBUG("↗️ Sending response via MQTT: %s", response.c_str());
publishToMqtt(response);
} else if (context.source == CommandHandler::MessageSource::WEBSOCKET) {
LOG_DEBUG("↗️ Sending response to WebSocket client #%u: %s", context.clientId, response.c_str());
_wsServer.sendToClient(context.clientId, response);
} else {
LOG_ERROR("❌ Unknown message source for response routing!");
}
}

118
vesper/src/Communication/CommunicationRouter/CommunicationRouter.hpp Normal file
View File

@@ -0,0 +1,118 @@
/*
* ═══════════════════════════════════════════════════════════════════════════════════
* COMMUNICATIONROUTER.HPP - Multi-Protocol Communication Router v4.0
* ═══════════════════════════════════════════════════════════════════════════════════
*
* 📡 THE COMMUNICATION ROUTER OF VESPER 📡
*
* Routes messages between protocols and command handlers:
* • MQTTAsyncClient: AsyncMqttClient for non-blocking MQTT
* • WebSocketServer: Multi-client WebSocket management
* • CommandHandler: Unified command processing
* • ResponseBuilder: Structured response generation
*
* 🏗️ ARCHITECTURE:
* • Message routing between protocols and handlers
* • MQTT on dedicated RTOS task (Core 0)
* • Unified command processing
* • Thread-safe message routing
*
* 📡 SUPPORTED PROTOCOLS:
* • MQTT: AsyncMqttClient for reliable async connectivity
* • WebSocket: Real-time multi-client web interface
* • UDP Discovery: Auto-discovery service
*
* 📋 VERSION: 5.0 (AsyncMqttClient)
* 📅 DATE: 2025-10-01
* 👨‍💻 AUTHOR: Advanced Bell Systems
* ═══════════════════════════════════════════════════════════════════════════════════
*/
#pragma once
#include <Arduino.h>
#include <ESPAsyncWebServer.h>
#include <AsyncUDP.h>
#include <ArduinoJson.h>
#include "../MQTTAsyncClient/MQTTAsyncClient.hpp"
#include "../WebSocketServer/WebSocketServer.hpp"
#include "../CommandHandler/CommandHandler.hpp"
#include "../ResponseBuilder/ResponseBuilder.hpp"
#include "../../ClientManager/ClientManager.hpp"
class ConfigManager;
class OTAManager;
class Player;
class FileManager;
class Timekeeper;
class Networking;
class FirmwareValidator;
class CommunicationRouter {
public:
explicit CommunicationRouter(ConfigManager& configManager,
OTAManager& otaManager,
Networking& networking,
AsyncWebServer& server,
AsyncWebSocket& webSocket,
AsyncUDP& udp);
~CommunicationRouter();
void begin();
void setPlayerReference(Player* player);
void setFileManagerReference(FileManager* fm);
void setTimeKeeperReference(Timekeeper* tk);
void setFirmwareValidatorReference(FirmwareValidator* fv);
void setupUdpDiscovery();
// Status methods
bool isMqttConnected() const;
bool hasActiveWebSocketClients() const;
size_t getWebSocketClientCount() const;
bool isHealthy() const;
// Broadcast methods
void broadcastStatus(const String& statusMessage);
void broadcastStatus(const JsonDocument& statusJson);
void broadcastToMasterClients(const String& message);
void broadcastToSecondaryClients(const String& message);
void broadcastToAllWebSocketClients(const String& message);
void broadcastToAllWebSocketClients(const JsonDocument& message);
void publishToMqtt(const String& data);
// Bell overload notification
void sendBellOverloadNotification(const std::vector<uint8_t>& bellNumbers,
const std::vector<uint16_t>& bellLoads,
const String& severity);
// Network connection callbacks
void onNetworkConnected();
void onNetworkDisconnected();
private:
// Dependencies
ConfigManager& _configManager;
OTAManager& _otaManager;
Networking& _networking;
AsyncWebServer& _server;
AsyncWebSocket& _webSocket;
AsyncUDP& _udp;
Player* _player;
FileManager* _fileManager;
Timekeeper* _timeKeeper;
FirmwareValidator* _firmwareValidator;
// Communication subsystems
MQTTAsyncClient _mqttClient;
ClientManager _clientManager;
WebSocketServer _wsServer;
CommandHandler _commandHandler;
// Message handlers
void onMqttMessage(const String& topic, const String& payload);
void onWebSocketMessage(uint32_t clientId, const JsonDocument& message);
// Response routing
void sendResponse(const String& response, const CommandHandler::MessageContext& context);
};

240
vesper/src/Communication/MQTTAsyncClient/MQTTAsyncClient.cpp Normal file
View File

@@ -0,0 +1,240 @@
/*
* MQTTASYNCCLIENT.CPP - MQTT Client Implementation with AsyncMqttClient
*/
#include "MQTTAsyncClient.hpp"
#include "../../ConfigManager/ConfigManager.hpp"
#include "../../Networking/Networking.hpp"
#include "../../Logging/Logging.hpp"
MQTTAsyncClient* MQTTAsyncClient::_instance = nullptr;
MQTTAsyncClient::MQTTAsyncClient(ConfigManager& configManager, Networking& networking)
: _configManager(configManager)
, _networking(networking)
, _messageCallback(nullptr)
, _mqttReconnectTimer(nullptr) {
_instance = this; // Set static instance pointer
// Create reconnection timer
_mqttReconnectTimer = xTimerCreate(
"mqttReconnect", // Timer name (for debugging)
pdMS_TO_TICKS(MQTT_RECONNECT_DELAY), // Period: 5000ms = 5 seconds
pdFALSE, // One-shot (false) or Auto-reload (true)
(void*)0, // Timer ID (can store data)
mqttReconnectTimerCallback // Callback function when timer expires
);
}
MQTTAsyncClient::~MQTTAsyncClient() {
if (_mqttReconnectTimer) {
xTimerDelete(_mqttReconnectTimer, portMAX_DELAY);
}
_mqttClient.disconnect();
}
void MQTTAsyncClient::begin() {
LOG_INFO("Initializing MQTT Async Client");
auto& mqttConfig = _configManager.getMqttConfig();
// Build topic strings (cache for performance)
String deviceUID = _configManager.getDeviceUID();
_controlTopic = "vesper/" + deviceUID + "/control";
_dataTopic = "vesper/" + deviceUID + "/data";
_clientId = "vesper-" + deviceUID;
LOG_INFO("MQTT Topics: control=%s, data=%s", _controlTopic.c_str(), _dataTopic.c_str());
// Setup event handlers
_mqttClient.onConnect([this](bool sessionPresent) {
this->onMqttConnect(sessionPresent);
});
_mqttClient.onDisconnect([this](AsyncMqttClientDisconnectReason reason) {
this->onMqttDisconnect(reason);
});
_mqttClient.onSubscribe([this](uint16_t packetId, uint8_t qos) {
this->onMqttSubscribe(packetId, qos);
});
_mqttClient.onUnsubscribe([this](uint16_t packetId) {
this->onMqttUnsubscribe(packetId);
});
_mqttClient.onMessage([this](char* topic, char* payload, AsyncMqttClientMessageProperties properties, size_t len, size_t index, size_t total) {
this->onMqttMessage(topic, payload, properties, len, index, total);
});
_mqttClient.onPublish([this](uint16_t packetId) {
this->onMqttPublish(packetId);
});
// Configure connection
_mqttClient.setServer(mqttConfig.host, mqttConfig.port);
_mqttClient.setCredentials(mqttConfig.user.c_str(), mqttConfig.password.c_str());
_mqttClient.setClientId(_clientId.c_str()); // Use member variable
_mqttClient.setKeepAlive(15);
_mqttClient.setCleanSession(true);
LOG_INFO("✅ MQTT Async Client initialized");
}
void MQTTAsyncClient::connect() {
if (_mqttClient.connected()) {
LOG_DEBUG("Already connected to MQTT");
return;
}
auto& mqttConfig = _configManager.getMqttConfig();
LOG_INFO("Free heap BEFORE MQTT connect: %d bytes", ESP.getFreeHeap());
_mqttClient.connect();
LOG_INFO("MQTT connect() called - waiting for async connection...");
}
void MQTTAsyncClient::disconnect() {
_mqttClient.disconnect();
LOG_INFO("Disconnected from MQTT broker");
}
uint16_t MQTTAsyncClient::publish(const String& topic, const String& payload, int qos, bool retain) {
// Build full topic (if relative)
String fullTopic = topic.startsWith("vesper/") ? topic : _dataTopic;
uint16_t packetId = _mqttClient.publish(fullTopic.c_str(), qos, retain, payload.c_str());
if (packetId > 0) {
LOG_DEBUG("Published to %s: %s (packetId=%d)", fullTopic.c_str(), payload.c_str(), packetId);
} else {
LOG_ERROR("Failed to publish to %s", fullTopic.c_str());
}
return packetId;
}
void MQTTAsyncClient::setCallback(MessageCallback callback) {
_messageCallback = callback;
}
bool MQTTAsyncClient::isConnected() const {
return _mqttClient.connected();
}
void MQTTAsyncClient::onNetworkConnected() {
LOG_DEBUG("Network connected - waiting 2 seconds for network stack to stabilize...");
// Small delay to ensure network stack is fully ready
delay(2000);
LOG_DEBUG("Network stable - connecting to MQTT");
connect();
}
void MQTTAsyncClient::onNetworkDisconnected() {
LOG_DEBUG("Network disconnected - MQTT will auto-reconnect when network returns");
if (_mqttClient.connected()) {
_mqttClient.disconnect(true);
}
}
void MQTTAsyncClient::subscribe() {
uint16_t packetId = _mqttClient.subscribe(_controlTopic.c_str(), 0);
LOG_INFO("📬 Subscribing to control topic: %s (packetId=%d)", _controlTopic.c_str(), packetId);
}
void MQTTAsyncClient::onMqttConnect(bool sessionPresent) {
LOG_INFO("✅ Connected to MQTT broker (session present: %s)", sessionPresent ? "yes" : "no");
LOG_INFO("🔍 Free heap AFTER MQTT connect: %d bytes", ESP.getFreeHeap());
// Subscribe to control topic
subscribe();
}
void MQTTAsyncClient::onMqttDisconnect(AsyncMqttClientDisconnectReason reason) {
const char* reasonStr;
switch(reason) {
case AsyncMqttClientDisconnectReason::TCP_DISCONNECTED:
reasonStr = "TCP disconnected";
break;
case AsyncMqttClientDisconnectReason::MQTT_UNACCEPTABLE_PROTOCOL_VERSION:
reasonStr = "Unacceptable protocol version";
break;
case AsyncMqttClientDisconnectReason::MQTT_IDENTIFIER_REJECTED:
reasonStr = "Identifier rejected";
break;
case AsyncMqttClientDisconnectReason::MQTT_SERVER_UNAVAILABLE:
reasonStr = "Server unavailable";
break;
case AsyncMqttClientDisconnectReason::MQTT_MALFORMED_CREDENTIALS:
reasonStr = "Malformed credentials";
break;
case AsyncMqttClientDisconnectReason::MQTT_NOT_AUTHORIZED:
reasonStr = "Not authorized";
break;
default:
reasonStr = "Unknown";
break;
}
LOG_ERROR("❌ Disconnected from MQTT broker - Reason: %s (%d)", reasonStr, static_cast<int>(reason));
if (_networking.isConnected()) {
LOG_INFO("Network still connected - scheduling MQTT reconnection in %d seconds", MQTT_RECONNECT_DELAY / 1000);
xTimerStart(_mqttReconnectTimer, 0);
} else {
LOG_INFO("Network is down - waiting for network to reconnect");
}
}
void MQTTAsyncClient::onMqttSubscribe(uint16_t packetId, uint8_t qos) {
LOG_INFO("✅ Subscribed to topic (packetId=%d, QoS=%d)", packetId, qos);
}
void MQTTAsyncClient::onMqttUnsubscribe(uint16_t packetId) {
LOG_DEBUG("Unsubscribed from topic (packetId=%d)", packetId);
}
void MQTTAsyncClient::onMqttMessage(char* topic, char* payload, AsyncMqttClientMessageProperties properties, size_t len, size_t index, size_t total) {
// Convert to String
String topicStr = String(topic);
String payloadStr = String(payload).substring(0, len);
LOG_DEBUG("MQTT message received - topic: %s, payload: %s", topicStr.c_str(), payloadStr.c_str());
// Call user callback
if (_messageCallback) {
_messageCallback(topicStr, payloadStr);
}
}
void MQTTAsyncClient::onMqttPublish(uint16_t packetId) {
LOG_DEBUG("MQTT publish acknowledged (packetId=%d)", packetId);
}
void MQTTAsyncClient::attemptReconnection() {
// Double-check network is still up
if (_networking.isConnected()) {
LOG_INFO("Attempting MQTT reconnection...");
connect();
} else {
LOG_WARNING("Network down during reconnect attempt - aborting");
}
}
void MQTTAsyncClient::mqttReconnectTimerCallback(TimerHandle_t xTimer) {
// Get the MQTT instance from the timer's ID (set during timer creation)
// For now, we'll use a static instance pointer (similar to Networking)
// You'll need to add: static MQTTAsyncClient* _instance; to header
if (MQTTAsyncClient::_instance) {
MQTTAsyncClient::_instance->attemptReconnection();
}
}

116
vesper/src/Communication/MQTTAsyncClient/MQTTAsyncClient.hpp Normal file
View File

@@ -0,0 +1,116 @@
/*
* ═══════════════════════════════════════════════════════════════════════════════════
* MQTTASYNCCLIENT.HPP - MQTT Client with AsyncMqttClient
* ═══════════════════════════════════════════════════════════════════════════════════
*
* 🔒 MQTT CONNECTION WITH ASYNC ARCHITECTURE 🔒
*
* This class manages MQTT connections using AsyncMqttClient library:
* • Fully async/non-blocking operation
* • No conflicts with AsyncWebServer
* • Auto-reconnection built-in
* • SSL/TLS support (optional)
* • Perfect for Mosquitto broker
*
* 📋 VERSION: 3.0 (AsyncMqttClient-based)
* 📅 DATE: 2025-01-04
* 👨‍💻 AUTHOR: Advanced Bell Systems
* ═══════════════════════════════════════════════════════════════════════════════════
*/
#pragma once
#include <Arduino.h>
#include <AsyncMqttClient.h>
class ConfigManager;
class Networking;
class MQTTAsyncClient {
public:
static MQTTAsyncClient* _instance;
// Message callback type
using MessageCallback = std::function<void(const String& topic, const String& payload)>;
explicit MQTTAsyncClient(ConfigManager& configManager, Networking& networking);
~MQTTAsyncClient();
/**
* @brief Initialize MQTT client
*/
void begin();
/**
* @brief Connect to MQTT broker
*/
void connect();
/**
* @brief Disconnect from MQTT broker
*/
void disconnect();
/**
* @brief Publish message to MQTT
* @param topic Topic to publish to (relative, will prepend "vesper/{deviceID}/")
* @param payload Message payload
* @param qos QoS level (0, 1, or 2)
* @param retain Retain flag
* @return Packet ID (0 if failed)
*/
uint16_t publish(const String& topic, const String& payload, int qos = 0, bool retain = false);
/**
* @brief Set message received callback
*/
void setCallback(MessageCallback callback);
/**
* @brief Check if connected to MQTT broker
*/
bool isConnected() const;
/**
* @brief Handle network connection callback
*/
void onNetworkConnected();
/**
* @brief Handle network disconnection callback
*/
void onNetworkDisconnected();
private:
ConfigManager& _configManager;
Networking& _networking;
AsyncMqttClient _mqttClient;
MessageCallback _messageCallback;
// Device topic strings (cached for performance)
String _controlTopic;
String _dataTopic;
String _clientId; // Store client ID to keep it alive
/**
* @brief Subscribe to control topic
*/
void subscribe();
/**
* @brief MQTT event handlers
*/
void onMqttConnect(bool sessionPresent);
void onMqttDisconnect(AsyncMqttClientDisconnectReason reason);
void onMqttSubscribe(uint16_t packetId, uint8_t qos);
void onMqttUnsubscribe(uint16_t packetId);
void onMqttMessage(char* topic, char* payload, AsyncMqttClientMessageProperties properties, size_t len, size_t index, size_t total);
void onMqttPublish(uint16_t packetId);
// Reconnection Timer
TimerHandle_t _mqttReconnectTimer;
static const unsigned long MQTT_RECONNECT_DELAY = 5000; // 5 seconds
void attemptReconnection();
static void mqttReconnectTimerCallback(TimerHandle_t xTimer);
};

2
vesper/src/Communication/ResponseBuilder.cpp → vesper/src/Communication/ResponseBuilder/ResponseBuilder.cpp
View File

@@ -1,5 +1,5 @@
#include "ResponseBuilder.hpp"
#include "../Logging/Logging.hpp"
#include "../../Logging/Logging.hpp"
// Static member initialization
StaticJsonDocument<512> ResponseBuilder::_responseDoc;

2
vesper/src/Communication/ResponseBuilder.hpp → vesper/src/Communication/ResponseBuilder/ResponseBuilder.hpp
View File

@@ -43,7 +43,7 @@
#include <Arduino.h>
#include <ArduinoJson.h>
#include "../Player/Player.hpp" // For PlayerStatus enum
#include "../../Player/Player.hpp" // For PlayerStatus enum
class ResponseBuilder {
public:

157
vesper/src/Communication/WebSocketServer/WebSocketServer.cpp Normal file
View File

@@ -0,0 +1,157 @@
/*
* WEBSOCKETSERVER.CPP - WebSocket Server Implementation
*/
#include "WebSocketServer.hpp"
#include "../../Logging/Logging.hpp"
#include "../ResponseBuilder/ResponseBuilder.hpp"
// Static instance for callback
WebSocketServer* WebSocketServer::_instance = nullptr;
WebSocketServer::WebSocketServer(AsyncWebSocket& webSocket, ClientManager& clientManager)
: _webSocket(webSocket)
, _clientManager(clientManager)
, _messageCallback(nullptr) {
_instance = this;
}
WebSocketServer::~WebSocketServer() {
_instance = nullptr;
}
void WebSocketServer::begin() {
_webSocket.onEvent(onEvent);
LOG_INFO("WebSocket server initialized on /ws");
// 🔥 CRITICAL: This line was missing - attach WebSocket to the AsyncWebServer
// Without this, the server doesn't know about the WebSocket handler!
// Note: We can't access _server here directly, so this must be done in CommunicationRouter
}
void WebSocketServer::setCallback(MessageCallback callback) {
_messageCallback = callback;
}
void WebSocketServer::sendToClient(uint32_t clientId, const String& message) {
_clientManager.sendToClient(clientId, message);
}
void WebSocketServer::broadcastToAll(const String& message) {
_clientManager.broadcastToAll(message);
LOG_DEBUG("Broadcast to all WebSocket clients: %s", message.c_str());
}
void WebSocketServer::broadcastToMaster(const String& message) {
_clientManager.sendToMasterClients(message);
LOG_DEBUG("Broadcast to master clients: %s", message.c_str());
}
void WebSocketServer::broadcastToSecondary(const String& message) {
_clientManager.sendToSecondaryClients(message);
LOG_DEBUG("Broadcast to secondary clients: %s", message.c_str());
}
bool WebSocketServer::hasClients() const {
return _clientManager.hasClients();
}
size_t WebSocketServer::getClientCount() const {
return _clientManager.getClientCount();
}
void WebSocketServer::onEvent(AsyncWebSocket* server, AsyncWebSocketClient* client,
AwsEventType type, void* arg, uint8_t* data, size_t len) {
if (!_instance) {
LOG_ERROR("WebSocketServer static instance is NULL - callback ignored!");
return;
}
switch (type) {
case WS_EVT_CONNECT:
_instance->onConnect(client);
break;
case WS_EVT_DISCONNECT:
_instance->onDisconnect(client);
break;
case WS_EVT_DATA:
_instance->onData(client, arg, data, len);
break;
case WS_EVT_ERROR:
LOG_ERROR("WebSocket client #%u error(%u): %s",
client->id(), *((uint16_t*)arg), (char*)data);
break;
default:
break;
}
}
void WebSocketServer::onConnect(AsyncWebSocketClient* client) {
LOG_INFO("WebSocket client #%u connected from %s",
client->id(), client->remoteIP().toString().c_str());
// Add client to manager (type UNKNOWN until they identify)
_clientManager.addClient(client, ClientManager::DeviceType::UNKNOWN);
// Send welcome message
String welcomeMsg = ResponseBuilder::success("connection", "Connected to Vesper");
_clientManager.sendToClient(client->id(), welcomeMsg);
}
void WebSocketServer::onDisconnect(AsyncWebSocketClient* client) {
LOG_INFO("WebSocket client #%u disconnected", client->id());
_clientManager.removeClient(client->id());
_clientManager.cleanupDisconnectedClients();
}
void WebSocketServer::onData(AsyncWebSocketClient* client, void* arg, uint8_t* data, size_t len) {
AwsFrameInfo* info = (AwsFrameInfo*)arg;
// Only handle complete, single-frame text messages
if (info->final && info->index == 0 && info->len == len && info->opcode == WS_TEXT) {
// Allocate buffer for payload
char* payload = (char*)malloc(len + 1);
if (!payload) {
LOG_ERROR("Failed to allocate memory for WebSocket payload");
String errorResponse = ResponseBuilder::error("memory_error", "Out of memory");
_clientManager.sendToClient(client->id(), errorResponse);
return;
}
memcpy(payload, data, len);
payload[len] = '\0';
LOG_DEBUG("WebSocket client #%u sent: %s", client->id(), payload);
// Parse JSON
StaticJsonDocument<2048> doc;
DeserializationError error = deserializeJson(doc, payload);
if (error) {
LOG_ERROR("Failed to parse WebSocket JSON from client #%u: %s", client->id(), error.c_str());
String errorResponse = ResponseBuilder::error("parse_error", "Invalid JSON");
_clientManager.sendToClient(client->id(), errorResponse);
} else {
// Update client last seen time
_clientManager.updateClientLastSeen(client->id());
// Call user callback if set
if (_messageCallback) {
LOG_DEBUG("Routing message from client #%u to callback handler", client->id());
_messageCallback(client->id(), doc);
} else {
LOG_WARNING("WebSocket message received but no callback handler is set!");
}
}
free(payload);
} else {
LOG_WARNING("Received fragmented or non-text WebSocket message from client #%u - ignoring", client->id());
}
}

103
vesper/src/Communication/WebSocketServer/WebSocketServer.hpp Normal file
View File

@@ -0,0 +1,103 @@
/*
* ═══════════════════════════════════════════════════════════════════════════════════
* WEBSOCKETSERVER.HPP - WebSocket Server Manager
* ═══════════════════════════════════════════════════════════════════════════════════
*
* 📡 WEBSOCKET MULTI-CLIENT MANAGER 📡
*
* Handles WebSocket connections with:
* • Multi-client support with device type identification
* • Automatic client cleanup
* • Broadcast and targeted messaging
* • Integration with ClientManager
*
* 📋 VERSION: 1.0
* 📅 DATE: 2025-10-01
* 👨‍💻 AUTHOR: Advanced Bell Systems
* ═══════════════════════════════════════════════════════════════════════════════════
*/
#pragma once
#include <Arduino.h>
#include <ESPAsyncWebServer.h>
#include <ArduinoJson.h>
#include "../../ClientManager/ClientManager.hpp"
class WebSocketServer {
public:
// Message callback type
using MessageCallback = std::function<void(uint32_t clientId, const JsonDocument& message)>;
explicit WebSocketServer(AsyncWebSocket& webSocket, ClientManager& clientManager);
~WebSocketServer();
/**
* @brief Initialize WebSocket server
*/
void begin();
/**
* @brief Set message received callback
*/
void setCallback(MessageCallback callback);
/**
* @brief Send message to specific client
*/
void sendToClient(uint32_t clientId, const String& message);
/**
* @brief Broadcast to all connected clients
*/
void broadcastToAll(const String& message);
/**
* @brief Broadcast to master devices only
*/
void broadcastToMaster(const String& message);
/**
* @brief Broadcast to secondary devices only
*/
void broadcastToSecondary(const String& message);
/**
* @brief Check if any clients are connected
*/
bool hasClients() const;
/**
* @brief Get number of connected clients
*/
size_t getClientCount() const;
private:
AsyncWebSocket& _webSocket;
ClientManager& _clientManager;
MessageCallback _messageCallback;
/**
* @brief Static WebSocket event handler
*/
static void onEvent(AsyncWebSocket* server, AsyncWebSocketClient* client,
AwsEventType type, void* arg, uint8_t* data, size_t len);
/**
* @brief Handle client connection
*/
void onConnect(AsyncWebSocketClient* client);
/**
* @brief Handle client disconnection
*/
void onDisconnect(AsyncWebSocketClient* client);
/**
* @brief Handle received data
*/
void onData(AsyncWebSocketClient* client, void* arg, uint8_t* data, size_t len);
// Static instance for callback routing
static WebSocketServer* _instance;
};