iot/spore
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feature/cluster-message #9

Merged
master merged 3 commits from feature/cluster-message into main 2025-09-28 13:43:54 +02:00
Conversation 0 Commits 3 Files Changed 6 +22 -10
2 changed files with 22 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit f4ccb1c7ef - Show all commits

15
examples/neopattern/NeoPatternService.cpp
View File

@@ -154,7 +154,7 @@ void NeoPatternService::handleControlRequest(AsyncWebServerRequest* request) {
updated = true; updated = true;
} }
// Broadcast to peers if requested // Broadcast to peers if requested (delegate to core broadcast handler)
if (broadcast && any) { if (broadcast && any) {
JsonDocument eventDoc; JsonDocument eventDoc;
eventDoc["event"] = "api/neopattern"; eventDoc["event"] = "api/neopattern";
@@ -163,15 +163,10 @@ void NeoPatternService::handleControlRequest(AsyncWebServerRequest* request) {
String eventJson; String eventJson;
serializeJson(eventDoc, eventJson); serializeJson(eventDoc, eventJson);
// Compute subnet-directed broadcast to improve delivery on some networks LOG_INFO("NeoPattern", String("Submitting cluster/broadcast for api/neopattern payloadLen=") + String(payloadStr.length()));
IPAddress ip = WiFi.localIP(); std::string ev = "cluster/broadcast";
IPAddress mask = WiFi.subnetMask(); String eventStr = eventJson;
IPAddress bcast(ip[0] | ~mask[0], ip[1] | ~mask[1], ip[2] | ~mask[2], ip[3] | ~mask[3]); ctx.fire(ev, &eventStr);
LOG_INFO("NeoPattern", String("Broadcasting CLUSTER_EVENT api/neopattern to ") + bcast.toString() + " payloadLen=" + String(payloadStr.length()));
ctx.udp->beginPacket(bcast, ctx.config.udp_port);
String msg = String(ClusterProtocol::CLUSTER_EVENT_MSG) + ":" + eventJson;
ctx.udp->write(msg.c_str());
ctx.udp->endPacket();
} }
// Return current state // Return current state

17
src/spore/core/ClusterManager.cpp
View File

@@ -8,6 +8,23 @@ ClusterManager::ClusterManager(NodeContext& ctx, TaskManager& taskMgr) : ctx(ctx
NodeInfo* node = static_cast<NodeInfo*>(data); NodeInfo* node = static_cast<NodeInfo*>(data);
this->addOrUpdateNode(node->hostname, node->ip); this->addOrUpdateNode(node->hostname, node->ip);
}); });
// Centralized broadcast handler: services fire 'cluster/broadcast' with CLUSTER_EVENT JSON payload
ctx.on("cluster/broadcast", [this](void* data) {
String* jsonStr = static_cast<String*>(data);
if (!jsonStr) {
LOG_WARN("Cluster", "cluster/broadcast called with null data");
return;
}
// Subnet-directed broadcast (more reliable than 255.255.255.255 on some networks)
IPAddress ip = WiFi.localIP();
IPAddress mask = WiFi.subnetMask();
IPAddress bcast(ip[0] | ~mask[0], ip[1] | ~mask[1], ip[2] | ~mask[2], ip[3] | ~mask[3]);
LOG_INFO("Cluster", String("Broadcasting CLUSTER_EVENT to ") + bcast.toString() + " len=" + String(jsonStr->length()));
this->ctx.udp->beginPacket(bcast, this->ctx.config.udp_port);
String msg = String(ClusterProtocol::CLUSTER_EVENT_MSG) + ":" + *jsonStr;
this->ctx.udp->write(msg.c_str());
this->ctx.udp->endPacket();
});
// Register tasks // Register tasks
registerTasks(); registerTasks();
initMessageHandlers(); initMessageHandlers();